cpython: b07b1e58582d (original) (raw)

Mercurial > cpython

changeset 74107:b07b1e58582d

Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Patch by Hynek Schlawack. [#12708]

Antoine Pitrou solipsis@pitrou.net
date	Wed, 21 Dec 2011 11:03:24 +0100
parents	b08bf8df8eec
children	1ab124a6f171
files	Doc/library/multiprocessing.rst Lib/multiprocessing/managers.py Lib/multiprocessing/pool.py Lib/test/test_multiprocessing.py Misc/ACKS Misc/NEWS
diffstat	6 files changed, 72 insertions(+), 3 deletions(-)[+] [-] Doc/library/multiprocessing.rst 18 Lib/multiprocessing/managers.py 3 Lib/multiprocessing/pool.py 32 Lib/test/test_multiprocessing.py 18 Misc/ACKS 1 Misc/NEWS 3

line wrap: on

line diff

--- a/Doc/library/multiprocessing.rst +++ b/Doc/library/multiprocessing.rst @@ -1669,6 +1669,24 @@ with the :class:Pool class. returned iterator should be considered arbitrary. (Only when there is only one worker process is the order guaranteed to be "correct".)

.. method:: starmap(func, iterable[, chunksize]) +

 Like :meth:`map` except that the elements of the `iterable` are expected[](#l1.9)

 to be iterables that are unpacked as arguments.[](#l1.10)

 Hence an `iterable` of `[(1,2), (3, 4)]` results in `[func(1,2),[](#l1.12)

```
 func(3,4)]`.[](#l1.13)
```

```
 .. versionadded:: 3.3[](#l1.15)
```

.. method:: starmap_async(func, iterable[, chunksize[, callback[, error_back]]]) +

 A combination of :meth:`starmap` and :meth:`map_async` that iterates over[](#l1.19)

 `iterable` of iterables and calls `func` with the iterables unpacked.[](#l1.20)

```
 Returns a result object.[](#l1.21)
```

```
 .. versionadded:: 3.3[](#l1.23)
```

+ .. method:: close() Prevents any more tasks from being submitted to the pool. Once all the

--- a/Lib/multiprocessing/managers.py +++ b/Lib/multiprocessing/managers.py @@ -1066,11 +1066,12 @@ ArrayProxy = MakeProxyType('ArrayProxy', PoolProxy = MakeProxyType('PoolProxy', ( 'apply', 'apply_async', 'close', 'imap', 'imap_unordered', 'join',

'map', 'map_async', 'terminate'

'map', 'map_async', 'starmap', 'starmap_async', 'terminate' )) PoolProxy.method_to_typeid = { 'apply_async': 'AsyncResult', 'map_async': 'AsyncResult',
'starmap_async': 'AsyncResult', 'imap': 'Iterator', 'imap_unordered': 'Iterator' }

--- a/Lib/multiprocessing/pool.py +++ b/Lib/multiprocessing/pool.py @@ -64,6 +64,9 @@ job_counter = itertools.count() def mapstar(args): return list(map(*args)) +def starmapstar(args):

return list(itertools.starmap(args[0], args[1]))

+ #

Code run by worker processes

# @@ -248,7 +251,25 @@ class Pool(object): in a list that is returned. ''' assert self._state == RUN

   return self.map_async(func, iterable, chunksize).get()[](#l3.17)

   return self._map_async(func, iterable, mapstar, chunksize).get()[](#l3.18)

def starmap(self, func, iterable, chunksize=None):
```
   '''[](#l3.21)
```

   Like `map()` method but the elements of the `iterable` are expected to[](#l3.22)

   be iterables as well and will be unpacked as arguments. Hence[](#l3.23)

   `func` and (a, b) becomes func(a, b).[](#l3.24)

```
   '''[](#l3.25)
```
```
   assert self._state == RUN[](#l3.26)
```

   return self._map_async(func, iterable, starmapstar, chunksize).get()[](#l3.27)

def starmap_async(self, func, iterable, chunksize=None, callback=None,
```
       error_callback=None):[](#l3.30)
```
```
   '''[](#l3.31)
```

   Asynchronous version of `starmap()` method.[](#l3.32)

```
   '''[](#l3.33)
```
```
   assert self._state == RUN[](#l3.34)
```

   return self._map_async(func, iterable, starmapstar, chunksize,[](#l3.35)

                          callback, error_callback)[](#l3.36)

def imap(self, func, iterable, chunksize=1): ''' @@ -302,6 +323,13 @@ class Pool(object): Asynchronous version of map() method. ''' assert self._state == RUN

   return self._map_async(func, iterable, mapstar, chunksize)[](#l3.44)

def _map_async(self, func, iterable, mapper, chunksize=None, callback=None,
```
       error_callback=None):[](#l3.47)
```
```
   '''[](#l3.48)
```

   Helper function to implement map, starmap and their async counterparts.[](#l3.49)

   '''[](#l3.50)
   if not hasattr(iterable, '__len__'):[](#l3.51)
       iterable = list(iterable)[](#l3.52)

@@ -315,7 +343,7 @@ class Pool(object): task_batches = Pool._get_tasks(func, iterable, chunksize) result = MapResult(self._cache, chunksize, len(iterable), callback, error_callback=error_callback)

   self._taskqueue.put((((result._job, i, mapstar, (x,), {})[](#l3.58)

   self._taskqueue.put((((result._job, i, mapper, (x,), {})[](#l3.59)
                         for i, x in enumerate(task_batches)), None))[](#l3.60)
   return result[](#l3.61)

--- a/Lib/test/test_multiprocessing.py +++ b/Lib/test/test_multiprocessing.py @@ -8,6 +8,7 @@ import unittest import queue as pyqueue import time import io +import itertools import sys import os import gc @@ -1125,6 +1126,9 @@ def sqr(x, wait=0.0): time.sleep(wait) return x*x +def mul(x, y):

return x*y

+ class _TestPool(BaseTestCase): def test_apply(self): @@ -1138,6 +1142,20 @@ class _TestPool(BaseTestCase): self.assertEqual(pmap(sqr, list(range(100)), chunksize=20), list(map(sqr, list(range(100)))))

def test_starmap(self):
```
   psmap = self.pool.starmap[](#l4.26)
```

   tuples = list(zip(range(10), range(9,-1, -1)))[](#l4.27)

   self.assertEqual(psmap(mul, tuples),[](#l4.28)

                    list(itertools.starmap(mul, tuples)))[](#l4.29)

   tuples = list(zip(range(100), range(99,-1, -1)))[](#l4.30)

   self.assertEqual(psmap(mul, tuples, chunksize=20),[](#l4.31)

                    list(itertools.starmap(mul, tuples)))[](#l4.32)

def test_starmap_async(self):

   tuples = list(zip(range(100), range(99,-1, -1)))[](#l4.35)

   self.assertEqual(self.pool.starmap_async(mul, tuples).get(),[](#l4.36)

                    list(itertools.starmap(mul, tuples)))[](#l4.37)

+ def test_map_chunksize(self): try: self.pool.map_async(sqr, [], chunksize=1).get(timeout=TIMEOUT1)

--- a/Misc/ACKS +++ b/Misc/ACKS @@ -878,6 +878,7 @@ Michael Scharf Andreas Schawo Neil Schemenauer David Scherer +Hynek Schlawack Bob Schmertz Gregor Schmid Ralf Schmitt

--- a/Misc/NEWS +++ b/Misc/NEWS @@ -419,6 +419,9 @@ Core and Builtins Library ------- +- Issue #12708: Add starmap() and starmap_async() methods (similar to

itertools.starmap()) to multiprocessing.Pool. Patch by Hynek Schlawack. +

Issue #1785: Fix inspect and pydoc with misbehaving descriptors.
Issue #13637: "a2b" functions in the binascii module now accept ASCII-only