1. import time
  2. import threading
  3. import logging
  4. import queue
  6. class TaskScheduler:
  7. def __init__(self, max_workers=5):
  8. self.task_queue = queue.Queue() # Queue for tasks
  9. self.workers = [] # List to hold worker threads
  10. self.max_workers = max_workers
  11. self.logging = logging.getLogger(__name__)
  12. self.logging.setLevel(logging.ERROR) #Set log level to ERROR to prevent excessive output
  13. self.handler = logging.StreamHandler()
  14. self.formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
  15. self.handler.setFormatter(self.formatter)
  16. self.logging.addHandler(self.handler)
  19. def add_task(self, task, *args, **kwargs):
  20. """Adds a task to the queue."""
  21. self.task_queue.put((task, args, kwargs))
  23. def worker(self):
  24. """Worker thread function to execute tasks."""
  25. while True:
  26. try:
  27. task, args, kwargs = self.task_queue.get(timeout=1) #Get a task from the queue with a timeout
  28. try:
  29. task(*args, **kwargs) # Execute the task
  30. except Exception as e:
  31. self.logging.error(f"Task failed: {e}", exc_info=True) # Log the error
  32. finally:
  33. self.task_queue.task_done()
  34. except queue.Empty:
  35. # Timeout reached, check if workers should exit
  36. break
  38. def start(self):
  39. """Starts the worker threads."""
  40. for _ in range(self.max_workers):
  41. worker = threading.Thread(target=self.worker)
  42. worker.daemon = True # Allow main thread to exit even if workers are running
  43. self.workers.append(worker)
  44. worker.start()
  46. def wait_completion(self):
  47. """Waits for all tasks in the queue to complete."""
  48. self.task_queue.join() #Block until all items in the queue have been gotten and processed.
  51. def stop(self):
  52. """Stops the worker threads."""
  53. for worker in self.workers:
  54. worker.join() #Wait for the threads to finish
  55. self.task_queue.join() #Ensure all remaining tasks are processed
  56. self.workers = []
  60. if __name__ == '__main__':
  61. logging.basicConfig(level=logging.INFO) #Set default logging level
  62. scheduler = TaskScheduler(max_workers=3)
  63. scheduler.start()
  65. def my_task(task_id):
  66. """A short-lived task."""
  67. print(f"Task {task_id} started at {time.time()}")
  68. time.sleep(2) # Simulate some work
  69. print(f"Task {task_id} finished at {time.time()}")
  71. # Add tasks to the scheduler
  72. for i in range(5):
  73. scheduler.add_task(my_task, i)
  75. scheduler.wait_completion()
  76. scheduler.stop()
  77. print("All tasks completed.")

Add your comment