tubestation/xpcom/io/nsStorageStream.cpp

/* -*- 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
 * file, You can obtain one at http://mozilla.org/MPL/2.0/. */

/*
 * 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--;
  }

  AutoTArray<mozilla::UniqueFreePtr<char>, 16> toFree;

  while (newLastSegmentNum < mLastSegmentNum) {
    toFree.AppendElement(mSegmentedBuffer->PopLastSegment());
    mLastSegmentNum--;
  }

  // If we removed a substantial number of segments, dispatch the `free(...)`
  // calls to a background thread to avoid potential calling thread jank.
  if (toFree.Length() > 128) {
    NS_DispatchBackgroundTask(NS_NewRunnableFunction(
        "nsStorageStream::SetLengthLocked",
        [toFree = std::move(toFree)]() mutable { toFree.Clear(); }));
  }

  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