Use processing pool and selector as context managers
Delegate closing both resources to the with instruction.
This commit is contained in:
1 parent
aa96c33cbc
commit
bd54284d6f
1 file changed
+10
-9
@@ -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):
|
||||
|
||||
Reference in new issue
Block a user