Tutorial 8: The critical path
On this page
Where we are
Chapter 7 computes the CRC per block in parallel and combines it in order. The serial work left is reading, writing and a combine per block.
The problem
The reader feeds the whole pipeline. Every time it waits for a read, it is suspended; when the read completes, it becomes runnable again – and goes to the back of the queue, behind every encode task that is waiting for a thread. Until a thread gets round to it, no new block is read, and when the encode tasks run out, threads sit idle:
reader │R8│ queued behind encode tasks │ R9 │ …
thread 1 │ encode 4 │ encode 6 │ encode 8 │ idle │ encode 9 │
thread 2 │ encode 5 │ encode 7 │ idle │ …The reader is on the critical path: any delay of it delays everything. The encode tasks are not: it does not matter which of them runs first.
The writer does not have this problem here: it is main, which runs on its own thread and does not
queue for a scheduler thread.
The building block
TaskPriority. A task runs at High or Normal. A runnable High task is
always taken before Normal work, whether it is new or resumed after a wait.
AddTask takes the priority as a second argument.
Why not make everything High? Then nothing is preferred, and nothing gained. And there is no aging: a steady stream of High work starves the Normal tasks. Keep High for the few tasks on the critical path, which spend most of their time waiting anyway.
Important
The trap. AddTask with TaskPriority::Default – the default
argument – gives the new task the priority of the task that adds it. The reader runs at High, so
every encode task it forks without a priority would run at High too, and the priorities would
change nothing. Fork the parallel work at Normal explicitly.
The change
Pack starts the reader at High:
class Packer
{
// ...
bool Pack(TC::ITaskScheduler& scheduler, const Arguments& arguments,
std::size_t slotCount)
{
// ...
// The reader is the serial stage that feeds everything else: High.
scheduler.AddTask(&reader_, TC::TaskPriority::High);
// ...
}
// ...
};The reader forks the encoding at Normal:
class Packer
{
// ...
void Read() noexcept
{
for (std::size_t block = 0;; ++block) {
freeSlots_.Wait();
Slot& slot = slots_[block % slotCount_];
const std::uint64_t offset = std::uint64_t{slot.buffer.size()} * block;
DWORD read = 0; // DWORD: Read's out parameter
const auto error
= input_.Read(slot.buffer.data(),
static_cast<DWORD>(slot.buffer.size()), read, offset);
if (Error::Failed(error) && error != Error::EndOfFile)
readOk_ = false;
slot.size = read;
if (slot.size == 0) {
readySlots_.Release();
return;
}
slot.task.input = std::span(slot.buffer).first(slot.size);
slot.encoded.AddTask(&slot.task);
// Default would inherit the reader's High: the encoding is Normal.
scheduler_->AddTask(&slot.task, TC::TaskPriority::Normal);
readySlots_.Release();
}
}
// ...
};The result
blockpack input.bin output.bpk -r reference.bpk
time: … s
OKThe effect depends on the machine and the disk. With the file in the cache and few threads it is small; it grows with the number of threads and with slower reads, when the reader is resumed often while many encode tasks are queued. What the priority guarantees is the order: the reader gets the next free thread.
YieldTask with YieldFlags::HigherPriority complements this: a long-running Normal task can give way to waiting High work at points of its choosing. The encode tasks here are short enough not to need it.
What limits us now
The program works – as long as nothing goes wrong. A write error, or a user who presses Ctrl+C, must stop the pipeline without leaving the reader waiting for a slot that never becomes free, and without destroying a slot while its encode task still runs.
Learn more
Show the whole program
// Tutorial chapter 8: the critical path - the reader runs at high priority, so
// that it gets a thread as soon as its read completes instead of queuing behind
// the encode tasks.
#include "tutorial-common.h"
#include <memory>
#include <ThinkMeta.ConcurrentTasks/Core.h>
import ThinkMeta.ConcurrentTasks.AsyncIo.AsyncFile;
import ThinkMeta.ConcurrentTasks.Synchronization.Semaphore;
import ThinkMeta.ConcurrentTasks.Synchronization.TaskGroup;
import ThinkMeta.ConcurrentTasks.Tasks.Task;
import ThinkMeta.ConcurrentTasks.Tasks.WaitableTask;
import ThinkMeta.ConcurrentTasks.Windows.Error;
namespace TC = ThinkMeta::ConcurrentTasks;
using TC::AsyncIo::AsyncFile;
using TC::Windows::Error;
using namespace Tutorial;
// Encodes one block and computes its CRC, as in chapter 7.
class EncodeTask : public TC::Tasks::Task
{
public:
void Run() noexcept override
{
EncodeBlock(input, encoded);
crc = BlockCrc(encoded);
}
std::span<const std::byte> input;
EncodedBlock encoded;
std::uint32_t crc = 0;
};
struct Slot {
std::vector<std::byte> buffer;
std::size_t size = 0;
EncodeTask task;
TC::Synchronization::TaskGroup_s encoded;
};
class Packer;
class ReaderTask : public TC::Tasks::WaitableTask
{
public:
explicit ReaderTask(Packer& packer) noexcept : packer_(packer) {}
void Run() noexcept override;
private:
Packer& packer_;
};
class Packer
{
public:
bool Pack(TC::ITaskScheduler& scheduler, const Arguments& arguments,
std::size_t slotCount)
{
if (Error::Failed(input_.Create(arguments.input, GENERIC_READ,
FILE_SHARE_READ, OPEN_EXISTING,
FILE_FLAG_SEQUENTIAL_SCAN)))
return false;
File output = File::Create(arguments.output);
if (!output.IsOpen())
return false;
scheduler_ = &scheduler;
slotCount_ = slotCount;
slots_ = std::make_unique<Slot[]>(slotCount);
for (std::size_t i = 0; i < slotCount; ++i)
slots_[i].buffer.resize(arguments.blockSize);
freeSlots_.Release(slotCount);
// The reader is the serial stage that feeds everything else: High.
scheduler.AddTask(&reader_, TC::TaskPriority::High);
std::uint32_t crc = 0;
bool ok = true;
for (std::size_t block = 0;; ++block) {
readySlots_.Wait();
Slot& slot = slots_[block % slotCount];
if (slot.size == 0)
break;
slot.encoded.Wait();
const EncodedBlock& encoded = slot.task.encoded;
ok = ok && WriteBlockData(output, encoded);
crc = CombineCrc32(crc, slot.task.crc, BlockBytes(encoded));
freeSlots_.Release();
}
reader_.Wait();
input_.Close();
return WriteFooter(output, crc) && ok && readOk_;
}
void Read() noexcept
{
for (std::size_t block = 0;; ++block) {
freeSlots_.Wait();
Slot& slot = slots_[block % slotCount_];
const std::uint64_t offset = std::uint64_t{slot.buffer.size()} * block;
DWORD read = 0; // DWORD: Read's out parameter
const auto error
= input_.Read(slot.buffer.data(),
static_cast<DWORD>(slot.buffer.size()), read, offset);
if (Error::Failed(error) && error != Error::EndOfFile)
readOk_ = false;
slot.size = read;
if (slot.size == 0) {
readySlots_.Release();
return;
}
slot.task.input = std::span(slot.buffer).first(slot.size);
slot.encoded.AddTask(&slot.task);
// Default would inherit the reader's High: the encoding is Normal.
scheduler_->AddTask(&slot.task, TC::TaskPriority::Normal);
readySlots_.Release();
}
}
private:
TC::ITaskScheduler* scheduler_ = nullptr;
AsyncFile input_;
std::size_t slotCount_ = 0;
std::unique_ptr<Slot[]> slots_;
TC::Synchronization::Semaphore_s freeSlots_;
TC::Synchronization::Semaphore_s readySlots_;
ReaderTask reader_{*this};
bool readOk_ = true;
};
void ReaderTask::Run() noexcept
{
packer_.Read();
}
int wmain(int argc, wchar_t* argv[])
{
Arguments arguments;
if (!ParseArguments(argc, argv, arguments))
return arguments.exitCode;
const std::uint32_t threads = arguments.threads != 0
? arguments.threads
: GetActiveProcessorCount(ALL_PROCESSOR_GROUPS);
const std::size_t slotCount = std::size_t{4} * threads;
TC::ConvertThreadToTask();
if (!TC::AsyncIo::EnableAsyncIo())
return 1;
TC::SchedulerInfo info{};
info.numberOfThreads = threads;
TC::ITaskScheduler* scheduler
= TC::CreateTaskScheduler(TC::SchedulerKind::Default, &info);
if (scheduler == nullptr)
return 1;
scheduler->Start();
bool packed = false;
{
Packer packer;
const Stopwatch total;
packed = packer.Pack(*scheduler, arguments, slotCount);
std::printf("time: %.2f s\n", total.Seconds());
scheduler->Stop();
}
scheduler->Destroy();
TC::AsyncIo::DisableAsyncIo();
TC::ConvertTaskToThread();
if (!packed) {
std::printf("cannot pack the file\n");
return 1;
}
return CheckOutput(arguments) ? 0 : 1;
}