If you haven't read my previous blog Multithreads of Madness, you probably should, because you'll get a chance to witness some real tragic comedy. Undoubtedly, at my generous expense.
If you already feel hooked to this one (i totally understand), the TLDR is that my multithreaded matrix multiplication function spawned a bunch of threads every time the function was called for a sufficiently big matrix. And for doing deep learning work, if that is not a clear death sentence, I don't know what is.
In hindsight, that explains why that blog performed so poorly lol.
The blog wasn't entirely rubbish though. It intuitively introduced multi-threading, went over major trade-offs (ironic) and had a killer icebreaker joke.
While the multiplication function clearly demonstrated desirable results, the problem was that it was isolated. In a way that it could only serve as a standalone demo for basic multithreading. If I wanted to implement it for a performance-focused project, it would become critical to implement multithreading as efficiently as possible.
I have been working on my C++ MNIST Neural Network for a while, and since the SGD training process presented clear opportunities to be parallelized, I was determined to get it right this time.
Note: This blog assumes basic knowledge of neural networks and backpropagation. However, the thread pool implementation is generally applicable. Its code is sourced from an SO answer and is slightly modified for my use case. eep in mind that this implementation doesn't support the necessary sophistication required by production systems (like exception handling and graceful shutdowns), nor is it intended to.
In my C++ neural network (which is heavily inspired by Nielsen's), the Stochastic Gradient Descent (SGD) algorithm looks roughly as follows -
For each epoch:
- Shuffle the training data
- Create mini-batches
- Perform backpropagation
- Update parameters
In terms of the functions in the program, here is the execution flow:
void SGD(...)calls...void update_mini_batch(...)calls...void backprop(...)
At first, I was running the pipeline serially. A single thread was going through each epoch one after another, one mini-batch after another and one training sample after another.
In order to expedite the training process, I naturally leaned towards implementing concurrency. The first question I asked was - Can I parallelize epochs?
I am using a synchronous SGD implementation, where each epoch begins from the parameters produced by the previous epoch, meaning that parallelizing epochs would be impossible. Here is a simple justification: Imagine a multi-dimensional parameter hyperplane. Just kidding. Instead, imagine the hilly land with multiple peaks and valleys. This landscape represents the loss function. We want to get to the lowest point in this landscape, as this position has the minimum loss.
Gradient descent works by finding the gradient and direction of steepest ascent given a reference point and descending in the opposite direction by a certain amount. If I am on point A, and I find the steepest direction and move downwards. But at the new point B, it is highly unnecessary that the next direction of descent will be the same. Which implies the following - I need my current location to descend to the next point and I am at my current location because of where I was before.
In each epoch, I am essentially judging my current location to decide where to move next. That is a clear sequential task.
It is a different story when go a level deeper. Common implementations of update_mini_batch() take the container full of batches of training samples and apply backpropagation to each batch. The accumulated gradients are then subtracted from the current parameters (or coordinates if the hill analogy resonated with you).
Now, it is the same thing if we calculate the loss function for the entire training data set or calculate the function for n batches and add them up. Since the derivative of sums is equal to the sum of derivatives, the gradient of the loss function for the entire dataset can be represented by the sum of gradients of the loss function for each batch.
In standard mini-batch SGD, the parameters (weights and biases) are updated after processing each mini-batch. Just like the sequential nature of epochs, the mini-batches must also be executed sequentially to ensure the integrity of each step "downwards".
(Asynchronous SGD does exist, but it introduces optimisation problems that simply aren't worth it for simple networks like the one discussed here. For this reason, we will take sequentiality for granted in this blog.)
Going a level deeper, notice that the parameters are updated only after processing an entire mini-batch. Therefore, the individual training samples within that mini-batch can be processed concurrently, as long as each worker computes its gradient contribution independently and the contributions are aggregated before the parameter update.
Here is the ThreadPool class declaration:
class ThreadPool {
public:
~ThreadPool();
void Start();
void QueueJob(const std::function<void()>& job);
void QueueBatch(const std::vector<std::function<void()>>& job_batch);
void Wait();
void Stop();
bool busy();
private:
bool should_terminate = false;
size_t pending_tasks = 0;
std::mutex queue_mutex;
std::condition_variable mutex_condition;
std::condition_variable wait_condition;
std::vector<std::thread> threads;
std::queue<std::function<void()>> jobs;
void ThreadLoop();
};
Let's go through each member function:
void ThreadPool::Start() {
const int num_threads = std::thread::hardware_concurrency();
threads.reserve(num_threads);
for (int i = 0; i < num_threads; i++)
{
threads.emplace_back(&ThreadPool::ThreadLoop, this);
}
}
std::thread::hardware_concurrency() gives us a rough idea about the number of hardware threads that can execute concurrently. (hardware_concurrency() can return 0 when the implementation cannot determine the value, so production code should provide a fallback.) Capacity for the corresponding number of std::thread objects is reserved in the vector. Each call to emplace_back() then constructs a new thread whose entry point is ThreadLoop(this).
void ThreadPool::ThreadLoop() {
while (true) {
std::function<void()> job;
{
std::unique_lock<std::mutex> lock(queue_mutex);
mutex_condition.wait(lock, [this] {
return !jobs.empty() || should_terminate;
});
if (should_terminate) {
return;
}
job = jobs.front();
jobs.pop();
}
job();
{
std::unique_lock<std::mutex> lock(queue_mutex);
pending_tasks--;
if (pending_tasks == 0) {
wait_condition.notify_all();
}
}
}
}
Each thread executes the same continuous loop until the thread pool is stopped. A function object job is initialized.
The braces might seem random and even unnecessary, but they serve a vital purpose that stems from one of the most fundamental language features. Each set of braces creates its own scope. All variables initialized within a set of braces will only exist within them. Leaving the braces results in them getting out of scope and being destructed.
Continuing with ThreadLoop, we can see braces being used to set a local scope. This is a common pattern applied when using RAII mutex guards like std::lock_guard, and std::unique_lock, as it allows the guard to automatically unlock the mutex when it goes out of scope. In other words, the braces limit the lifetime of the lock guard so that the mutex is automatically released at the desired point. RAII for life! Notice that we wanted the guard to only be active for a particular section of the function, namely where we access the shared jobs queue to take a task.
mutex_condition.wait() blocks the thread until either (1) the jobs queue is non-empty or (2) the signal to terminate the pool has been sent out. It does so by releasing the mutex and putting the thread to sleep. When the condition variable is notified (or a spurious wakeup occurs), the thread reacquires the mutex and evaluates the predicate again. Using std::condition_variable avoids the unnecessary CPU consumption of active polling.
When the predicate evaluates to true, the worker copies the job out of the queue, leaves the scope, and releases queue_mutex before executing the job. It is important to understand that a thread will only get to this point when it actually has received a job. Until then, the thread will essentially be suspended by wait().
After the job is finished, we create another local scope where the pending_tasks shared counter is decremented. pending_tasks counts every submitted task that has not yet completed—meaning it includes both tasks waiting in jobs and tasks currently being executed by workers. Finally, if there are no remaining tasks (pending_tasks == 0), we notify all threads waiting on wait_condition.
void ThreadPool::QueueJob(const std::function<void()>& job) {
{
std::unique_lock<std::mutex> lock(queue_mutex);
jobs.push(job);
pending_tasks++;
}
mutex_condition.notify_one();
}
Again, the local scope exists to control the lifetime of lock. We acquire the mutex lock, push the new function to jobs, and increment pending_tasks. The mutex is released automatically when the scope ends.
When a new task is enqueued, only one waiting thread is notified via notify_one() to wake up and take the task.
void ThreadPool::QueueBatch(const std::vector<std::function<void()>>& job_batch) {
if (job_batch.empty()) return;
{
std::unique_lock<std::mutex> lock(queue_mutex);
for (const auto& job : job_batch) {
jobs.push(job);
}
pending_tasks += job_batch.size();
}
mutex_condition.notify_all();
}
This is simply a batched version of ThreadPool::QueueJob. The only relevant difference is that instead of notifying a single thread, we call notify_all() so that all waiting threads are notified.
The way this works: Threads are woken from sleep when notify_all() is called. They land back on to the wait() function from ThreadLoop, where they need to wait their turn to access the mutex. Once it is acquired, they check whether they can take up a task via the predicate. If yes, then they get on with that task. If the predicate still isn't fulfilled, then they go back to sleep and give up the mutex for another thread to try.
bool ThreadPool::busy() {
std::unique_lock<std::mutex> lock(queue_mutex);
return pending_tasks > 0;
}
Simply put, if any thread is still executing its task(s), or the jobs shared queue is non-empty, busy() will return true. Otherwise false. pending_tasks counts both currently executing tasks as well as tasks inside the shared queue.
void ThreadPool::Wait() {
std::unique_lock<std::mutex> lock(queue_mutex);
wait_condition.wait(lock, [this] { return pending_tasks == 0; });
}
Naturally, we need a way to ensure that ThreadPool finishes all of the assigned tasks before a certain point. This is similar to using std::thread::join() (see Multithreads of Madness for how that works), except that we need to actually keep the threads alive to execute future tasks.
This works by acquiring the mutex and checking the predicate. If pending_tasks > 0, the mutex is released and the calling thread (in this case, the main thread) is suspended. When a worker finishes a job, it acquires the mutex, decrements pending_tasks, and calls wait_condition.notify_all(). Wait() wakes up, reacquires the mutex, and re-evaluates the predicate. Once pending_tasks == 0, Wait() returns and the calling thread can continue execution.
Note that this function doesn't prevent other threads from enqueuing tasks. It works under the assumption that no further tasks will be added after Wait() is called.
Initially, I implemented this function as follows:
void ThreadPool::Wait() {
while (busy());
}
A simple and intuitive polling loop that actually has a deep resource cost. Not only is the mutex constantly acquired and released when busy() is called, but the active polling also overloads the CPU and wastes resources.
void ThreadPool::Stop() {
{
std::unique_lock<std::mutex> lock(queue_mutex);
should_terminate = true;
}
mutex_condition.notify_all();
for (std::thread& active_thread : threads) {
active_thread.join();
}
threads.clear();
}
To stop the pool, we acquire queue_mutex within a local scope, set the shared flag should_terminate = true, and release the lock. We then call mutex_condition.notify_all() so all waiting threads wake up and check their predicate.
It is important to understand the shutdown semantics here: Stop() prevents workers from taking additional queued jobs. Workers that are already executing a job will finish it before checking should_terminate at the top of the loop and exiting. However, workers that are currently waiting will wake up, observe should_terminate == true, and immediately return—meaning any unstarted tasks remaining in jobs may be abandoned. For this reason, it is essential to call Wait() before Stop() if you want all enqueued tasks to complete.
Rule of thumb: when shared state changes in a way that could make a waiting predicate true, notify the corresponding condition variable.
References:
- https://en.cppreference.com/cpp/thread/condition_variable
- https://medium.com/@litombeg/concurrency-in-c-conditional-variables-9c688298c4b9
- Deep Learning for Electrophysiological Investigation and Estimation of Anesthetic-Induced Unconsciousness - Scientific Figure on ResearchGate. Available from: https://www.researchgate.net/figure/Gradient-descent-visualization-Amini-et-al-2018-At-each-training-step-epoch-the_fig9_364197638 [accessed 17 Aug 2026]