import itertools
import threading

class LockCounter:
"""Protects an int with a threading.Lock — the direct equivalent of
Go's sync.Mutex. Correct regardless of what CPython's GIL does or
doesn't guarantee, and the only safe choice once a critical section
spans more than one statement."""
def __init__(self):
    self._lock = threading.Lock()
    self._value = 0

def inc(self):
    with self._lock:
        self._value += 1

def value(self):
    with self._lock:
        return self._value

class NaiveCounter:
"""BROKEN — and NOT something to rely on even under the classic GIL,
let alone free-threaded Python 3.13+ (--disable-gil). self._value += 1
compiles to three bytecodes (load, add, store). The GIL can switch
threads between any of them, so two threads can both read the same old
value before either writes back — a lost update. This is the race
both sync.Mutex and sync/atomic prevent in Go; Python's GIL does NOT
prevent it, because the GIL only guarantees one bytecode runs at a
time, not one statement."""
def __init__(self):
    self._value = 0

def inc(self):
    self._value += 1  # NOT atomic — classic lost-update race

def value(self):
    return self._value

class GilAtomicCounter:
"""The closest Python gets to Go's lock-free sync/atomic — not by
making += 1 atomic (nothing does that without a lock), but by using
an operation CPython itself implements as one C-level, GIL-uninterrupted
step: itertools.count().next(). This is the honest answer to "does
Python have an atomic increment?" — there is no general-purpose,
lock-free, single-value atomic primitive at the language level the way
sync/atomic is in Go. A handful of individual C-implemented operations
(this one, list.append, dict.setitem) happen to be safe under the
GIL; arbitrary compound operations like += on a plain int are not."""
def __init__(self):
    self._counter = itertools.count()
    self._last = 0

def inc(self):
    self._last = next(self._counter)  # atomic: one C-level step

def value(self):
    return self._last</code></pre>
  <p><strong>Benchmark comparison</strong></p>
  <pre><code class="language-python">import threading

import time
def bench(counter, n_threads=50, increments=20000):
threads = []
def worker():
    for _ in range(increments):
        counter.inc()

start = time.perf_counter()
for _ in range(n_threads):
    t = threading.Thread(target=worker)
    threads.append(t)
    t.start()
for t in threads:
    t.join()
elapsed = time.perf_counter() - start

expected = n_threads * increments
print(f"{type(counter).__name__}: {elapsed:.3f}s "
      f"value={counter.value()} expected={expected}")

bench(LockCounter())        # value == expected, every time
bench(NaiveCounter())       # value < expected — lost updates, reproducibly
bench(GilAtomicCounter())   # value == expected, but see the note below
GIL honesty check: this benchmark measures CORRECTNESS, not speedup.
Unlike Go's benchmark (real OS threads, real parallel cores), CPython's
GIL means only one thread executes bytecode at a time — these 50
threads take turns, they don't run concurrently on 50 cores. LockCounter
will often look no slower than GilAtomicCounter here, sometimes faster,
because lock acquisition inside the GIL is cheap when there's no true
hardware contention to begin with. To see a real throughput difference
in Python you need multiprocessing (separate processes, separate GILs)
or free-threaded (--disable-gil) builds — not threading.

</div>
<div class="tab-panel" data-tab-panel="counter-java">
  <pre><code class="language-java">package concurrency;

import java.util.concurrent.atomic.AtomicLong; // SynchronizedCounter protects a long with the built-in monitor lock — the // direct equivalent of Go's sync.Mutex. class SynchronizedCounter { private long value; public synchronized void inc() { value++; } public synchronized long value() { return value; } } // AtomicCounter uses AtomicLong — no lock, just a CPU-level atomic // instruction (CAS or fetch-and-add depending on architecture), exactly // like Go's sync/atomic. Unlike Python, where the honest answer is "no // general-purpose lock-free atomic exists, only GIL-protected C-level // ops," Java's java.util.concurrent.atomic types are genuinely // hardware-atomic — real CAS/fetch-and-add instructions, real parallel // threads across real cores, the same guarantee sync/atomic gives Go. class AtomicCounter { private final AtomicLong value = new AtomicLong(); public void inc() { value.incrementAndGet(); } public long value() { return value.get(); } }

Benchmark comparison

package concurrency;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
public class CounterBenchmark {
static void bench(String name, Runnable inc, java.util.function.LongSupplier value)
throws InterruptedException {
int nThreads = 50;
int increments = 20000;
CountDownLatch latch = new CountDownLatch(nThreads);
long start = System.nanoTime();
List<Thread> threads = new ArrayList<>();
for (int i = 0; i < nThreads; i++) {
Thread t = new Thread(() -> {
for (int j = 0; j < increments; j++) {
inc.run();
}
latch.countDown();
});
threads.add(t);
t.start();
}
latch.await();
double elapsed = (System.nanoTime() - start) / 1e9;
long expected = (long) nThreads * increments;
System.out.printf("%s: %.3fs value=%d expected=%d%n",
name, elapsed, value.getAsLong(), expected);
}
public static void main(String[] args) throws InterruptedException {
SynchronizedCounter sc = new SynchronizedCounter();
bench("SynchronizedCounter", sc::inc, sc::value);
AtomicCounter ac = new AtomicCounter();
bench("AtomicCounter", ac::inc, ac::value);
// Unlike Python's GIL-serialized threading benchmark, these 50
// threads really do run in parallel across OS threads/cores here —
// AtomicCounter's lock-free increment can show a genuine
// throughput edge over SynchronizedCounter under real contention,
// not just a correctness difference like in the Python version.
}
}

import queue
import threading
from dataclasses import dataclass
from typing import Callable, Optional

@dataclass
class Job:
id: int
input: int
@dataclass
class Result:
job_id: int
output: int
err: Optional[Exception] = None
class WorkerPool:
"""Processes jobs with a fixed number of worker threads, bounding
resource usage regardless of how many jobs are submitted — a direct
port of the Go WorkerPool, using queue.Queue instead of channels.
One real API gap vs. Go: a channel has a built-in closed state that
range-over-channel detects automatically. queue.Queue has no close()
at all — the idiomatic Python fix is a sentinel value (None, below),
one per worker, so every worker's blocking get() eventually receives
its own "stop" signal instead of an implicit closed-channel event."""

def __init__(self, num_workers: int, queue_size: int):
    self.num_workers = num_workers
    self.jobs: "queue.Queue[Optional[Job]]" = queue.Queue(maxsize=queue_size)
    self.results: "queue.Queue[Result]" = queue.Queue(maxsize=queue_size)
    self._threads = []

def start(self, process: Callable[[Job], int]):
    """Launches the fixed worker threads. Call once before submit()."""
    for _ in range(self.num_workers):
        t = threading.Thread(target=self._worker, args=(process,))
        t.start()
        self._threads.append(t)

def _worker(self, process):
    while True:
        job = self.jobs.get()
        if job is None:  # sentinel: no more work, exit
            break
        try:
            output = process(job)
            self.results.put(Result(job.id, output))
        except Exception as e:  # noqa: BLE001 — mirror Go's err return
            self.results.put(Result(job.id, 0, e))

def submit(self, job: Job):
    """Blocks if the queue is full — backpressure, same as Go's
    buffered-channel send."""
    self.jobs.put(job)

def close(self):
    """Signals no more jobs, then waits for all in-flight jobs to
    finish. Sends one None sentinel per worker instead of Go's
    single close(p.jobs) call."""
    for _ in range(self.num_workers):
        self.jobs.put(None)
    for t in self._threads:
        t.join()

Idiomatic alternative for the common case: concurrent.futures.
ThreadPoolExecutor gives you the same bounded-worker-count guarantee
with far less boilerplate, at the cost of the explicit Job/Result
shapes above — reach for this first unless you need the custom
queue semantics (e.g. bounded backpressure on submission itself).
from concurrent.futures import ThreadPoolExecutor, as_completed
def process(job: Job) -> int:
return job.input * 2
def run_with_thread_pool_executor():
with ThreadPoolExecutor(max_workers=4) as pool:
futures = {pool.submit(process, Job(i, i)): i for i in range(20)}
for future in as_completed(futures):
result = future.result()  # consume as they complete, unordered

Usage example

pool = WorkerPool(4, 100)  # 4 workers, queue capacity 100
pool.start(lambda job: job.input * 2)

