Skip to content

Instantly share code, notes, and snippets.

@haowen-xu
Created Sep 28, 2021
Embed
What would you like to do?
import threading
class Task(object):
def __init__(self):
self.lock = threading.RLock()
self.cv = threading.Condition(self.lock)
self.running = True
def task1(self):
with self.lock:
running = self.running
while running:
print("starting task1")
with self.lock:
self.cv.wait(3)
running = self.running
print("task1 completed!")
def task2(self):
with self.lock:
running = self.running
while running:
print("starting task2")
with self.lock:
self.cv.wait(3)
running = self.running
print("task2 completed!")
def stop(self):
with self.lock:
self.running = False
self.cv.notify_all()
def start(self):
self.thread1 = threading.Thread(target=self.task1)
self.thread2 = threading.Thread(target=self.task2)
self.thread1.start()
self.thread2.start()
def join(self):
self.thread1.join()
self.thread2.join()
if __name__ == "__main__":
task = Task()
task.start()
try:
task.join()
except KeyboardInterrupt:
print("KeyboardInterrupt has been caught.")
finally:
task.stop()
task.join()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment