michael@0: /* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
michael@0: /* vim:set ts=2 sw=2 sts=2 et cindent: */
michael@0: /* This Source Code Form is subject to the terms of the Mozilla Public
michael@0:  * License, v. 2.0. If a copy of the MPL was not distributed with this file,
michael@0:  * You can obtain one at http://mozilla.org/MPL/2.0/. */
michael@0: 
michael@0: #include "FileBlockCache.h"
michael@0: #include "VideoUtils.h"
michael@0: #include "prio.h"
michael@0: #include <algorithm>
michael@0: 
michael@0: namespace mozilla {
michael@0: 
michael@0: nsresult FileBlockCache::Open(PRFileDesc* aFD)
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "Only call on main thread");
michael@0:   NS_ENSURE_TRUE(aFD != nullptr, NS_ERROR_FAILURE);
michael@0:   {
michael@0:     MonitorAutoLock mon(mFileMonitor);
michael@0:     mFD = aFD;
michael@0:   }
michael@0:   {
michael@0:     MonitorAutoLock mon(mDataMonitor);
michael@0:     nsresult res = NS_NewThread(getter_AddRefs(mThread),
michael@0:                                 nullptr,
michael@0:                                 MEDIA_THREAD_STACK_SIZE);
michael@0:     mIsOpen = NS_SUCCEEDED(res);
michael@0:     return res;
michael@0:   }
michael@0: }
michael@0: 
michael@0: FileBlockCache::FileBlockCache()
michael@0:   : mFileMonitor("MediaCache.Writer.IO.Monitor"),
michael@0:     mFD(nullptr),
michael@0:     mFDCurrentPos(0),
michael@0:     mDataMonitor("MediaCache.Writer.Data.Monitor"),
michael@0:     mIsWriteScheduled(false),
michael@0:     mIsOpen(false)
michael@0: {
michael@0:   MOZ_COUNT_CTOR(FileBlockCache);
michael@0: }
michael@0: 
michael@0: FileBlockCache::~FileBlockCache()
michael@0: {
michael@0:   NS_ASSERTION(!mIsOpen, "Should Close() FileBlockCache before destroying");
michael@0:   {
michael@0:     // Note, mThread will be shutdown by the time this runs, so we won't
michael@0:     // block while taking mFileMonitor.
michael@0:     MonitorAutoLock mon(mFileMonitor);
michael@0:     if (mFD) {
michael@0:       PRStatus prrc;
michael@0:       prrc = PR_Close(mFD);
michael@0:       if (prrc != PR_SUCCESS) {
michael@0:         NS_WARNING("PR_Close() failed.");
michael@0:       }
michael@0:       mFD = nullptr;
michael@0:     }
michael@0:   }
michael@0:   MOZ_COUNT_DTOR(FileBlockCache);
michael@0: }
michael@0: 
michael@0: 
michael@0: void FileBlockCache::Close()
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "Only call on main thread");
michael@0:   MonitorAutoLock mon(mDataMonitor);
michael@0: 
michael@0:   mIsOpen = false;
michael@0: 
michael@0:   if (mThread) {
michael@0:     // We must shut down the thread in another runnable. This is called
michael@0:     // while we're shutting down the media cache, and nsIThread::Shutdown()
michael@0:     // can cause events to run before it completes, which could end up
michael@0:     // opening more streams, while the media cache is shutting down and
michael@0:     // releasing memory etc! Also note we close mFD in the destructor so
michael@0:     // as to not disturb any IO that's currently running.
michael@0:     nsCOMPtr<nsIRunnable> event = new ShutdownThreadEvent(mThread);
michael@0:     mThread = nullptr;
michael@0:     NS_DispatchToMainThread(event);
michael@0:   }
michael@0: }
michael@0: 
michael@0: nsresult FileBlockCache::WriteBlock(uint32_t aBlockIndex, const uint8_t* aData)
michael@0: {
michael@0:   MonitorAutoLock mon(mDataMonitor);
michael@0: 
michael@0:   if (!mIsOpen)
michael@0:     return NS_ERROR_FAILURE;
michael@0: 
michael@0:   // Check if we've already got a pending write scheduled for this block.
michael@0:   mBlockChanges.EnsureLengthAtLeast(aBlockIndex + 1);
michael@0:   bool blockAlreadyHadPendingChange = mBlockChanges[aBlockIndex] != nullptr;
michael@0:   mBlockChanges[aBlockIndex] = new BlockChange(aData);
michael@0: 
michael@0:   if (!blockAlreadyHadPendingChange || !mChangeIndexList.Contains(aBlockIndex)) {
michael@0:     // We either didn't already have a pending change for this block, or we
michael@0:     // did but we didn't have an entry for it in mChangeIndexList (we're in the process
michael@0:     // of writing it and have removed the block's index out of mChangeIndexList
michael@0:     // in Run() but not finished writing the block to file yet). Add the blocks
michael@0:     // index to the end of mChangeIndexList to ensure the block is written as
michael@0:     // as soon as possible.
michael@0:     mChangeIndexList.PushBack(aBlockIndex);
michael@0:   }
michael@0:   NS_ASSERTION(mChangeIndexList.Contains(aBlockIndex), "Must have entry for new block");
michael@0: 
michael@0:   EnsureWriteScheduled();
michael@0: 
michael@0:   return NS_OK;
michael@0: }
michael@0: 
michael@0: void FileBlockCache::EnsureWriteScheduled()
michael@0: {
michael@0:   mDataMonitor.AssertCurrentThreadOwns();
michael@0: 
michael@0:   if (!mIsWriteScheduled) {
michael@0:     mThread->Dispatch(this, NS_DISPATCH_NORMAL);
michael@0:     mIsWriteScheduled = true;
michael@0:   }
michael@0: }
michael@0: 
michael@0: nsresult FileBlockCache::Seek(int64_t aOffset)
michael@0: {
michael@0:   mFileMonitor.AssertCurrentThreadOwns();
michael@0: 
michael@0:   if (mFDCurrentPos != aOffset) {
michael@0:     int64_t result = PR_Seek64(mFD, aOffset, PR_SEEK_SET);
michael@0:     if (result != aOffset) {
michael@0:       NS_WARNING("Failed to seek media cache file");
michael@0:       return NS_ERROR_FAILURE;
michael@0:     }
michael@0:     mFDCurrentPos = result;
michael@0:   }
michael@0:   return NS_OK;
michael@0: }
michael@0: 
michael@0: nsresult FileBlockCache::ReadFromFile(int64_t aOffset,
michael@0:                                       uint8_t* aDest,
michael@0:                                       int32_t aBytesToRead,
michael@0:                                       int32_t& aBytesRead)
michael@0: {
michael@0:   mFileMonitor.AssertCurrentThreadOwns();
michael@0: 
michael@0:   nsresult res = Seek(aOffset);
michael@0:   if (NS_FAILED(res)) return res;
michael@0:   
michael@0:   aBytesRead = PR_Read(mFD, aDest, aBytesToRead);
michael@0:   if (aBytesRead <= 0)
michael@0:     return NS_ERROR_FAILURE;
michael@0:   mFDCurrentPos += aBytesRead;
michael@0:   
michael@0:   return NS_OK;
michael@0: }
michael@0: 
michael@0: nsresult FileBlockCache::WriteBlockToFile(int32_t aBlockIndex,
michael@0:                                           const uint8_t* aBlockData)
michael@0: {
michael@0:   mFileMonitor.AssertCurrentThreadOwns();
michael@0: 
michael@0:   nsresult rv = Seek(BlockIndexToOffset(aBlockIndex));
michael@0:   if (NS_FAILED(rv)) return rv;
michael@0: 
michael@0:   int32_t amount = PR_Write(mFD, aBlockData, BLOCK_SIZE);
michael@0:   if (amount < BLOCK_SIZE) {
michael@0:     NS_WARNING("Failed to write media cache block!");
michael@0:     return NS_ERROR_FAILURE;
michael@0:   }
michael@0:   mFDCurrentPos += BLOCK_SIZE;
michael@0: 
michael@0:   return NS_OK;
michael@0: }
michael@0: 
michael@0: nsresult FileBlockCache::MoveBlockInFile(int32_t aSourceBlockIndex,
michael@0:                                          int32_t aDestBlockIndex)
michael@0: {
michael@0:   mFileMonitor.AssertCurrentThreadOwns();
michael@0: 
michael@0:   uint8_t buf[BLOCK_SIZE];
michael@0:   int32_t bytesRead = 0;
michael@0:   if (NS_FAILED(ReadFromFile(BlockIndexToOffset(aSourceBlockIndex),
michael@0:                              buf,
michael@0:                              BLOCK_SIZE,
michael@0:                              bytesRead))) {
michael@0:     return NS_ERROR_FAILURE;
michael@0:   }
michael@0:   return WriteBlockToFile(aDestBlockIndex, buf);
michael@0: }
michael@0: 
michael@0: nsresult FileBlockCache::Run()
michael@0: {
michael@0:   MonitorAutoLock mon(mDataMonitor);
michael@0:   NS_ASSERTION(!NS_IsMainThread(), "Don't call on main thread");
michael@0:   NS_ASSERTION(!mChangeIndexList.IsEmpty(), "Only dispatch when there's work to do");
michael@0:   NS_ASSERTION(mIsWriteScheduled, "Should report write running or scheduled.");
michael@0: 
michael@0:   while (!mChangeIndexList.IsEmpty()) {
michael@0:     if (!mIsOpen) {
michael@0:       // We've been closed, abort, discarding unwritten changes.
michael@0:       mIsWriteScheduled = false;
michael@0:       return NS_ERROR_FAILURE;
michael@0:     }
michael@0: 
michael@0:     // Process each pending change. We pop the index out of the change
michael@0:     // list, but leave the BlockChange in mBlockChanges until the change
michael@0:     // is written to file. This is so that any read which happens while
michael@0:     // we drop mDataMonitor to write will refer to the data's source in
michael@0:     // memory, rather than the not-yet up to date data written to file.
michael@0:     // This also ensures we will insert a new index into mChangeIndexList
michael@0:     // when this happens.
michael@0: 
michael@0:     // Hold a reference to the change, in case another change
michael@0:     // overwrites the mBlockChanges entry for this block while we drop
michael@0:     // mDataMonitor to take mFileMonitor.
michael@0:     int32_t blockIndex = mChangeIndexList.PopFront();
michael@0:     nsRefPtr<BlockChange> change = mBlockChanges[blockIndex];
michael@0:     NS_ABORT_IF_FALSE(change,
michael@0:       "Change index list should only contain entries for blocks with changes");
michael@0:     {
michael@0:       MonitorAutoUnlock unlock(mDataMonitor);
michael@0:       MonitorAutoLock lock(mFileMonitor);
michael@0:       if (change->IsWrite()) {
michael@0:         WriteBlockToFile(blockIndex, change->mData.get());
michael@0:       } else if (change->IsMove()) {
michael@0:         MoveBlockInFile(change->mSourceBlockIndex, blockIndex);
michael@0:       }
michael@0:     }
michael@0:     // If a new change has not been made to the block while we dropped
michael@0:     // mDataMonitor, clear reference to the old change. Otherwise, the old
michael@0:     // reference has been cleared already.
michael@0:     if (mBlockChanges[blockIndex] == change) {
michael@0:       mBlockChanges[blockIndex] = nullptr;
michael@0:     }
michael@0:   }
michael@0: 
michael@0:   mIsWriteScheduled = false;
michael@0: 
michael@0:   return NS_OK;
michael@0: }
michael@0: 
michael@0: nsresult FileBlockCache::Read(int64_t aOffset,
michael@0:                               uint8_t* aData,
michael@0:                               int32_t aLength,
michael@0:                               int32_t* aBytes)
michael@0: {
michael@0:   MonitorAutoLock mon(mDataMonitor);
michael@0: 
michael@0:   if (!mFD || (aOffset / BLOCK_SIZE) > INT32_MAX)
michael@0:     return NS_ERROR_FAILURE;
michael@0: 
michael@0:   int32_t bytesToRead = aLength;
michael@0:   int64_t offset = aOffset;
michael@0:   uint8_t* dst = aData;
michael@0:   while (bytesToRead > 0) {
michael@0:     int32_t blockIndex = static_cast<int32_t>(offset / BLOCK_SIZE);
michael@0:     int32_t start = offset % BLOCK_SIZE;
michael@0:     int32_t amount = std::min(BLOCK_SIZE - start, bytesToRead);
michael@0: 
michael@0:     // If the block is not yet written to file, we can just read from
michael@0:     // the memory buffer, otherwise we need to read from file.
michael@0:     int32_t bytesRead = 0;
michael@0:     nsRefPtr<BlockChange> change = mBlockChanges[blockIndex];
michael@0:     if (change && change->IsWrite()) {
michael@0:       // Block isn't yet written to file. Read from memory buffer.
michael@0:       const uint8_t* blockData = change->mData.get();
michael@0:       memcpy(dst, blockData + start, amount);
michael@0:       bytesRead = amount;
michael@0:     } else {
michael@0:       if (change && change->IsMove()) {
michael@0:         // The target block is the destination of a not-yet-completed move
michael@0:         // action, so read from the move's source block from file. Note we
michael@0:         // *don't* follow a chain of moves here, as a move's source index
michael@0:         // is resolved when MoveBlock() is called, and the move's source's
michael@0:         // block could be have itself been subject to a move (or write)
michael@0:         // which happened *after* this move was recorded.
michael@0:         blockIndex = mBlockChanges[blockIndex]->mSourceBlockIndex;
michael@0:       }
michael@0:       // Block has been written to file, either as the source block of a move,
michael@0:       // or as a stable (all changes made) block. Read the data directly
michael@0:       // from file.
michael@0:       nsresult res;
michael@0:       {
michael@0:         MonitorAutoUnlock unlock(mDataMonitor);
michael@0:         MonitorAutoLock lock(mFileMonitor);
michael@0:         res = ReadFromFile(BlockIndexToOffset(blockIndex) + start,
michael@0:                            dst,
michael@0:                            amount,
michael@0:                            bytesRead);
michael@0:       }
michael@0:       NS_ENSURE_SUCCESS(res,res);
michael@0:     }
michael@0:     dst += bytesRead;
michael@0:     offset += bytesRead;
michael@0:     bytesToRead -= bytesRead;
michael@0:   }
michael@0:   *aBytes = aLength - bytesToRead;
michael@0:   return NS_OK;
michael@0: }
michael@0: 
michael@0: nsresult FileBlockCache::MoveBlock(int32_t aSourceBlockIndex, int32_t aDestBlockIndex)
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "Only call on main thread");
michael@0:   MonitorAutoLock mon(mDataMonitor);
michael@0: 
michael@0:   if (!mIsOpen)
michael@0:     return NS_ERROR_FAILURE;
michael@0: 
michael@0:   mBlockChanges.EnsureLengthAtLeast(std::max(aSourceBlockIndex, aDestBlockIndex) + 1);
michael@0: 
michael@0:   // The source block's contents may be the destination of another pending
michael@0:   // move, which in turn can be the destination of another pending move,
michael@0:   // etc. Resolve the final source block, so that if one of the blocks in
michael@0:   // the chain of moves is overwritten, we don't lose the reference to the
michael@0:   // contents of the destination block.
michael@0:   int32_t sourceIndex = aSourceBlockIndex;
michael@0:   BlockChange* sourceBlock = nullptr;
michael@0:   while ((sourceBlock = mBlockChanges[sourceIndex]) &&
michael@0:           sourceBlock->IsMove()) {
michael@0:     sourceIndex = sourceBlock->mSourceBlockIndex;
michael@0:   }
michael@0: 
michael@0:   if (mBlockChanges[aDestBlockIndex] == nullptr ||
michael@0:       !mChangeIndexList.Contains(aDestBlockIndex)) {
michael@0:     // Only add another entry to the change index list if we don't already
michael@0:     // have one for this block. We won't have an entry when either there's
michael@0:     // no pending change for this block, or if there is a pending change for
michael@0:     // this block and we're in the process of writing it (we've popped the
michael@0:     // block's index out of mChangeIndexList in Run() but not finished writing
michael@0:     // the block to file yet.
michael@0:     mChangeIndexList.PushBack(aDestBlockIndex);
michael@0:   }
michael@0:   
michael@0:   // If the source block hasn't yet been written to file then the dest block
michael@0:   // simply contains that same write. Resolve this as a write instead.
michael@0:   if (sourceBlock && sourceBlock->IsWrite()) {
michael@0:     mBlockChanges[aDestBlockIndex] = new BlockChange(sourceBlock->mData.get());
michael@0:   } else {
michael@0:     mBlockChanges[aDestBlockIndex] = new BlockChange(sourceIndex);
michael@0:   }
michael@0: 
michael@0:   EnsureWriteScheduled();
michael@0: 
michael@0:   NS_ASSERTION(mChangeIndexList.Contains(aDestBlockIndex),
michael@0:     "Should have scheduled block for change");
michael@0: 
michael@0:   return NS_OK;
michael@0: }
michael@0: 
michael@0: } // End namespace mozilla.