Language-Level Concurrency and Multithreading Questions

Direct answer

Concurrency means several tasks are in progress during overlapping time periods, so a program is structured to deal with many things at once, even if only one of them is executing at any instant (one CPU core switching between them). Parallelism means several tasks are literally executing at the same instant on different CPU cores. Concurrency is about structure and waiting; parallelism is about hardware. A program can be concurrent without being parallel (one thread running an event loop), and parallel work needs more than one core.

The practical consequence: I/O-bound work (mostly waiting on network, disk or a database) benefits from concurrency alone, because while one task waits another can run. CPU-bound work (mostly computing) only speeds up with real parallelism, and then only up to the number of cores.

Python, Go and Java side by side

Terms in the table: an event loop is a loop on one thread that waits for any of many pending operations to be ready and runs the code waiting on it; a coroutine is a function that can pause at an await and be resumed later by the loop; a virtual thread (Java 21) is a very lightweight thread managed by the JVM rather than the OS, so thousands can block cheaply on a few OS threads; a ForkJoinPool is a Java thread pool that splits a big job into pieces for its worker threads. In an interview, one pair per language is enough: a concurrent-but-not-parallel example and a truly-parallel one.

The GIL (global interpreter lock) is a mutex inside CPython that lets only one thread execute Python bytecode (the low-level instructions the interpreter runs) at a time. The Python 3.12 glossary says it exists to make the object model safe against concurrent access, at the expense of much of the parallelism of multi-processor machines, and that it is always released while doing I/O (and by some extension modules during heavy computation such as compression or hashing). So in CPython threads give concurrency for I/O-bound work but not parallel speedup for pure-Python CPU-bound code. A free-threaded build (PEP 703, the Python design document that removes the GIL in a separate build) exists as an optional build variant starting with Python 3.13, and it is not the default distribution, so do not assume it in an interview answer without saying which build you mean.

Runnable examples

Python (3.12). Asyncio overlaps three coroutines on one thread; a thread pool overlaps four blocking calls; a process pool computes the CPU-bound part in other processes:

import asyncio
import os
import threading
import time
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor


def cpu_task(n):
    """CPU-bound: sum of squares below n. Returns (worker pid, result)."""
    return os.getpid(), sum(i * i for i in range(n))


async def io_coroutine(name, log):
    log.append((name, "start", threading.get_ident()))
    await asyncio.sleep(0.1)          # stands in for a network call; yields to the event loop
    log.append((name, "end", threading.get_ident()))


async def run_coroutines():
    log = []
    await asyncio.gather(*(io_coroutine(f"c{i}", log) for i in range(3)))
    starts_before_first_end = [e[1] for e in log].index("end")
    thread_ids = {e[2] for e in log}
    print("asyncio: events before the first 'end':", starts_before_first_end,
          "| distinct threads used:", len(thread_ids))


def blocking_io(state):
    with state["lock"]:
        state["now"] += 1
        state["max"] = max(state["max"], state["now"])
    time.sleep(0.1)                   # blocking call; the GIL is released while sleeping
    with state["lock"]:
        state["now"] -= 1


def run_threads():
    state = {"lock": threading.Lock(), "now": 0, "max": 0}
    with ThreadPoolExecutor(max_workers=4) as pool:
        list(pool.map(blocking_io, [state] * 8))
    print("threads: most blocking calls in flight at once:", state["max"], "(pool size 4)")


def run_processes():
    sizes = [200_000] * 8
    with ProcessPoolExecutor(max_workers=4) as pool:
        results = list(pool.map(cpu_task, sizes))
    expected = sum(i * i for i in range(200_000))
    pids = {pid for pid, _ in results}
    print("processes: all results correct:", all(r == expected for _, r in results),
          "| ran in other processes:", os.getpid() not in pids,
          "| distinct worker processes <= 4:", len(pids) <= 4)


if __name__ == "__main__":
    asyncio.run(run_coroutines())
    run_threads()
    run_processes()

The log holds a start and an end entry per coroutine. If the coroutines ran one after another, the first end would come at index 1 (right after c0 started). Three start entries before any end means all three were started and waiting at the same time, and one distinct thread id means they did it without a second thread: concurrency without parallelism. In blocking_io, a counter (now) is raised on entry and lowered on exit under a lock, and max remembers its highest value, so 4 means four blocking calls were in flight together even though the pool had to cycle through 8 jobs.

asyncio: events before the first 'end': 3 | distinct threads used: 1
threads: most blocking calls in flight at once: 4 (pool size 4)
processes: all results correct: True | ran in other processes: True | distinct worker processes <= 4: True

Go (run with go run -race main.go, Go 1.22 or later: each goroutine below captures the loop variable c, and only from Go 1.22 does every loop iteration get its own c; under an older language version, for example a go.mod saying go 1.21 with the package run as go run . (naming main.go directly does not apply the go.mod language version), the same program shares one c, which is a data race, and typically panics with an index out of range). Four blocking goroutines overlap; the same CPU-bound split is run with GOMAXPROCS set to 1 and then to the CPU count, and the answer is identical, which is the point: the code is concurrent by structure, and whether it is parallel depends on the setting and the hardware:

package main

import (
	"fmt"
	"runtime"
	"sync"
	"sync/atomic"
	"time"
)

func sumSquares(lo, hi int) int {
	s := 0
	for i := lo; i < hi; i++ {
		s += i * i
	}
	return s
}

func main() {
	// Concurrency: 4 goroutines that block (simulated I/O) overlap in time.
	var now, max int32
	var wg sync.WaitGroup
	for i := 0; i < 4; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			cur := atomic.AddInt32(&now, 1)
			for {
				m := atomic.LoadInt32(&max)
				if cur <= m || atomic.CompareAndSwapInt32(&max, m, cur) {
					break
				}
			}
			time.Sleep(50 * time.Millisecond)
			atomic.AddInt32(&now, -1)
		}()
	}
	wg.Wait()
	fmt.Println("blocking goroutines in flight at once:", max)

	// Parallelism: split CPU-bound work into 4 chunks; the same code runs with 1 or many OS threads.
	const n = 4_000_000
	want := sumSquares(0, n)
	for _, procs := range []int{1, runtime.NumCPU()} {
		runtime.GOMAXPROCS(procs)
		parts := make([]int, 4)
		for c := 0; c < 4; c++ {
			wg.Add(1)
			go func() {
				defer wg.Done()
				parts[c] = sumSquares(c*n/4, (c+1)*n/4)
			}()
		}
		wg.Wait()
		got := parts[0] + parts[1] + parts[2] + parts[3]
		fmt.Printf("GOMAXPROCS=%d result correct: %v\n", runtime.GOMAXPROCS(0), got == want)
	}
	fmt.Println("CPUs visible to this container:", runtime.NumCPU())
}

blocking goroutines in flight at once: 4
GOMAXPROCS=1 result correct: true
GOMAXPROCS=14 result correct: true
CPUs visible to this container: 14

(The 14 is the CPU count of the Linux container these programs ran in; it differs on other machines.)

The loop inside each goroutine records the highest number of goroutines in flight. now counts goroutines currently inside the sleep. A plain if cur > max { max = cur } would be a data race (two goroutines read and write max at once, and one could overwrite a larger value with a smaller one). So the loop uses compare-and-swap (CAS): CompareAndSwapInt32(&max, m, cur) sets max to cur only if max still equals the m that was just read, and returns false if another goroutine changed it first, in which case the loop reloads and tries again. When it ends, max holds the largest cur any goroutine saw. The race detector (the -race flag, which instruments the program and reports unsynchronized accesses at runtime) reported nothing.

Java 21. Four tasks that wait for each other can only finish if all four are in flight at once: each task counts a CountDownLatch (a counter that releases all waiters when it reaches zero) down by one and then waits on it, so the latch reaches zero only after all four tasks have started and are waiting together. If the pool ran them one at a time, the first would wait forever and the program would hang. The second part starts 1,000 virtual threads that all block at once, which shows that blocking is cheap; the parallel stream splits the CPU-bound sum across the common fork-join pool:

import java.util.concurrent.*;
import java.util.stream.IntStream;

public class ConcurrencyDemo {
    public static void main(String[] args) throws Exception {
        // Concurrency with few threads: 4 tasks that all wait for each other prove 4 are in flight at once.
        CountDownLatch fourStarted = new CountDownLatch(4);
        try (ExecutorService pool = Executors.newFixedThreadPool(4)) {
            for (int i = 0; i < 4; i++) {
                pool.submit(() -> { fourStarted.countDown(); fourStarted.await(); return null; });
            }
        }
        System.out.println("fixed pool of 4: all 4 tasks were in flight together");

        // Many cheap blocking tasks: 1,000 virtual threads (Java 21) all waiting on one latch.
        int n = 1000;
        CountDownLatch allStarted = new CountDownLatch(n);
        try (ExecutorService vt = Executors.newVirtualThreadPerTaskExecutor()) {
            for (int i = 0; i < n; i++) {
                vt.submit(() -> { allStarted.countDown(); allStarted.await(); return null; });
            }
        }
        System.out.println("virtual threads: all " + n + " tasks were blocked at once");

        // Parallelism: a CPU-bound sum split across the common ForkJoinPool.
        long expected = 0;
        for (int i = 0; i < 4_000_000; i++) expected += (long) i * i;
        long got = IntStream.range(0, 4_000_000).parallel().mapToLong(i -> (long) i * i).sum();
        System.out.println("parallel stream result correct: " + (got == expected)
            + " | common pool parallelism: " + ForkJoinPool.getCommonPoolParallelism()
            + " | availableProcessors: " + Runtime.getRuntime().availableProcessors());
    }
}

fixed pool of 4: all 4 tasks were in flight together
virtual threads: all 1000 tasks were blocked at once
parallel stream result correct: true | common pool parallelism: 13 | availableProcessors: 14

Run repeatedly, each program prints the same lines apart from the machine-specific CPU numbers (the 13 and 14 in the Java line and the 14 in the Go lines). On a container limited to one CPU the Java line reports a common pool parallelism of 1 and one available processor: the stream is still correct, but with a single core nothing can execute at the same instant (the pool's one worker and the calling thread only take turns), which is the point that parallelism depends on the hardware. What these outputs show: tasks overlap in time (concurrency), the work splits correctly, and the process pool really runs in other processes. They do not show a speedup; that depends on core count and load, so measure it with a timer on your own machine rather than quoting a number.

What this means for CPU-bound versus I/O-bound work

Worked example: a service makes 50 calls that each wait 200 ms on the network. Run one after another that is 50 x 200 ms = 10 s of waiting. With 50 concurrent tasks the waits overlap and the total is close to one call's 200 ms plus overhead, on a single core. Now take 8 CPU-heavy jobs of 1 s each on a 4-core machine: with real parallelism the best case is 8 / 4 = 2 s of work time, while CPython threads running pure-Python code would still take roughly 8 s (this is the arithmetic of the example, not a timing from the demos above) because only one thread executes bytecode at a time. Mixed workloads split the two: fetch concurrently, compute in parallel.

Pitfalls

• Calling a blocking function inside an async def freezes the event loop, which turns "concurrent" into "stuck".
• Virtual threads are not faster threads: JEP 444 says they do not run code faster than platform threads and they help throughput of waiting tasks, not CPU-bound work. They also should not be pooled; limit concurrency with a semaphore instead.
• Shared mutable state between parallel workers needs synchronization in Go and Java (Go's race detector flagged none here because each goroutine wrote its own slot), whereas separate Python processes share nothing by default and must send results back.

Direct answer

For a script that polls thousands of sockets (waiting on the network, almost no CPU per socket), use asyncio: one thread runs an event loop (a loop that waits for any of many sockets to become ready, then runs the code waiting on that socket) and handles all of them with little memory and no locking. The code that waits is written as a coroutine: a function declared with async def that can pause at each await (for example await reader.readline()) and hand control back to the loop, which resumes it when the awaited thing is ready. Use threads when the code you must call is blocking and has no async version. Use multiprocessing only when the work is CPU-bound Python and you need several cores.

What the GIL changes

The GIL (Global Interpreter Lock) is a lock in the standard CPython interpreter that lets only one thread execute Python bytecode (the low-level instructions the interpreter runs) at a time. Consequences:

• Threads running pure Python do not run in parallel: two CPU-bound threads take turns.
• A thread waiting on I/O (socket.recv, time.sleep, file reads) releases the GIL, so other threads run while it waits. Threads therefore do help I/O-bound work.
• Some C extensions (NumPy, hashing, compression) release the GIL during heavy work, so threads can speed those up too.
• Processes each have their own interpreter and their own GIL, so they run in parallel across cores, at the cost of process start-up, memory per process, and pickling (serializing an object to bytes with Python's pickle module so it can cross a process boundary) every argument and result.
• asyncio is single-threaded, so the GIL is not even a factor; the cost is that any blocking call inside a coroutine freezes the entire loop. A tiny demonstration: a ticker prints every 0.2 s while another coroutine does time.sleep(1.0) (a blocking call) after 0.3 s:

import asyncio, time

async def ticker(t0):
    for _ in range(5):
        print(f"tick at {time.perf_counter() - t0:.1f} s")
        await asyncio.sleep(0.2)

async def blocker(t0):
    await asyncio.sleep(0.3)
    time.sleep(1.0)            # blocking call: freezes the whole loop

async def main():
    t0 = time.perf_counter()
    await asyncio.gather(ticker(t0), blocker(t0))

asyncio.run(main())

A sample run in a python:3.12-slim container printed:

tick at 0.0 s
tick at 0.2 s
tick at 1.3 s
tick at 1.5 s
tick at 1.7 s

The ticks at 0.4 s to 1.2 s are missing: the whole loop stood still for the second the blocking call held it. Replacing time.sleep(1.0) with await asyncio.sleep(1.0) in blocker gave ticks at 0.0, 0.2, 0.4, 0.6 and 0.8 s with no gap. (Exact tick times shift by a few hundredths of a second between runs; the missing stretch is what matters.)

Status of the GIL (this is background depth; a PEP is a Python Enhancement Proposal, the design document for a language change): PEP 703 added an optional free-threaded build (no GIL) of CPython, experimental in 3.13. PEP 779 (status Final) made that build officially supported but still optional from Python 3.14; the default build still has the GIL, and extension compatibility and single-thread speed are the trade-offs (the 3.14 release notes put the single-thread penalty at roughly 5 to 10 percent, depending on platform and compiler). So do not design around a GIL-free interpreter being present: on Python 3.13 and later, sys._is_gil_enabled() tells you whether the interpreter you will deploy to is running with the GIL.

Measured comparison

import asyncio, time, threading
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

def cpu_work(n):
    s = 0
    for i in range(n):
        s += i * i
    return s

def io_work(_):
    time.sleep(0.5)          # stands in for a blocking network call; releases the GIL

async def aio_work(_):
    await asyncio.sleep(0.5)

def timed(label, fn):
    t = time.perf_counter(); fn(); print(f"{label:<28}{time.perf_counter() - t:6.2f} s")

def main():
    N, W = 4_000_000, 4
    print("CPU-bound, 4 jobs of 4M iterations")
    timed("  sequential", lambda: [cpu_work(N) for _ in range(W)])
    with ThreadPoolExecutor(W) as ex:
        timed("  4 threads", lambda: list(ex.map(cpu_work, [N] * W)))
    with ProcessPoolExecutor(W) as ex:
        list(ex.map(cpu_work, [1] * W))    # warm up the worker processes
        timed("  4 processes", lambda: list(ex.map(cpu_work, [N] * W)))
    print("I/O-bound, 40 waits of 0.5 s")
    timed("  sequential", lambda: [io_work(0) for _ in range(40)])
    with ThreadPoolExecutor(40) as ex:
        timed("  40 threads", lambda: list(ex.map(io_work, range(40))))
    async def many(): await asyncio.gather(*(aio_work(i) for i in range(40)))
    timed("  asyncio, 40 tasks", lambda: asyncio.run(many()))

if __name__ == "__main__":
    main()

One sample run, Python 3.12.15 in a python:3.12-slim container on a 14-core aarch64 host:

CPU-bound, 4 jobs of 4M iterations
  sequential                  0.42 s
  4 threads                   0.41 s
  4 processes                 0.15 s
I/O-bound, 40 waits of 0.5 s
  sequential                 20.04 s
  40 threads                  0.51 s
  asyncio, 40 tasks           0.51 s

Reading it: ex.map(f, items) runs f on every item in the pool and returns results in order. The warm-up line sends tiny jobs first so process start-up is not counted in the timing, and asyncio.gather(*coros) starts all the coroutines at once and waits for all of them. For CPU-bound work, threads gave no speed-up (0.41 s against 0.42 s sequential) because of the GIL, and processes were about 2.8 times faster (0.15 s; with 4 workers the ideal would be 4 times, and the gap is process overhead on a short job). That speed-up needs at least 4 free cores: the same script in a container limited to 1 CPU showed the 4 processes no faster than sequential. For waiting, threads and asyncio both collapse 40 waits of 0.5 s to the length of one wait, 0.51 s. Timings vary run to run; the pattern is the finding.

Polling thousands of sockets

import asyncio, resource, time

N = 3000   # concurrent connections; 3000 client + 3000 server sockets = 6000 descriptors

# Many shells start with a soft limit of 1,024: raise it as far as the hard limit allows.
soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
if soft < 8192:
    resource.setrlimit(resource.RLIMIT_NOFILE, (8192 if hard == resource.RLIM_INFINITY else min(8192, hard), hard))

async def handle(reader, writer):
    data = await reader.readline()
    writer.write(data.upper()); await writer.drain()
    writer.close()

async def client(port, i, ok):
    r, w = await asyncio.open_connection("127.0.0.1", port)
    w.write(f"ping {i}\n".encode()); await w.drain()
    reply = await r.readline()
    ok.append(reply.startswith(b"PING"))
    w.close()

async def main():
    server = await asyncio.start_server(handle, "127.0.0.1", 0, backlog=N)
    port = server.sockets[0].getsockname()[1]
    ok = []
    t = time.perf_counter()
    await asyncio.gather(*(client(port, i, ok) for i in range(N)))
    print(f"{len(ok)} of {N} sockets answered, all correct: {all(ok)}, {time.perf_counter() - t:.2f} s, one thread")
    server.close()

print("fd limit:", resource.getrlimit(resource.RLIMIT_NOFILE)[0])
asyncio.run(main())

In the same container (soft limit already 20,480, so no raise was needed) it printed:

fd limit: 20480
3000 of 3000 sockets answered, all correct: True, 0.26 s, one thread

Walk through the program. asyncio.start_server(handle, host, 0, backlog=N) listens on a free port (0 lets the OS choose) and runs handle as a new coroutine for each accepted connection; backlog is how many not-yet-accepted connections the OS may queue, raised so a burst of 3,000 is not refused. reader.readline() waits for one line without blocking the loop. writer.write(...) queues bytes and await writer.drain() waits until the send buffer has room. asyncio.open_connection is the client side and returns a reader and writer pair. The 3,000 client coroutines run concurrently under gather. That is 3,000 client connections and 3,000 matching server connections (6,000 file descriptors) served from one thread on loopback. The file-descriptor limit printed first matters: each socket uses one file descriptor (the small integer the OS uses to name an open file or socket), and many systems default to 1,024, so a script that polls thousands of sockets must raise ulimit -n (the shell command that sets the per-process descriptor limit) or call resource.setrlimit, as the program above does, regardless of the concurrency model. Without that, this program fails with OSError: [Errno 24] Too many open files when the soft limit is 1,024. Loopback with an instant echo is the easy case; against real remote servers the elapsed time is dominated by network latency, which is exactly what concurrency hides.

Choosing, with the numbers

For the socket-polling script, recommend asyncio with a cap on concurrent connections (an asyncio.Semaphore) and timeouts (asyncio.timeout, available from Python 3.11, or asyncio.wait_for, which also works on older versions) so one dead peer cannot hang the run. On Linux, asyncio uses epoll (a kernel facility that reports which of many sockets are ready without scanning them all) through the selectors module (Python's wrapper over such facilities), whereas a hand-written select() loop cannot watch descriptors numbered 1,024 or higher (the FD_SETSIZE limit, a fixed size of the descriptor set that select() accepts). You do not need these names to use asyncio; they explain why it scales. If part of the job is blocking or CPU-heavy (parsing large replies), hand that part to loop.run_in_executor or asyncio.to_thread, so the loop keeps answering sockets. What would flip the choice: an existing blocking client library with no async version (threads), or heavy per-reply computation (add a process pool behind the loop).

Direct answer

The same word means three different things:

• Java volatile is a concurrency tool. A write to a volatile field happens-before every later read of that field (happens-before is the Java Memory Model's guarantee that one action's effects are visible to another). It gives visibility between threads, an ordering guarantee, and atomic reads and writes even for long and double. It does not make compound operations like count++ atomic.
• C++ volatile (before C++11, and still today) is not a concurrency tool. It tells the compiler that every access must really happen (no caching in a register, no removal), which is what you need for memory-mapped hardware registers. It gives no atomicity and no ordering between threads, and a data race on a volatile variable is still undefined behaviour (UB: the language places no requirements on what the program does, so results can be wrong in ways that vary by compiler and machine).
• C++11 std::atomic<T> is the C++ equivalent of what Java volatile does: atomic operations with defined ordering rules, selected by a memory_order argument. The default, seq_cst (sequentially consistent: all threads agree on one global order of atomic operations), is the strongest and the right choice unless you have a measured reason to weaken it; the weaker orders (relaxed, acquire, release) exist for expert tuning.

What Java volatile guarantees

From the Java Language Specification (chapter 17):

1. Visibility and happens-before. "A write to a volatile field (§8.3.1.4) happens-before every subsequent read of that field." Everything the writing thread did before the write (including writes to ordinary fields) is visible to a thread that reads the new value.
2. Order. All volatile accesses are part of one total synchronization order (a single sequence of all volatile reads and writes that every thread agrees on) that is consistent with each thread's program order (the order written in that thread's code). In practice, volatile accesses are not reordered with each other, and ordinary reads and writes cannot be moved across them in a way that breaks the happens-before edge.
3. Atomic reads and writes. Reads and writes of volatile long and double are always atomic; for non-volatile ones the specification allows them to be treated as two 32-bit halves.

It does not give: atomic read-modify-write (x++ is read, add, write, so two threads can lose an update), mutual exclusion over several fields, or any guarantee about non-volatile fields that are not published through a volatile write.

Demonstration

public class Visibility {
    static boolean plainStop = false;          // no volatile
    static volatile boolean volStop = false;
    static int payload;                         // plain field published via the volatile flag
    static volatile boolean ready;

    static void spinPlain() throws Exception {
        Thread t = new Thread(() -> { long n = 0; while (!plainStop) n++; });
        t.start();
        Thread.sleep(200);
        plainStop = true;
        t.join(2000);
        System.out.println("plain flag: worker " + (t.isAlive() ? "STILL RUNNING after 2 s" : "stopped"));
        if (t.isAlive()) System.exit(0);
    }

    static void spinVolatile() throws Exception {
        Thread t = new Thread(() -> { long n = 0; while (!volStop) n++; });
        t.start();
        Thread.sleep(200);
        volStop = true;
        t.join(2000);
        System.out.println("volatile flag: worker " + (t.isAlive() ? "STILL RUNNING" : "stopped"));
    }

    static void publish() throws Exception {
        Thread reader = new Thread(() -> {
            while (!ready) { }
            System.out.println("reader saw payload = " + payload); // guaranteed 42
        });
        reader.start();
        payload = 42;   // plain write, ordered before the volatile write below
        ready = true;   // volatile write: happens-before every later read that sees true
        reader.join();
    }

    public static void main(String[] a) throws Exception {
        if (a.length > 0 && a[0].equals("plain")) spinPlain(); else { spinVolatile(); publish(); }
    }
}

Run with java Visibility.java plain on OpenJDK 21.0.12.1 (container eclipse-temurin:21-jdk, aarch64), it printed plain flag: worker STILL RUNNING after 2 s at 2, 4 and 8 CPUs. Limited to one CPU the outcome varied from run to run: some runs printed stopped and others STILL RUNNING, because whether the JIT has compiled the loop within the 200 ms depends on how much CPU time its compiler threads get. Once the loop is compiled by the JIT (the JVM's just-in-time compiler, which turns hot bytecode into machine code while the program runs), the non-volatile flag is read once and the loop never sees the update. That is behaviour this JVM exhibited, not a guarantee: the specification permits the worker to see the change or not, and the timing makes it vary. Run with no argument it printed:

volatile flag: worker stopped
reader saw payload = 42

The second line is the use case that matters: payload is a plain field, yet it is guaranteed to be 42 because the write to it comes before the volatile write to ready and the reader saw ready == true.

C++: volatile is not atomic

#include <atomic>
#include <cstdio>
#include <thread>
#include <vector>

volatile int vcount = 0;            // volatile int: NOT atomic
std::atomic<int> acount{0};

int main() {
    std::vector<std::thread> ts;
    for (int t = 0; t < 4; t++)
        ts.emplace_back([] {
            for (int i = 0; i < 100000; i++) {
                vcount = vcount + 1;                         // load, add, store: races
                acount.fetch_add(1, std::memory_order_relaxed);
            }
        });
    for (auto& t : ts) t.join();
    std::printf("volatile: %d  atomic: %d  (expected 400000)\n", (int)vcount, acount.load());
}

Compiled with g++ -std=c++20 -O2 -pthread (GCC 14.4, aarch64 container), the atomic total is 400000 at every CPU count, because fetch_add is atomic. The volatile total depends on how many CPUs the process may use: with the container limited to 2 or more CPUs it usually came out below 400000 and changed from run to run (on 2 CPUs some runs still printed exactly 400000; with 8 CPUs it was typically far lower). Limited to a single CPU it normally printed 400000, because the four threads rarely got interleaved in the middle of an increment, yet that is still a data race. The volatile total is never guaranteed to be low (a run in which the threads happen not to overlap can reach 400000); the point is that nothing makes it reliable. One sample run printed:

volatile: 119392  atomic: 400000  (expected 400000)

Read it as: four threads each tried 100,000 increments, so 400,000 is the correct total. The atomic figure matches it. The volatile figure is the number of increments that survived; every missing one is a case where two threads loaded the same old value and both stored old+1. Compiled with -fsanitize=thread, ThreadSanitizer (a compiler-inserted checker that reports data races while the program runs) printed WARNING: ThreadSanitizer: data race pointing at the volatile line, including on a run limited to one CPU whose volatile total happened to be correct. (The atomic uses memory_order_relaxed because a pure counter needs atomicity but no ordering; that is safe only because nothing else is published through it. For a flag that publishes data, use release on the store and acquire on the load (a pair that makes everything written before the store visible after the load), or simply leave the default seq_cst, which is what a beginner should write.)

Comparison

Use cases for Java volatile: a stop or status flag written by one thread and read by others; safely publishing an immutable or effectively immutable object; the one reference field in double-checked locking. When you need atomic read-modify-write or compare-and-set, use AtomicInteger and friends or VarHandle (the java.lang.invoke API that gives atomic and ordered operations on a field or array element); when you need to guard several variables together, use synchronized or a Lock.

Direct answer

Use a buffered channel of tasks as the queue (a Go channel is a typed pipe between goroutines, the language's lightweight threads; a buffered one has a fixed capacity, a send succeeds immediately while there is room and blocks when the buffer is full, and a receive blocks while it is empty), N worker goroutines that range over it, and a Pool struct with three methods: Submit (blocks when the queue is full), TrySubmit (returns ErrQueueFull instead of waiting for queue space) and Shutdown (stops intake, drains, waits). The queue capacity is the backpressure: once it is full, the producer either waits or is told no, so memory cannot grow without bound. Graceful shutdown is close(tasks) after a flag stops new submissions: workers keep receiving until the channel is both closed and empty, so every accepted task still runs, and sync.WaitGroup tells Shutdown when the last worker has exited.

Signatures

type Task func()

type Pool struct { /* tasks chan Task; wg sync.WaitGroup; mu sync.RWMutex; closed bool */ }

func NewPool(n, queueSize int) *Pool
func (p *Pool) Submit(t Task) error     // blocks while the queue is full; ErrClosed after Shutdown
func (p *Pool) TrySubmit(t Task) error  // never waits for queue space; ErrQueueFull when full, ErrClosed after Shutdown
func (p *Pool) Shutdown()               // stop accepting, finish queued and running tasks, then return

Goroutine behaviour

• Workers (N goroutines): for t := range p.tasks { t() }, where range over a channel receives values one after another and ends when the channel is closed and empty. A receive from a closed channel only reports "done" once the buffer is empty, so closing the channel does not lose queued tasks. defer wg.Done() marks exit: defer schedules a call to run when the surrounding function returns, and sync.WaitGroup is a counter that Add(n) raises, each Done() lowers, and Wait() blocks until it reaches zero, so Shutdown knows when every worker has exited.
• Producers (any goroutine calling Submit): take a read lock, check closed, then send. A full buffer blocks the send, which is the backpressure. Many producers can hold the read lock at once: sync.RWMutex allows any number of readers together but only one writer, and here the "readers" are the senders. They do not modify shared state; they only need closed to stay false while they send. Using a plain mutex instead would make senders take turns, so one producer blocked on a full queue would hold up every other producer's TrySubmit check, and Shutdown is the only party that needs exclusive access.
• TrySubmit: the same check, then a select with a default case. select waits on several channel operations and runs whichever can proceed; with a default case it never waits on the channel, so if the send cannot proceed immediately (the buffer is full) the default branch runs and returns ErrQueueFull. It does still take the read lock first, and that lock can make it wait: if Shutdown is already waiting for the write lock because some Submit is parked on a full queue, Go's sync.RWMutex makes new readers queue behind the waiting writer, so a fresh TrySubmit waits until the queue drains and Shutdown has taken and released the write lock, and then returns ErrClosed. The test TestTrySubmitWaitsBehindShutdown, shown after the stress test below, sets this up with one worker, a queue of one, one parked Submit and a pending Shutdown, and shows the new TrySubmit still waiting until the worker is released.
• Shutdown: takes the write lock (waits for in-progress sends to finish), sets closed = true, calls close(tasks) once, releases the lock, then wg.Wait().

The lock exists for one reason: sending on a closed channel panics, and Go has no way to ask "is this channel closed?" atomically with the send. The read-write lock makes "check flag, then send" and "set flag, then close" mutually exclusive, so a producer can never send into a channel that Shutdown is closing.

Runnable code

package main

import (
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
)

var (
	ErrQueueFull = errors.New("queue full")
	ErrClosed    = errors.New("pool closed")
)

type Task func()

type Pool struct {
	tasks  chan Task
	wg     sync.WaitGroup
	mu     sync.RWMutex // makes "check closed, then send" exclusive with "set closed, then close"
	closed bool
}

func NewPool(n, queueSize int) *Pool {
	p := &Pool{tasks: make(chan Task, queueSize)}
	p.wg.Add(n)
	for i := 0; i < n; i++ {
		go func() {
			defer p.wg.Done()
			for t := range p.tasks { // ends only when tasks is closed AND drained
				t()
			}
		}()
	}
	return p
}

func (p *Pool) Submit(t Task) error {
	p.mu.RLock()
	defer p.mu.RUnlock()
	if p.closed {
		return ErrClosed
	}
	p.tasks <- t
	return nil
}

func (p *Pool) TrySubmit(t Task) error {
	p.mu.RLock()
	defer p.mu.RUnlock()
	if p.closed {
		return ErrClosed
	}
	select {
	case p.tasks <- t:
		return nil
	default:
		return ErrQueueFull
	}
}

func (p *Pool) Shutdown() {
	p.mu.Lock()
	if !p.closed {
		p.closed = true
		close(p.tasks)
	}
	p.mu.Unlock()
	p.wg.Wait()
}

func main() {
	p := NewPool(2, 2)
	var done atomic.Int32
	release := make(chan struct{})
	slow := func() { <-release; done.Add(1) }   // blocks until `release` is closed

	// 2 workers pick up 2 tasks, 2 more fill the queue; the 5th is rejected.
	for i := 1; i <= 5; i++ {
		time.Sleep(20 * time.Millisecond) // let a worker dequeue before the next submit
		fmt.Printf("task %d: %v\n", i, p.TrySubmit(slow))
	}
	go func() { time.Sleep(100 * time.Millisecond); close(release) }()
	p.Shutdown()
	fmt.Println("completed after shutdown:", done.Load())
	fmt.Println("submit after shutdown:", p.Submit(slow))
}

The program uses atomic.Int32 and atomic.Int64, which need Go 1.19 or later. Run with go run -race in a golang:1.23 container, it prints:

task 1: <nil>
task 2: <nil>
task 3: <nil>
task 4: <nil>
task 5: queue full
completed after shutdown: 4
submit after shutdown: pool closed

Two workers hold tasks 1 and 2 (blocked on release), tasks 3 and 4 fill the capacity-2 queue, task 5 is refused. The 20 ms sleeps are only there so the first two tasks are dequeued before the next submit; they make the demo's output stable and are not part of the pool. After release closes, all four accepted tasks finish before Shutdown returns, and a later submit gets ErrClosed.

A stress test checks the shutdown race itself: 8 producer goroutines call Submit 50 times each while the main goroutine calls Shutdown concurrently, repeated 200 times. It lives in pool_test.go next to the pool:

package main

import (
	"sync"
	"sync/atomic"
	"testing"
)

func TestShutdownRacesWithSubmit(t *testing.T) {
	for iter := 0; iter < 200; iter++ {
		p := NewPool(4, 8)
		var ran, accepted atomic.Int64
		var producers sync.WaitGroup
		for g := 0; g < 8; g++ {
			producers.Add(1)
			go func() {
				defer producers.Done()
				for i := 0; i < 50; i++ {
					if err := p.Submit(func() { ran.Add(1) }); err == nil {
						accepted.Add(1)
					}
				}
			}()
		}
		p.Shutdown() // runs while producers are still submitting
		producers.Wait()
		if ran.Load() != accepted.Load() {
			t.Fatalf("iteration %d: ran %d tasks but %d submits succeeded", iter, ran.Load(), accepted.Load())
		}
	}
}

Run with go test -race -count=1 in the golang:1.23 container, on one CPU and on several, it prints ok with no race reported. To confirm the test can fail, delete the RLock/RUnlock pair from Submit and TrySubmit: the same test then reports a data race. That equality is the graceful-shutdown property: accepted means it runs. (The check runs after producers.Wait(), so it does not by itself notice a Shutdown that returns before the workers finish; the program in main above does, because it prints the completed count immediately after Shutdown returns.) (After Shutdown starts, later Submit calls return ErrClosed and are not counted as accepted.) The equality check also catches a Shutdown that throws away queued tasks, but only when several CPUs let producers and workers interleave: on a single CPU such a bug can pass unnoticed, so run this test with more than one CPU available.

The TrySubmit behaviour described above is checked by a second test in the same package, in its own file such as wait_test.go:

package main

import (
	"testing"
	"time"
)

// A TrySubmit that starts while Shutdown is waiting for the write lock queues behind it.
func TestTrySubmitWaitsBehindShutdown(t *testing.T) {
	p := NewPool(1, 1)
	release := make(chan struct{})
	block := func() { <-release }

	p.Submit(block) // the single worker takes it and blocks
	time.Sleep(20 * time.Millisecond)
	p.Submit(block)    // fills the queue of one
	go p.Submit(block) // parks: the queue is full
	time.Sleep(20 * time.Millisecond)
	go p.Shutdown() // waits for the write lock, which waits for the parked Submit
	time.Sleep(20 * time.Millisecond)

	result := make(chan error, 1)
	go func() { result <- p.TrySubmit(block) }()
	select {
	case err := <-result:
		t.Fatalf("TrySubmit returned %v while Shutdown was pending", err)
	case <-time.After(200 * time.Millisecond):
		// still waiting, as described
	}
	close(release) // the worker drains the queue, the parked Submit finishes, Shutdown proceeds
	if err := <-result; err != ErrClosed {
		t.Fatalf("after release TrySubmit returned %v, want ErrClosed", err)
	}
}

Run with go test -race -count=1 -run TestTrySubmitWaitsBehindShutdown, it passes on one CPU and on many. If TrySubmit skipped the read lock, the test fails with "TrySubmit returned queue full while Shutdown was pending", which confirms the test can fail.

Design choices and what would change them

• Block or reject? Blocking (Submit) suits an internal pipeline where slowing the producer is the desired effect. Rejecting (TrySubmit) suits an HTTP handler, where you would rather return 429 or 503 quickly than hold a connection. Recommend exposing both and letting the caller choose; for a request path default to TrySubmit, and stop routing traffic to the instance (or stop calling Submit from other paths) before calling Shutdown, because a TrySubmit that arrives while Shutdown is waiting on a parked Submit waits too.
• Queue size. The buffer only absorbs bursts. A queue of 1,000 with slow tasks just delays the 503 and adds latency; size it to a short burst (a small multiple of N), not to your memory.
• Shutdown can wait forever if a task never returns. In production add a deadline: run Shutdown in a goroutine and select on its completion channel versus ctx.Done(), and pass a context.Context into tasks so they can be told to stop. Without it, one stuck task hangs a deploy.
• A panic in a task kills the whole process, because it unwinds a worker goroutine with no recover. Wrap t() in a function with defer recover() and log, unless crashing is what you want.
• Blocked Submit calls during Shutdown are allowed to complete: Shutdown waits for the write lock, which waits for the blocked senders, who unblock as workers drain. That is correct for "finish what was accepted", but it means shutdown latency includes those producers, and any TrySubmit that starts in that window waits behind Shutdown as well.

Direct answer

Structured concurrency means every concurrent child task lives inside a block of code that waits for it: when the block exits, all children have finished, failed or been cancelled, and a failure in one child cancels its siblings and surfaces in the parent. Nested callbacks have no such owner, so a failure in one branch leaves other branches running, errors land in whichever callback happens to run, and resources leak. The refactor is: wrap each legacy callback API as an awaitable, put one scope (Python asyncio.TaskGroup, Java StructuredTaskScope) at each request boundary, move one request path at a time, give the whole request one deadline, and prove with a leak test that nothing outlives the request.

Terms used here: a callback is a function you hand to an API to be called later; cancellation means asking running work to stop (Python raises CancelledError in the task, Java interrupts the thread); an ExceptionGroup (Python 3.11+) is one exception object that carries several failures at once, and except* ("except star") is the matching syntax: except* RuntimeError as eg catches the RuntimeErrors inside a group, and eg.exceptions lists them; a future is an object that stands for a result that does not exist yet and is filled in later by set_result or set_exception, which await then returns or raises; a virtual thread is a cheap JVM thread managed by the runtime, which is what makes one-thread-per-subtask affordable. asyncio.TaskGroup, asyncio.timeout and except* all need Python 3.11 or newer; the Python blocks below were run on 3.12.

What goes wrong with callbacks

The Python run below starts two calls from a legacy callback style: orders fails after 0.1 s, user would finish after 0.4 s. The handler gives up once it sees the error, and the callback for user still runs later and mutates state nobody owns. The second half does the same with a TaskGroup.

import asyncio, threading, time

# ---- legacy style: callbacks, no owner for the work ----
def fetch_cb(name, delay, ok, on_done, on_error):
    def work():
        time.sleep(delay)
        (on_error if not ok else on_done)(name)
    threading.Thread(target=work, daemon=True).start()

def legacy_handler():
    state = {"results": [], "failed": False}
    def done(n): state["results"].append(n); print(f"  legacy: {n} finished")
    def error(n): state["failed"] = True; print(f"  legacy: {n} FAILED, returning an error")
    fetch_cb("user", 0.4, True, done, error)
    fetch_cb("orders", 0.1, False, done, error)
    time.sleep(0.2)                    # handler "returns" once it sees the first error
    return state

# ---- structured style: the scope owns every child ----
async def call(name, delay, ok=True):
    try:
        await asyncio.sleep(delay)
        if not ok:
            raise RuntimeError(f"{name} failed")
        return name
    except asyncio.CancelledError:
        print(f"  structured: {name} cancelled before the scope exited")
        raise

async def structured_handler():
    async with asyncio.TaskGroup() as tg:
        user = tg.create_task(call("user", 0.4))
        orders = tg.create_task(call("orders", 0.1, ok=False))
    return user.result(), orders.result()

print("legacy:")
s = legacy_handler()
print("  handler returned; failed =", s["failed"], "; results so far =", s["results"])
time.sleep(0.4)
print("  later, a leaked callback ran: results =", s["results"])

print("structured:")
try:
    asyncio.run(structured_handler())
except* RuntimeError as eg:
    print("  caught:", [str(e) for e in eg.exceptions])
print("  nothing is still running after the scope exits")

Run in a python:3.12-slim container (Python 3.12.15), it printed:

legacy:
  legacy: orders FAILED, returning an error
  handler returned; failed = True ; results so far = []
  legacy: user finished
  later, a leaked callback ran: results = ['user']
structured:
  structured: user cancelled before the scope exited
  caught: ['orders failed']
  nothing is still running after the scope exits

The legacy handler returned while user was still working (the leak), and its callback wrote into state after the request was over. In the structured version the failure of orders cancelled user before the async with block ended, the error came out as one ExceptionGroup (caught with except*), and no work survived the block.

The Java version

Java's counterpart is StructuredTaskScope. It is a preview API in JDK 21, so it needs --enable-preview. It remains a preview API in JDK 25 (JEP 505, fifth preview) and JDK 26 (JEP 525, sixth preview), and the API has changed between previews: JDK 25 and later open a scope with StructuredTaskScope.open(...) and a Joiner instead of the new StructuredTaskScope.ShutdownOnFailure() constructor used here. Check the JDK version before copying code. The concepts carry across versions: a scope you fork subtasks into, a join that waits for them, failure of one cancelling the rest, and closing the scope guaranteeing nothing survives. The code below is the JDK 21 form; on JDK 25 and later the same structure is written with open(...) and a Joiner such as one that fails fast, instead of the ShutdownOnFailure subclass. (A JEP is a JDK Enhancement Proposal, the document that defines a JDK feature; a preview API is a finished-but-not-final API that must be switched on with --enable-preview and may change.)

import java.time.Duration;
import java.util.concurrent.*;
import java.util.concurrent.StructuredTaskScope.Subtask;

public class Fanout {
    static String call(String name, long millis, boolean ok) throws InterruptedException {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            System.out.println("  " + name + " interrupted (cancelled by the scope)");
            throw e;
        }
        if (!ok) throw new IllegalStateException(name + " failed");
        return name;
    }

    public static void main(String[] args) throws Exception {
        try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
            Subtask<String> user   = scope.fork(() -> call("user", 400, true));
            Subtask<String> orders = scope.fork(() -> call("orders", 100, false));
            scope.join().throwIfFailed();
            System.out.println(user.get() + " " + orders.get());
        } catch (ExecutionException e) {
            System.out.println("caught: " + e.getCause());
        }
        System.out.println("scope closed: no subtask outlives the try block");

        try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
            Subtask<String> a = scope.fork(() -> call("a", 100, true));
            Subtask<String> b = scope.fork(() -> call("b", 150, true));
            scope.joinUntil(java.time.Instant.now().plus(Duration.ofSeconds(2))).throwIfFailed();
            System.out.println("both ok: " + a.get() + b.get());
        }
    }
}

Compiled with javac --release 21 --enable-preview Fanout.java and run with java --enable-preview Fanout on Temurin 21.0.12.1, it printed:

  user interrupted (cancelled by the scope)
caught: java.lang.IllegalStateException: orders failed
scope closed: no subtask outlives the try block
both ok: ab

ShutdownOnFailure interrupted user as soon as orders failed, join().throwIfFailed() rethrew the first failure wrapped in ExecutionException, and leaving the try block guaranteed every subtask had ended. joinUntil adds a deadline for the whole group. Each fork runs on a new virtual thread by default.

A refactoring plan

1. Inventory. List every callback entry point with its lifetime owner: who starts it, who should stop it, what resource it holds (connection, file, lock, timer). The result is a table; rows with no owner are the leaks. An illustrative inventory:

The first two rows are the leaks to fix first.
2. Characterization tests first. (A characterization test records what the code does today, right or wrong, so you can tell later whether the refactor changed behaviour.) For each request path, write a test that records today's observable result and a leak check: thread or task count and open connections before and after, including a client that disconnects mid-request. Run it against the legacy code so you know the baseline.
3. Wrap, do not rewrite. Turn each legacy callback API into an awaitable (Python: a future completed with loop.call_soon_threadsafe; Java: a CompletableFuture or a blocking call on a virtual thread). The adapter below is the Python one, run end to end. Read it in three parts. fut = loop.create_future() makes an empty placeholder for the result. The two lambdas are the callbacks handed to the legacy API; they run on the legacy thread, and asyncio objects may only be touched from the loop's thread, so each calls loop.call_soon_threadsafe(...), which schedules a function to run on the loop's thread. That function sets the future's result or exception, guarded by fut.done() or ..., which does nothing if the future is already finished (for example already cancelled), because a or b only evaluates b when a is false. Finally await fut suspends the task until a callback fills the placeholder:

import asyncio, threading, time

# Legacy API we cannot change: runs on its own thread and reports through callbacks.
def legacy_fetch(name, delay, ok, on_done, on_error):
    def work():
        time.sleep(delay)
        on_done(name) if ok else on_error(RuntimeError(f"{name} failed"))
    threading.Thread(target=work, daemon=True).start()

# Adapter: turn the callback pair into an awaitable, so a scope can own it.
async def fetch(name, delay, ok=True):
    loop = asyncio.get_running_loop()
    fut = loop.create_future()
    # callbacks run on another thread, so hand results to the loop thread-safely
    legacy_fetch(name, delay, ok,
                 lambda r: loop.call_soon_threadsafe(lambda: fut.done() or fut.set_result(r)),
                 lambda e: loop.call_soon_threadsafe(lambda: fut.done() or fut.set_exception(e)))
    return await fut           # cancelling the awaiting task cancels the future

async def handler(orders_ok=True, deadline=1.0):
    result = None
    try:
        async with asyncio.timeout(deadline):          # one deadline for the whole request
            try:
                async with asyncio.TaskGroup() as tg:
                    user = tg.create_task(fetch("user", 0.2))
                    orders = tg.create_task(fetch("orders", 0.1, ok=orders_ok))
                result = {"user": user.result(), "orders": orders.result()}
            except* RuntimeError as eg:                # a group of failures; return is not allowed here
                result = {"error": [str(e) for e in eg.exceptions]}
    except TimeoutError:
        result = {"error": "deadline exceeded"}
    return result

print(asyncio.run(handler()))
print(asyncio.run(handler(orders_ok=False)))
print(asyncio.run(handler(deadline=0.05)))

Run in the same container, it printed:

{'user': 'user', 'orders': 'orders'}
{'error': ['orders failed']}
{'error': 'deadline exceeded'}

The three lines are the success path, a child failure, and a 0.05 s deadline that fires before the 0.2 s user call returns. return is not allowed inside an except* block (Python rejects it at compile time), so the handler assigns a variable and returns after the block.

Cancellation reaches only the awaitable, not the legacy thread. When the 0.05 s deadline fires, fut is cancelled and the guarded callbacks discard the late result, but the legacy work thread keeps running until it returns on its own; in a program whose loop has already closed, its call_soon_threadsafe call raises RuntimeError: Event loop is closed inside that thread. If the legacy API has a cancel or close call, register it with fut.add_done_callback so cancellation reaches the real work; if it has none, the thread count only returns to baseline once the legacy call finishes.
4. Introduce one scope per request boundary. Start at the outermost handler and replace the nesting inside it with a TaskGroup or StructuredTaskScope block. Convert one request path per change, so each can be reviewed, deployed and reverted alone.
5. One deadline per request. Use asyncio.timeout (Python) or joinUntil (Java) at the boundary instead of a timer per callback, so cancellation reaches every child at once.
6. Error mapping. Decide once how a group of failures becomes an HTTP status and log line (for example: first failure sets the status, all are logged). Do not catch CancelledError / InterruptedException without re-raising.
7. Resource lifetime. Put connections and files in async with / try-with-resources inside the scope, so cleanup runs when a sibling fails.
8. Verify and roll out. Re-run the leak test from step 2 against the new path (task and thread counts back to baseline, for wrapped legacy calls once those calls return as step 3 explains, and no result delivered after the response). Roll out behind a flag at a small percentage of traffic, compare error rate, p99 latency (the latency that 99 percent of requests come in under, so the slowest 1 percent are above it) and open-connection count with the legacy path, then remove the legacy path.

Not every callback becomes a child task. A truly long-lived background job (a metrics flusher, a consumer loop) belongs to a scope that matches the service's lifetime, started at startup and cancelled at shutdown, not to a request.