Python GIL & Threading with Queue

The Story That Explains the GIL 🎤GIL समझाने वाली कहानी 🎤一个讲清 GIL 的故事 🎤

The GIL is a mutex (mutual exclusion lock) inside CPython. It stops several threads from running Python bytecode at the same time. This keeps CPython simple and fast for single-threaded programs, but it limits CPU-heavy multithreaded programs.

GIL CPython के अंदर एक mutex (mutual exclusion lock) है। यह कई threads को एक ही समय पर Python bytecode चलाने से रोकता है। इससे CPython single-threaded programs के लिए सरल और तेज़ रहता है, पर CPU-भारी multithreaded programs सीमित हो जाते हैं।

GIL 是 CPython 内部的一个 互斥锁（mutex）。它阻止多个线程同时执行 Python 字节码。这让 CPython 在单线程程序中保持简单、快速，但限制了 CPU 密集型的多线程程序。

What Exactly Is the GIL? 🔒GIL असल में क्या है? 🔒GIL 到底是什么？🔒

The Global Interpreter Lock is one process-wide lock inside the CPython interpreter. A thread must hold the GIL before it can run any Python bytecode. While one thread holds it, the others wait.

Global Interpreter Lock CPython interpreter के अंदर पूरे process का एक lock है। कोई भी Python bytecode चलाने से पहले thread के पास GIL होना चाहिए। जब एक thread इसे रखे है, बाक़ी इंतज़ार करते हैं।

全局解释器锁是 CPython 解释器中整个进程共用的一把锁。线程必须持有 GIL 才能执行任何 Python 字节码。一个线程持有它时，其他线程只能等待。

You can ask Python about its GIL and switch interval directly:

आप Python से उसके GIL और switch interval के बारे में सीधे पूछ सकते हैं:

你可以直接向 Python 查询它的 GIL 和切换间隔：

import threading, sys

print("GIL enabled?     ", sys._is_gil_enabled())      # Python 3.13+
print("switch interval: ", sys.getswitchinterval(), "s")
print("threads running: ", threading.active_count())

Why Does the GIL Exist? 🤔GIL क्यों है? 🤔为什么会有 GIL？🤔

The GIL was added in the early 1990s to solve one problem: memory-management safety. CPython frees objects using reference counting. Without a global lock, two threads could change an object's reference count at the same moment, corrupt memory and crash the interpreter.

GIL 1990 के दशक की शुरुआत में एक समस्या हल करने के लिए जोड़ा गया: memory-management की सुरक्षा। CPython objects को reference counting से मुक्त करता है। Global lock के बिना दो threads एक ही पल में किसी object का reference count बदल सकते थे, memory बिगाड़ सकते थे और interpreter crash कर सकते थे।

GIL 在 1990 年代初被加入，为了解决一个问题：内存管理的安全。CPython 用 引用计数 来释放对象。没有全局锁的话，两个线程可能同时修改一个对象的引用计数，破坏内存并让解释器崩溃。

Behind the Scenes — Threads and PyThreadState 🛠️पर्दे के पीछे — Threads और PyThreadState 🛠️幕后 — 线程与 PyThreadState 🛠️

When you call threading.Thread(target=…).start(), Python does surprisingly little. It creates a C structure called PyThreadState, starts a native OS thread (a POSIX thread on Linux/macOS, a Windows thread on Windows) and hands it your function. The operating system decides which core runs the thread and when — Python has no thread scheduler of its own. The GIL only decides which thread may run Python code.

जब आप threading.Thread(target=…).start() बुलाते हैं, Python हैरानी की हद तक कम करता है। यह PyThreadState नाम की एक C structure बनाता है, एक native OS thread शुरू करता है (Linux/macOS पर POSIX thread, Windows पर Windows thread) और उसे आपका function दे देता है। Thread कौन-सा core और कब चलाए, यह operating system तय करता है — Python का अपना कोई thread scheduler नहीं है। GIL सिर्फ़ तय करता है कि कौन-सा thread Python code चला सकता है।

当你调用 threading.Thread(target=…).start() 时，Python 做的事情少得惊人。它创建一个名为 PyThreadState 的 C 结构体，启动一个原生操作系统线程（Linux/macOS 上是 POSIX 线程，Windows 上是 Windows 线程），并把你的函数交给它。操作系统 决定线程何时在哪个核心上运行 — Python 自己没有线程调度器。GIL 只决定哪个线程可以运行 Python 代码。

Proof that they are real OS threads — each one gets its own operating-system thread id:

सबूत कि ये असली OS threads हैं — हर एक को अपना operating-system thread id मिलता है:

证明它们是真正的操作系统线程 — 每个线程都有自己的操作系统线程 ID：

import threading

def show():
    t = threading.current_thread()
    print(f"{t.name:10}  Python ident={t.ident % 100000:>5}  OS thread id={threading.get_native_id()}")

show()                                   # the main thread
ts = [threading.Thread(target=show) for _ in range(3)]
for t in ts: t.start()
for t in ts: t.join()

The PyThreadState structurePyThreadState structurePyThreadState 结构体

Each Python thread carries its own state: the current stack frame, recursion depth, the exception being handled, tracing hooks and a thread-local dictionary. Here is a simplified sketch based on older CPython sources:

हर Python thread अपनी state रखता है: current stack frame, recursion depth, handle हो रहा exception, tracing hooks और एक thread-local dictionary। यह पुराने CPython source पर आधारित एक सरल ढाँचा है:

每个 Python 线程都带着自己的状态：当前栈帧、递归深度、正在处理的异常、跟踪钩子和一个线程局部字典。下面是基于旧版 CPython 源码的 简化示意：

/* Include/pystate.h - simplified sketch, older CPython (2.x / early 3.x) */
typedef struct _ts {
    struct _ts          *next;
    PyInterpreterState  *interp;
    struct _frame       *frame;            // current stack frame
    int                  recursion_depth;
    int                  tracing;
    Py_tracefunc         c_profilefunc;
    Py_tracefunc         c_tracefunc;
    PyObject            *curexc_type;      // exception being raised
    PyObject            *curexc_value;
    PyObject            *curexc_traceback;
    PyObject            *dict;             // thread-local storage
    int                  tick_counter;     // Python 2 "check interval" ticks
    int                  gilstate_counter;
    PyObject            *async_exc;
    long                 thread_id;
} PyThreadState;

/* older CPython: points to the state of the thread holding the GIL */
PyThreadState *_PyThreadState_Current = NULL;

Ticks, Check Intervals and the Modern Switch ⏲️Ticks, Check Intervals और आधुनिक Switch ⏲️Tick、检查间隔与现代切换机制 ⏲️

How does Python take the GIL away from a CPU-heavy thread that never does I/O? It checks from time to time. The way it checks changed a lot between Python 2 and Python 3.2.

जो CPU-भारी thread कभी I/O नहीं करता, उससे Python GIL कैसे छीनता है? यह बीच-बीच में जाँच करता है। जाँच का तरीक़ा Python 2 और Python 3.2 के बीच बहुत बदला।

Python 如何从一个从不做 I/O 的 CPU 密集线程手中拿走 GIL？它会时不时地检查一下。检查的方式在 Python 2 和 Python 3.2 之间变化很大。

What is a "tick"?"Tick" क्या है?什么是"tick"？

In Python 2 a tick was roughly one virtual-machine instruction. You can see the instructions of any function with the dis module. Here is the real output on Python 3.13.16:

Python 2 में एक tick मोटे तौर पर एक virtual-machine instruction था। किसी भी function के instructions dis module से देख सकते हैं। यह Python 3.13.16 का असली output है:

在 Python 2 中，一个 tick 大致对应一条虚拟机指令。你可以用 dis 模块查看任意函数的指令。下面是 Python 3.13.16 上的真实输出：

import dis

def countdown(n):
    while n > 0:
        print(n)
        n -= 1

dis.dis(countdown)

The check itself — Python 2 ceval.c (simplified)जाँच ख़ुद — Python 2 ceval.c (सरल)检查本身 — Python 2 的 ceval.c（简化）

/* Python/ceval.c - Python 2 era, simplified */
if (--_Py_Ticker < 0) {
    _Py_Ticker = _Py_CheckInterval;          /* reset: 100 ticks */
    if (things_to_do) {
        if (Py_MakePendingCalls() < 0) { ... }   /* e.g. signal handlers */
    }
    if (interpreter_lock) {
        /* Give another thread a chance */
        PyThread_release_lock(interpreter_lock);
        /* Other threads may run now */
        PyThread_acquire_lock(interpreter_lock, 1);
    }
}

Look at the last two lines: release, then immediately acquire again. On a multicore machine the releasing thread often wins the race back — the root of the "GIL battle" in Section 09. Python 3.2 replaced this code.

आख़िरी दो lines देखिए: release, फिर तुरंत फिर से acquire। Multicore machine पर छोड़ने वाला thread अक्सर race फिर जीत जाता है — Section 09 के "GIL battle" की जड़ यही है। Python 3.2 ने यह code बदल दिया।

看最后两行：释放，然后 立刻 再次获取。在多核机器上，释放锁的线程常常又抢回了锁 — 这就是第 09 节"GIL 争夺战"的根源。Python 3.2 替换了这段代码。

⏱️ Simulator: GIL Timeline (real traces)⏱️ Simulator: GIL Timeline (असली traces)⏱️ 模拟器：GIL 时间线（真实追踪）

Each picture is a real 60 ms recording on Python 3.13.16. Every thread logged when it was running Python code. The top lane shows who held the GIL. Red blocks are an I/O thread that has finished waiting but still has no GIL. Change the workload and the switch interval.

