Pull, Don't Push: Fixing Load Imbalance with a Shared Task Queue

Sources this series builds on: Chili's (ChiliTomatoNoodle) multithreading
playlist,
CMU 15-213 (Computer Systems: A Programmer's Perspective), various CppCon talks,
and LLMs for restructuring and sanity-checking. Everything here is my own code,
compiled and run — not just transcribed.

---

Recap

Part 8 measured the actual cost of a bad assumption: split a chunk of tasks
into 4 equal-count slices, hand one slice to each worker, and hope the
work inside those slices is roughly equal too. In the Stacked layout it
wasn't — one worker got every heavy task, the other three got none, and
three of four workers sat idle ~85% of the time waiting on the one that
didn't. The fix isn't a smarter way to pre-split the pile. It's to stop
pre-splitting at all.

---

The actual problem with fixed slicing

Every version in Part 8 decided, up front, before a single task ran, exactly
which tasks each worker would get. That decision was made with zero
information about how expensive any individual task actually was. Once a
worker's slice is heavy, there's no way to hand some of that work to an
idle neighbor mid-chunk — the assignment was locked in before anyone knew
it was a bad one.

flowchart LR
    subgraph Push["Push model (Part 8)"]
        direction LR
        P["Pile of tasks"] --> PS["Pre-split into 4<br/>fixed slices, up front"]
        PS --> PW1["Worker 1's slice"]
        PS --> PW2["Worker 2's slice"]
        PS --> PW3["Worker 3's slice"]
        PS --> PW4["Worker 4's slice<br/>(happens to be heavy)"]
    end

    style PW4 fill:#FAECE7,stroke:#993C1D,color:#4A1B0C

The fix: instead of deciding assignments in advance, let each worker pull
the next task from a shared pile the moment it's free. A worker that got
lucky with cheap tasks finishes fast and immediately comes back for more.
A worker stuck on an expensive task just takes longer on that one task —
but it isn't sitting on a whole pre-assigned pile of them anymore, because
there never was one.

flowchart LR
    subgraph Pull["Pull model (this article)"]
        direction LR
        Q["Shared task queue<br/>(one pile, shared cursor)"] --> QW1["Worker 1 pulls<br/>when free"]
        Q --> QW2["Worker 2 pulls<br/>when free"]
        Q --> QW3["Worker 3 pulls<br/>when free"]
        Q --> QW4["Worker 4 pulls<br/>when free"]
    end

    style Q fill:#E1F5EE,stroke:#0F6E56,color:#04342C

Nobody decides in advance who gets which task. The pile naturally drains
faster through whichever workers happen to be free at any given moment.

---

The real implementation: tq::Master and tq::Worker

The shared cursor

class Master {
private:
    std::span<Task> chunk;
    std::condition_variable cv;
    std::mutex mtx;
    std::atomic<int> idx;
    int doneCount = 0;
public:
    Master() {};
    void setChunk(std::span<Task> c) {
        {
            std::lock_guard<std::mutex> lock(mtx);
            chunk = c;
            idx = 0;
        }
    }

    void setDone() {
        bool notify = false;
        {
            std::lock_guard<std::mutex> lock(mtx);
            ++doneCount;
            notify = (doneCount == WORKER_COUNT);
        }
        if (notify) cv.notify_one();
    }

    const Task* getTask() {
        int currentIdx = idx++;
        if (currentIdx >= CHUNK_SIZE) return nullptr;
        return &chunk[currentIdx];
    }

    void wait_for_all() {
        std::unique_lock<std::mutex> lock(mtx);
        cv.wait(lock, [this]() ->  bool { return doneCount == WORKER_COUNT; });
        doneCount = 0;
    }
};

getTask() is the entire mechanism. idx is a std::atomic<int>, and
idx++ on it is a single locked hardware instruction (Part 6) — every
worker calling getTask() at the same moment gets a genuinely unique,
non-overlapping index back, with no mutex around the hot path at all.
Whichever worker happens to call it first gets index 0; the next caller,
whoever that is, gets index 1 — assignment order is decided by whichever
worker actually shows up asking, not by anything fixed in advance. Once
idx passes CHUNK_SIZE, every subsequent caller gets nullptr, which is
how a worker knows the pile is empty and it's time to stop.

The mutex/condition_variable pair here is doing a completely different
job from idx — it's not protecting the task dispatch itself, it's used
for setChunk (loading a new chunk between rounds) and wait_for_all
(blocking the main thread until every worker reports done). The hot,
frequently-called path — getTask() — never touches the mutex at all.

The worker loop

class Worker {
private:
    Master* master;
    bool working;
    bool dying;
    std::condition_variable cv;
    std::mutex mtx;
    std::jthread thread;
    unsigned int accumulation;
    int heavyProcessed = 0;
    float timeSpentPerWorker = 0.;
    Timer* t;

private:
    void Run_() {
        std::unique_lock<std::mutex> lock(mtx);
        while (true) {
            cv.wait(lock, [this]() -> bool { return working || dying; });
            if (dying) {
                return;
            }
            t->start();
            while (const Task* task = master->getTask()) {
                accumulation += task->process();
                heavyProcessed += task->heavy ? 1 : 0;
            }
            timeSpentPerWorker = t->stop("Worker Processing Time", false);

            working = false;
            master->setDone();
        }
    }
public:
    Worker(Master* master, Timer* t) : master(master), working(false), dying(false), accumulation(0), thread(&Worker::Run_, this), t(t) {};
    void startWorking() {
        heavyProcessed = 0;
        {
            std::unique_lock<std::mutex> lock(mtx);
            working = true;
        }
        cv.notify_one();
    }
    void Kill()
    {
        {
            std::lock_guard<std::mutex> lg(mtx);
            dying = true;
        }
        cv.notify_one();
    }

    int getHeavyCount() const { return heavyProcessed; }
    float getTimeSpentPerWorker() const { return timeSpentPerWorker; }

    double GetResult() const { return accumulation; }
};

The inner loop is the whole idea in one line:

while (const Task* task = master->getTask()) {
    accumulation += task->process();
    heavyProcessed += task->heavy ? 1 : 0;
}

No pre-assigned span, no fixed slice — just "keep asking for the next task
until there isn't one." A worker that burns through five cheap tasks in the
time another worker spends on one heavy task will simply have called
getTask() five more times. The queue doesn't know or care about heavy
at all; it never needs to, because it isn't trying to predict cost, it's
just handing out the next item to whoever's free.

Wiring it up

void Thread::simpleMultiThreadTaskQueue() {
    std::vector<tq::Chunk> chunks = generateStackedDataSeti();
    t.start();
    tq::Master master;
    std::vector<std::unique_ptr<tq::Worker>> workers;
    std::vector<ChunkTimingInfo> chunkTimings;
    chunkTimings.reserve(CHUNK_COUNT);
    for (int i = 0; i < WORKER_COUNT; i++) {
        workers.push_back(std::make_unique<tq::Worker>(&master, &t));
    }

    for (auto& chunk : chunks) {
        t.start();
        master.setChunk(chunk);
        for (auto& worker : workers) {
            worker->startWorking();
        }
        master.wait_for_all();
        float totalTime = t.stop("Chunk Processing Time", false);

        ChunkTimingInfo cTiming;
        for (int i = 0; i < WORKER_COUNT; i++) {
            cTiming.timeSpentPerWorker[i] = workers[i]->getTimeSpentPerWorker();
            cTiming.heavyCountPerWorker[i] = workers[i]->getHeavyCount();
        }
        cTiming.totalChunkTime = totalTime;
        chunkTimings.push_back(cTiming);
    }
    unsigned int answer = 0.0;
    for (auto& w : workers) answer += w->GetResult();
    std::cout << "Total sum: " << answer << "\n";

    for (auto& w : workers) w->Kill();
    workers.clear();
    t.stop("Multi Thread Task Queue", true);
    // ... CSV output, same ChunkTimingInfo format as Part 8
}

Deliberately run against generateStackedDataSeti() — the exact worst-case
layout from Part 8, all the heavy tasks clustered together. If the pull
model actually fixes the imbalance, this is the hardest dataset it could be
asked to prove that on. Same ChunkTimingInfo struct, same per-worker
timing and heavy-count tracking as Part 8's WorkerController, so the
results are directly comparable, chunk for chunk.

---

Why the atomic cursor specifically, not a mutex-protected queue

It would also work to protect idx (or a real std::queue) with a plain
std::mutex, locked and unlocked on every getTask() call. The reason
std::atomic<int> is used instead ties straight back to Part 6: idx++
compiles to a single lock-prefixed instruction. A mutex, by contrast,
means every worker blocks and potentially sleeps/wakes through the OS
scheduler just to grab the next index — real overhead, paid on every single
task dispatch, for work that doesn't need anything close to a full mutex's
guarantees. This is genuinely the shape of problem atomics exist for: a
tiny, hot, frequently-contended piece of shared state, where the operation
needed is simple enough that a locked hardware instruction covers it
completely.

---

Results — same worst-case dataset, 100 chunks

Part 8's Stacked layout — the exact same 167-heavy-task cluster used
here — produced 63.68% idle time with fixed slicing. Run the identical
dataset through the task queue instead:

Approach	Avg chunk time	Idle as % of available worker-time
Fixed slicing (Part 8, Stacked)	0.007561s	63.68%
Pull-based queue (this article)	0.002982s	0.67%

Same total work, same worker count, same worst-case task arrangement.
Idle time drops from 63.68% to 0.67% — and chunk time itself drops by
2.54x, purely from letting workers self-balance instead of guessing
their assignments up front.

The per-worker breakdown shows exactly why. Despite the queue having no
concept of heavy whatsoever, every worker ends up with almost the same
number of heavy tasks:

Worker	Avg time	Avg idle	Idle %	Avg heavy tasks
0	0.002962s	0.000020s	0.67%	42.0
1	0.002962s	0.000020s	0.68%	41.9
2	0.002962s	0.000020s	0.67%	42.0
3	0.002962s	0.000020s	0.66%	41.0

Compare that to Part 8's Stacked breakdown: one worker with all 167 heavy
tasks, three with zero. Here the same 167 heavy tasks split almost exactly
41–42 per worker, without the queue ever checking a single task's heavy
flag before handing it out. The balance isn't the result of a smarter
scheduling decision — it's a side effect of pulling one task at a time
instead of committing to a block of them in advance. A worker that
happens to draw several light tasks in a row simply calls getTask() more
often in the same span of time, and naturally ends up drawing into the
heavy cluster more too, since the cluster is still sitting in the same
shared pile everyone is pulling from.

---

Takeaways

• Fixed, pre-decided slicing (Part 8) locks in an assignment before anyone knows if it's a good one. A pull-based queue never has to guess — workers self-balance by simply asking for more work exactly when they're free.
• On the exact same worst-case dataset, idle time dropped from 63.68% (fixed slicing) to 0.67% (pull-based queue) — and chunk time itself dropped 2.54x.
• The queue never checks a task's heavy flag, yet every worker ended up with almost the same heavy-task count (~41–42 each, versus 167-vs-0 with fixed slicing). The balance is a side effect of pulling one task at a time, not a scheduling decision anyone made on purpose.
• The mechanism is a single std::atomic<int> cursor, incremented with a locked instruction on every request — not a queue data structure guarded by a mutex. Simpler, and cheaper on the hot path.
• Tested against the exact Stacked layout that produced ~85% idle time on individual workers in Part 8 — the hardest case available, not a favorable one picked to make the fix look better than it is.