From b62d950bb8bf248586e07ba5fc3710c8eadca5db Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Honza=20Kr=C3=A1l?= Date: Wed, 9 Dec 2015 18:35:14 +0100 Subject: [PATCH] Make sure we always close the pool when doing parallel_bulk --- elasticsearch/helpers/__init__.py | 18 ++++++++++-------- 1 file changed, 10 insertions(+), 8 deletions(-) diff --git a/elasticsearch/helpers/__init__.py b/elasticsearch/helpers/__init__.py index 143894b3..01d0b6cf 100644 --- a/elasticsearch/helpers/__init__.py +++ b/elasticsearch/helpers/__init__.py @@ -222,15 +222,17 @@ def parallel_bulk(client, actions, thread_count=4, chunk_size=500, pool = Pool(thread_count) - for result in pool.imap( - lambda chunk: list(_process_bulk_chunk(client, chunk, **kwargs)), - _chunk_actions(actions, chunk_size, max_chunk_bytes, client.transport.serializer) - ): - for item in result: - yield item + try: + for result in pool.imap( + lambda chunk: list(_process_bulk_chunk(client, chunk, **kwargs)), + _chunk_actions(actions, chunk_size, max_chunk_bytes, client.transport.serializer) + ): + for item in result: + yield item - pool.close() - pool.join() + finally: + pool.close() + pool.join() def scan(client, query=None, scroll='5m', raise_on_error=True, preserve_order=False, **kwargs): """