From 916cb2693eb26e08947062dd961443c49da7d387 Mon Sep 17 00:00:00 2001 From: Raz Date: Tue, 16 Oct 2018 09:15:14 +0300 Subject: [PATCH 1/5] Add name parameter to proccess and thread pool Contributed byt Allegro.ai --- Lib/multiprocessing/pool.py | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/Lib/multiprocessing/pool.py b/Lib/multiprocessing/pool.py index 574b5db5afb6846..8a6ab96c12398c7 100644 --- a/Lib/multiprocessing/pool.py +++ b/Lib/multiprocessing/pool.py @@ -154,7 +154,7 @@ def Process(ctx, *args, **kwds): return ctx.Process(*args, **kwds) def __init__(self, processes=None, initializer=None, initargs=(), - maxtasksperchild=None, context=None): + maxtasksperchild=None, context=None, name='Process-Pool'): self._ctx = context or get_context() self._setup_queues() self._taskqueue = queue.SimpleQueue() @@ -163,6 +163,7 @@ def __init__(self, processes=None, initializer=None, initargs=(), self._maxtasksperchild = maxtasksperchild self._initializer = initializer self._initargs = initargs + self._name = name if processes is None: processes = os.cpu_count() or 1 @@ -177,6 +178,7 @@ def __init__(self, processes=None, initializer=None, initargs=(), self._repopulate_pool() self._worker_handler = threading.Thread( + name='{}-Worker-Handler'.format(self._name), target=Pool._handle_workers, args=(self._cache, self._taskqueue, self._ctx, self.Process, self._processes, self._pool, self._inqueue, self._outqueue, @@ -187,7 +189,9 @@ def __init__(self, processes=None, initializer=None, initargs=(), self._worker_handler._state = RUN self._worker_handler.start() + self._task_handler = threading.Thread( + name='{}-Task-Handler'.format(self._name), target=Pool._handle_tasks, args=(self._taskqueue, self._quick_put, self._outqueue, self._pool, self._cache) @@ -197,6 +201,7 @@ def __init__(self, processes=None, initializer=None, initargs=(), self._task_handler.start() self._result_handler = threading.Thread( + name='{}-Result-Handler'.format(self._name), target=Pool._handle_results, args=(self._outqueue, self._quick_get, self._cache) ) @@ -245,7 +250,8 @@ def _repopulate_pool_static(ctx, Process, processes, pool, inqueue, for use after reaping workers which have exited. """ for i in range(processes - len(pool)): - w = Process(ctx, target=worker, + w = self.Process(name='{}-Worker-{}'.format(self._name, i), + target=worker, args=(inqueue, outqueue, initializer, initargs, maxtasksperchild, @@ -824,8 +830,8 @@ def Process(ctx, *args, **kwds): from .dummy import Process return Process(*args, **kwds) - def __init__(self, processes=None, initializer=None, initargs=()): - Pool.__init__(self, processes, initializer, initargs) + def __init__(self, processes=None, initializer=None, initargs=(), name='Thread-Pool'): + Pool.__init__(self, processes, initializer, initargs, name=name) def _setup_queues(self): self._inqueue = queue.SimpleQueue() From de6489a42c3ec064d386beb0a73fe73cc30bf919 Mon Sep 17 00:00:00 2001 From: Raz Date: Wed, 17 Oct 2018 09:02:09 +0300 Subject: [PATCH 2/5] Fix usage of self in static method --- Lib/multiprocessing/pool.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/Lib/multiprocessing/pool.py b/Lib/multiprocessing/pool.py index 8a6ab96c12398c7..1c59abc7a9c76d7 100644 --- a/Lib/multiprocessing/pool.py +++ b/Lib/multiprocessing/pool.py @@ -240,17 +240,18 @@ def _repopulate_pool(self): self._outqueue, self._initializer, self._initargs, self._maxtasksperchild, - self._wrap_exception) + self._wrap_exception, + self._name) @staticmethod def _repopulate_pool_static(ctx, Process, processes, pool, inqueue, outqueue, initializer, initargs, - maxtasksperchild, wrap_exception): + maxtasksperchild, wrap_exception, name): """Bring the number of pool processes up to the specified number, for use after reaping workers which have exited. """ for i in range(processes - len(pool)): - w = self.Process(name='{}-Worker-{}'.format(self._name, i), + w = self.Process(name='{}-Worker-{}'.format(name, i), target=worker, args=(inqueue, outqueue, initializer, From 82ab9d41bee2323e7f6cd0ff75ac7b04bf669ff5 Mon Sep 17 00:00:00 2001 From: Raz Date: Wed, 17 Oct 2018 09:11:42 +0300 Subject: [PATCH 3/5] Add news entry --- .../NEWS.d/next/Library/2018-10-17-09-09-51.bpo-34996.lzdfsg.rst | 1 + 1 file changed, 1 insertion(+) create mode 100644 Misc/NEWS.d/next/Library/2018-10-17-09-09-51.bpo-34996.lzdfsg.rst diff --git a/Misc/NEWS.d/next/Library/2018-10-17-09-09-51.bpo-34996.lzdfsg.rst b/Misc/NEWS.d/next/Library/2018-10-17-09-09-51.bpo-34996.lzdfsg.rst new file mode 100644 index 000000000000000..5dc6427e68a079d --- /dev/null +++ b/Misc/NEWS.d/next/Library/2018-10-17-09-09-51.bpo-34996.lzdfsg.rst @@ -0,0 +1 @@ +Add name paramter to process and thread pool From 3101bf195689010ba1ea1e49ebefaacfbd6754c4 Mon Sep 17 00:00:00 2001 From: Raz Date: Wed, 17 Oct 2018 09:43:48 +0300 Subject: [PATCH 4/5] Remove another self usage from static method --- Lib/multiprocessing/pool.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Lib/multiprocessing/pool.py b/Lib/multiprocessing/pool.py index 1c59abc7a9c76d7..6f2ff5b0cda821c 100644 --- a/Lib/multiprocessing/pool.py +++ b/Lib/multiprocessing/pool.py @@ -251,8 +251,8 @@ def _repopulate_pool_static(ctx, Process, processes, pool, inqueue, for use after reaping workers which have exited. """ for i in range(processes - len(pool)): - w = self.Process(name='{}-Worker-{}'.format(name, i), - target=worker, + w = Process(ctx, target=worker, + name='{}-Worker-{}'.format(name, i), args=(inqueue, outqueue, initializer, initargs, maxtasksperchild, From 290b6c71a6ca5ac3b7846a523182bea766243f31 Mon Sep 17 00:00:00 2001 From: Raz Date: Tue, 23 Oct 2018 11:19:19 +0300 Subject: [PATCH 5/5] Fix alignment issue --- Lib/multiprocessing/pool.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Lib/multiprocessing/pool.py b/Lib/multiprocessing/pool.py index 6f2ff5b0cda821c..075905d4943024e 100644 --- a/Lib/multiprocessing/pool.py +++ b/Lib/multiprocessing/pool.py @@ -252,7 +252,7 @@ def _repopulate_pool_static(ctx, Process, processes, pool, inqueue, """ for i in range(processes - len(pool)): w = Process(ctx, target=worker, - name='{}-Worker-{}'.format(name, i), + name='{}-Worker-{}'.format(name, i), args=(inqueue, outqueue, initializer, initargs, maxtasksperchild,