हर तस्वीर Python 3.13.16 पर असली 60 ms की recording है। हर thread ने दर्ज किया कि कब वह Python code चला रहा था। सबसे ऊपर की lane दिखाती है कि GIL किसके पास था। लाल blocks वह I/O thread हैं जिसका इंतज़ार ख़त्म हो गया पर GIL अभी नहीं मिला। Workload और switch interval बदलिए।

每张图都是 Python 3.13.16 上 真实的 60 毫秒记录。每个线程都记录了自己何时在运行 Python 代码。最上面一行显示 GIL 在谁手里。红色块表示 I/O 线程已经等完了，却还没拿到 GIL。试着切换工作负载和切换间隔。

GIL Behaviour — Releasing It for I/O 🔄GIL का व्यवहार — I/O के लिए छोड़ना 🔄GIL 的行为 — 为 I/O 释放 🔄

For I/O-bound threads the rule is simple: a thread holds the GIL while it runs Python code, and releases it whenever it blocks for I/O. Another ready thread takes the GIL and runs.

I/O-bound threads के लिए नियम सरल है: thread Python code चलाते समय GIL रखता है, और I/O के लिए रुकते ही उसे छोड़ देता है। कोई दूसरा तैयार thread GIL लेकर चलता है।

对 I/O 密集型线程，规则很简单：线程运行 Python 代码时持有 GIL，一旦因 I/O 阻塞就释放它。另一个就绪的线程拿到 GIL 并运行。

CPU-Bound vs I/O-Bound — When the GIL Bites 📈CPU-Bound बनाम I/O-Bound — GIL कब काटता है 📈CPU 密集 vs I/O 密集 — GIL 何时"咬人" 📈

🏁 Simulator: Speed Lab (real benchmarks)🏁 Simulator: Speed Lab (असली benchmarks)🏁 模拟器：速度实验室（真实基准测试）

The same amount of work, done three ways on our 2-core machine: one after another, with threads, and with processes. Every bar is the middle of three real runs.

उतना ही काम, हमारी 2-core machine पर तीन तरह से: एक के बाद एक, threads से, और processes से। हर bar तीन असली runs का बीच वाला मान है।

同样的工作量，在我们的双核机器上用三种方式完成：依次执行、用线程、用进程。每个条形都是三次真实运行的中位数。

The Beazley Experiment — When 2 Threads Were Slower Than 1 🐢Beazley Experiment — जब 2 Threads 1 से धीमे थे 🐢Beazley 实验 — 当 2 个线程比 1 个还慢 🐢

Re-running the experiment todayआज यह experiment फिर चलाना今天重新运行这个实验

import time
from threading import Thread

def count(n):
    while n > 0:
        n -= 1

COUNT = 20_000_000          # Beazley used 100_000_000

# --- Sequential: count twice, one after the other ---
start = time.perf_counter()
count(COUNT)
count(COUNT)
print(f"Sequential : {time.perf_counter() - start:.2f}s")

# --- Two threads: each thread counts the full COUNT ---
start = time.perf_counter()
t1 = Thread(target=count, args=(COUNT,))
t2 = Thread(target=count, args=(COUNT,))
t1.start(); t2.start()
t1.join();  t2.join()
print(f"Threaded   : {time.perf_counter() - start:.2f}s")

The Multicore GIL Battle (Old GIL) ⚔️Multicore GIL Battle (पुराना GIL) ⚔️多核 GIL 争夺战（旧 GIL）⚔️

On a single core the OS runs one thread at a time, so the GIL is barely contested. On several cores the OS happily runs threads on different cores at the same moment, and with the old (Python 2) GIL they started fighting over the one lock.

एक core पर OS एक समय में एक thread चलाता है, इसलिए GIL के लिए मुश्किल से होड़ होती है। कई cores पर OS ख़ुशी-ख़ुशी threads को एक ही पल में अलग cores पर चलाता है, और पुराने (Python 2) GIL के साथ वे एक lock के लिए लड़ने लगते थे।

在单核上，操作系统一次只运行一个线程，所以 GIL 几乎没有竞争。在多核上，操作系统乐于把线程同时放到不同核心上运行，而在 旧（Python 2）GIL 下，它们就开始争抢这唯一的锁。

Beazley's instrumented trace (Python 2)Beazley का instrumented trace (Python 2)Beazley 的插桩追踪（Python 2）

# t1, t2 = thread ids. Middle number = ticks left. Right = total checks.
t2 100 5392 ENTRY
t2 100 5392 ACQUIRE
t2 100 5393 RELEASE
t1 100 5393 ACQUIRE     # thread switch - t1 wins
t2 100 5393 ENTRY
t2  27 5393 BUSY        # t2 tries to run but the GIL is taken by t1
t1 100 5394 RELEASE
t1 100 5394 ENTRY       # t1 releases and IMMEDIATELY re-enters
t1 100 5394 ACQUIRE     # t1 re-acquires before t2 can wake
t2  74 5394 RETRY       # t2 wakes up, fails, goes back to sleep
t1 100 5395 RELEASE
t1 100 5395 ENTRY
t1 100 5395 ACQUIRE     # t1 wins AGAIN
t2  83 5395 RETRY       # t2 fails AGAIN
...                     # hundreds of failed retries before t2 gets a turn

Signalling Overhead — Where the Time Went 📡Signalling Overhead — समय कहाँ गया 📡信号开销 — 时间都去哪了 📡

The GIL is not a plain mutex. On Unix it is built from a mutex plus a condition variable (older builds used a semaphore). Every hand-off involves signalling, and signalling costs system calls. Beazley counted them in 2009:

GIL एक साधारण mutex नहीं है। Unix पर यह एक mutex और एक condition variable से बना है (पुराने builds में semaphore)। हर hand-off में signalling होती है, और signalling की क़ीमत system calls है। Beazley ने 2009 में इन्हें गिना:

GIL 不是一个普通的互斥锁。在 Unix 上它由一个互斥锁加一个条件变量构成（较旧的版本用信号量）。每次交接都涉及 发信号，而发信号要付出系统调用的代价。Beazley 在 2009 年统计过：

Priority Inversion — When CPU Threads Slow Down I/O Threads 🚨Priority Inversion — जब CPU Threads I/O Threads को धीमा करें 🚨优先级反转 — 当 CPU 线程拖慢 I/O 线程 🚨

Now a mixed program: one thread crunches numbers, another waits on a socket. A packet arrives — the I/O thread wants to run now — but it must first get the GIL from the busy CPU thread.

अब एक मिला-जुला program: एक thread numbers crunch करता है, दूसरा socket पर इंतज़ार करता है। एक packet आता है — I/O thread अभी चलना चाहता है — पर उसे पहले व्यस्त CPU thread से GIL लेना होगा।

现在是一个混合程序：一个线程在做数值计算，另一个在套接字上等待。一个数据包到达 — I/O 线程想 立刻 运行 — 但它必须先从忙碌的 CPU 线程手里拿到 GIL。

Measuring it on Python 3.13.16Python 3.13.16 पर इसे मापना在 Python 3.13.16 上测量

import threading, time, statistics

def io_worker(results):
    for _ in range(100):
        t0 = time.perf_counter()
        time.sleep(0.001)                 # wait 1 ms for a "network packet"
        results.append((time.perf_counter() - t0) * 1000)

stop = False
def cpu_worker():
    n = 0
    while not stop:                       # pure Python number crunching
        n += 1

for busy in (False, True):
    stop, results = False, []
    if busy:
        c = threading.Thread(target=cpu_worker); c.start()
    t = threading.Thread(target=io_worker, args=(results,))
    t.start(); t.join()
    stop = True
    if busy: c.join()
    print(f"CPU thread running: {busy!s:5}  "
          f"avg wake-up {statistics.mean(results):.2f} ms  "
          f"worst {max(results):.2f} ms")

Signals, Ctrl-C and Threaded Programs ⌨️Signals, Ctrl-C और Threaded Programs ⌨️信号、Ctrl-C 与多线程程序 ⌨️

What really happens — a real run with two Ctrl-C pressesअसल में क्या होता है — दो Ctrl-C के साथ असली run实际情况 — 按两次 Ctrl-C 的真实运行

import threading, time

def worker():
    while True:                     # never stops on its own
        time.sleep(0.1)

t = threading.Thread(target=worker)
t.start()
print("main: waiting on t.join() ...")
t.join()                            # bare join, no timeout

Safe pattern — an interruptible threaded programसुरक्षित pattern — interrupt होने वाला threaded program安全模式 — 可中断的多线程程序

import threading
import time

stop_event = threading.Event()

def worker():
    while not stop_event.is_set():
        # do real work, but check stop_event periodically
        time.sleep(0.1)
    print("Worker exiting cleanly.")

t = threading.Thread(target=worker)
t.start()

try:
    # Main thread stays responsive to signals
    while t.is_alive():
        t.join(timeout=0.5)     # timeout lets main thread wake to check signals
except KeyboardInterrupt:
    print("\nCtrl-C received - asking worker to stop")
    stop_event.set()
    t.join()

Proving the GIL — A CPU-Bound Benchmark 🧪GIL साबित करना — एक CPU-Bound Benchmark 🧪证明 GIL 的存在 — CPU 密集基准测试 🧪

import time
import threading
from multiprocessing import Process

# A pure Python CPU-bound task
def count_down(n):
    while n > 0:
        n -= 1

COUNT = 50_000_000

