Skip to content

Instantly share code, notes, and snippets.

@plamut

plamut/test.py Secret

Last active May 1, 2019 23:30
Show Gist options
  • Save plamut/b8937bdf2f6663e3e9fc503c5ccf1983 to your computer and use it in GitHub Desktop.
Save plamut/b8937bdf2f6663e3e9fc503c5ccf1983 to your computer and use it in GitHub Desktop.
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()
@anguillanneuf
Copy link

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

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment