#include <MissingBlocksUploader.h>

Inheritance diagram for audacity::cloud::audiocom::sync::MissingBlocksUploader:

Collaboration diagram for audacity::cloud::audiocom::sync::MissingBlocksUploader:

Classes
struct	ProducedItem

struct	Tag

Public Member Functions
	MissingBlocksUploader (Tag, const ServiceConfig &serviceConfig)

	~MissingBlocksUploader ()

Public Member Functions inherited from audacity::concurrency::ICancellable
virtual	~ICancellable ()=default

virtual void	Cancel ()=0

Static Public Member Functions
static std::shared_ptr< MissingBlocksUploader >	Create (CancellationContextPtr cancellationContex, const ServiceConfig &serviceConfig, std::vector< BlockUploadTask > uploadTasks, MissingBlocksUploadProgressCallback progress)

Static Public Attributes
static constexpr auto	NUM_PRODUCERS = 3

static constexpr auto	NUM_UPLOADERS = 6

static constexpr auto	RING_BUFFER_SIZE = 16

Private Member Functions
void	Start (CancellationContextPtr cancellationContex, std::vector< BlockUploadTask > uploadTasks, MissingBlocksUploadProgressCallback progress)

void	Cancel () override

ProducedItem	ProduceBlock ()

void	ConsumeBlock (ProducedItem item)

void	PushBlockToQueue (ProducedItem item)

ProducedItem	PopBlockFromQueue ()

void	ConfirmBlock (BlockUploadTask task)

void	HandleFailedBlock (const ResponseResult &result, BlockUploadTask task)

void	ProducerThread ()

void	ConsumerThread ()

Private Attributes
const ServiceConfig &	mServiceConfig

std::vector< BlockUploadTask >	mUploadTasks

MissingBlocksUploadProgressCallback	mProgressCallback

std::atomic_bool	mIsRunning { true }

std::thread	mProducerThread [NUM_PRODUCERS]

std::thread	mConsumerThread

std::mutex	mBlocksMutex

size_t	mFirstUnprocessedBlockIndex { 0 }

std::mutex	mUploadsMutex

std::condition_variable	mUploadsNotFull

size_t	mConcurrentUploads { 0 }

std::mutex	mRingBufferMutex

std::condition_variable	mRingBufferNotEmpty

std::condition_variable	mRingBufferNotFull

std::array< ProducedItem, RING_BUFFER_SIZE >	mRingBuffer

size_t	mRingBufferWriteIndex { 0 }

size_t	mRingBufferReadIndex { 0 }

std::mutex	mProgressDataMutex

MissingBlocksUploadProgress	mProgressData

CancellationContextPtr	mCancellationContext

Detailed Description

Definition at line 63 of file MissingBlocksUploader.h.

Constructor & Destructor Documentation

◆ MissingBlocksUploader()

audacity::cloud::audiocom::sync::MissingBlocksUploader::MissingBlocksUploader	(	Tag	,
		const ServiceConfig &	serviceConfig
	)

Definition at line 21 of file MissingBlocksUploader.cpp.

    : mServiceConfig { serviceConfig }
{
}

◆ ~MissingBlocksUploader()

audacity::cloud::audiocom::sync::MissingBlocksUploader::~MissingBlocksUploader ( )

Definition at line 47 of file MissingBlocksUploader.cpp.

{
   Cancel();
}

References Cancel().

Here is the call graph for this function:

Member Function Documentation

◆ Cancel()

void audacity::cloud::audiocom::sync::MissingBlocksUploader::Cancel ( )

overrideprivatevirtual

Implements audacity::concurrency::ICancellable.

Definition at line 71 of file MissingBlocksUploader.cpp.

{
   if (!mIsRunning.exchange(false))
      return;
 
   mRingBufferNotEmpty.notify_all();
   mRingBufferNotFull.notify_all();
   mUploadsNotFull.notify_all();
 
   for (auto& thread : mProducerThread)
      thread.join();
 
   mConsumerThread.join();
 
   std::lock_guard lock(mProgressDataMutex);
}