if __name__ == "__main__":          # needed for multiprocessing on Windows/macOS
    # --- 1. Single thread (baseline) ---
    start = time.perf_counter()
    count_down(COUNT)
    print(f"Single thread:  {time.perf_counter() - start:.2f}s")

    # --- 2. Two threads (splitting the work) ---
    start = time.perf_counter()
    t1 = threading.Thread(target=count_down, args=(COUNT // 2,))
    t2 = threading.Thread(target=count_down, args=(COUNT // 2,))
    t1.start(); t2.start()
    t1.join(); t2.join()
    print(f"Two threads:    {time.perf_counter() - start:.2f}s")

    # --- 3. Two processes (bypasses the GIL) ---
    start = time.perf_counter()
    p1 = Process(target=count_down, args=(COUNT // 2,))
    p2 = Process(target=count_down, args=(COUNT // 2,))
    p1.start(); p2.start()
    p1.join(); p2.join()
    print(f"Two processes:  {time.perf_counter() - start:.2f}s")

Threads Shine for I/O — Same Benchmark, Different Work 🌐Threads I/O में चमकते हैं — वही Benchmark, अलग काम 🌐线程在 I/O 上大放异彩 — 同样的测试，不同的工作 🌐

import time
import threading

def fake_get(url):
    """Stands in for requests.get(url): waits 1 s like a slow server."""
    time.sleep(1)                   # sleep releases the GIL, just like socket I/O

URLS = ["https://httpbin.org/delay/1"] * 10

def fetch(url):
    fake_get(url)                   # real code: requests.get(url)

# --- Sequential ---
start = time.perf_counter()
for url in URLS:
    fetch(url)
print(f"Sequential:  {time.perf_counter() - start:.2f}s")

# --- 10 threads in parallel ---
start = time.perf_counter()
threads = [threading.Thread(target=fetch, args=(u,)) for u in URLS]
for t in threads: t.start()
for t in threads: t.join()
print(f"10 threads:  {time.perf_counter() - start:.2f}s")

While each thread waits, the GIL is free, so the other threads start their own waits. All 10 waits overlap, and the total time shrinks to about the time of one call — exactly 10× faster here. This is why threads paired with a Queue are perfect for I/O work.

जब हर thread इंतज़ार करता है, GIL ख़ाली रहता है, इसलिए बाक़ी threads अपना इंतज़ार शुरू कर देते हैं। सभी 10 इंतज़ार एक साथ होते हैं, और कुल समय लगभग एक call के बराबर रह जाता है — यहाँ ठीक 10× तेज़। इसीलिए Queue के साथ threads I/O काम के लिए एकदम सही हैं।

每个线程等待时 GIL 是空闲的，所以其他线程也开始各自的等待。10 次等待全部重叠，总时间缩短到大约一次调用的时间 — 这里正好 快 10 倍。这就是为什么线程配合 Queue 非常适合 I/O 工作。

Enter the Queue — Thread-Safe Communication 📨Queue आता है — Thread-Safe बातचीत 📨队列登场 — 线程安全的通信 📨

Queue API — The Essential Methods 🧰Queue API — ज़रूरी Methods 🧰Queue API — 必备方法 🧰

🧪 Simulator: Queue Playground🧪 Simulator: Queue Playground🧪 模拟器：Queue 游乐场

Press buttons to call put_nowait(), get_nowait() and task_done(). Watch the items, the unfinished-task counter that q.join() waits on, and the exceptions. Every rule here matched 300 random sequences run on a real queue.Queue.

Buttons दबाकर put_nowait(), get_nowait() और task_done() बुलाइए। Items, q.join() जिसका इंतज़ार करता है वह unfinished-task counter, और exceptions देखिए। यहाँ का हर नियम असली queue.Queue पर चलाए गए 300 random sequences से मिलता है।

按按钮调用 put_nowait()、get_nowait() 和 task_done()。观察元素、q.join() 所等待的 未完成任务计数器 以及异常。这里的每条规则都与在真实 queue.Queue 上运行的 300 个随机序列一致。

Your First Producer-Consumer With Queue 🏭Queue के साथ आपका पहला Producer-Consumer 🏭你的第一个基于 Queue 的生产者-消费者 🏭

import threading
import queue
import time
import random

# A thread-safe, bounded FIFO queue holding at most 5 items
q = queue.Queue(maxsize=5)

def producer(name):
    for i in range(10):
        item = f"{name}-item-{i}"
        q.put(item)                          # blocks if queue is full
        print(f"[P {name}] produced {item} (queue size={q.qsize()})")
        time.sleep(random.uniform(0.05, 0.2))

def consumer(name):
    while True:
        item = q.get()                       # blocks if queue is empty
        if item is None:                     # poison pill = shutdown signal
            q.task_done()
            break
        print(f"    [C {name}] consumed {item}")
        time.sleep(random.uniform(0.1, 0.3))
        q.task_done()                        # signal completion

# Start 2 producers and 3 consumers
producers = [threading.Thread(target=producer, args=(f"P{i}",)) for i in range(2)]
consumers = [threading.Thread(target=consumer, args=(f"C{i}",)) for i in range(3)]
for t in producers + consumers:
    t.start()

for t in producers: t.join()      # wait for producers to finish
q.join()                          # wait until all produced items are processed

for _ in consumers:               # one poison pill per consumer
    q.put(None)
for t in consumers: t.join()
print("All work complete.")

Read it in order: two producers add 20 items in total; three consumers take them as they arrive. q.join() returns only after every item got its task_done(). Then one None per consumer tells each consumer to stop.

क्रम से पढ़िए: दो producers कुल 20 items डालते हैं; तीन consumers आते ही उन्हें ले लेते हैं। q.join() तभी लौटता है जब हर item का task_done() हो जाए। फिर हर consumer के लिए एक None उसे रुकने को कहता है।

按顺序看：两个生产者共放入 20 个元素；三个消费者一有元素就取走。只有每个元素都得到 task_done() 后，q.join() 才会返回。然后给每个消费者放一个 None，通知它停止。

Real-World Example — A Concurrent Fetcher 🕸️असली उदाहरण — एक Concurrent Fetcher 🕸️实战示例 — 并发抓取器 🕸️

Something practical: download 50 URLs using 10 worker threads. One queue holds the URLs; workers take them, "download" them (with fake_get() from Section 14) and put results into a second queue.

कुछ काम का: 10 worker threads से 50 URLs download करना। एक queue में URLs हैं; workers उन्हें लेकर "download" करते हैं (Section 14 वाले fake_get() से) और नतीजे दूसरी queue में डालते हैं।

来点实用的：用 10 个工作线程下载 50 个 URL。一个队列存放 URL；工作线程取出它们，"下载"（使用第 14 节的 fake_get()），再把结果放进第二个队列。

import threading
import queue
import time

def fake_get(url):
    """Stands in for requests.get(url, timeout=10)."""
    time.sleep(1)                   # sleep releases the GIL, just like socket I/O
    return 200, b"x" * 308

URL_LIST = [f"https://httpbin.org/delay/1?id={i}" for i in range(50)]
NUM_WORKERS = 10

task_queue   = queue.Queue()
result_queue = queue.Queue()

def worker(worker_id):
    while True:
        url = task_queue.get()
        if url is None:                     # shutdown
            task_queue.task_done()
            break
        try:
            status, content = fake_get(url)
            result_queue.put((url, status, len(content)))
            print(f"[W{worker_id}] {url} -> {status}")
        except Exception as e:
            result_queue.put((url, "ERROR", str(e)))
        finally:
            task_queue.task_done()

# Feed the queue
for url in URL_LIST:
    task_queue.put(url)

# Start workers
start = time.perf_counter()
workers = [threading.Thread(target=worker, args=(i,), daemon=True)
           for i in range(NUM_WORKERS)]
for w in workers: w.start()

task_queue.join()                     # wait for all URLs to be processed

for _ in workers:                     # shut down the workers
    task_queue.put(None)
for w in workers: w.join()
elapsed = time.perf_counter() - start

results = []                          # drain the results
while not result_queue.empty():
    results.append(result_queue.get())

print(f"\nFetched {len(results)} URLs in {elapsed:.2f}s using {NUM_WORKERS} threads")
print(f"Sequential would take ~{len(URL_LIST)}s")

50 one-second "downloads" finished in 5 seconds: 10 workers × 5 rounds. Each worker always calls task_done() in finally, so even a failed download is counted and task_queue.join() cannot hang. Notice also that some output lines are glued together: when many threads call print() at once, the text and the newline can interleave. That is why the real results travel through result_queue, not through printed text.

50 एक-second वाले "downloads" 5 seconds में पूरे: 10 workers × 5 rounds। हर worker हमेशा finally में task_done() बुलाता है, इसलिए fail हुआ download भी गिना जाता है और task_queue.join() अटक नहीं सकता। यह भी देखिए कि कुछ output lines आपस में चिपक गई हैं: जब कई threads एक साथ print() बुलाते हैं, तो text और newline आपस में मिल सकते हैं। इसीलिए असली नतीजे printed text से नहीं, result_queue से भेजे जाते हैं।

50 次各 1 秒的"下载"在 5 秒内完成：10 个工作线程 × 5 轮。每个工作线程都在 finally 中调用 task_done()，所以即使下载失败也会被计数，task_queue.join() 不会卡住。另外注意，有些输出行粘在了一起：当许多线程同时调用 print() 时，文字和换行符可能交错。这就是真正的结果要通过 result_queue 传递、而不是靠打印文本的原因。

PriorityQueue — When Some Tasks Matter More 🏆PriorityQueue — जब कुछ Tasks ज़्यादा ज़रूरी हों 🏆PriorityQueue — 当有些任务更重要时 🏆

A support-ticket system where urgent tickets jump the line. The priority number goes first; the lowest number comes out first.

एक support-ticket system जहाँ ज़रूरी tickets कतार तोड़कर आगे जाते हैं। Priority number पहले आता है; सबसे छोटा number पहले निकलता है।

一个支持工单系统，紧急工单可以插队。优先级数字放在最前面；数字最小 的先出。

import threading
import queue
import time
from dataclasses import dataclass, field
from typing import Any

# Wrap payload so equal-priority items don't try to compare data
@dataclass(order=True)
class Ticket:
    priority: int
    data: Any = field(compare=False)

pq = queue.PriorityQueue()

# Producers add tickets with varied priority (1=urgent, 5=low)
pq.put(Ticket(3, "Fix typo on homepage"))
pq.put(Ticket(1, "PROD DOWN - payment gateway failing!"))
pq.put(Ticket(5, "Update the copyright year"))
pq.put(Ticket(2, "Enterprise customer login broken"))
pq.put(Ticket(1, "Data breach alert!"))

def handler():
    while not pq.empty():
        ticket = pq.get()
        print(f"[P{ticket.priority}] {ticket.data}")
        pq.task_done()
        time.sleep(0.2)

threading.Thread(target=handler).start()

Tie-breaking — the error and the fixTie-breaking — error और fix平局决胜 — 错误与修复

import queue

pq = queue.PriorityQueue()
pq.put((1, {"task": "backup"}))
pq.put((1, {"task": "deploy"}))      # same priority -> compares the dicts!

import queue, itertools

counter = itertools.count()            # 0, 1, 2, ... never repeats
pq = queue.PriorityQueue()

for prio, data in [(3, "Fix typo on homepage"), (1, "PROD DOWN - payment gateway failing!"),
                   (5, "Update the copyright year"), (2, "Enterprise customer login broken"),
                   (1, "Data breach alert!"), (1, {"task": "a dict is fine now"})]:
    pq.put((prio, next(counter), data))   # (priority, tie-breaker, data)

while not pq.empty():
    prio, n, data = pq.get()
    print(f"[P{prio}] #{n} {data}")

The counter is unique, so Python never needs to compare the data — even a dict works — and equal priorities come out in the order they were put (#1, #4, #5).

Counter अद्वितीय है, इसलिए Python को कभी data की तुलना नहीं करनी पड़ती — dict भी चलता है — और बराबर priorities उसी क्रम में निकलती हैं जिसमें डाली गईं (#1, #4, #5)।

计数器是唯一的，所以 Python 永远不需要比较数据 — 即使是字典也没问题 — 相同优先级的元素按放入顺序取出（#1、#4、#5）。

🏆 Simulator: PriorityQueue Lab🏆 Simulator: PriorityQueue Lab🏆 模拟器：PriorityQueue 实验室

Add tickets with priority 1, 2 or 3 and call get(). The tree shows the real heap inside the queue. Compare the dataclass style with the counter style, or press "show unfair order". Checked against 300 real runs of queue.PriorityQueue.

Priority 1, 2 या 3 के tickets डालिए और get() बुलाइए। Tree queue के अंदर का असली heap दिखाता है। Dataclass तरीक़े की counter तरीक़े से तुलना कीजिए, या "अनुचित क्रम दिखाएँ" दबाइए। queue.PriorityQueue के 300 असली runs से जाँचा गया।

添加优先级为 1、2 或 3 的工单并调用 get()。树形图显示队列内部真实的堆。比较 dataclass 写法和计数器写法，或者点击"演示不公平顺序"。已与 queue.PriorityQueue 的 300 次真实运行核对。

The Modern Way — ThreadPoolExecutor 🏊आधुनिक तरीक़ा — ThreadPoolExecutor 🏊现代做法 — ThreadPoolExecutor 🏊

concurrent.futures.ThreadPoolExecutor is a thread pool with a queue inside and a clean API. It manages the queue, the worker threads and the results for you. For most new batch jobs, this is what to reach for.

concurrent.futures.ThreadPoolExecutor अंदर queue वाला एक thread pool है, साफ़ API के साथ। यह queue, worker threads और नतीजे आपके लिए संभालता है। ज़्यादातर नए batch jobs के लिए यही लीजिए।

concurrent.futures.ThreadPoolExecutor 是一个内含队列、接口简洁的线程池。它替你管理队列、工作线程和结果。对大多数新的批处理任务，首选它。

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def fake_get(url):
    """Stands in for requests.get(url, timeout=10)."""
    time.sleep(1)                   # sleep releases the GIL, just like socket I/O
    return 200, b"x" * 308

urls = [f"https://httpbin.org/delay/1?id={i}" for i in range(20)]

def fetch(url):
    code, content = fake_get(url)
    return url, code, len(content)

start = time.perf_counter()

# The executor manages an internal queue + worker threads
with ThreadPoolExecutor(max_workers=10) as pool:
    # submit() returns a Future - puts a task on the queue
    futures = {pool.submit(fetch, u): u for u in urls}

    # as_completed yields futures as they finish
    for future in as_completed(futures):
        url, code, size = future.result()
        print(f"{code} {size:6d}B  {url}")

print(f"\nDone in {time.perf_counter() - start:.2f}s")

Pipeline Pattern — Chained Queues 🔗Pipeline Pattern — जुड़ी हुई Queues 🔗流水线模式 — 串联的队列 🔗

Real systems often have stages: download → parse → save. Put a queue between stages, and give each stage its own group of threads.

असली systems में अक्सर stages होते हैं: download → parse → save। Stages के बीच एक queue रखिए, और हर stage को threads का अपना समूह दीजिए।

真实系统常常分成几个阶段：下载 → 解析 → 保存。在阶段之间放一个队列，并给每个阶段配一组自己的线程。

import threading
import queue
import time

download_q = queue.Queue()
parse_q    = queue.Queue()
save_q     = queue.Queue()

def downloader():
    while True:
        url = download_q.get()
        if url is None: break
        time.sleep(0.3)                     # simulated network
        html = f"<html>from {url}</html>"
        parse_q.put((url, html))
        download_q.task_done()

def parser():
    while True:
        item = parse_q.get()
        if item is None: break
        url, html = item
        time.sleep(0.1)                     # simulated parsing
        record = {"url": url, "len": len(html)}
        save_q.put(record)
        parse_q.task_done()

def saver():
    while True:
        rec = save_q.get()
        if rec is None: break
        print(f"SAVED {rec}")
        save_q.task_done()

# Start each stage as a thread pool
threads = []
for _ in range(4): threads.append(threading.Thread(target=downloader))
for _ in range(2): threads.append(threading.Thread(target=parser))
for _ in range(1): threads.append(threading.Thread(target=saver))
for t in threads: t.start()

start = time.perf_counter()
for i in range(20):                         # feed URLs
    download_q.put(f"https://example.com/page{i}")

# Wait for each stage to fully drain, then send poison pills
download_q.join()
for _ in range(4): download_q.put(None)
parse_q.join()
for _ in range(2): parse_q.put(None)
save_q.join()
save_q.put(None)

for t in threads: t.join()
print(f"Pipeline complete in {time.perf_counter() - start:.2f}s.")

🔗 Simulator: Pipeline Lab (real runs)🔗 Simulator: Pipeline Lab (असली runs)🔗 模拟器：流水线实验室（真实运行）

12 pages flow through the same pipeline (download 0.3 s, parse 0.1 s). Choose how many downloaders and parsers to start. Each bar is one page in one worker, from a real run.

12 pages उसी pipeline से गुज़रते हैं (download 0.3 s, parse 0.1 s)। चुनिए कितने downloaders और parsers शुरू हों। हर bar एक worker में एक page है, असली run से।

12 个页面流经同样的流水线（下载 0.3 秒，解析 0.1 秒）。选择启动多少个下载线程和解析线程。每个条形表示某个工作线程处理的一个页面，来自真实运行。

Getting Around the GIL — Your Options 🚀GIL से बचने के रास्ते — आपके विकल्प 🚀绕过 GIL — 你的选项 🚀

Common Pitfalls and Fixes 🩺आम ग़लतियाँ और उनके Fix 🩺常见陷阱与修复 🩺

import queue, threading

def process(item):                           # your real work
    if item == "bad":
        raise ValueError("cannot process 'bad'")
    print("processed", item)

def safe_worker(work_q, error_q):
    while True:
        item = work_q.get()
        try:
            if item is None:                 # poison pill
                break
            process(item)
        except Exception as e:
            error_q.put((item, repr(e)))     # never let exceptions vanish
        finally:
            work_q.task_done()               # ALWAYS signal completion

work_q, error_q = queue.Queue(), queue.Queue()
t = threading.Thread(target=safe_worker, args=(work_q, error_q))
t.start()
for item in ["a", "bad", "b", None]:
    work_q.put(item)
work_q.join()                                # returns - every get() had a task_done()
t.join()
print("errors:", list(error_q.queue))

🩺 Simulator: Bug Doctor🩺 Simulator: Bug Doctor🩺 模拟器：Bug 医生

Six real mistakes from this tutorial. Buggy code on the left, fixed code on the right, changed lines marked. Every output is from a real run (programs that would hang were checked with a 1-second timeout).

इस tutorial की छह असली ग़लतियाँ। बाईं ओर buggy code, दाईं ओर ठीक किया हुआ, बदली lines चिह्नित। हर output असली run से है (जो programs अटक जाते उन्हें 1-second timeout से जाँचा गया)।

本教程中的六个真实错误。左边是有 bug 的代码，右边是修正后的代码，改动的行已标出。每个输出都来自真实运行（会挂起的程序用 1 秒超时进行检查）。

Threads vs Processes vs asyncio — Decision Table ⚖️Threads बनाम Processes बनाम asyncio — फ़ैसले की Table ⚖️线程 vs 进程 vs asyncio — 决策表 ⚖️

Golden Rules and Quiz 🏆सुनहरे नियम और Quiz 🏆黄金法则与小测验 🏆

Quick Quiz — 8 Questionsछोटा Quiz — 8 सवाल小测验 — 8 道题