From bd54284d6f5c3cd9407ca76147bed289ee4498d7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dani=20M=C3=B6ller?= Date: Sun, 4 Feb 2024 19:45:59 +0100 Subject: [PATCH] Use processing pool and selector as context managers Delegate closing both resources to the with instruction. --- trytond/trytond/worker.py | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) diff --git a/trytond/trytond/worker.py b/trytond/trytond/worker.py index 5f24602b44..86c34a2fda 100644 --- a/trytond/trytond/worker.py +++ b/trytond/trytond/worker.py @@ -73,12 +73,15 @@ def work(options): if sys.version_info < (3, 11): del executor_options["max_tasks_per_child"] + with \ + futures.ProcessPoolExecutor(**executor_options) as executor, \ + selectors.DefaultSelector() as selector: + queues = [Queue(name, executor) for name in options.database_names] + tasks = TaskList() + + for queue in queues: + selector.register(queue.connection, selectors.EVENT_READ) - tasks = TaskList() - selector = selectors.DefaultSelector() - for queue in queues: - selector.register(queue.connection, selectors.EVENT_READ) - try: while True: timeout = options.timeout # Add some randomness to avoid concurrent pulling @@ -86,7 +89,8 @@ def work(options): while len(tasks.filter()) >= processes: futures.wait(tasks, return_when=futures.FIRST_COMPLETED) - # Probe process pool is still operative before pulling a new task + # Probe process pool is still operative + # before pulling a new task executor.submit(_noop).result() for queue in queues: @@ -105,9 +109,6 @@ def work(options): connection.poll() while connection.notifies: connection.notifies.pop(0) - finally: - executor.shutdown() - selector.close() def initializer(database_names, worker=True):