def submit_all():
for i in range(20):
pool.submit(Job(i, i))
pool.close()  # safe to call from a separate thread once all submits are done
threading.Thread(target=submit_all).start()
No closed-channel signal to range over, so the consumer has to know how
many results to expect — here, exactly 20 (one per submitted job).
for _ in range(20):
result = pool.results.get()
_ = result  # consume results as they complete, unordered across workers

package concurrency;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.function.Function;
record Job(int id, int input) {}
record Result(int jobId, int output, Exception err) {}
// WorkerPool processes jobs with a fixed number of worker threads, bounding
// resource usage regardless of how many jobs are submitted — a direct port
// of the Go/Python WorkerPool, using BlockingQueue instead of channels.
//
// Same API gap as Python's queue.Queue: a Go channel has a built-in closed
// state that range-over-channel detects automatically. BlockingQueue has no
// close() at all either, so the fix here is the same sentinel-value trick
// Python uses — one poison pill per worker.
class WorkerPool {
private final int numWorkers;
private final BlockingQueue<Job> jobs;
private final BlockingQueue<Result> results;
private final List<Thread> workers = new ArrayList<>();
private static final Job POISON_PILL = new Job(-1, -1);
public WorkerPool(int numWorkers, int queueSize) {
this.numWorkers = numWorkers;
this.jobs = new ArrayBlockingQueue<>(queueSize);
this.results = new ArrayBlockingQueue<>(queueSize);
}
// start launches the fixed worker threads. Call once before submit().
public void start(Function<Job, Integer> process) {
for (int i = 0; i < numWorkers; i++) {
Thread t = new Thread(() -> worker(process));
t.start();
workers.add(t);
}
}
private void worker(Function<Job, Integer> process) {
try {
while (true) {
Job job = jobs.take(); // blocks until a job or the poison pill arrives
if (job == POISON_PILL) break; // sentinel: no more work, exit
try {
int output = process.apply(job);
results.put(new Result(job.id(), output, null));
} catch (Exception e) {
results.put(new Result(job.id(), 0, e));
}
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
// submit enqueues a job. Blocks if the queue is full (backpressure).
public void submit(Job j) throws InterruptedException {
jobs.put(j);
}
// close signals no more jobs will be submitted, then waits for all
// in-flight jobs to finish.
public void close() throws InterruptedException {
for (int i = 0; i < numWorkers; i++) {
jobs.put(POISON_PILL);
}
for (Thread t : workers) {
t.join();
}
}
public BlockingQueue<Result> results() {
return results;
}
}
// Idiomatic alternative for the common case: ThreadPoolExecutor configured
// with a bounded queue and an explicit rejection policy gives you the same
// bounded-worker-count guarantee with far less boilerplate, at the cost of
// losing the explicit Job/Result shapes above — reach for this first unless
// you need the custom queue semantics (e.g. bounded backpressure on
// submission itself, which ThreadPoolExecutor's default rejection policy
// does NOT give you: AbortPolicy throws instead of blocking the caller).
class ExecutorAlternative {
static java.util.concurrent.ThreadPoolExecutor boundedExecutor(int numWorkers, int queueSize) {
return new java.util.concurrent.ThreadPoolExecutor(
numWorkers, numWorkers,
0L, java.util.concurrent.TimeUnit.MILLISECONDS,
new ArrayBlockingQueue<>(queueSize),
new java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy());
}
}

Usage example

WorkerPool pool = new WorkerPool(4, 100); // 4 workers, queue capacity 100
pool.start(job -> job.input() * 2);
Thread submitter = new Thread(() -> {
try {
for (int i = 0; i < 20; i++) {
pool.submit(new Job(i, i));
}
pool.close(); // safe to call from a separate thread once all submits are done
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
submitter.start();
// No closed-queue signal to range over, so the consumer has to know how
// many results to expect — here, exactly 20 (one per submitted job).
for (int i = 0; i < 20; i++) {
Result res = pool.results().take();
// consume results as they complete, unordered across workers
}
submitter.join();

import asyncio

class PubSub:
"""asyncio equivalent of the channel-based PubSub above. Each
subscriber gets its own bounded asyncio.Queue; publishing never blocks
on a slow subscriber past that subscriber's buffer — messages are
dropped past that point, exactly like the Go version's select/default.
Same close() gap as the worker pool: asyncio.Queue has no close()
either, so shutdown uses a sentinel value per subscriber instead of a
built-in closed state."""

def __init__(self):
    self._subs: dict[str, list[asyncio.Queue]] = {}
    self._closed = False

def subscribe(self, topic: str, buffer_size: int) -> asyncio.Queue:
    q = asyncio.Queue(maxsize=buffer_size)
    self._subs.setdefault(topic, []).append(q)
    return q

def publish(self, topic: str, msg: str):
    """Non-blocking per subscriber: put_nowait mirrors Go's
    select/default — if a subscriber's queue is full, the message is
    dropped for that subscriber rather than blocking the publisher."""
    if self._closed:
        return
    for q in self._subs.get(topic, []):
        try:
            q.put_nowait(msg)
        except asyncio.QueueFull:
            pass  # subscriber buffer full — drop, a production system
                  # would count/log this as a metric, same as the Go version

def close(self):
    self._closed = True
    for queues in self._subs.values():
        for q in queues:
            q.put_nowait(None)  # sentinel — see the class docstring

Simpler alternative when you don't need buffering or backpressure at
all: a plain dict of topic -> list of callbacks, invoked synchronously
inside publish(). No asyncio, no queues — but the publisher now blocks
on however long each subscriber's callback takes, with zero isolation
between a fast subscriber and a slow one. Fine for small, trusted,
fast handlers; wrong the moment one subscriber does I/O.
class CallbackPubSub:
def init(self):
self._subs: dict[str, list] = {}
def subscribe(self, topic: str, callback):
    self._subs.setdefault(topic, []).append(callback)

def publish(self, topic: str, msg: str):
    for callback in self._subs.get(topic, []):
        callback(msg)  # runs inline — a slow callback stalls every publish</code></pre>
  <p><strong>Usage example</strong></p>
  <pre><code class="language-python">async def main():
ps = PubSub()
sub1 = ps.subscribe("orders", 10)
sub2 = ps.subscribe("orders", 10)

async def consume(sub, name):
    while True:
        msg = await sub.get()
        if msg is None:  # sentinel from close()
            break
        _ = msg  # handle order event

task1 = asyncio.create_task(consume(sub1, "sub1"))
task2 = asyncio.create_task(consume(sub2, "sub2"))

ps.publish("orders", "order-123-created")
ps.close()
await asyncio.gather(task1, task2)

asyncio.run(main())

package concurrency;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
// BlockingQueue-based fan-out, chosen over Flow.SubmissionPublisher because
// it mirrors Go channel semantics far more directly: a bounded
// ArrayBlockingQueue per subscriber is the same shape as a buffered Go
// channel, offer() is a non-blocking send exactly like Go's
// select/default, and take() is a blocking receive exactly like ranging
// over a channel. SubmissionPublisher is push-based with its own
// reactive-streams backpressure protocol (subscribers call request(n),
// buffer-full behavior is a configurable BufferOverflowStrategy or an
// exception) — a genuinely different concurrency model, not a
// line-for-line match for "each subscriber owns a bounded buffer."
class PubSub {
private final Map<String, List<BlockingQueue<String>>> subs = new HashMap<>();
private boolean closed = false;
private static final String POISON_PILL = "\0__CLOSE__";
// subscribe returns a queue that receives all messages published to
// topic. bufferSize controls how many messages can queue before
// publish drops them for this slow subscriber.
public synchronized BlockingQueue<String> subscribe(String topic, int bufferSize) {
BlockingQueue<String> q = new ArrayBlockingQueue<>(bufferSize);
subs.computeIfAbsent(topic, k -> new ArrayList<>()).add(q);
return q;
}
// publish sends msg to every subscriber of topic. Non-blocking per
// subscriber: if a subscriber's buffer is full, that message is
// dropped for that subscriber rather than blocking the publisher.
public synchronized void publish(String topic, String msg) {
if (closed) return;
for (BlockingQueue<String> q : subs.getOrDefault(topic, List.of())) {
if (!q.offer(msg)) {
// subscriber buffer full — drop rather than block the
// publisher. A production system would count/log this.
}
}
}
// close shuts down the broker, pushing a poison pill to every
// subscriber queue so consumers terminate — same sentinel trick as
// the worker pool, since BlockingQueue has no closed state either.
public synchronized void close() {
closed = true;
for (List<BlockingQueue<String>> queues : subs.values()) {
for (BlockingQueue<String> q : queues) {
q.offer(POISON_PILL);
}
}
}
static boolean isPoisonPill(String msg) {
return POISON_PILL.equals(msg);
}
}

Usage example

PubSub ps = new PubSub();
BlockingQueue<String> sub1 = ps.subscribe("orders", 10);
BlockingQueue<String> sub2 = ps.subscribe("orders", 10);
new Thread(() -> consumeOrders(sub1)).start();
new Thread(() -> consumeOrders(sub2)).start();
ps.publish("orders", "order-123-created");
ps.close();
static void consumeOrders(BlockingQueue<String> sub) {
try {
while (true) {
String msg = sub.take();
if (PubSub.isPoisonPill(msg)) break; // sentinel from close()
handleOrderEvent(msg); // handle order event
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}

import threading
import time

def debounce(wait: float, fn):
"""Delays invoking fn until wait seconds have elapsed since the
last call — repeated calls within the window reset the timer, only
the final call in a burst actually fires fn. Built on
threading.Timer instead of Go's time.AfterFunc, same semantics."""
lock = threading.Lock()
timer = None
def debounced(*args, **kwargs):
    nonlocal timer
    with lock:
        if timer is not None:
            timer.cancel()
        timer = threading.Timer(wait, fn, args=args, kwargs=kwargs)
        timer.start()

return debounced

def throttle(interval: float, fn):
"""Invokes fn at most once per interval seconds, regardless of call
frequency. The first call in a window fires immediately; the rest are
dropped until the window elapses."""
lock = threading.Lock()
last_run = 0.0
def throttled(*args, **kwargs):
    nonlocal last_run
    with lock:
        now = time.monotonic()
        if now - last_run < interval:
            return  # still within the throttle window — drop this call
        last_run = now
        fn(*args, **kwargs)

return throttled</code></pre>
  <p><strong>Usage example</strong></p>
  <pre><code class="language-python">def on_search_fire():
pass  # e.g., fires the search query only after typing pauses for 300ms

debounced_search = debounce(0.3, on_search_fire)
for keystroke in ["g", "go", "gol", "gola", "golang"]:
_ = keystroke
debounced_search()  # only the last call actually executes fn, ~300ms after it
def flush_metrics():
pass  # e.g., flush a metrics buffer at most once per second even under
# a tight loop calling this every microsecond
throttled_flush = throttle(1.0, flush_metrics)
for _ in range(1000):
throttled_flush()  # only fires roughly once per second across the whole loop

package concurrency;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
class DebounceThrottle {
// Debounce: delays invoking task until waitMs has elapsed since the
// last call. Repeated calls within the window cancel the pending
// schedule and reschedule — only the final call in a burst actually
// fires. Built on ScheduledExecutorService instead of Go's
// time.AfterFunc, same cancel-and-reschedule semantics.
static Runnable debounce(ScheduledExecutorService scheduler, long waitMs, Runnable task) {
Object lock = new Object();
ScheduledFuture<?>[] pending = new ScheduledFuture<?>[1];
return () -> {
synchronized (lock) {
if (pending[0] != null) {
pending[0].cancel(false);
}
pending[0] = scheduler.schedule(task, waitMs, TimeUnit.MILLISECONDS);
}
};
}
// Throttle: invokes task at most once per intervalMs, regardless of
// call frequency, using a nanoTime rate-gate rather than a scheduled
// timer — the first call in a window fires immediately; subsequent
// calls within the window are dropped.
static Runnable throttle(long intervalMs, Runnable task) {
long intervalNanos = TimeUnit.MILLISECONDS.toNanos(intervalMs);
Object lock = new Object();
long[] lastRun = new long[] { Long.MIN_VALUE / 2 };
return () -> {
synchronized (lock) {
long now = System.nanoTime();
if (now - lastRun[0] < intervalNanos) {
return; // still within the throttle window — drop this call
}
lastRun[0] = now;
}
task.run();
};
}
}

Usage example

ScheduledExecutorService scheduler = java.util.concurrent.Executors.newScheduledThreadPool(1);
Runnable debouncedSearch = DebounceThrottle.debounce(scheduler, 300, () -> {
// e.g., fires the search query only after typing pauses for 300ms
});
for (String keystroke : new String[] { "g", "go", "gol", "gola", "golang" }) {
debouncedSearch.run(); // only the last call actually executes fn, ~300ms after it
}
Runnable throttledFlush = DebounceThrottle.throttle(1000, () -> {
// e.g., flush a metrics buffer at most once per second even under
// a tight loop calling this every microsecond
});
for (int i = 0; i < 1000; i++) {
throttledFlush.run(); // only fires roughly once per second across the whole loop
}

The same burst under a 1s throttle window: the very first call fires immediately, then every call until the window elapses is dropped — fn runs on a predictable cadence instead of waiting for quiet.

graph LR
    A["t=0ms call, fn fires"] --> B["t=50ms call, dropped"]
    B --> C["t=300ms call, dropped"]
    C --> D["t=600ms call, dropped"]
    D --> E["t=1000ms call, fn fires again"]
    E --> F["t=1050ms call, dropped"]

import threading

def fetch_with_timeout_buggy(timeout: float, work):
"""BUGGY: leaves an unjoined, untracked thread running on every call
where the caller times out before work() finishes — Python's
equivalent of the leaked goroutine. The failure mechanism is
different (no blocked channel send here; Event/dict just aren't read
again), but the outcome is the same class of bug: a background worker
nobody is tracking or waiting for."""
done = threading.Event()
box = {}
def runner():
    box["result"] = work()
    done.set()  # nobody may ever check this again

threading.Thread(target=runner).start()  # not joined, not tracked

if done.wait(timeout):
    return box["result"], None
return None, TimeoutError("timed out")
# The thread above is still running: it will eventually finish work()
# and call done.set(), but nothing will ever read `done` or `box`
# again. Worse than Go here: threading.Thread defaults to non-daemon,
# so a leaked thread that never finishes actually keeps the whole
# Python process alive past the point where main() returns and tries
# to exit — Go's runtime has no such rule, the process exits when
# main() returns regardless of live goroutines.</code></pre>
</div>
<div class="tab-panel" data-tab-panel="leak-buggy-java">
  <pre><code class="language-java">package concurrency;

import java.util.concurrent.Callable;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
class LeakyFetch {
// BUGGY: leaks a non-daemon thread on every call where the caller
// times out before the worker finishes and hands off its result.
// SynchronousQueue is a zero-capacity rendezvous — put() blocks until
// another thread take()s/poll()s, exactly mirroring Go's unbuffered
// channel send semantics.
static int fetchWithTimeoutBuggy(long timeoutMs, Callable<Integer> work) throws Exception {
SynchronousQueue<Integer> resultCh = new SynchronousQueue<>();
Thread t = new Thread(() -> {
try {
int result = work.call();
resultCh.put(result); // BLOCKS FOREVER if nobody ever takes
} catch (Exception ignored) {
// work() itself failed, or the thread was interrupted
}
});
t.start(); // non-daemon by default — keeps the JVM alive if leaked
Integer result = resultCh.poll(timeoutMs, TimeUnit.MILLISECONDS);
if (result == null) {
throw new TimeoutException("timed out");
// The thread above is now leaked: it will eventually finish
// work(), then block forever on resultCh.put(result), because
// nothing will ever poll()/take() from resultCh again. Worse
// than Go here: Thread defaults to non-daemon, so a leaked
// thread that never finishes keeps the whole JVM process alive
// past the point where main() returns — Go's runtime has no
// such rule, the process exits when main() returns regardless
// of live goroutines.
}
return result;
}
}

import threading
import time

def test_leak_demonstration():
before = threading.active_count()
for _ in range(100):
    fetch_with_timeout_buggy(0.01, lambda: (time.sleep(0.05), 42)[1])

time.sleep(0.1)  # let any leaked threads finish their sleep
after = threading.active_count()

print(f"threads before={before} after={after}")
if after - before < 50:  # expect most of the 100 to still be leaked
    print("leak not reliably reproduced in this run — timing dependent, "
          "see fixed version below")</code></pre>
</div>
<div class="tab-panel" data-tab-panel="leak-test-java">
  <pre><code class="language-java">package concurrency;

public class LeakDemonstrationTest {
public static void main(String[] args) throws InterruptedException {
int before = Thread.activeCount();
for (int i = 0; i < 100; i++) {
try {
LeakyFetch.fetchWithTimeoutBuggy(10, () -> {
Thread.sleep(50); // always slower than the timeout
return 42;
});
} catch (Exception ignored) {
// expected: TimeoutException
}
}
Thread.sleep(100); // let any leaked threads finish their sleep
int after = Thread.activeCount();
System.out.println("threads before=" + before + " after=" + after);
if (after - before < 50) {
System.out.println("leak not reliably reproduced in this run — "
+ "timing dependent, see fixed version below");
}
// NOTE: unlike a Go test binary (exits regardless of leaked
// goroutines), this main() never actually returns control to the
// shell — the ~100 leaked, non-daemon threads above are still
// parked on resultCh.put(result) and keep the JVM alive forever.
// That hang IS the leak, demonstrated live rather than just
// logged; killing the process is the only way out, exactly the
// "worse than Go" case called out above.
}
}

from concurrent.futures import ThreadPoolExecutor
from concurrent.futures import TimeoutError as FutureTimeout

Bounded — caps how many threads can ever exist, unlike a fresh
threading.Thread() per call above.
_pool = ThreadPoolExecutor(max_workers=50)
def fetch_with_timeout_pooled(timeout: float, work):
"""FIXED (bounded, but NOT cancelled): a flood of timeouts can no
longer create unbounded threads, because the executor caps concurrent
threads at max_workers and reuses them — exactly like Go's
buffered-channel fix bounding memory. But exactly like that Go fix,
work() still runs to completion inside its pool thread even after
this function has already returned a timeout error to the caller —
the thread isn't leaked (the pool owns and will reuse it once free),
the wasted work is simply not cancelled."""
future = _pool.submit(work)
try:
return future.result(timeout=timeout), None
except FutureTimeout:
return None, TimeoutError("timed out")
# future.cancel() would return False here — Python can only
# cancel a Future that hasn't started running yet, never one
# that's already executing. There is no forced-stop for a
# running thread at all, unlike Go where at least the language
# doesn't stop you from trying (ctx cancellation below is the
# correct, cooperative way to actually do it).

package concurrency;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
class FixedFetch {
// FIXED: a buffered queue of capacity 1 means the worker thread's
// offer() never blocks, even if nobody ever reads the result. The
// thread always completes and exits, so it can be garbage collected —
// the same fix shape as Go's buffered channel.
static int fetchWithTimeoutFixed(long timeoutMs, Callable<Integer> work) throws Exception {
BlockingQueue<Integer> resultCh = new ArrayBlockingQueue<>(1);
Thread t = new Thread(() -> {
try {
int result = work.call();
resultCh.offer(result); // always succeeds immediately, buffer absorbs it
} catch (Exception ignored) {
}
});
t.start();
Integer result = resultCh.poll(timeoutMs, TimeUnit.MILLISECONDS);
if (result == null) {
throw new TimeoutException("timed out");
// Thread is NOT leaked: it will complete work(), offer into the
// buffer (succeeds instantly, capacity=1), then exit normally.
}
return result;
}
// Bounded — caps how many threads can ever exist, unlike a fresh
// Thread per call above. Mirrors Python's ThreadPoolExecutor fix.
private static final ExecutorService pool = Executors.newFixedThreadPool(50);
static int fetchWithTimeoutPooled(long timeoutMs, Callable<Integer> work) throws Exception {
Future<Integer> future = pool.submit(work);
try {
return future.get(timeoutMs, TimeUnit.MILLISECONDS);
} catch (TimeoutException e) {
future.cancel(false); // returns false here — a running task can't
// be force-stopped, only one not yet started
throw e;
}
}
}

import threading

def fetch_with_cancellation(timeout: float, work):
"""BETTER: propagate a cancellation signal into work() itself so
wasted CPU/IO actually stops — the same idea as Go's
context.Context, and just as cooperative: work() must check
cancel_event periodically on its own; nothing forces it to stop.
This cooperative check is not just the *preferred* way to cancel a
Python thread, it is the *only* way — Python has no equivalent of
Go goroutines being safely abandoned at a blocked channel op, nor any
Thread.stop(). An ignored threading.Event, unlike an ignored
ctx.Done() in a goroutine that's merely parked, keeps a real OS
thread burning CPU/GIL time indefinitely if work() never checks it."""
cancel_event = threading.Event()
result_box = {}

def runner():
    result_box["value"] = work(cancel_event)

t = threading.Thread(target=runner)
t.start()
t.join(timeout)

if t.is_alive():
    cancel_event.set()  # ask work() to stop — it must check this itself
    return None, TimeoutError("timed out")
return result_box.get("value"), None

asyncio angle: fire-and-forget tasks are the other common Python leak.
import asyncio
BUGGY: task reference is dropped as soon as this function returns, so
nobody can await or cancel it — it keeps running in the background for
the life of the event loop, and if it raises, the exception is silently
swallowed until garbage collection logs "Task exception was never
retrieved". This is the asyncio-flavored version of the same bug class.
async def fetch_buggy_asyncio(timeout, work):
asyncio.create_task(work())  # fire-and-forget — leaked, untracked
await asyncio.sleep(timeout)
return "timed out"
FIXED: asyncio.wait_for actually cancels the inner coroutine on
timeout — it delivers a real CancelledError into it. Unlike Go's
buffered-channel fix and unlike Python's own ThreadPoolExecutor,
asyncio genuinely can stop already-started work, not just avoid
leaking memory around it.
async def fetch_fixed_asyncio(timeout, work):
try:
return await asyncio.wait_for(work(), timeout=timeout)
except asyncio.TimeoutError:
return "timed out"  # work()'s task was already cancelled for us

package concurrency;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
interface InterruptibleWork {
int run() throws InterruptedException;
}
class CancellableFetch {
// BETTER: propagate cancellation into work() itself via
// Thread.interrupt() — Java's built-in cooperative cancellation
// signal — so wasted CPU/IO actually stops instead of just not
// leaking memory. work() must check Thread.interrupted() (or let a
// blocking call like Thread.sleep throw InterruptedException)
// periodically on its own; nothing forces it to stop. Same
// cooperative contract as Go's ctx.Done() and Python's
// threading.Event, though Java's version is a language-level signal
// rather than a hand-rolled flag — this cooperative check is not just
// the preferred way to cancel a Java thread, it is the only way:
// Java has no forced-stop for a running thread either (Thread.stop()
// exists but has been deprecated for removal since it can leave
// shared state corrupted mid-update).
static int fetchWithCancellation(long timeoutMs, InterruptibleWork work) throws Exception {
BlockingQueue<Object> resultCh = new ArrayBlockingQueue<>(1);
Thread t = new Thread(() -> {
try {
int result = work.run();
resultCh.offer(result);
} catch (InterruptedException e) {
// interrupted during cancellation — exit quietly
} catch (Exception e) {
resultCh.offer(e);
}
});
t.start();
Object result = resultCh.poll(timeoutMs, TimeUnit.MILLISECONDS);
if (result == null) {
t.interrupt(); // ask work() to stop — it must check this itself
throw new TimeoutException("timed out");
}
if (result instanceof Exception) {
throw (Exception) result;
}
return (Integer) result;
}
}
class ExecutorLeakExample {
// BUGGY: an ExecutorService created but never shutdown() keeps its
// non-daemon worker threads alive for the life of the JVM — the
// executor-flavored version of the same bug class. Unlike a single
// leaked Thread, this can keep dozens of pool threads (and the JVM
// process) alive indefinitely even after every submitted task
// finishes.
static final ExecutorService leakyPool = Executors.newFixedThreadPool(4); // never shutdown — leak
static void fetchBuggyExecutor(Runnable work) {
leakyPool.submit(work); // fire-and-forget, executor itself is never shutdown
}
// FIXED: shutdown() (or shutdownNow() for cancellation) releases the
// pool's threads once work completes, letting the JVM exit normally.
static int fetchFixedExecutor(long timeoutMs, java.util.concurrent.Callable<Integer> work) throws Exception {
ExecutorService pool = Executors.newFixedThreadPool(4);
try {
Future<Integer> f = pool.submit(work);
return f.get(timeoutMs, TimeUnit.MILLISECONDS);
} finally {
pool.shutdown(); // or shutdownNow() to interrupt in-flight tasks
}
}
}