test.py

import logging
from logging import handlers
import os
import random
import threading
import time


script_dir = os.path.dirname(os.path.realpath(__file__))
log_filename = os.path.join(script_dir, "test.log")
handler = handlers.TimedRotatingFileHandler(
    filename=log_filename, when="midnight", utc=True
)

logging.basicConfig(
    level=logging.INFO,
    format="%(levelname)-8s [%(asctime)s] %(threadName)-16s: %(message)s",
    #handlers=[handler],
)

logger = logging.getLogger(__name__)


global_number = 1


def worker(condition):
    while True:
        with condition:
            logger.debug("acquired condition lock")
            #while global_number != 10:
            if global_number != 10:  # <--- incorrect!
                logger.info("the number is not 10, will wait for condition")
                condition.wait()
            #condition.wait_for(lambda: global_number == 10)  # <--- alternative
            logger.info(f"I see the number {global_number}")
            logger.debug("releasing condition lock")
        time.sleep(1)


def number_changer(condition):
    global global_number
    while True:
        with condition:
            logger.debug("acquired condition lock")
            new_number = random.randint(1, 10)
            logger.info(f"Changing number from {global_number:>2} to {new_number:>2}")
            global_number = new_number
            if new_number == 10:
                logger.info("New number == 10, notifying others")
                condition.notify_all()
            logger.debug("releasing condition lock")
        time.sleep(1)


def main():
    logger.info("main() started")
    number_eq_10 = threading.Condition()
    
    worker_thread = threading.Thread(
        name="Thread-Worker",
        target=worker,
        args=(number_eq_10,),
        daemon=True,
    )

    changer_threads = [
        threading.Thread(
            name=f"Thread-Changer-{i}",
            target=number_changer,
            args=(number_eq_10,),
            daemon=True,
        )
        for i in range(5)
    ]

    worker_thread.start()
    for t in changer_threads:
        t.start()

    while True:
        try:
            time.sleep(60)
        except KeyboardInterrupt:
            logger.info("Main thread exiting")
            break
    _


if __name__ == "__main__":
    main()

What's the expected behavior when we set while global_number != 10:? I still see pretty much the same behavior (with the changer thread changing the number after it's been changed to 10).

@anguillanneuf The expected behavior is that the worker thread always prints out "I see the number 10" (the desired condition), and never "I see <N != 10 >". In other words, it should only proceed when the condition number == 10 is fulfilled.

In the bidi.py case, this would translate to BackgroundConsumer always resuming when self._paused == False, and never resuming when self._paused == True is set by another thread.

FWIW, if the number is set to 10, another thread might still quickly change it to something else, i.e. before the worker sees it. That will cause the worker to keep waiting (if using while), which is fine. We want the worker to only proceed when the condition number == 10 definitely holds.

Excellent code+explanation. Took me a while but I get it now. Thank you!