Concurrency and Modules

Queue Worker

A Queue hands work from one thread to another. The source stays in one pane, and replay uses thread colors to show producer and consumer events.

queue handoff `put()` adds work, and `get()` removes it. A small sentinel value can tell the consumer that no more work is coming.

Producer and Consumer

count
queue_worker.py
Replay: real traced execution (multi-file project)
import queue
import threading


work = queue.Queue()


def producer(count):
    for number in range(1, count + 1):
        item = "job-" + str(number)
        work.put(item)
        print("queued=" + item)
    work.put("done")


def consumer():
    while True:
        item = work.get()
        if item == "done":
            print("stopped")
            break
        print("handled=" + item)


count = 2

producer_thread = threading.Thread(target=producer, args=(count,))
consumer_thread = threading.Thread(target=consumer)

producer_thread.start()
consumer_thread.start()
producer_thread.join()
consumer_thread.join()

import queue
import threading


work = queue.Queue()


def producer(count):
    for number in range(1, count + 1):
        item = "job-" + str(number)
        work.put(item)
        print("queued=" + item)
    work.put("done")


def consumer():
    while True:
        item = work.get()
        if item == "done":
            print("stopped")
            break
        print("handled=" + item)


count = 3

producer_thread = threading.Thread(target=producer, args=(count,))
consumer_thread = threading.Thread(target=consumer)

producer_thread.start()
consumer_thread.start()
producer_thread.join()
consumer_thread.join()

  1. work ← ⟨Queue A⟩, count ← 2, producer_thread ← <Thread(Thread-1 (producer), initial)>

    5work→ ⟨Queue A⟩ = queue<module 'queue' from '/usr/local/lib/python3.12/queue.py'>.Queue()678def producer(count):9    for number in range(1, count + 1):10        item = "job-" + str(number)11        work.put(item)12        print("queued=" + item)13    work.put("done")141516def consumer():17    while True:18        item = work.get()19        if item == "done":20            print("stopped")21            break22        print("handled=" + item)232425count→ 2 = 2  #@count=32627producer_thread→ <Thread(Thread-1 (producer), initial)> = threading<module 'threading' from '/usr/local/lib/python3.12/threading.py'>.Thread(target=producer⟨function producer B⟩, args=(count2,))28consumer_thread→ <Thread(Thread-2 (consumer), initial)> = threading<module 'threading' from '/usr/local/lib/python3.12/threading.py'>.Thread(target=consumer⟨function consumer C⟩)2930producer_thread<Thread(Thread-1 (producer), initial)>.start()31consumer_thread.start()

  2. def producer(count):

    8def producer(count2):9    for number in range(1, count + 1):10        item = "job-" + str(number)

  3. item ← job-1

    pass 1 of 2
    8def producer(count):9    for number1 in range(1, count2 + 1):10        item→ job-1 = "job-" + str(number1)11        work⟨Queue A⟩.put(itemjob-1)12        print("queued=" + itemjob-1)13    work.put("done")
    outputqueued=job-1

  4. item ← job-2

    pass 2 of 2
    8def producer(count):9    for number2 in range(1, count2 + 1):10        item→ job-2 = "job-" + str(number2)11        work⟨Queue A⟩.put(itemjob-2)12        print("queued=" + itemjob-2)13    work.put("done")
    outputqueued=job-2

  5. work.put("done")

    12    print("queued=" + item)13work⟨Queue A⟩.put("done")

  6. producer_thread ← <Thread(Thread-1 (producer), stopped 139437626431168)>

    30producer_thread→ <Thread(Thread-1 (producer), stopped 139437626431168)>.start()31consumer_thread<Thread(Thread-2 (consumer), initial)>.start()32producer_thread.join()

  7. item ← job-1

    pass 1 of 3
    16def consumer():17    while True:18        item→ job-1 = work⟨Queue A⟩.get()19        if item == "done":20            print("stopped")21            break22        print("handled=" + itemjob-1)
    outputhandled=job-1
    All 3 passes — pass 1 is the card above
    passitem
    1job-1
    2job-2
    3done

  8. if item == "done":

    18item = work.get()19if itemdone == "done":20    print("stopped")21    break22print("handled=" + item)
    outputstopped

  9. consumer_thread ← <Thread(Thread-2 (consumer), stopped 139437626431168)>

    30producer_thread.start()31consumer_thread→ <Thread(Thread-2 (consumer), stopped 139437626431168)>.start()32producer_thread<Thread(Thread-1 (producer), stopped 139437626431168)>.join()33consumer_thread<Thread(Thread-2 (consumer), stopped 139437626431168)>.join()

  1. work ← ⟨Queue A⟩, count ← 3, producer_thread ← <Thread(Thread-1 (producer), initial)>

    5work→ ⟨Queue A⟩ = queue<module 'queue' from '/usr/local/lib/python3.12/queue.py'>.Queue()678def producer(count):9    for number in range(1, count + 1):10        item = "job-" + str(number)11        work.put(item)12        print("queued=" + item)13    work.put("done")141516def consumer():17    while True:18        item = work.get()19        if item == "done":20            print("stopped")21            break22        print("handled=" + item)232425count→ 3 = 32627producer_thread→ <Thread(Thread-1 (producer), initial)> = threading<module 'threading' from '/usr/local/lib/python3.12/threading.py'>.Thread(target=producer⟨function producer B⟩, args=(count3,))28consumer_thread→ <Thread(Thread-2 (consumer), initial)> = threading<module 'threading' from '/usr/local/lib/python3.12/threading.py'>.Thread(target=consumer⟨function consumer C⟩)2930producer_thread<Thread(Thread-1 (producer), initial)>.start()31consumer_thread.start()

  2. def producer(count):

    8def producer(count3):9    for number in range(1, count + 1):10        item = "job-" + str(number)

  3. item ← job-1

    pass 1 of 3
    8def producer(count):9    for number1 in range(1, count3 + 1):10        item→ job-1 = "job-" + str(number1)11        work⟨Queue A⟩.put(itemjob-1)12        print("queued=" + itemjob-1)13    work.put("done")
    outputqueued=job-1
    All 3 passes — pass 1 is the card above
    passnumberitem
    11job-1
    22job-2
    33job-3

  4. work.put("done")

    12    print("queued=" + item)13work⟨Queue A⟩.put("done")

  5. producer_thread ← <Thread(Thread-1 (producer), stopped 132846169237184)>

    30producer_thread→ <Thread(Thread-1 (producer), stopped 132846169237184)>.start()31consumer_thread<Thread(Thread-2 (consumer), initial)>.start()32producer_thread.join()

  6. item ← job-1

    pass 1 of 4
    16def consumer():17    while True:18        item→ job-1 = work⟨Queue A⟩.get()19        if item == "done":20            print("stopped")21            break22        print("handled=" + itemjob-1)
    outputhandled=job-1
    All 4 passes — pass 1 is the card above
    passitem
    1job-1
    2job-2
    3job-3
    4—

  7. consumer_thread ← <Thread(Thread-2 (consumer), started 132846169237184)>

    17    while True:18        item→ done = work⟨Queue A⟩.get()19        if item == "done":20            print("stopped")21            break22        print("handled=" + item)232425count = 32627producer_thread = threading.Thread(target=producer, args=(count,))28consumer_thread = threading.Thread(target=consumer)2930producer_thread.start()31consumer_thread→ <Thread(Thread-2 (consumer), started 132846169237184)>.start()32producer_thread<Thread(Thread-1 (producer), stopped 132846169237184)>.join()33consumer_thread<Thread(Thread-2 (consumer), started 132846169237184)>.join()

  8. if item == "done":

    18item = work.get()19if itemdone == "done":20    print("stopped")21    break22print("handled=" + item)
    outputstopped

  9. consumer_thread ← <Thread(Thread-2 (consumer), stopped 132846169237184)>

    32producer_thread.join()33consumer_thread→ <Thread(Thread-2 (consumer), stopped 132846169237184)>.join()