Source code
Revision control
Copy as Markdown
Other Tools
/* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
/* vim: set ts=8 sts=2 et sw=2 tw=80: */
/* This Source Code Form is subject to the terms of the Mozilla Public
* License, v. 2.0. If a copy of the MPL was not distributed with this
/*
* The storage stream provides an internal buffer that can be filled by a
* client using a single output stream. One or more independent input streams
* can be created to read the data out non-destructively. The implementation
* uses a segmented buffer internally to avoid realloc'ing of large buffers,
* with the attendant performance loss and heap fragmentation.
*/
#include "mozilla/Mutex.h"
#include "nsAlgorithm.h"
#include "nsStorageStream.h"
#include "nsSegmentedBuffer.h"
#include "nsStreamUtils.h"
#include "nsCOMPtr.h"
#include "nsICloneableInputStream.h"
#include "nsIInputStream.h"
#include "nsIIPCSerializableInputStream.h"
#include "nsISeekableStream.h"
#include "mozilla/Logging.h"
#include "mozilla/Attributes.h"
#include "mozilla/Likely.h"
#include "mozilla/MathAlgorithms.h"
#include "mozilla/ipc/InputStreamUtils.h"
using mozilla::MutexAutoLock;
using mozilla::ipc::InputStreamParams;
using mozilla::ipc::StringInputStreamParams;
//
// Log module for StorageStream logging...
//
// To enable logging (see prlog.h for full details):
//
// set MOZ_LOG=StorageStreamLog:5
// set MOZ_LOG_FILE=storage.log
//
// This enables LogLevel::Debug level information and places all output in
// the file storage.log.
//
static mozilla::LazyLogModule sStorageStreamLog("nsStorageStream");
#ifdef LOG
# undef LOG
#endif
#define LOG(args) MOZ_LOG(sStorageStreamLog, mozilla::LogLevel::Debug, args)
nsStorageStream::nsStorageStream() {
LOG(("Creating nsStorageStream [%p].\n", this));
}
nsStorageStream::~nsStorageStream() { delete mSegmentedBuffer; }
NS_IMPL_ISUPPORTS(nsStorageStream, nsIStorageStream, nsIOutputStream)
NS_IMETHODIMP
nsStorageStream::Init(uint32_t aSegmentSize, uint32_t aMaxSize) {
MutexAutoLock lock(mMutex);
mSegmentedBuffer = new nsSegmentedBuffer();
mSegmentSize = aSegmentSize;
mSegmentSizeLog2 = mozilla::FloorLog2(aSegmentSize);
mMaxLogicalLength = aMaxSize;
// Segment size must be a power of two
if (mSegmentSize != ((uint32_t)1 << mSegmentSizeLog2)) {
return NS_ERROR_INVALID_ARG;
}
return mSegmentedBuffer->Init(aSegmentSize);
}
NS_IMETHODIMP
nsStorageStream::GetOutputStream(int32_t aStartingOffset,
nsIOutputStream** aOutputStream) {
if (NS_WARN_IF(!aOutputStream)) {
return NS_ERROR_INVALID_ARG;
}
MutexAutoLock lock(mMutex);
if (NS_WARN_IF(!mSegmentedBuffer)) {
return NS_ERROR_NOT_INITIALIZED;
}
if (mWriteInProgress) {
return NS_ERROR_NOT_AVAILABLE;
}
if (mActiveSegmentBorrows > 0) {
return NS_ERROR_NOT_AVAILABLE;
}
nsresult rv = Seek(aStartingOffset);
if (NS_FAILED(rv)) {
return rv;
}
// Enlarge the last segment in the buffer so that it is the same size as
// all the other segments in the buffer. (It may have been realloc'ed
// smaller in the Close() method.)
if (mLastSegmentNum >= 0)
if (mSegmentedBuffer->ReallocLastSegment(mSegmentSize)) {
// Need to re-Seek, since realloc changed segment base pointer
rv = Seek(aStartingOffset);
if (NS_FAILED(rv)) {
return rv;
}
}
NS_ADDREF(this);
*aOutputStream = static_cast<nsIOutputStream*>(this);
mWriteInProgress = true;
return NS_OK;
}
NS_IMETHODIMP
nsStorageStream::Close() {
MutexAutoLock lock(mMutex);
if (NS_WARN_IF(!mSegmentedBuffer)) {
return NS_ERROR_NOT_INITIALIZED;
}
mWriteInProgress = false;
int32_t segmentOffset = SegOffset(mLogicalLength);
// Shrink the final segment in the segmented buffer to the minimum size
// needed to contain the data, so as to conserve memory.
if (segmentOffset && !mActiveSegmentBorrows) {
mSegmentedBuffer->ReallocLastSegment(segmentOffset);
}
mWriteCursor = 0;
mSegmentEnd = 0;
LOG(("nsStorageStream [%p] Close mWriteCursor=%p mSegmentEnd=%p\n", this,
mWriteCursor, mSegmentEnd));
return NS_OK;
}
NS_IMETHODIMP
nsStorageStream::Flush() { return NS_OK; }
NS_IMETHODIMP
nsStorageStream::StreamStatus() {
MutexAutoLock lock(mMutex);
if (!mSegmentedBuffer) {
return NS_ERROR_NOT_INITIALIZED;
}
return NS_OK;
}
NS_IMETHODIMP
nsStorageStream::Write(const char* aBuffer, uint32_t aCount,
uint32_t* aNumWritten) {
if (NS_WARN_IF(!aNumWritten) || NS_WARN_IF(!aBuffer)) {
return NS_ERROR_INVALID_ARG;
}
MutexAutoLock lock(mMutex);
if (NS_WARN_IF(!mSegmentedBuffer)) {
return NS_ERROR_NOT_INITIALIZED;
}
if (NS_WARN_IF(mLogicalLength >= mMaxLogicalLength)) {
return NS_ERROR_OUT_OF_MEMORY;
}
LOG(("nsStorageStream [%p] Write mWriteCursor=%p mSegmentEnd=%p aCount=%d\n",
this, mWriteCursor, mSegmentEnd, aCount));
uint32_t remaining = aCount;
const char* readCursor = aBuffer;
remaining = std::min(remaining, mMaxLogicalLength - mLogicalLength);
auto onExit = mozilla::MakeScopeExit([&] {
mMutex.AssertCurrentThreadOwns();
*aNumWritten = aCount - remaining;
mLogicalLength += *aNumWritten;
LOG(
("nsStorageStream [%p] Wrote mWriteCursor=%p mSegmentEnd=%p "
"numWritten=%d\n",
this, mWriteCursor, mSegmentEnd, *aNumWritten));
});
// If no segments have been created yet, create one even if we don't have
// to write any data; this enables creating an input stream which reads from
// the very end of the data for any amount of data in the stream (i.e.
// this stream contains N bytes of data and newInputStream(N) is called),
// even for N=0 (with the caveat that we require .write("", 0) be called to
// initialize internal buffers).
bool firstTime = mSegmentedBuffer->GetSegmentCount() == 0;
while (remaining || MOZ_UNLIKELY(firstTime)) {
firstTime = false;
uint32_t availableInSegment = mSegmentEnd - mWriteCursor;
if (!availableInSegment) {
mWriteCursor = mSegmentedBuffer->AppendNewSegment();
if (!mWriteCursor) {
mSegmentEnd = 0;
return NS_ERROR_OUT_OF_MEMORY;
}
mLastSegmentNum++;
mSegmentEnd = mWriteCursor + mSegmentSize;
availableInSegment = mSegmentEnd - mWriteCursor;
LOG(
("nsStorageStream [%p] Write (new seg) mWriteCursor=%p "
"mSegmentEnd=%p\n",
this, mWriteCursor, mSegmentEnd));
}
uint32_t count = XPCOM_MIN(availableInSegment, remaining);
memcpy(mWriteCursor, readCursor, count);
remaining -= count;
readCursor += count;
mWriteCursor += count;
LOG(
("nsStorageStream [%p] Writing mWriteCursor=%p mSegmentEnd=%p "
"count=%d\n",
this, mWriteCursor, mSegmentEnd, count));
}
return NS_OK;
}
NS_IMETHODIMP
nsStorageStream::WriteFrom(nsIInputStream* aInStr, uint32_t aCount,
uint32_t* aResult) {
return NS_ERROR_NOT_IMPLEMENTED;
}
NS_IMETHODIMP
nsStorageStream::WriteSegments(nsReadSegmentFun aReader, void* aClosure,
uint32_t aCount, uint32_t* aResult) {
return NS_ERROR_NOT_IMPLEMENTED;
}
NS_IMETHODIMP
nsStorageStream::IsNonBlocking(bool* aNonBlocking) {
*aNonBlocking = false;
return NS_OK;
}
NS_IMETHODIMP
nsStorageStream::GetLength(uint32_t* aLength) {
MutexAutoLock lock(mMutex);
*aLength = mLogicalLength;
return NS_OK;
}
NS_IMETHODIMP
nsStorageStream::SetLength(uint32_t aLength) {
MutexAutoLock lock(mMutex);
return SetLengthLocked(aLength);
}
// Truncate the buffer by deleting the end segments
nsresult nsStorageStream::SetLengthLocked(uint32_t aLength) {
if (NS_WARN_IF(!mSegmentedBuffer)) {
return NS_ERROR_NOT_INITIALIZED;
}
if (mWriteInProgress) {
return NS_ERROR_NOT_AVAILABLE;
}
if (mActiveSegmentBorrows) {
return NS_ERROR_NOT_AVAILABLE;
}
if (aLength > mLogicalLength) {
return NS_ERROR_INVALID_ARG;
}
int32_t newLastSegmentNum = SegNum(aLength);
int32_t segmentOffset = SegOffset(aLength);
if (segmentOffset == 0) {
newLastSegmentNum--;
}
while (newLastSegmentNum < mLastSegmentNum) {
mSegmentedBuffer->DeleteLastSegment();
mLastSegmentNum--;
}
mLogicalLength = aLength;
return NS_OK;
}
NS_IMETHODIMP
nsStorageStream::GetWriteInProgress(bool* aWriteInProgress) {
MutexAutoLock lock(mMutex);
*aWriteInProgress = mWriteInProgress;
return NS_OK;
}
nsresult nsStorageStream::Seek(int32_t aPosition) {
if (NS_WARN_IF(!mSegmentedBuffer)) {
return NS_ERROR_NOT_INITIALIZED;
}
// An argument of -1 means "seek to end of stream"
if (aPosition == -1) {
aPosition = mLogicalLength;
}
// Seeking beyond the buffer end is illegal
if ((uint32_t)aPosition > mLogicalLength) {
return NS_ERROR_INVALID_ARG;
}
// Seeking backwards in the write stream results in truncation
SetLengthLocked(aPosition);
// Special handling for seek to start-of-buffer
if (aPosition == 0) {
mWriteCursor = 0;
mSegmentEnd = 0;
LOG(("nsStorageStream [%p] Seek mWriteCursor=%p mSegmentEnd=%p\n", this,
mWriteCursor, mSegmentEnd));
return NS_OK;
}
// Segment may have changed, so reset pointers
mWriteCursor = mSegmentedBuffer->GetSegment(mLastSegmentNum);
NS_ASSERTION(mWriteCursor, "null mWriteCursor");
mSegmentEnd = mWriteCursor + mSegmentSize;
// Adjust write cursor for current segment offset. This test is necessary
// because SegNum may reference the next-to-be-allocated segment, in which
// case we need to be pointing at the end of the last segment.
int32_t segmentOffset = SegOffset(aPosition);
if (segmentOffset == 0 && (SegNum(aPosition) > (uint32_t)mLastSegmentNum)) {
mWriteCursor = mSegmentEnd;
} else {
mWriteCursor += segmentOffset;
}
LOG(("nsStorageStream [%p] Seek mWriteCursor=%p mSegmentEnd=%p\n", this,
mWriteCursor, mSegmentEnd));
return NS_OK;
}
////////////////////////////////////////////////////////////////////////////////
// There can be many nsStorageInputStreams for a single nsStorageStream
class nsStorageInputStream final : public nsIInputStream,
public nsISeekableStream,
public nsIIPCSerializableInputStream,
public nsICloneableInputStream {
public:
nsStorageInputStream(nsStorageStream* aStorageStream, uint32_t aSegmentSize)
: mStorageStream(aStorageStream),
mReadCursor(0),
mSegmentEnd(0),
mSegmentNum(0),
mSegmentSize(aSegmentSize),
mLogicalCursor(0),
mStatus(NS_OK) {}
NS_DECL_THREADSAFE_ISUPPORTS
NS_DECL_NSIINPUTSTREAM
NS_DECL_NSISEEKABLESTREAM
NS_DECL_NSITELLABLESTREAM
NS_DECL_NSIIPCSERIALIZABLEINPUTSTREAM
NS_DECL_NSICLONEABLEINPUTSTREAM
private:
~nsStorageInputStream() = default;
protected:
nsresult Seek(uint32_t aPosition) MOZ_REQUIRES(mStorageStream->mMutex);
friend class nsStorageStream;
private:
RefPtr<nsStorageStream> mStorageStream;
uint32_t mReadCursor; // Next memory location to read byte, or 0
uint32_t mSegmentEnd; // One byte past end of current buffer segment
uint32_t mSegmentNum; // Segment number containing read cursor
uint32_t mSegmentSize; // All segments, except the last, are of this size
uint32_t mLogicalCursor; // Logical offset into stream
nsresult mStatus;
uint32_t SegNum(uint32_t aPosition) MOZ_REQUIRES(mStorageStream->mMutex) {
return aPosition >> mStorageStream->mSegmentSizeLog2;
}
uint32_t SegOffset(uint32_t aPosition) {
return aPosition & (mSegmentSize - 1);
}
};
NS_IMPL_ISUPPORTS(nsStorageInputStream, nsIInputStream, nsISeekableStream,
nsITellableStream, nsIIPCSerializableInputStream,
nsICloneableInputStream)
NS_IMETHODIMP
nsStorageStream::NewInputStream(int32_t aStartingOffset,
nsIInputStream** aInputStream) {
MutexAutoLock lock(mMutex);
if (NS_WARN_IF(!mSegmentedBuffer)) {
return NS_ERROR_NOT_INITIALIZED;
}
RefPtr<nsStorageInputStream> inputStream =
new nsStorageInputStream(this, mSegmentSize);
inputStream->mStorageStream->mMutex.AssertCurrentThreadOwns();
nsresult rv = inputStream->Seek(aStartingOffset);
if (NS_FAILED(rv)) {
return rv;
}
inputStream.forget(aInputStream);
return NS_OK;
}
NS_IMETHODIMP
nsStorageInputStream::Close() {
mStatus = NS_BASE_STREAM_CLOSED;
return NS_OK;
}
NS_IMETHODIMP
nsStorageInputStream::Available(uint64_t* aAvailable) {
if (NS_FAILED(mStatus)) {
return mStatus;
}
MutexAutoLock lock(mStorageStream->mMutex);
*aAvailable = mStorageStream->mLogicalLength - mLogicalCursor;
return NS_OK;
}
NS_IMETHODIMP
nsStorageInputStream::StreamStatus() { return mStatus; }
NS_IMETHODIMP
nsStorageInputStream::Read(char* aBuffer, uint32_t aCount, uint32_t* aNumRead) {
return ReadSegments(NS_CopySegmentToBuffer, aBuffer, aCount, aNumRead);
}
NS_IMETHODIMP
nsStorageInputStream::ReadSegments(nsWriteSegmentFun aWriter, void* aClosure,
uint32_t aCount, uint32_t* aNumRead) {
*aNumRead = 0;
if (mStatus == NS_BASE_STREAM_CLOSED) {
return NS_OK;
}
if (NS_FAILED(mStatus)) {
return mStatus;
}
uint32_t count, availableInSegment, remainingCapacity, bytesConsumed;
nsresult rv;
remainingCapacity = aCount;
while (remainingCapacity) {
const char* cur = nullptr;
{
MutexAutoLock lock(mStorageStream->mMutex);
availableInSegment = mSegmentEnd - mReadCursor;
if (!availableInSegment) {
uint32_t available = mStorageStream->mLogicalLength - mLogicalCursor;
if (!available) {
break;
}
// We have data in the stream, but if mSegmentEnd is zero, then we
// were likely constructed prior to any data being written into
// the stream. Therefore, if mSegmentEnd is non-zero, we should
// move into the next segment; otherwise, we should stay in this
// segment so our input state can be updated and we can properly
// perform the initial read.
if (mSegmentEnd > 0) {
mSegmentNum++;
}
mReadCursor = 0;
mSegmentEnd = XPCOM_MIN(mSegmentSize, available);
availableInSegment = mSegmentEnd;
}
cur = mStorageStream->mSegmentedBuffer->GetSegment(mSegmentNum);
mStorageStream->mActiveSegmentBorrows++;
}
auto dropBorrow = mozilla::MakeScopeExit([&] {
MutexAutoLock lock(mStorageStream->mMutex);
mStorageStream->mActiveSegmentBorrows--;
});
count = XPCOM_MIN(availableInSegment, remainingCapacity);
rv = aWriter(this, aClosure, cur + mReadCursor, aCount - remainingCapacity,
count, &bytesConsumed);
if (NS_FAILED(rv) || (bytesConsumed == 0)) {
break;
}
remainingCapacity -= bytesConsumed;
mReadCursor += bytesConsumed;
mLogicalCursor += bytesConsumed;
}
*aNumRead = aCount - remainingCapacity;
bool isWriteInProgress = false;
if (NS_FAILED(mStorageStream->GetWriteInProgress(&isWriteInProgress))) {
isWriteInProgress = false;
}
if (*aNumRead == 0 && isWriteInProgress) {
return NS_BASE_STREAM_WOULD_BLOCK;
}
return NS_OK;
}
NS_IMETHODIMP
nsStorageInputStream::IsNonBlocking(bool* aNonBlocking) {
// TODO: This class should implement nsIAsyncInputStream so that callers
// have some way of dealing with NS_BASE_STREAM_WOULD_BLOCK errors.
*aNonBlocking = true;
return NS_OK;
}
NS_IMETHODIMP
nsStorageInputStream::Seek(int32_t aWhence, int64_t aOffset) {
if (NS_FAILED(mStatus)) {
return mStatus;
}
MutexAutoLock lock(mStorageStream->mMutex);
int64_t pos = aOffset;
switch (aWhence) {
case NS_SEEK_SET:
break;
case NS_SEEK_CUR:
pos += mLogicalCursor;
break;
case NS_SEEK_END:
pos += mStorageStream->mLogicalLength;
break;
default:
MOZ_ASSERT_UNREACHABLE("unexpected whence value");
return NS_ERROR_UNEXPECTED;
}
if (pos == int64_t(mLogicalCursor)) {
return NS_OK;
}
return Seek(pos);
}
NS_IMETHODIMP
nsStorageInputStream::Tell(int64_t* aResult) {
if (NS_FAILED(mStatus)) {
return mStatus;
}
*aResult = mLogicalCursor;
return NS_OK;
}
NS_IMETHODIMP
nsStorageInputStream::SetEOF() {
MOZ_ASSERT_UNREACHABLE("nsStorageInputStream::SetEOF");
return NS_ERROR_NOT_IMPLEMENTED;
}
nsresult nsStorageInputStream::Seek(uint32_t aPosition) {
uint32_t length = mStorageStream->mLogicalLength;
if (aPosition > length) {
return NS_ERROR_INVALID_ARG;
}
if (length == 0) {
return NS_OK;
}
mSegmentNum = SegNum(aPosition);
mReadCursor = SegOffset(aPosition);
uint32_t available = length - aPosition;
mSegmentEnd = mReadCursor + XPCOM_MIN(mSegmentSize - mReadCursor, available);
mLogicalCursor = aPosition;
return NS_OK;
}
void nsStorageInputStream::SerializedComplexity(uint32_t aMaxSize,
uint32_t* aSizeUsed,
uint32_t* aPipes,
uint32_t* aTransferables) {
uint64_t remaining = 0;
mozilla::DebugOnly<nsresult> rv = Available(&remaining);
MOZ_ASSERT(NS_SUCCEEDED(rv));
if (remaining >= aMaxSize) {
*aPipes = 1;
} else {
*aSizeUsed = remaining;
}
}
void nsStorageInputStream::Serialize(InputStreamParams& aParams,
uint32_t aMaxSize, uint32_t* aSizeUsed) {
MOZ_ASSERT(aSizeUsed);
*aSizeUsed = 0;
uint64_t remaining = 0;
mozilla::DebugOnly<nsresult> rv = Available(&remaining);
MOZ_ASSERT(NS_SUCCEEDED(rv));
if (remaining >= aMaxSize) {
mozilla::ipc::InputStreamHelper::SerializeInputStreamAsPipe(this, aParams);
return;
}
*aSizeUsed = remaining;
nsCString combined;
int64_t offset;
rv = Tell(&offset);
MOZ_ASSERT(NS_SUCCEEDED(rv));
auto handleOrErr = combined.BulkWrite(remaining, 0, false);
MOZ_ASSERT(!handleOrErr.isErr());
auto handle = handleOrErr.unwrap();
uint32_t numRead = 0;
rv = Read(handle.Elements(), remaining, &numRead);
MOZ_ASSERT(NS_SUCCEEDED(rv));
MOZ_ASSERT(numRead == remaining);
handle.Finish(numRead, false);
rv = Seek(NS_SEEK_SET, offset);
MOZ_ASSERT(NS_SUCCEEDED(rv));
StringInputStreamParams params;
params.data() = combined;
aParams = params;
}
bool nsStorageInputStream::Deserialize(const InputStreamParams& aParams) {
MOZ_ASSERT_UNREACHABLE(
"We should never attempt to deserialize a storage "
"input stream.");
return false;
}
NS_IMETHODIMP
nsStorageInputStream::GetCloneable(bool* aCloneableOut) {
*aCloneableOut = true;
return NS_OK;
}
NS_IMETHODIMP
nsStorageInputStream::Clone(nsIInputStream** aCloneOut) {
return mStorageStream->NewInputStream(mLogicalCursor, aCloneOut);
}
nsresult NS_NewStorageStream(uint32_t aSegmentSize, uint32_t aMaxSize,
nsIStorageStream** aResult) {
RefPtr<nsStorageStream> storageStream = new nsStorageStream();
nsresult rv = storageStream->Init(aSegmentSize, aMaxSize);
if (NS_FAILED(rv)) {
return rv;
}
storageStream.forget(aResult);
return NS_OK;
}
// Undefine LOG, so that other .cpp files (or their includes) won't complain
// about it already being defined, when we build in unified mode.
#undef LOG