Skip to content

Instantly share code, notes, and snippets.

@victorusachev
Created October 23, 2019 22:00
Show Gist options
  • Select an option

  • Save victorusachev/c1a3f07d2aaa0d2210567764921dd057 to your computer and use it in GitHub Desktop.

Select an option

Save victorusachev/c1a3f07d2aaa0d2210567764921dd057 to your computer and use it in GitHub Desktop.
import logging
import threading
def worker(event: threading.Event, /, *, sleep: float = 1):
while not event.is_set():
logging.debug("worker thread checking in")
event.wait(sleep)
def main(thread_count=2):
logging.basicConfig(
level=logging.DEBUG,
format="%(relativeCreated)6d %(threadName)s %(message)s"
)
event = threading.Event()
threads = [
threading.Thread(
name=f'Thread {number}',
target=worker,
args=[event],
kwargs={'sleep': (number + 1) * 2},
)
for number in range(thread_count)
]
for thread in threads:
thread.start()
while not event.is_set():
try:
logging.debug("Checking in from main thread")
event.wait(0.75)
except KeyboardInterrupt:
event.set()
logging.debug('Stopping')
break
for thread in threads:
thread.join()
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment