Untitled
unknown
python
a year ago
3.7 kB
5
Indexable
import threading import time import random from datetime import datetime # Constants ACTIVITY_DELAY_RANGE = (0.1, 0.15) #delay range GRAB_TIMEOUT = 1 NUM_PHILOSOPHERS = 5 #Number of PHILOSOPHERS RUN_TIME = 10 # run for 10 seconds print_lock = threading.Lock() def timestamped_print(message): """Custom print function for timestamped messages.""" with print_lock: print(f"{datetime.now().strftime('%H:%M:%S.%f')[:-3]}-- {message}") class Philosopher(threading.Thread): running = True def __init__(self, name, left_fork, right_fork, position): super().__init__() self.name = name self.left_fork = left_fork self.right_fork = right_fork self.position = position self.hungry_times = 0 self.eating_times = 0 def run(self): while Philosopher.running: time.sleep(random.uniform(*ACTIVITY_DELAY_RANGE)) self.hungry_times += 1 timestamped_print(f"{self.name} is hungry.") self.dine() def dine(self): fork1, fork2 = self.left_fork, self.right_fork # Philosopher with an odd position number reverses the order of fork picking if self.position % 2: fork1, fork2 = fork2, fork1 while True: time.sleep(random.uniform(*ACTIVITY_DELAY_RANGE)) # Attempt to grab the first fork fork1_grabbed = fork1.grab(timeout=GRAB_TIMEOUT) if fork1_grabbed: timestamped_print(f"{self.name} grabbed fork {fork1.position}") time.sleep(random.uniform(*ACTIVITY_DELAY_RANGE)) # Attempt to grab the second fork fork2_grabbed = fork2.grab(timeout=GRAB_TIMEOUT) if fork2_grabbed: timestamped_print(f"{self.name} grabbed fork {fork2.position}") break # Both forks acquired; break from the loop to eat else: # If the second fork cannot be grabbed, release the first and retry fork1.put_down() timestamped_print(f"{self.name} released fork {fork1.position} and is thinking.") # Eating process self.eating_times += 1 timestamped_print(f"{self.name} is eating.") time.sleep(random.uniform(*ACTIVITY_DELAY_RANGE)) # Put down both forks after eating fork1.put_down() timestamped_print(f"{self.name} put down fork {fork1.position}.") fork2.put_down() timestamped_print(f"{self.name} put down fork {fork2.position} and is done eating.") class Fork: def __init__(self, position): self.position = position self.lock = threading.Lock() def grab(self, timeout): return self.lock.acquire(timeout=timeout) def put_down(self): self.lock.release() if __name__ == "__main__": forks = [Fork(i + 1) for i in range(NUM_PHILOSOPHERS)] philosophers = [Philosopher(f"P{i + 1}", forks[i], forks[(i + 1) % NUM_PHILOSOPHERS], i + 1) for i in range(NUM_PHILOSOPHERS)] start_time = time.time() for philosopher in philosophers: philosopher.start() while time.time() - start_time < RUN_TIME: time.sleep(0.1) # Signal all threads to stop Philosopher.running = False summary_message = f"---------------Summary-------------\n" for philosopher in philosophers: philosopher.join() # Ensure thread has finished summary_message += f" Philosopher {philosopher.name}: Hungry times: {philosopher.hungry_times} && Eating times: {philosopher.eating_times} \n" print(summary_message) timestamped_print("Program terminated")
Editor is loading...
Leave a Comment