References mConsumerThread, mIsRunning, mProducerThread, mProgressDataMutex, mRingBufferNotEmpty, mRingBufferNotFull, and mUploadsNotFull.

Referenced by ~MissingBlocksUploader().

Here is the caller graph for this function:

◆ ConfirmBlock()

void audacity::cloud::audiocom::sync::MissingBlocksUploader::ConfirmBlock ( BlockUploadTask task )

private

Definition at line 175 of file MissingBlocksUploader.cpp.

{
   MissingBlocksUploadProgress progressData;
   {
      std::lock_guard<std::mutex> lock(mProgressDataMutex);
      mProgressData.UploadedBlocks++;
      progressData = mProgressData;
   }
 
   mProgressCallback(progressData, item.Block, {});
 
   {
      std::lock_guard<std::mutex> lock(mUploadsMutex);
      --mConcurrentUploads;
      mUploadsNotFull.notify_one();
   }
}

References audacity::cloud::audiocom::sync::BlockUploadTask::Block, mConcurrentUploads, mProgressCallback, mProgressData, mProgressDataMutex, mUploadsMutex, mUploadsNotFull, and audacity::cloud::audiocom::sync::MissingBlocksUploadProgress::UploadedBlocks.

Referenced by ConsumeBlock().

Here is the caller graph for this function:

◆ ConsumeBlock()

void audacity::cloud::audiocom::sync::MissingBlocksUploader::ConsumeBlock ( ProducedItem item )

private

Definition at line 96 of file MissingBlocksUploader.cpp.

{
   {
      std::unique_lock<std::mutex> lock(mUploadsMutex);
      mUploadsNotFull.wait(
         lock,
         [this]
         {
            return mConcurrentUploads < NUM_UPLOADERS ||
                   !mIsRunning.load(std::memory_order_consume);
         });
 
      if (!mIsRunning.load(std::memory_order_relaxed))
         return;
 
      ++mConcurrentUploads;
   }
 
   DataUploader::Get().Upload(
      mCancellationContext, mServiceConfig, item.Task.BlockUrls,
      std::move(item.CompressedData),
      [this, task = item.Task,
       weakThis = weak_from_this()](ResponseResult result)
      {
         auto lock = weakThis.lock();
 
         if (!lock)
            return;
 
         if (result.Code != SyncResultCode::Success)
            HandleFailedBlock(result, task);
         else
            ConfirmBlock(task);
      });
}

References audacity::cloud::audiocom::sync::BlockUploadTask::BlockUrls, audacity::cloud::audiocom::ResponseResult::Code, audacity::cloud::audiocom::sync::MissingBlocksUploader::ProducedItem::CompressedData, ConfirmBlock(), audacity::cloud::audiocom::sync::DataUploader::Get(), HandleFailedBlock(), mCancellationContext, mConcurrentUploads, mIsRunning, mServiceConfig, mUploadsMutex, mUploadsNotFull, NUM_UPLOADERS, audacity::cloud::audiocom::Success, audacity::cloud::audiocom::sync::MissingBlocksUploader::ProducedItem::Task, and audacity::cloud::audiocom::sync::DataUploader::Upload().

Here is the call graph for this function:

◆ ConsumerThread()

void audacity::cloud::audiocom::sync::MissingBlocksUploader::ConsumerThread ( )

private

Definition at line 255 of file MissingBlocksUploader.cpp.

{
   while (mIsRunning.load(std::memory_order_consume))
   {
      auto item = PopBlockFromQueue();
      ConsumeBlock(std::move(item));
   }
}

Referenced by Start().

Here is the caller graph for this function:

◆ Create()

std::shared_ptr< MissingBlocksUploader > audacity::cloud::audiocom::sync::MissingBlocksUploader::Create	(	CancellationContextPtr	cancellationContex,
		const ServiceConfig &	serviceConfig,
		std::vector< BlockUploadTask >	uploadTasks,
		MissingBlocksUploadProgressCallback	progress
	)

static

Definition at line 27 of file MissingBlocksUploader.cpp.

{
   auto uploader =
      std::make_shared<MissingBlocksUploader>(Tag {}, serviceConfig);
 
   if (!cancellationContex)
      cancellationContex = concurrency::CancellationContext::Create();
 
   cancellationContex->OnCancelled(uploader);
 
   uploader->Start(
      std::move(cancellationContex), std::move(uploadTasks),
      std::move(progressCallback));
 
   return uploader;
}

References audacity::concurrency::CancellationContext::Create().

Referenced by audacity::cloud::audiocom::sync::LocalProjectSnapshot::OnSnapshotCreated(), and audacity::cloud::audiocom::sync::anonymous_namespace{ResumedSnaphotUploadOperation.cpp}::ResumedSnaphotUploadOperation::UploadBlocks().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ HandleFailedBlock()

void audacity::cloud::audiocom::sync::MissingBlocksUploader::HandleFailedBlock	(	const ResponseResult &	result,
		BlockUploadTask	task
	)

private

Definition at line 193 of file MissingBlocksUploader.cpp.

{
   MissingBlocksUploadProgress progressData;
   {
      std::lock_guard<std::mutex> lock(mProgressDataMutex);
 
      mProgressData.FailedBlocks++;
      mProgressData.UploadErrors.push_back(result);
      progressData = mProgressData;
   }
 
   mProgressCallback(progressData, task.Block, result);
 
   {
      std::lock_guard<std::mutex> lock(mUploadsMutex);
      --mConcurrentUploads;
      mUploadsNotFull.notify_one();
   }
 
   mCancellationContext->Cancel();
}

References audacity::cloud::audiocom::sync::BlockUploadTask::Block, audacity::cloud::audiocom::sync::MissingBlocksUploadProgress::FailedBlocks, mCancellationContext, mConcurrentUploads, mProgressCallback, mProgressData, mProgressDataMutex, mUploadsMutex, mUploadsNotFull, and audacity::cloud::audiocom::sync::MissingBlocksUploadProgress::UploadErrors.

Referenced by ConsumeBlock().

Here is the caller graph for this function:

◆ PopBlockFromQueue()

MissingBlocksUploader::ProducedItem audacity::cloud::audiocom::sync::MissingBlocksUploader::PopBlockFromQueue ( )

private

Definition at line 153 of file MissingBlocksUploader.cpp.

{
   std::unique_lock<std::mutex> lock(mRingBufferMutex);
   mRingBufferNotEmpty.wait(
      lock,
      [this]
      {
         return mRingBufferWriteIndex != mRingBufferReadIndex ||
                !mIsRunning.load(std::memory_order_consume);
      });
 
   if (!mIsRunning.load(std::memory_order_relaxed))
      return {};
 
   auto item            = std::move(mRingBuffer[mRingBufferReadIndex]);
   mRingBufferReadIndex = (mRingBufferReadIndex + 1) % RING_BUFFER_SIZE;
 
   mRingBufferNotFull.notify_one();
 
   return std::move(item);
}

References mIsRunning, mRingBuffer, mRingBufferMutex, mRingBufferNotEmpty, mRingBufferNotFull, mRingBufferReadIndex, mRingBufferWriteIndex, and RING_BUFFER_SIZE.

◆ ProduceBlock()

MissingBlocksUploader::ProducedItem audacity::cloud::audiocom::sync::MissingBlocksUploader::ProduceBlock ( )

private

Definition at line 88 of file MissingBlocksUploader.cpp.

{
   const auto index = mFirstUnprocessedBlockIndex++;
   const auto& task = mUploadTasks[index];
 
   return { task, CompressBlock(task.Block) };
}

References audacity::cloud::audiocom::sync::CompressBlock(), mFirstUnprocessedBlockIndex, and mUploadTasks.

Here is the call graph for this function:

◆ ProducerThread()

void audacity::cloud::audiocom::sync::MissingBlocksUploader::ProducerThread ( )

private

Definition at line 216 of file MissingBlocksUploader.cpp.

{
   while (mIsRunning.load(std::memory_order_consume))
   {
      BlockUploadTask task;
 
      {
         std::lock_guard<std::mutex> lock(mBlocksMutex);
 
         if (mFirstUnprocessedBlockIndex >= mUploadTasks.size())
            return;
 
         const auto index = mFirstUnprocessedBlockIndex++;
         task = std::move(mUploadTasks[index]);
      }
 
      auto compressedData = CompressBlock(task.Block);
 
      if (compressedData.empty())
      {
         MissingBlocksUploadProgress progressData;
         {
            std::lock_guard<std::mutex> lock(mProgressDataMutex);
            mProgressData.FailedBlocks++;
            progressData = mProgressData;
         }
 
         mProgressCallback(
            progressData, task.Block,
            { SyncResultCode::InternalClientError, {} });
      }
      else
      {
         PushBlockToQueue(
            ProducedItem { std::move(task), std::move(compressedData) });
      }
   }
}

References audacity::cloud::audiocom::sync::BlockUploadTask::Block, audacity::cloud::audiocom::sync::CompressBlock(), audacity::cloud::audiocom::sync::MissingBlocksUploadProgress::FailedBlocks, mBlocksMutex, mFirstUnprocessedBlockIndex, mIsRunning, mProgressCallback, mProgressData, mProgressDataMutex, mUploadTasks, and PushBlockToQueue().

Referenced by Start().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ PushBlockToQueue()

void audacity::cloud::audiocom::sync::MissingBlocksUploader::PushBlockToQueue ( ProducedItem item )

private

Definition at line 132 of file MissingBlocksUploader.cpp.

{
   std::unique_lock<std::mutex> lock(mRingBufferMutex);
   mRingBufferNotFull.wait(
      lock,
      [this]
      {
         return ((mRingBufferWriteIndex + 1) % RING_BUFFER_SIZE) !=
                   mRingBufferReadIndex ||
                !mIsRunning.load(std::memory_order_consume);
      });
 
   if (!mIsRunning.load(std::memory_order_relaxed))
      return;
 
   mRingBuffer[mRingBufferWriteIndex] = std::move(item);
   mRingBufferWriteIndex = (mRingBufferWriteIndex + 1) % RING_BUFFER_SIZE;
 
   mRingBufferNotEmpty.notify_one();
}

References mIsRunning, mRingBuffer, mRingBufferMutex, mRingBufferNotEmpty, mRingBufferNotFull, mRingBufferReadIndex, mRingBufferWriteIndex, and RING_BUFFER_SIZE.

Referenced by ProducerThread().

Here is the caller graph for this function:

◆ Start()

void audacity::cloud::audiocom::sync::MissingBlocksUploader::Start	(	CancellationContextPtr	cancellationContex,
		std::vector< BlockUploadTask >	uploadTasks,
		MissingBlocksUploadProgressCallback	progress
	)

private

Definition at line 52 of file MissingBlocksUploader.cpp.

{
   mCancellationContext = std::move(cancellationContex);
   mUploadTasks         = std::move(uploadTasks);
   mProgressCallback    = std::move(progressCallback);
   if (!mProgressCallback)
      mProgressCallback = [](auto...) {};
 
   mProgressData.TotalBlocks = mUploadTasks.size();
 
   for (auto& thread : mProducerThread)
      thread = std::thread([this] { ProducerThread(); });
 
   mConsumerThread = std::thread([this] { ConsumerThread(); });
}

References ConsumerThread(), mCancellationContext, mConsumerThread, mProducerThread, mProgressCallback, mProgressData, mUploadTasks, ProducerThread(), and audacity::cloud::audiocom::sync::MissingBlocksUploadProgress::TotalBlocks.

Here is the call graph for this function:

Member Data Documentation

◆ mBlocksMutex

std::mutex audacity::cloud::audiocom::sync::MissingBlocksUploader::mBlocksMutex

private

Definition at line 121 of file MissingBlocksUploader.h.

Referenced by ProducerThread().

◆ mCancellationContext

CancellationContextPtr audacity::cloud::audiocom::sync::MissingBlocksUploader::mCancellationContext

private

Definition at line 140 of file MissingBlocksUploader.h.

Referenced by ConsumeBlock(), HandleFailedBlock(), and Start().

◆ mConcurrentUploads

size_t audacity::cloud::audiocom::sync::MissingBlocksUploader::mConcurrentUploads { 0 }

private

Definition at line 126 of file MissingBlocksUploader.h.

Referenced by ConfirmBlock(), ConsumeBlock(), and HandleFailedBlock().

◆ mConsumerThread

std::thread audacity::cloud::audiocom::sync::MissingBlocksUploader::mConsumerThread

private

Definition at line 119 of file MissingBlocksUploader.h.

Referenced by Cancel(), and Start().

◆ mFirstUnprocessedBlockIndex

size_t audacity::cloud::audiocom::sync::MissingBlocksUploader::mFirstUnprocessedBlockIndex { 0 }

private

Definition at line 122 of file MissingBlocksUploader.h.

Referenced by ProduceBlock(), and ProducerThread().

◆ mIsRunning

std::atomic_bool audacity::cloud::audiocom::sync::MissingBlocksUploader::mIsRunning { true }

private

Definition at line 116 of file MissingBlocksUploader.h.

Referenced by Cancel(), ConsumeBlock(), PopBlockFromQueue(), ProducerThread(), and PushBlockToQueue().

◆ mProducerThread

std::thread audacity::cloud::audiocom::sync::MissingBlocksUploader::mProducerThread[NUM_PRODUCERS]

private

Definition at line 118 of file MissingBlocksUploader.h.

Referenced by Cancel(), and Start().

◆ mProgressCallback

MissingBlocksUploadProgressCallback audacity::cloud::audiocom::sync::MissingBlocksUploader::mProgressCallback

private

Definition at line 114 of file MissingBlocksUploader.h.

Referenced by ConfirmBlock(), HandleFailedBlock(), ProducerThread(), and Start().

◆ mProgressData

MissingBlocksUploadProgress audacity::cloud::audiocom::sync::MissingBlocksUploader::mProgressData

private

Definition at line 138 of file MissingBlocksUploader.h.

Referenced by ConfirmBlock(), HandleFailedBlock(), ProducerThread(), and Start().

◆ mProgressDataMutex

std::mutex audacity::cloud::audiocom::sync::MissingBlocksUploader::mProgressDataMutex

private

Definition at line 137 of file MissingBlocksUploader.h.

Referenced by Cancel(), ConfirmBlock(), HandleFailedBlock(), and ProducerThread().

◆ mRingBuffer

std::array<ProducedItem, RING_BUFFER_SIZE> audacity::cloud::audiocom::sync::MissingBlocksUploader::mRingBuffer

private

Definition at line 133 of file MissingBlocksUploader.h.

Referenced by PopBlockFromQueue(), and PushBlockToQueue().

◆ mRingBufferMutex

std::mutex audacity::cloud::audiocom::sync::MissingBlocksUploader::mRingBufferMutex

private

Definition at line 128 of file MissingBlocksUploader.h.

Referenced by PopBlockFromQueue(), and PushBlockToQueue().

◆ mRingBufferNotEmpty

std::condition_variable audacity::cloud::audiocom::sync::MissingBlocksUploader::mRingBufferNotEmpty

private

Definition at line 130 of file MissingBlocksUploader.h.

Referenced by Cancel(), PopBlockFromQueue(), and PushBlockToQueue().

◆ mRingBufferNotFull

std::condition_variable audacity::cloud::audiocom::sync::MissingBlocksUploader::mRingBufferNotFull

private

Definition at line 131 of file MissingBlocksUploader.h.

