Four small building blocks that every low-level design with threads reaches for: a queue that says no, workers you reuse, a lock that lets readers in together, and an update that only succeeds if nobody got there first.
An order service gets bursts: orders arrive on request threads faster than a downstream system can price them, so something has to hold them. And a price list is read by every request but changed once an hour. The interviewer's version of both: what happens when there is more work than workers, and what happens when many threads touch one thing?
Four answers recur in almost every low-level design with threads. A bounded queue (producer–consumer) separates making work from doing it, and says no when full. A thread pool reuses a fixed number of workers. A read–write lock lets readers in together and a writer in alone. And compare-and-set, or a single writer, avoids the lock. You will meet them in the logger, the task scheduler, ride sharing, the order book and the key-value store.
Concurrent output is usually random, so every program here is built so that what it prints cannot vary: threads are joined, results are sorted or indexed, and where order matters the program waits on a counter, never on a timer. Each program ran 40 times in each language on a busy machine and printed the same text every time, but the guarantee comes from that construction, not from the runs.
A kitchen's serving counter is a bounded queue: cooks wait when it is full, waiters wait when it is empty, and neither needs to know the other.
The first version: a list and a loop#
The first version is a plain list that a producer appends to and a consumer polls. It is what you write in the first five minutes:
python
# orders_v1.py: a shared list, a producer, and a consumer that polls it
import threading
import time
orders = [] # no limit, no signal
produced_all = threading.Event()
processed = []
peak = 0
def producer(n):
global peak
for i in range(1, n + 1):
orders.append(i)
peak = max(peak, len(orders))
produced_all.set()
def consumer():
while True:
if orders:
processed.append(orders.pop(0))
elif produced_all.is_set():
return
else:
time.sleep(0.001) # nothing to do: poll again
# The consumer is slow to start, so the producer runs far ahead of it.
p = threading.Thread(target=producer, args=(1000,))
p.start()
p.join()
c = threading.Thread(target=consumer)
c.start()
c.join()
print("produced 1000, processed", len(processed))
print("peak queue depth", peak)
print("in order:", processed == sorted(processed))java
// OrdersV1.java: a shared list, a producer, and a consumer that polls it
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
class OrdersV1 {
static final LinkedList<Integer> orders = new LinkedList<>(); // no limit, no signal
static final AtomicBoolean producedAll = new AtomicBoolean();
static final List<Integer> processed = new ArrayList<>();
static int peak = 0;
static void producer(int n) {
for (int i = 1; i <= n; i++) {
synchronized (orders) {
orders.add(i);
peak = Math.max(peak, orders.size());
}
}
producedAll.set(true);
}
static void consumer() throws InterruptedException {
while (true) {
Integer next;
synchronized (orders) {
next = orders.pollFirst();
}
if (next != null) processed.add(next);
else if (producedAll.get()) return;
else Thread.sleep(1); // nothing to do: poll again
}
}
public static void main(String[] args) throws InterruptedException {
// The consumer is slow to start, so the producer runs far ahead of it.
Thread p = new Thread(() -> producer(1000));
p.start();
p.join();
Thread c = new Thread(() -> {
try {
consumer();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
c.start();
c.join();
boolean inOrder = true;
for (int i = 1; i < processed.size(); i++) inOrder &= processed.get(i - 1) < processed.get(i);
System.out.println("produced 1000, processed " + processed.size());
System.out.println("peak queue depth " + peak);
System.out.println("in order: " + (inOrder ? "True" : "False"));
}
}c++
// orders_v1.cpp: a shared list, a producer, and a consumer that polls it
#include <algorithm>
#include <atomic>
#include <chrono>
#include <deque>
#include <iostream>
#include <mutex>
#include <thread>
#include <vector>
std::deque<int> orders; // no limit, no signal
std::mutex m;
std::atomic<bool> producedAll{false};
std::vector<int> processed;
size_t peak = 0;
void producer(int n) {
for (int i = 1; i <= n; i++) {
std::lock_guard<std::mutex> lock(m);
orders.push_back(i);
peak = std::max(peak, orders.size());
}
producedAll = true;
}
void consumer() {
while (true) {
bool got = false;
int next = 0;
{
std::lock_guard<std::mutex> lock(m);
if (!orders.empty()) {
next = orders.front();
orders.pop_front();
got = true;
}
}
if (got) processed.push_back(next);
else if (producedAll) return;
else std::this_thread::sleep_for(std::chrono::milliseconds(1)); // nothing to do: poll again
}
}
int main() {
// The consumer is slow to start, so the producer runs far ahead of it.
std::thread p(producer, 1000);
p.join();
std::thread c(consumer);
c.join();
std::cout << "produced 1000, processed " << processed.size() << "\n";
std::cout << "peak queue depth " << peak << "\n";
std::cout << "in order: " << (std::is_sorted(processed.begin(), processed.end()) ? "True" : "False") << "\n";
}javascript
// orders_v1.js: a shared list, a producer, and a consumer that polls it
// JavaScript runs one thread, so the two take turns; the unbounded list is the same problem.
const orders = []; // no limit, no signal
const processed = [];
let producedAll = false;
let peak = 0;
async function producer(n) {
for (let i = 1; i <= n; i++) {
orders.push(i);
peak = Math.max(peak, orders.length);
}
producedAll = true;
}
async function consumer() {
while (true) {
if (orders.length > 0) processed.push(orders.shift());
else if (producedAll) return;
else await new Promise((resolve) => setTimeout(resolve, 1)); // nothing to do: poll again
}
}
async function main() {
// The consumer is slow to start, so the producer runs far ahead of it.
await producer(1000);
await consumer();
console.log("produced 1000, processed", processed.length);
console.log("peak queue depth", peak);
console.log("in order:", processed.every((v, i) => i === 0 || processed[i - 1] < v) ? "True" : "False");
}
main();text
produced 1000, processed 1000
peak queue depth 1000
in order: TrueIt works. But nothing in it ever says no: the producer made 1,000 orders before the consumer started, so all 1,000 sat in memory at once. This consumer was late on purpose; a real one is late whenever it is slower than its producers:
- The queue has no limit, and nobody is told "slow down". The peak depth above was 1,000, and it would have been 1,000,000 just as happily. The producer finds out the consumer is behind when the process runs out of memory.
- Polling burns time. The consumer wakes every millisecond to look at an empty list, so it either wastes CPU (poll fast) or adds delay (poll slowly).
- Stopping is a guess. A
doneflag and a poll mean the consumer might exit with an order still on its way in.
Your turn · checkpoint 1not answered
Before the refactor: what should a full queue do?
The queue has room for 2 orders and a third arrives. What are the choices, and what does each cost the producer?
Think it through, or jot it down. Then compare.
One strong answer
Wait, drop, or refuse. Wait until a consumer takes one: the producer slows down, which is the point. Drop (and count it): the caller never slows, but work is lost, which suits log lines, not payments. Refuse with an error such as "busy": the decision moves up a level. What you cannot do is grow without limit and call it a queue.
The shapes#
| Block | In the order service | Its one job |
|---|---|---|
| Bounded queue | BoundedQueue(capacity) with put, offer, get, close |
Hold at most capacity items; make producers wait when full and consumers wait when empty |
| Producer / consumer | the code that calls put, and the worker loop that calls get |
Make work, and do work, without knowing about each other |
| Thread pool | ThreadPool(workers, capacity) with submit and shutdown |
Run many tasks on a fixed number of reused threads |
| Read–write lock | ReadWriteLock with acquire_read, acquire_write |
Let any number of readers in together, or one writer alone |
| Compare-and-set | Price.compare_and_set(expected, new) |
Change a value only if it is still what the caller read |
DiagramThe concurrency blocks and how they connect
%% title: The concurrency blocks and how they connect
classDiagram
class BoundedQueue {
+capacity int
+put(item)
+offer(item) bool
+get() item
+close()
}
class QueueClosed {
<<exception>>
}
class ThreadPool {
+threads list
+results map
+submit(index, fn)
+shutdown()
}
class PoolClosed {
<<exception>>
}
class ReadWriteLock {
+acquire_read()
+release_read()
+acquire_write()
+release_write()
}
class Price {
+get() int
+compare_and_set(expected, new) bool
}
ThreadPool *-- BoundedQueue : owns
BoundedQueue ..> QueueClosed : raises
ThreadPool ..> QueueClosed : catches
ThreadPool ..> PoolClosed : raisesRead the arrows. The pool owns a queue and turns a closed queue into a closed pool, so a caller never sees QueueClosed. ReadWriteLock and Price stand alone: one pessimistic (wait your turn), one optimistic (try, and retry if you lost). Producers and consumers are not classes; they are whoever calls put and get.
Producer–consumer: a bounded queue#
A bounded queue has one lock and one condition. A producer that finds it full waits on the condition, and so does a consumer that finds it empty. Every change wakes everyone, and each waiter re-checks. Closing the queue refuses producers and lets consumers drain what is left and stop, so nothing needs a done flag.
python
# order_queue.py: a bounded queue: producers wait when it is full, consumers wait when it is empty
import threading
class QueueClosed(Exception):
pass
class BoundedQueue:
def __init__(self, capacity):
self.capacity, self.items, self.closed = capacity, [], False
self.max_depth, self.taken = 0, [] # for the demo: the deepest it got, and the order items left
self._cond = threading.Condition() # one lock, one condition, shared by both sides
def put(self, item): # blocks while full: this is the back-pressure
with self._cond:
while len(self.items) >= self.capacity and not self.closed:
self._cond.wait()
if self.closed:
raise QueueClosed("queue is closed")
self._add(item)
def offer(self, item): # never blocks: False when full or closed
with self._cond:
if self.closed or len(self.items) >= self.capacity:
return False
self._add(item)
return True
def get(self): # blocks while empty; raises once the queue is closed and drained
with self._cond:
while not self.items and not self.closed:
self._cond.wait()
if not self.items:
raise QueueClosed("queue is closed")
item = self.items.pop(0)
self.taken.append(item)
self._cond.notify_all()
return item
def close(self):
with self._cond:
self.closed = True
self._cond.notify_all()
def _add(self, item): # the caller holds the lock
self.items.append(item)
self.max_depth = max(self.max_depth, len(self.items))
self._cond.notify_all()
q = BoundedQueue(2)
done, done_lock = [], threading.Lock()
def producer(first): # five orders: first .. first + 4
for i in range(first, first + 5):
q.put(i)
def consumer():
while True:
try:
order = q.get()
except QueueClosed:
return
with done_lock:
done.append(order)
producers = [threading.Thread(target=producer, args=(first,)) for first in (1, 6)]
consumers = [threading.Thread(target=consumer) for _ in range(2)]
for t in producers + consumers:
t.start()
for t in producers:
t.join()
q.close() # no more orders: consumers finish what is left, then stop
for t in consumers:
t.join()
print("processed:", " ".join(str(o) for o in sorted(done)))
print("depth never above capacity:", q.max_depth <= q.capacity)
print("each producer's orders left in order:", [o for o in q.taken if o <= 5] == [1, 2, 3, 4, 5] and [o for o in q.taken if o > 5] == [6, 7, 8, 9, 10])
small = BoundedQueue(3)
accepted = [i for i in range(1, 6) if small.offer(i)]
print("offer accepted", ", ".join(str(i) for i in accepted) + ", rejected", ", ".join(str(i) for i in range(1, 6) if i not in accepted))
small.close()
try:
small.put(9)
except QueueClosed as e:
print("put after close: refused (" + str(e) + ")")
print("drained after close:", " ".join(str(small.get()) for _ in range(3)))
try:
small.get()
except QueueClosed:
print("then: closed")java
// OrderQueue.java: a bounded queue: producers wait when it is full, consumers wait when it is empty
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.stream.Collectors;
class OrderQueue {
static final class QueueClosed extends RuntimeException {
QueueClosed(String message) { super(message); }
}
static final class BoundedQueue {
final int capacity;
final LinkedList<Integer> items = new LinkedList<>();
final List<Integer> taken = new ArrayList<>(); // for the demo: the order items left
int maxDepth; // for the demo: the deepest it got
boolean closed;
BoundedQueue(int capacity) { this.capacity = capacity; }
// One lock (the monitor) and one wait set, shared by both sides.
synchronized void put(int item) throws InterruptedException { // blocks while full: the back-pressure
while (items.size() >= capacity && !closed) wait();
if (closed) throw new QueueClosed("queue is closed");
add(item);
}
synchronized boolean offer(int item) { // never blocks: false when full or closed
if (closed || items.size() >= capacity) return false;
add(item);
return true;
}
synchronized int get() throws InterruptedException { // blocks while empty; throws once closed and drained
while (items.isEmpty() && !closed) wait();
if (items.isEmpty()) throw new QueueClosed("queue is closed");
int item = items.removeFirst();
taken.add(item);
notifyAll();
return item;
}
synchronized void close() {
closed = true;
notifyAll();
}
private void add(int item) { // the caller holds the lock
items.addLast(item);
maxDepth = Math.max(maxDepth, items.size());
notifyAll();
}
}
public static void main(String[] args) throws InterruptedException {
BoundedQueue q = new BoundedQueue(2);
List<Integer> done = new ArrayList<>();
List<Thread> producers = new ArrayList<>();
for (int first : new int[] {1, 6}) { // five orders each: first .. first + 4
producers.add(new Thread(() -> {
try {
for (int i = first; i < first + 5; i++) q.put(i);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}));
}
List<Thread> consumers = new ArrayList<>();
for (int k = 0; k < 2; k++) {
consumers.add(new Thread(() -> {
try {
while (true) {
int order = q.get();
synchronized (done) { done.add(order); }
}
} catch (QueueClosed e) {
// closed and drained: stop
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}));
}
producers.forEach(Thread::start);
consumers.forEach(Thread::start);
for (Thread t : producers) t.join();
q.close(); // no more orders: consumers finish what is left, then stop
for (Thread t : consumers) t.join();
List<Integer> sorted = done.stream().sorted().collect(Collectors.toList());
System.out.println("processed: " + sorted.stream().map(String::valueOf).collect(Collectors.joining(" ")));
System.out.println("depth never above capacity: " + (q.maxDepth <= q.capacity ? "True" : "False"));
List<Integer> low = q.taken.stream().filter(o -> o <= 5).collect(Collectors.toList());
List<Integer> high = q.taken.stream().filter(o -> o > 5).collect(Collectors.toList());
boolean fifo = low.equals(List.of(1, 2, 3, 4, 5)) && high.equals(List.of(6, 7, 8, 9, 10));
System.out.println("each producer's orders left in order: " + (fifo ? "True" : "False"));
BoundedQueue small = new BoundedQueue(3);
List<Integer> accepted = new ArrayList<>();
List<Integer> rejected = new ArrayList<>();
for (int i = 1; i <= 5; i++) (small.offer(i) ? accepted : rejected).add(i);
System.out.println("offer accepted " + accepted.stream().map(String::valueOf).collect(Collectors.joining(", "))
+ ", rejected " + rejected.stream().map(String::valueOf).collect(Collectors.joining(", ")));
small.close();
try {
small.put(9);
} catch (QueueClosed e) {
System.out.println("put after close: refused (" + e.getMessage() + ")");
}
System.out.println("drained after close: " + small.get() + " " + small.get() + " " + small.get());
try {
small.get();
} catch (QueueClosed e) {
System.out.println("then: closed");
}
}
}c++
// order_queue.cpp: a bounded queue: producers wait when it is full, consumers wait when it is empty
#include <algorithm>
#include <condition_variable>
#include <deque>
#include <iostream>
#include <mutex>
#include <stdexcept>
#include <string>
#include <thread>
#include <vector>
struct QueueClosed : std::runtime_error {
explicit QueueClosed(const std::string& m) : std::runtime_error(m) {}
};
class BoundedQueue {
std::mutex m_;
std::condition_variable cond_; // one lock, one condition, shared by both sides
std::deque<int> items_;
bool closed_ = false;
void add(int item) { // the caller holds the lock
items_.push_back(item);
maxDepth = std::max(maxDepth, items_.size());
cond_.notify_all();
}
public:
const size_t capacity;
size_t maxDepth = 0; // for the demo: the deepest it got
std::vector<int> taken; // for the demo: the order items left
explicit BoundedQueue(size_t cap) : capacity(cap) {}
void put(int item) { // blocks while full: the back-pressure
std::unique_lock<std::mutex> lock(m_);
cond_.wait(lock, [&] { return items_.size() < capacity || closed_; });
if (closed_) throw QueueClosed("queue is closed");
add(item);
}
bool offer(int item) { // never blocks: false when full or closed
std::lock_guard<std::mutex> lock(m_);
if (closed_ || items_.size() >= capacity) return false;
add(item);
return true;
}
int get() { // blocks while empty; throws once the queue is closed and drained
std::unique_lock<std::mutex> lock(m_);
cond_.wait(lock, [&] { return !items_.empty() || closed_; });
if (items_.empty()) throw QueueClosed("queue is closed");
int item = items_.front();
items_.pop_front();
taken.push_back(item);
cond_.notify_all();
return item;
}
void close() {
std::lock_guard<std::mutex> lock(m_);
closed_ = true;
cond_.notify_all();
}
};
static std::string join(const std::vector<int>& v, const char* sep) {
std::string out;
for (size_t i = 0; i < v.size(); i++) out += (i ? sep : "") + std::to_string(v[i]);
return out;
}
int main() {
BoundedQueue q(2);
std::vector<int> done;
std::mutex doneMutex;
std::vector<std::thread> producers, consumers;
for (int first : {1, 6}) { // five orders each: first .. first + 4
producers.emplace_back([&q, first] {
for (int i = first; i < first + 5; i++) q.put(i);
});
}
for (int k = 0; k < 2; k++) {
consumers.emplace_back([&] {
try {
while (true) {
int order = q.get();
std::lock_guard<std::mutex> lock(doneMutex);
done.push_back(order);
}
} catch (const QueueClosed&) {
// closed and drained: stop
}
});
}
for (auto& t : producers) t.join();
q.close(); // no more orders: consumers finish what is left, then stop
for (auto& t : consumers) t.join();
std::sort(done.begin(), done.end());
std::cout << "processed: " << join(done, " ") << "\n";
std::cout << "depth never above capacity: " << (q.maxDepth <= q.capacity ? "True" : "False") << "\n";
std::vector<int> low, high;
for (int o : q.taken) (o <= 5 ? low : high).push_back(o);
bool fifo = low == std::vector<int>{1, 2, 3, 4, 5} && high == std::vector<int>{6, 7, 8, 9, 10};
std::cout << "each producer's orders left in order: " << (fifo ? "True" : "False") << "\n";
BoundedQueue small(3);
std::vector<int> accepted, rejected;
for (int i = 1; i <= 5; i++) (small.offer(i) ? accepted : rejected).push_back(i);
std::cout << "offer accepted " << join(accepted, ", ") << ", rejected " << join(rejected, ", ") << "\n";
small.close();
try {
small.put(9);
} catch (const QueueClosed& e) {
std::cout << "put after close: refused (" << e.what() << ")\n";
}
int a = small.get(), b = small.get(), c = small.get();
std::cout << "drained after close: " << a << " " << b << " " << c << "\n";
try {
small.get();
} catch (const QueueClosed&) {
std::cout << "then: closed\n";
}
}javascript
// order_queue.js: a bounded queue: producers wait when it is full, consumers wait when it is empty
// JavaScript runs one thread, so "waiting" means awaiting a promise; the queue's job is the same.
class QueueClosed extends Error {}
class BoundedQueue {
constructor(capacity) {
Object.assign(this, { capacity, items: [], closed: false, maxDepth: 0, taken: [], waiters: [] });
// for the demo: maxDepth is the deepest it got, taken the order items left
}
async put(item) { // waits while full: the back-pressure
while (this.items.length >= this.capacity && !this.closed) await this.#sleep();
if (this.closed) throw new QueueClosed("queue is closed");
this.#add(item);
}
offer(item) { // never waits: false when full or closed
if (this.closed || this.items.length >= this.capacity) return false;
this.#add(item);
return true;
}
async get() { // waits while empty; throws once the queue is closed and drained
while (this.items.length === 0 && !this.closed) await this.#sleep();
if (this.items.length === 0) throw new QueueClosed("queue is closed");
const item = this.items.shift();
this.taken.push(item);
this.#wake();
return item;
}
close() {
this.closed = true;
this.#wake();
}
#add(item) {
this.items.push(item);
this.maxDepth = Math.max(this.maxDepth, this.items.length);
this.#wake();
}
#sleep() { // park this caller until something changes
return new Promise((resolve) => this.waiters.push(resolve));
}
#wake() { // everyone re-checks their condition, as with notify-all
const waiting = this.waiters.splice(0);
waiting.forEach((resolve) => resolve());
}
}
async function main() {
const q = new BoundedQueue(2);
const done = [];
const producer = async (first) => { // five orders: first .. first + 4
for (let i = first; i < first + 5; i++) await q.put(i);
};
const consumer = async () => {
try {
while (true) done.push(await q.get());
} catch (e) {
if (!(e instanceof QueueClosed)) throw e; // closed and drained: stop
}
};
const consumers = [consumer(), consumer()];
await Promise.all([producer(1), producer(6)]);
q.close(); // no more orders: consumers finish what is left, then stop
await Promise.all(consumers);
console.log("processed:", [...done].sort((a, b) => a - b).join(" "));
console.log("depth never above capacity:", q.maxDepth <= q.capacity ? "True" : "False");
const low = q.taken.filter((o) => o <= 5);
const high = q.taken.filter((o) => o > 5);
console.log("each producer's orders left in order:", low.join() === "1,2,3,4,5" && high.join() === "6,7,8,9,10" ? "True" : "False");
const small = new BoundedQueue(3);
const accepted = [];
const rejected = [];
for (let i = 1; i <= 5; i++) (small.offer(i) ? accepted : rejected).push(i);
console.log("offer accepted " + accepted.join(", ") + ", rejected " + rejected.join(", "));
small.close();
try {
await small.put(9);
} catch (e) {
console.log("put after close: refused (" + e.message + ")");
}
console.log("drained after close:", [await small.get(), await small.get(), await small.get()].join(" "));
try {
await small.get();
} catch (e) {
console.log("then: closed");
}
}
main();All four print the same thing:
text
processed: 1 2 3 4 5 6 7 8 9 10
depth never above capacity: True
each producer's orders left in order: True
offer accepted 1, 2, 3, rejected 4, 5
put after close: refused (queue is closed)
drained after close: 1 2 3
then: closedTwo producers put orders 1 to 5 and 6 to 10 into a queue of two, and two consumers take all ten out exactly once. Which consumer got which order changes from run to run, so the list is sorted for printing. What cannot change is that the queue never held more than two, and that each producer's orders left in the order it put them in. Then offer on a queue of three: five in, three accepted, two rejected, no waiting. After close, a put is refused, the three orders left are still handed out, and only then does get report that the queue is closed.
What each language brings:
- Python:
threading.Conditionholds the lock and the wait set;wait()releases the lock while it sleeps. In production,queue.Queue(maxsize)is this queue (Python 3.13 addsshutdown()for the close). - Java:
synchronizedwithwait()andnotifyAll(): the object is the lock and the condition. In production,ArrayBlockingQueue. - C++:
std::mutexandstd::condition_variable;waittakes a predicate, which writes thewhileloop for you. - JavaScript: one thread, so no lock. Waiting is awaiting a promise resolved when something changes, with the same check-again loop.
A put on a full queue waits, asleep and not polling, until a consumer takes an item and the change wakes the producer to re-check and finish.
Back-pressure: wait, drop or refuse#
The size limit is a design decision, not a memory setting. When the queue is full:
Wait: put
- The producer is slowed to the consumer's speed
- No work is lost
- Fits payments, orders, anything that must be done
- Risk: the producer's own caller now waits too, so put a timeout on it
Drop or refuse: offer
- The producer never waits
- Work is lost or bounced, so count it
- Fits log lines, metrics, anything that can be redone
- The logger does this: its
AsyncSinkhas a queue of 10,000, one worker thread, and a full queue means the record is dropped and counted, because a request thread must never wait for a log line
Refusing is the same idea one level up: the service returns "busy". That is the back-pressure of Week 3, Day 3: tell callers to slow down instead of absorbing the overload until you fall over. Whichever you choose, say the limit and what happens at it.
Thread pool: a fixed number of workers#
A queue plus a fixed set of threads that loop on get is a thread pool. Callers submit a task and move on. A thread per task would create 8,000 threads for 8,000 orders; a pool creates its threads once and reuses them.
python
# order_pool.py: a fixed number of workers take tasks from a bounded queue
import threading
class QueueClosed(Exception):
pass
class PoolClosed(Exception):
pass
class BoundedQueue: # the queue from the previous program, without the demo counters
def __init__(self, capacity):
self.capacity, self.items, self.closed = capacity, [], False
self._cond = threading.Condition()
def put(self, item):
with self._cond:
while len(self.items) >= self.capacity and not self.closed:
self._cond.wait()
if self.closed:
raise QueueClosed("queue is closed")
self.items.append(item)
self._cond.notify_all()
def get(self):
with self._cond:
while not self.items and not self.closed:
self._cond.wait()
if not self.items:
raise QueueClosed("queue is closed")
item = self.items.pop(0)
self._cond.notify_all()
return item
def close(self):
with self._cond:
self.closed = True
self._cond.notify_all()
class ThreadPool:
def __init__(self, workers, capacity):
self.queue, self.results, self._lock = BoundedQueue(capacity), {}, threading.Lock()
self.threads = [threading.Thread(target=self._work) for _ in range(workers)]
for t in self.threads:
t.start()
def submit(self, index, fn): # blocks when the queue is full: the caller slows down, memory does not grow
try:
self.queue.put((index, fn))
except QueueClosed:
raise PoolClosed("pool is shut down")
def _work(self):
while True:
try:
index, fn = self.queue.get()
except QueueClosed:
return
try:
outcome = str(fn())
except Exception as e: # a failing task must not kill its worker
outcome = f"failed ({e})"
with self._lock:
self.results[index] = outcome
def shutdown(self): # finish everything already accepted, then stop the workers
self.queue.close()
for t in self.threads:
t.join()
def price_order(i):
if i == 5:
raise ValueError("card declined")
return 50 + 25 * i
pool = ThreadPool(workers=3, capacity=4)
for i in range(1, 9):
pool.submit(i, lambda i=i: price_order(i))
pool.shutdown()
for i in range(1, 9):
print(f"order {i}: {pool.results[i]}")
print(f"finished {len(pool.results)} of 8 with {len(pool.threads)} threads")
try:
pool.submit(9, lambda: price_order(9))
except PoolClosed as e:
print("submit after shutdown: refused (" + str(e) + ")")java
// OrderPool.java: a fixed number of workers take tasks from a bounded queue
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.Callable;
class OrderPool {
static final class QueueClosed extends RuntimeException {
QueueClosed(String message) { super(message); }
}
static final class PoolClosed extends RuntimeException {
PoolClosed(String message) { super(message); }
}
record Task(int index, Callable<Integer> fn) {}
static final class BoundedQueue { // the queue from the previous program, without the demo counters
private final int capacity;
private final LinkedList<Task> items = new LinkedList<>();
private boolean closed;
BoundedQueue(int capacity) { this.capacity = capacity; }
synchronized void put(Task task) throws InterruptedException {
while (items.size() >= capacity && !closed) wait();
if (closed) throw new QueueClosed("queue is closed");
items.addLast(task);
notifyAll();
}
synchronized Task get() throws InterruptedException {
while (items.isEmpty() && !closed) wait();
if (items.isEmpty()) throw new QueueClosed("queue is closed");
Task task = items.removeFirst();
notifyAll();
return task;
}
synchronized void close() {
closed = true;
notifyAll();
}
}
static final class ThreadPool {
private final BoundedQueue queue;
final Map<Integer, String> results = new TreeMap<>();
final List<Thread> threads = new ArrayList<>();
ThreadPool(int workers, int capacity) {
queue = new BoundedQueue(capacity);
for (int i = 0; i < workers; i++) {
Thread t = new Thread(this::work);
threads.add(t);
t.start();
}
}
// Blocks when the queue is full: the caller slows down, memory does not grow.
void submit(int index, Callable<Integer> fn) throws InterruptedException {
try {
queue.put(new Task(index, fn));
} catch (QueueClosed e) {
throw new PoolClosed("pool is shut down");
}
}
private void work() {
try {
while (true) {
Task task = queue.get();
String outcome;
try {
outcome = String.valueOf(task.fn().call());
} catch (Exception e) { // a failing task must not kill its worker
outcome = "failed (" + e.getMessage() + ")";
}
synchronized (results) { results.put(task.index(), outcome); }
}
} catch (QueueClosed e) {
// closed and drained: stop
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
void shutdown() throws InterruptedException { // finish everything already accepted, then stop the workers
queue.close();
for (Thread t : threads) t.join();
}
}
static int priceOrder(int i) {
if (i == 5) throw new IllegalArgumentException("card declined");
return 50 + 25 * i;
}
public static void main(String[] args) throws InterruptedException {
ThreadPool pool = new ThreadPool(3, 4);
for (int i = 1; i <= 8; i++) {
int n = i;
pool.submit(n, () -> priceOrder(n));
}
pool.shutdown();
for (int i = 1; i <= 8; i++) System.out.println("order " + i + ": " + pool.results.get(i));
System.out.println("finished " + pool.results.size() + " of 8 with " + pool.threads.size() + " threads");
try {
pool.submit(9, () -> priceOrder(9));
} catch (PoolClosed e) {
System.out.println("submit after shutdown: refused (" + e.getMessage() + ")");
}
}
}c++
// order_pool.cpp: a fixed number of workers take tasks from a bounded queue
#include <condition_variable>
#include <deque>
#include <functional>
#include <iostream>
#include <map>
#include <mutex>
#include <stdexcept>
#include <string>
#include <thread>
#include <utility>
#include <vector>
struct QueueClosed : std::runtime_error {
explicit QueueClosed(const std::string& m) : std::runtime_error(m) {}
};
struct PoolClosed : std::runtime_error {
explicit PoolClosed(const std::string& m) : std::runtime_error(m) {}
};
using Task = std::pair<int, std::function<int()>>;
class BoundedQueue { // the queue from the previous program, without the demo counters
std::mutex m_;
std::condition_variable cond_;
std::deque<Task> items_;
size_t capacity_;
bool closed_ = false;
public:
explicit BoundedQueue(size_t capacity) : capacity_(capacity) {}
void put(Task task) {
std::unique_lock<std::mutex> lock(m_);
cond_.wait(lock, [&] { return items_.size() < capacity_ || closed_; });
if (closed_) throw QueueClosed("queue is closed");
items_.push_back(std::move(task));
cond_.notify_all();
}
Task get() {
std::unique_lock<std::mutex> lock(m_);
cond_.wait(lock, [&] { return !items_.empty() || closed_; });
if (items_.empty()) throw QueueClosed("queue is closed");
Task task = std::move(items_.front());
items_.pop_front();
cond_.notify_all();
return task;
}
void close() {
std::lock_guard<std::mutex> lock(m_);
closed_ = true;
cond_.notify_all();
}
};
class ThreadPool {
BoundedQueue queue_;
std::mutex lock_;
void work() {
try {
while (true) {
Task task = queue_.get();
std::string outcome;
try {
outcome = std::to_string(task.second());
} catch (const std::exception& e) { // a failing task must not kill its worker
outcome = std::string("failed (") + e.what() + ")";
}
std::lock_guard<std::mutex> guard(lock_);
results[task.first] = outcome;
}
} catch (const QueueClosed&) {
// closed and drained: stop
}
}
public:
std::map<int, std::string> results;
std::vector<std::thread> threads;
ThreadPool(int workers, size_t capacity) : queue_(capacity) {
for (int i = 0; i < workers; i++) threads.emplace_back([this] { work(); });
}
// Blocks when the queue is full: the caller slows down, memory does not grow.
void submit(int index, std::function<int()> fn) {
try {
queue_.put({index, std::move(fn)});
} catch (const QueueClosed&) {
throw PoolClosed("pool is shut down");
}
}
void shutdown() { // finish everything already accepted, then stop the workers
queue_.close();
for (auto& t : threads) t.join();
}
};
int priceOrder(int i) {
if (i == 5) throw std::invalid_argument("card declined");
return 50 + 25 * i;
}
int main() {
ThreadPool pool(3, 4);
for (int i = 1; i <= 8; i++) pool.submit(i, [i] { return priceOrder(i); });
pool.shutdown();
for (int i = 1; i <= 8; i++) std::cout << "order " << i << ": " << pool.results[i] << "\n";
std::cout << "finished " << pool.results.size() << " of 8 with " << pool.threads.size() << " threads\n";
try {
pool.submit(9, [] { return priceOrder(9); });
} catch (const PoolClosed& e) {
std::cout << "submit after shutdown: refused (" << e.what() << ")\n";
}
}javascript
// order_pool.js: a fixed number of workers take tasks from a bounded queue
// JavaScript runs one thread: these workers are async loops that take turns at each await.
// For CPU-heavy tasks, the equivalent is a pool of worker_threads.
class QueueClosed extends Error {}
class PoolClosed extends Error {}
class BoundedQueue { // the queue from the previous program, without the demo counters
constructor(capacity) {
Object.assign(this, { capacity, items: [], closed: false, waiters: [] });
}
async put(item) {
while (this.items.length >= this.capacity && !this.closed) await this.#sleep();
if (this.closed) throw new QueueClosed("queue is closed");
this.items.push(item);
this.#wake();
}
async get() {
while (this.items.length === 0 && !this.closed) await this.#sleep();
if (this.items.length === 0) throw new QueueClosed("queue is closed");
const item = this.items.shift();
this.#wake();
return item;
}
close() {
this.closed = true;
this.#wake();
}
#sleep() {
return new Promise((resolve) => this.waiters.push(resolve));
}
#wake() {
this.waiters.splice(0).forEach((resolve) => resolve());
}
}
class ThreadPool {
constructor(workers, capacity) {
this.queue = new BoundedQueue(capacity);
this.results = new Map();
this.threads = Array.from({ length: workers }, () => this.#work());
}
// Waits when the queue is full: the caller slows down, memory does not grow.
async submit(index, fn) {
try {
await this.queue.put({ index, fn });
} catch (e) {
if (e instanceof QueueClosed) throw new PoolClosed("pool is shut down");
throw e;
}
}
async #work() {
while (true) {
let task;
try {
task = await this.queue.get();
} catch (e) {
if (e instanceof QueueClosed) return; // closed and drained: stop
throw e;
}
try {
this.results.set(task.index, String(await task.fn()));
} catch (e) { // a failing task must not kill its worker
this.results.set(task.index, `failed (${e.message})`);
}
}
}
async shutdown() { // finish everything already accepted, then stop the workers
this.queue.close();
await Promise.all(this.threads);
}
}
async function priceOrder(i) {
await Promise.resolve(); // the task yields once, as real work would
if (i === 5) throw new Error("card declined");
return 50 + 25 * i;
}
async function main() {
const pool = new ThreadPool(3, 4);
for (let i = 1; i <= 8; i++) await pool.submit(i, () => priceOrder(i));
await pool.shutdown();
for (let i = 1; i <= 8; i++) console.log(`order ${i}: ${pool.results.get(i)}`);
console.log(`finished ${pool.results.size} of 8 with ${pool.threads.length} threads`);
try {
await pool.submit(9, () => priceOrder(9));
} catch (e) {
console.log("submit after shutdown: refused (" + e.message + ")");
}
}
main();All four print the same thing:
text
order 1: 75
order 2: 100
order 3: 125
order 4: 150
order 5: failed (card declined)
order 6: 200
order 7: 225
order 8: 250
finished 8 of 8 with 3 threads
submit after shutdown: refused (pool is shut down)The tasks price eight orders at 50 + 25 × the order number: order 1 is 75, order 8 is 250. Order 5 fails with "card declined", and the other seven carry on: the worker caught the failure, recorded it and went back for the next task. Three threads were started, once, for eight tasks. Results are stored by order number, so the print order does not depend on which worker finished first. Once four tasks are waiting, submit blocks, so the caller slows down instead of piling up work. After shutdown, every accepted task has finished and new ones are refused.
- Python: the workers are
threading.Threads looping onget. The library'sconcurrent.futures.ThreadPoolExecutorhas no limit on its queue, so bound what you submit (a semaphore aroundsubmitwill do). - Java: the same loop on
Threads, withCallabletasks that may throw.Executors.newFixedThreadPoolalso queues without limit; for a bounded pool, build aThreadPoolExecutorwith anArrayBlockingQueue. - C++:
std::functionis the task. The standard library has no thread pool, so real projects use a library. - JavaScript: the "workers" are async loops that take turns at each
await, which is all I/O needs; CPU-heavy work needs a pool ofworker_threads.
Three decisions come with every pool, and an interviewer asks about each:
- How many workers? About the core count for CPU-bound tasks; more for tasks that mostly wait on a database or network, up to what that system can take. Start from a number, measure, and make it a setting. (In standard Python, the global interpreter lock runs one thread's Python code at a time, so CPU-bound work needs processes.)
- How big is the queue? It is the pool's memory limit and its back-pressure signal: a small one pushes back early, a big one hides an overload until the delay is long.
- What does shutdown mean? Drain (finish what was accepted, as here) or drop (stop now and discard what is waiting). Say which.
A task that throws is caught per task and recorded, and its worker takes the next task, so one bad order cannot shrink the pool.
Read–write lock: many readers, one writer#
A price list is read by every request and written once an hour. A plain lock makes readers wait for each other, though readers cannot hurt each other. A read–write lock lets any number of readers in together and makes a writer wait for them all to leave, then runs it alone.
python
# price_lock.py: many readers at once, one writer alone, and a waiting writer is not starved
import threading
import time
class ReadWriteLock:
"""Writer-preferring: once a writer is waiting, new readers wait behind it."""
def __init__(self):
self._cond = threading.Condition()
self.readers, self.writing = 0, False
self.waiting_writers, self.waiting_readers = 0, 0
def acquire_read(self):
with self._cond:
self.waiting_readers += 1
while self.writing or self.waiting_writers:
self._cond.wait()
self.waiting_readers -= 1
self.readers += 1
def release_read(self):
with self._cond:
self.readers -= 1
if self.readers == 0:
self._cond.notify_all()
def acquire_write(self):
with self._cond:
self.waiting_writers += 1
while self.writing or self.readers:
self._cond.wait()
self.waiting_writers -= 1
self.writing = True
def release_write(self):
with self._cond:
self.writing = False
self._cond.notify_all()
def counts(self): # for the demo: readers inside, writers waiting, readers waiting
with self._cond:
return self.readers, self.waiting_writers, self.waiting_readers
lock = ReadWriteLock()
prices = {"version": 1, "tea": 40}
log, log_lock, leave = [], threading.Lock(), threading.Event()
def note(text):
with log_lock:
log.append(text)
def reader(name):
lock.acquire_read()
try:
note(f"{name} in, sees version {prices['version']}")
leave.wait() # keeps the read lock until the demo says to leave
finally:
lock.release_read()
def writer():
lock.acquire_write()
try:
note(f"writer in, readers inside: {lock.counts()[0]}")
prices["version"] += 1
prices["tea"] = 45
finally:
lock.release_write()
def wait_until(condition):
while not condition():
time.sleep(0.001)
threads = [threading.Thread(target=reader, args=(f"R{n}",)) for n in (1, 2, 3)]
for t in threads:
t.start()
wait_until(lambda: lock.counts()[0] == 3) # only possible if three readers hold the lock at once
print("3 readers held the read lock together:", lock.counts()[0] == 3)
w = threading.Thread(target=writer)
w.start()
wait_until(lambda: lock.counts()[1] == 1) # the writer is waiting for the readers
late = threading.Thread(target=reader, args=("R4",))
late.start()
wait_until(lambda: lock.counts()[2] == 1) # a reader arrived behind the waiting writer
leave.set()
for t in threads + [w, late]:
t.join()
print("\n".join(sorted(log[:3])))
print("after the readers left:", "; ".join(log[3:]))java
// PriceLock.java: many readers at once, one writer alone, and a waiting writer is not starved
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.function.BooleanSupplier;
class PriceLock {
/** Writer-preferring: once a writer is waiting, new readers wait behind it. */
static final class ReadWriteLock {
private int readers, waitingWriters, waitingReaders;
private boolean writing;
synchronized void acquireRead() throws InterruptedException {
waitingReaders++;
while (writing || waitingWriters > 0) wait();
waitingReaders--;
readers++;
}
synchronized void releaseRead() {
readers--;
if (readers == 0) notifyAll();
}
synchronized void acquireWrite() throws InterruptedException {
waitingWriters++;
while (writing || readers > 0) wait();
waitingWriters--;
writing = true;
}
synchronized void releaseWrite() {
writing = false;
notifyAll();
}
synchronized int[] counts() { // for the demo: readers inside, writers waiting, readers waiting
return new int[] {readers, waitingWriters, waitingReaders};
}
}
static final ReadWriteLock lock = new ReadWriteLock();
static int version = 1;
static int tea = 40;
static final List<String> log = Collections.synchronizedList(new ArrayList<>());
static final CountDownLatch leave = new CountDownLatch(1);
static Thread reader(String name) {
return new Thread(() -> {
try {
lock.acquireRead();
try {
log.add(name + " in, sees version " + version);
leave.await(); // keeps the read lock until the demo says to leave
} finally {
lock.releaseRead();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
}
static Thread writer() {
return new Thread(() -> {
try {
lock.acquireWrite();
try {
log.add("writer in, readers inside: " + lock.counts()[0]);
version++;
tea = 45;
} finally {
lock.releaseWrite();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
}
static void waitUntil(BooleanSupplier condition) throws InterruptedException {
while (!condition.getAsBoolean()) Thread.sleep(1);
}
public static void main(String[] args) throws InterruptedException {
List<Thread> threads = new ArrayList<>();
for (int n = 1; n <= 3; n++) threads.add(reader("R" + n));
threads.forEach(Thread::start);
waitUntil(() -> lock.counts()[0] == 3); // only possible if three readers hold the lock at once
System.out.println("3 readers held the read lock together: " + (lock.counts()[0] == 3 ? "True" : "False"));
Thread w = writer();
w.start();
waitUntil(() -> lock.counts()[1] == 1); // the writer is waiting for the readers
Thread late = reader("R4");
late.start();
waitUntil(() -> lock.counts()[2] == 1); // a reader arrived behind the waiting writer
leave.countDown();
threads.add(w);
threads.add(late);
for (Thread t : threads) t.join();
List<String> first = new ArrayList<>(log.subList(0, 3));
Collections.sort(first);
System.out.println(String.join("\n", first));
System.out.println("after the readers left: " + String.join("; ", log.subList(3, log.size())));
}
}c++
// price_lock.cpp: many readers at once, one writer alone, and a waiting writer is not starved
#include <algorithm>
#include <array>
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <functional>
#include <iostream>
#include <mutex>
#include <string>
#include <thread>
#include <vector>
// Writer-preferring: once a writer is waiting, new readers wait behind it.
class ReadWriteLock {
std::mutex m_;
std::condition_variable cond_;
int readers_ = 0, waitingWriters_ = 0, waitingReaders_ = 0;
bool writing_ = false;
public:
void acquireRead() {
std::unique_lock<std::mutex> lock(m_);
waitingReaders_++;
cond_.wait(lock, [&] { return !writing_ && waitingWriters_ == 0; });
waitingReaders_--;
readers_++;
}
void releaseRead() {
std::lock_guard<std::mutex> lock(m_);
if (--readers_ == 0) cond_.notify_all();
}
void acquireWrite() {
std::unique_lock<std::mutex> lock(m_);
waitingWriters_++;
cond_.wait(lock, [&] { return !writing_ && readers_ == 0; });
waitingWriters_--;
writing_ = true;
}
void releaseWrite() {
std::lock_guard<std::mutex> lock(m_);
writing_ = false;
cond_.notify_all();
}
std::array<int, 3> counts() { // for the demo: readers inside, writers waiting, readers waiting
std::lock_guard<std::mutex> lock(m_);
return {readers_, waitingWriters_, waitingReaders_};
}
};
ReadWriteLock rw;
int version = 1, tea = 40;
std::vector<std::string> logLines;
std::mutex logMutex;
std::atomic<bool> leave{false};
void note(const std::string& text) {
std::lock_guard<std::mutex> lock(logMutex);
logLines.push_back(text);
}
void reader(std::string name) {
rw.acquireRead();
note(name + " in, sees version " + std::to_string(version));
while (!leave) std::this_thread::sleep_for(std::chrono::milliseconds(1)); // keeps the read lock until the demo says to leave
rw.releaseRead();
}
void writer() {
rw.acquireWrite();
note("writer in, readers inside: " + std::to_string(rw.counts()[0]));
version++;
tea = 45;
rw.releaseWrite();
}
void waitUntil(const std::function<bool()>& condition) {
while (!condition()) std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
int main() {
std::vector<std::thread> threads;
for (int n = 1; n <= 3; n++) threads.emplace_back(reader, "R" + std::to_string(n));
waitUntil([] { return rw.counts()[0] == 3; }); // only possible if three readers hold the lock at once
std::cout << "3 readers held the read lock together: " << (rw.counts()[0] == 3 ? "True" : "False") << "\n";
std::thread w(writer);
waitUntil([] { return rw.counts()[1] == 1; }); // the writer is waiting for the readers
std::thread late(reader, "R4");
waitUntil([] { return rw.counts()[2] == 1; }); // a reader arrived behind the waiting writer
leave = true;
for (auto& t : threads) t.join();
w.join();
late.join();
std::vector<std::string> first(logLines.begin(), logLines.begin() + 3);
std::sort(first.begin(), first.end());
for (const auto& line : first) std::cout << line << "\n";
std::cout << "after the readers left:";
for (size_t i = 3; i < logLines.size(); i++) std::cout << (i > 3 ? ";" : "") << " " << logLines[i];
std::cout << "\n";
}javascript
// price_lock.js: many readers at once, one writer alone, and a waiting writer is not starved
// JavaScript runs one thread, so a lock matters only where a critical section awaits (I/O, a timer).
class ReadWriteLock {
// Writer-preferring: once a writer is waiting, new readers wait behind it.
readers = 0;
writing = false;
waitingWriters = 0;
waitingReaders = 0;
#waiters = [];
async acquireRead() {
this.waitingReaders++;
while (this.writing || this.waitingWriters > 0) await this.#sleep();
this.waitingReaders--;
this.readers++;
}
releaseRead() {
this.readers--;
if (this.readers === 0) this.#wake();
}
async acquireWrite() {
this.waitingWriters++;
while (this.writing || this.readers > 0) await this.#sleep();
this.waitingWriters--;
this.writing = true;
}
releaseWrite() {
this.writing = false;
this.#wake();
}
#sleep() {
return new Promise((resolve) => this.#waiters.push(resolve));
}
#wake() {
this.#waiters.splice(0).forEach((resolve) => resolve());
}
}
const lock = new ReadWriteLock();
const prices = { version: 1, tea: 40 };
const log = [];
let leave;
const left = new Promise((resolve) => (leave = resolve));
async function reader(name) {
await lock.acquireRead();
try {
log.push(`${name} in, sees version ${prices.version}`);
await left; // keeps the read lock until the demo says to leave
} finally {
lock.releaseRead();
}
}
async function writer() {
await lock.acquireWrite();
try {
log.push(`writer in, readers inside: ${lock.readers}`);
prices.version += 1;
prices.tea = 45;
} finally {
lock.releaseWrite();
}
}
async function waitUntil(condition) {
while (!condition()) await new Promise((resolve) => setTimeout(resolve, 1));
}
async function main() {
const tasks = [1, 2, 3].map((n) => reader(`R${n}`));
await waitUntil(() => lock.readers === 3); // only possible if three readers hold the lock at once
console.log("3 readers held the read lock together:", lock.readers === 3 ? "True" : "False");
tasks.push(writer());
await waitUntil(() => lock.waitingWriters === 1); // the writer is waiting for the readers
tasks.push(reader("R4"));
await waitUntil(() => lock.waitingReaders === 1); // a reader arrived behind the waiting writer
leave();
await Promise.all(tasks);
console.log(log.slice(0, 3).sort().join("\n"));
console.log("after the readers left:", log.slice(3).join("; "));
}
main();All four print the same thing:
text
3 readers held the read lock together: True
R1 in, sees version 1
R2 in, sees version 1
R3 in, sees version 1
after the readers left: writer in, readers inside: 0; R4 in, sees version 2Three readers took the read lock and held it, and the program waited until all three were inside at once, which a plain lock could never allow, so "True" is a fact about the lock. Each saw version 1 (the lines are sorted, since their order of entry is not fixed). A writer then had to wait for the three, and a fourth reader arrived behind the waiting writer. When the three left, the writer went first, found no readers inside and bumped the version, and the late reader got in afterwards and saw version 2.
That last step is a policy choice. This lock is writer-preferring: once a writer is waiting, new readers queue behind it. Without that rule, a steady stream of readers could keep the count above zero for ever and the writer would starve. The price is the mirror image: readers wait while a writer is queued, and a steady stream of writers would starve them, which is fine for a price list written once an hour.
In a writer-preferring read–write lock, a reader that arrives while a writer waits queues behind it, so the writer cannot starve.
What each language brings:
- Python: the standard library has no read–write lock, so you build one from a
Condition, as here. - Java:
ReentrantReadWriteLockis built in. Made fair (new ReentrantReadWriteLock(true)), it holds a new reader back while a writer waits, as this one does; the default mode promises no order. - C++:
std::shared_mutex(C++17) is a read–write lock, used withstd::shared_lockfor readers andstd::unique_lockfor the writer. The standard leaves its reader-or-writer preference to the implementation, so if starvation matters, check yours. - JavaScript: one thread, so a lock matters only where a critical section contains an
await(a timer, a request). This one is built from promises.
Three limits to know before you reach for it:
- No upgrade. A reader that asks for the write lock waits for the readers to leave, and it is one of them, so it waits for ever. Release the read lock, take the write lock, and re-check what you read.
- A read that changes things is not a read. In chess,
legal_moves()looks like a query but plays and takes back every candidate on the shared board, so it needs the exclusive lock. The key-value store reads under its one lock because a read may delete an expired key and reorder the eviction list. - It is slower than a plain lock when the work inside is tiny. Counting readers is work. The file system lesson names a readers–writer lock as its next step, taken once you have measured contention.
A read–write lock around a read that changes state is a mistake: chess's legal moves plays and undoes every candidate move, so it needs the exclusive lock.
Two ways to avoid the lock#
Compare-and-set is optimistic: read a value, work out the new one, and write it only if the value is still what you read. If someone changed it first, your write fails and you retry from the fresh value. Nobody waits while holding anything.
python
# price_cas.py: compare-and-set: change a value only if it is still what you read
import threading
class Price:
def __init__(self, value):
# Python has no compare-and-set instruction, so a lock makes the compare and the set one step.
self._value, self._lock = value, threading.Lock()
def get(self):
with self._lock:
return self._value
def compare_and_set(self, expected, new):
with self._lock:
if self._value != expected:
return False
self._value = new
return True
price = Price(50)
a_read, b_read = price.get(), price.get()
print(f"A and B both read {a_read}")
print("A sets 55:", price.compare_and_set(a_read, 55))
print(f"B sets 60: {price.compare_and_set(b_read, 60)} (the price is no longer {b_read})")
fresh = price.get()
print(f"B retries from {fresh}:", price.compare_and_set(fresh, 60))
print("final price", price.get())java
// PriceCas.java: compare-and-set: change a value only if it is still what you read
import java.util.concurrent.atomic.AtomicInteger;
class PriceCas {
public static void main(String[] args) {
AtomicInteger price = new AtomicInteger(50); // compareAndSet is one atomic step
int aRead = price.get(), bRead = price.get();
System.out.println("A and B both read " + aRead);
System.out.println("A sets 55: " + (price.compareAndSet(aRead, 55) ? "True" : "False"));
System.out.println("B sets 60: " + (price.compareAndSet(bRead, 60) ? "True" : "False") + " (the price is no longer " + bRead + ")");
int fresh = price.get();
System.out.println("B retries from " + fresh + ": " + (price.compareAndSet(fresh, 60) ? "True" : "False"));
System.out.println("final price " + price.get());
}
}c++
// price_cas.cpp: compare-and-set: change a value only if it is still what you read
#include <atomic>
#include <iostream>
// compare_exchange_strong is one atomic step; on failure it writes the current value into `expected`,
// so each call gets its own copy of what the caller read.
bool compareAndSet(std::atomic<int>& value, int expected, int next) {
return value.compare_exchange_strong(expected, next);
}
int main() {
std::atomic<int> price{50};
int aRead = price.load(), bRead = price.load();
std::cout << "A and B both read " << aRead << "\n";
std::cout << "A sets 55: " << (compareAndSet(price, aRead, 55) ? "True" : "False") << "\n";
std::cout << "B sets 60: " << (compareAndSet(price, bRead, 60) ? "True" : "False") << " (the price is no longer " << bRead << ")\n";
int fresh = price.load();
std::cout << "B retries from " << fresh << ": " << (compareAndSet(price, fresh, 60) ? "True" : "False") << "\n";
std::cout << "final price " << price.load() << "\n";
}javascript
// price_cas.js: compare-and-set: change a value only if it is still what you read
const price = new Int32Array(new SharedArrayBuffer(4)); // Atomics works on shared typed arrays
price[0] = 50;
// Atomics.compareExchange is one atomic step: it returns the value it found.
const compareAndSet = (expected, next) => Atomics.compareExchange(price, 0, expected, next) === expected;
const bool = (b) => (b ? "True" : "False");
const aRead = Atomics.load(price, 0);
const bRead = Atomics.load(price, 0);
console.log(`A and B both read ${aRead}`);
console.log("A sets 55:", bool(compareAndSet(aRead, 55)));
console.log(`B sets 60: ${bool(compareAndSet(bRead, 60))} (the price is no longer ${bRead})`);
const fresh = Atomics.load(price, 0);
console.log(`B retries from ${fresh}:`, bool(compareAndSet(fresh, 60)));
console.log("final price", Atomics.load(price, 0));All four print the same thing:
text
A and B both read 50
A sets 55: True
B sets 60: False (the price is no longer 50)
B retries from 55: True
final price 60Two updaters read the price 50. A sets 55 and wins. B tries to set 60, but the price is no longer 50, so its write fails. B reads again, gets 55, and sets 60. No update was lost: a plain write would have let B overwrite A's 55 with a value worked out from a stale 50. Java, C++ and JavaScript use the processor's atomic instruction; Python's standard library has none, so a lock makes the compare and the set one step, which is the same contract. Ride sharing keeps one driver on one trip this way, and the key-value store does it at the scale of a transaction.
The demo takes turns so you can read it; real code writes B's retry as a loop (read, compute, compare-and-set, repeat on failure), which Java's updateAndGet writes for you. The trap: the value can change and change back (50, then 55, then 50) and your compare still passes. When that matters, compare a version number, not the value.
The single writer removes the contention by design: only one thread ever changes the data, and everyone else sends it a message through a queue. The order book has one matcher thread per symbol and no locks inside the book, so the same input always gives the same trades.
| Plain lock | Read–write lock | Compare-and-set | Single writer | |
|---|---|---|---|---|
| Threads wait | Whenever another holds it | Writers for readers; readers for a waiting writer | No: a loser retries | Producers wait only if the queue is full |
| Good when | Short critical sections, mixed traffic | Many reads, rare writes, long reads | One small value, rare conflict | Many changes that must be applied in order |
| Bad when | Many readers contend | Reads that mutate, tiny sections | High conflict (everyone retries) | The writer is the bottleneck |
| In the course | lld-01, lld-10, lld-17, lld-18 | the next step in lld-17 | lld-14, lld-22 | lld-20 |
Concurrency and its look-alikes#
| Pattern | Looks like it because | The difference |
|---|---|---|
| Observer (dp-17) | one side announces, the other reacts | Observer calls listeners on the caller's thread. A queue decouples when: the producer returns at once and a worker acts later. The logger's AsyncSink puts a slow listener behind a queue. |
| Message broker (Week 3) | producers, consumers and a queue | The same shape between processes, with durability and delivery guarantees. The in-process queue here is lost if the process stops. |
| Command (dp-18) | a task is a request that waits | The pool's tasks are commands: Command is the object, the pool is the machinery that runs it on a thread. |
| Async and await | many things in flight, few threads | Async is concurrency on one thread, switching at each await. A pool runs tasks on several threads at once, where the runtime allows it. |
When not to use them#
- There is no second thread. A queue, a pool or a lock around single-threaded code is cost for nothing.
- One lock is enough. Several rounds (the parking lot, chess, the file system, the editor) are correct with one lock around the shared state. Say so, and name the finer option for when the interviewer pushes on throughput.
- A thread per task is fine for a script that starts three tasks. A pool is for many tasks over time.
- The work cannot be repeated. Compare-and-set retries; if the work before the set has side effects (a payment, an email), a retry runs them twice. Put the side effect after the successful set.
- You can use the library.
ArrayBlockingQueue,ThreadPoolExecutor,std::shared_mutexandReentrantReadWriteLockare tested far more than your first version. Know how they work so you can choose their settings (the queue limit first), and use them.
Where it shows up in the case studies#
| Case study | The pattern | What it is |
|---|---|---|
| Parking lot | One lock | Find-and-occupy is one step under one lock, so two gates cannot take the same spot. A lock per floor is the follow-up. |
| Movie booking | Holds that expire | InMemorySeatLocks.lock: every seat or none, checked and taken under one lock, and each hold lapses after its TTL, so an abandoned checkout frees its seats. |
| Pub-sub logger | Producer–consumer, drop don't block | AsyncSink: a bounded queue and one worker thread, with a full queue dropping and counting records. |
| Chess, file system, text editor | One re-entrant lock | Every public method takes it, reads included: chess because legal_moves() mutates the board; the editor's autosave takes a snapshot under the lock and writes it outside. |
| Ride sharing | Compare-and-set | Driver.try_reserve: "if still free, take", under that driver's own lock, so a stale list of nearby drivers is harmless. |
| Order book | Single writer | One matcher thread per symbol, no locks inside the book. |
| Task scheduler | Worker pool, condition variable | Workers sleep on one condition variable until there is work or a delayed task is due: 40 tasks on 4 workers. |
| Key-value store | Optimistic concurrency | Transactions record the versions they read and raise Conflict at commit if any moved. |
In the interview#
Say the pattern with the reason, in one breath. For the queue: "Orders arrive faster than we can price them, so I'll put a bounded queue between them: producers wait when it is full, so the system pushes back instead of running out of memory, and a fixed pool of workers drains it." For the lock: "The price list is read constantly and written rarely, so a read–write lock lets readers run together, and I'll prefer the writer so it cannot starve."
Then expect the follow-ups:
- "What if the queue is full?" Block (with a timeout, since the caller now waits), drop and count (only for work that can be lost), or refuse with an error.
- "How many workers? What if a task throws? How do you shut down?" Cores for CPU work, more for I/O, as a measured setting. Catch per task, since a dead worker silently shrinks the pool. Drain or drop, and say which.
- "Why not a thread per task?" Threads cost time and memory, and nothing stops 100,000 of them. A pool bounds both.
- "Could a writer starve? Can a reader upgrade?" Not here, because a waiting writer holds back new readers. No upgrade: release the read lock, take the write lock, and re-check.
- "Can you avoid the lock?" Compare-and-set for one small value (costs retries under conflict), or a single writer fed by a queue (the writer can be the bottleneck).
Common mistakes#
- An unbounded queue. Say the limit and what happens at it.
- Checking a condition with
ifinstead ofwhile. A woken thread must check again. - Polling with a sleep where a blocking
getwould do, and a pool whose workers die on an exception. - A read–write lock around reads that change something, or a reader asking for the write lock.
- Compare-and-set on a value that can change and change back, when a version number was needed.
- Printing or testing concurrent output that is not deterministic. Join the threads, sort or index the results, and wait on counters, never on a sleep.
Quick check4 questions · +5 XP each
Concurrency patterns
Q1/4 In the first version, 1,000 orders were produced before the consumer started and all 1,000 sat in the queue. What does a bounded queue change?
Answer & explanation
Answer: C
The limit turns a slow consumer into back-pressure on the producers instead of unbounded memory. What the producer does at the limit (wait, drop or refuse) is the design decision.
Q2/4 A worker thread takes a task that throws an exception. What should the pool do?
Answer & explanation
Answer: C
A dead worker silently shrinks the pool. Catching per task keeps the pool at its size and keeps the failure visible in that task's result.
Q3/4 Three readers hold a read–write lock and a writer is waiting. A fourth reader arrives. In a writer-preferring lock, what happens?
Answer & explanation
Answer: B
Letting new readers in while a writer waits could keep the reader count above zero for ever. Queueing the new reader behind the writer costs it a short wait and guarantees the writer a turn.
Q4/4 Two updaters both read a price of 50. A compare-and-sets 50 to 55 and succeeds. B compare-and-sets 50 to 60. What happens, and what should B do?
Answer & explanation
Answer: C
Compare-and-set writes only if the value is still what you read. A failure means someone else changed it first, and the fix is to re-read and try again, so no update is lost.
this lesson in shorts · 3
play them all →One idea from this lesson in about a minute each, made for your phone.
End of Pattern 25 · Concurrency Patterns: Producer–Consumer, Thread Pool, Read–Write Lock
Finish Pattern 25Stages clear as you read them (0/1 so far). Skipping ahead doesn't count: marking complete unlocks after ~5 more min of reading.
New lessons by email
New lessons, side quests and case files as they ship, at most once a week. No spam, unsubscribe anytime.
Discussion · …
Public: everyone reading this lesson sees comments. Be kind and keep it on the material.
Sign in to ask a question or answer one. Reading stays open to everyone.
loading the discussion…