Referenced by Cancel(), PopBlockFromQueue(), and PushBlockToQueue().

◆ mRingBufferReadIndex

size_t audacity::cloud::audiocom::sync::MissingBlocksUploader::mRingBufferReadIndex { 0 }

private

Definition at line 135 of file MissingBlocksUploader.h.

Referenced by PopBlockFromQueue(), and PushBlockToQueue().

◆ mRingBufferWriteIndex

size_t audacity::cloud::audiocom::sync::MissingBlocksUploader::mRingBufferWriteIndex { 0 }

private

Definition at line 134 of file MissingBlocksUploader.h.

Referenced by PopBlockFromQueue(), and PushBlockToQueue().

◆ mServiceConfig

const ServiceConfig& audacity::cloud::audiocom::sync::MissingBlocksUploader::mServiceConfig

private

Definition at line 111 of file MissingBlocksUploader.h.

Referenced by ConsumeBlock().

◆ mUploadsMutex

std::mutex audacity::cloud::audiocom::sync::MissingBlocksUploader::mUploadsMutex

private

Definition at line 124 of file MissingBlocksUploader.h.

Referenced by ConfirmBlock(), ConsumeBlock(), and HandleFailedBlock().

◆ mUploadsNotFull

std::condition_variable audacity::cloud::audiocom::sync::MissingBlocksUploader::mUploadsNotFull

private

Definition at line 125 of file MissingBlocksUploader.h.

Referenced by Cancel(), ConfirmBlock(), ConsumeBlock(), and HandleFailedBlock().

◆ mUploadTasks

std::vector<BlockUploadTask> audacity::cloud::audiocom::sync::MissingBlocksUploader::mUploadTasks

private

Definition at line 113 of file MissingBlocksUploader.h.

Referenced by ProduceBlock(), ProducerThread(), and Start().

◆ NUM_PRODUCERS

constexpr auto audacity::cloud::audiocom::sync::MissingBlocksUploader::NUM_PRODUCERS = 3

staticconstexpr

Definition at line 72 of file MissingBlocksUploader.h.

◆ NUM_UPLOADERS

constexpr auto audacity::cloud::audiocom::sync::MissingBlocksUploader::NUM_UPLOADERS = 6

staticconstexpr

Definition at line 73 of file MissingBlocksUploader.h.

Referenced by ConsumeBlock().

◆ RING_BUFFER_SIZE

constexpr auto audacity::cloud::audiocom::sync::MissingBlocksUploader::RING_BUFFER_SIZE = 16

staticconstexpr

Definition at line 74 of file MissingBlocksUploader.h.

Referenced by PopBlockFromQueue(), and PushBlockToQueue().

The documentation for this class was generated from the following files:

Classes

Public Member Functions

Static Public Member Functions

Static Public Attributes

Private Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

◆ MissingBlocksUploader()

◆ ~MissingBlocksUploader()

Member Function Documentation

◆ Cancel()

◆ ConfirmBlock()

◆ ConsumeBlock()

◆ ConsumerThread()

◆ Create()

◆ HandleFailedBlock()

◆ PopBlockFromQueue()

◆ ProduceBlock()

◆ ProducerThread()

◆ PushBlockToQueue()

◆ Start()

Member Data Documentation

◆ mBlocksMutex

◆ mCancellationContext

◆ mConcurrentUploads

◆ mConsumerThread

◆ mFirstUnprocessedBlockIndex

◆ mIsRunning

◆ mProducerThread

◆ mProgressCallback

◆ mProgressData

◆ mProgressDataMutex

◆ mRingBuffer

◆ mRingBufferMutex

◆ mRingBufferNotEmpty

◆ mRingBufferNotFull

◆ mRingBufferReadIndex

◆ mRingBufferWriteIndex

◆ mServiceConfig

◆ mUploadsMutex

◆ mUploadsNotFull

◆ mUploadTasks

◆ NUM_PRODUCERS

◆ NUM_UPLOADERS

◆ RING_BUFFER_SIZE