diff -r a7365994f4d6 Doc/library/multiprocessing.rst --- a/Doc/library/multiprocessing.rst Sun Aug 07 17:13:13 2011 +0200 +++ b/Doc/library/multiprocessing.rst Mon Aug 08 12:24:14 2011 +0200 @@ -1664,6 +1664,20 @@ 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 + to be tuples that are unpacked as arguments. + + Hence an `iterable` of `[(1,2), (3, 4)]` results in `[func(1,2), + func(3,4)]`. + + .. method:: starmap_async(func, iterable[, chunksize[, callback[, error_back]]]) + + A combination of :meth:`starmap` and :meth:`map_async` that iterates over + `iterable` of tuples and calls `func` with the tuples unpacked. Returns a + result object. + .. method:: close() Prevents any more tasks from being submitted to the pool. Once all the diff -r a7365994f4d6 Lib/multiprocessing/managers.py --- a/Lib/multiprocessing/managers.py Sun Aug 07 17:13:13 2011 +0200 +++ b/Lib/multiprocessing/managers.py Mon Aug 08 12:24:14 2011 +0200 @@ -1070,11 +1070,12 @@ 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' } diff -r a7365994f4d6 Lib/multiprocessing/pool.py --- a/Lib/multiprocessing/pool.py Sun Aug 07 17:13:13 2011 +0200 +++ b/Lib/multiprocessing/pool.py Mon Aug 08 12:24:14 2011 +0200 @@ -64,6 +64,9 @@ 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 @@ in a list that is returned. ''' assert self._state == RUN - return self.map_async(func, iterable, chunksize).get() + return self._map_async(func, iterable, mapstar, chunksize).get() + + def starmap(self, func, iterable, chunksize=None): + ''' + Like `map()` method but the elements of the `iterable` are expected to + be tuples which will be unpacked as arguments. Hence `func` and (a, b) + becomes func(a, b). + ''' + assert self._state == RUN + return self._map_async(func, iterable, starmapstar, chunksize).get() + + def starmap_async(self, func, iterable, chunksize=None, callback=None, + error_callback=None): + ''' + Asynchronous version of `starmap()` method. + ''' + assert self._state == RUN + return self._map_async(func, iterable, starmapstar, chunksize, + callback, error_callback) def imap(self, func, iterable, chunksize=1): ''' @@ -302,6 +323,13 @@ Asynchronous version of `map()` method. ''' assert self._state == RUN + return self._map_async(func, iterable, mapstar, chunksize) + + def _map_async(self, func, iterable, mapper, chunksize=None, callback=None, + error_callback=None): + ''' + Helper function to implement map, starmap and their async counterparts. + ''' if not hasattr(iterable, '__len__'): iterable = list(iterable) @@ -315,7 +343,7 @@ 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,), {}) + self._taskqueue.put((((result._job, i, mapper, (x,), {}) for i, x in enumerate(task_batches)), None)) return result diff -r a7365994f4d6 Lib/test/test_multiprocessing.py --- a/Lib/test/test_multiprocessing.py Sun Aug 07 17:13:13 2011 +0200 +++ b/Lib/test/test_multiprocessing.py Mon Aug 08 12:24:14 2011 +0200 @@ -8,6 +8,7 @@ import queue as pyqueue import time import io +import itertools import sys import os import gc @@ -1085,6 +1086,9 @@ time.sleep(wait) return x*x +def mul(x, y): + return x*y + class _TestPool(BaseTestCase): def test_apply(self): @@ -1098,6 +1102,20 @@ self.assertEqual(pmap(sqr, list(range(100)), chunksize=20), list(map(sqr, list(range(100))))) + def test_starmap(self): + psmap = self.pool.starmap + tuples = list(zip(range(10), range(9,-1, -1))) + self.assertEqual(psmap(mul, tuples), + list(itertools.starmap(mul, tuples))) + tuples = list(zip(range(100), range(99,-1, -1))) + self.assertEqual(psmap(mul, tuples, chunksize=20), + list(itertools.starmap(mul, tuples))) + + def test_starmap_async(self): + tuples = list(zip(range(100), range(99,-1, -1))) + self.assertEqual(self.pool.starmap_async(mul, tuples).get(), + list(itertools.starmap(mul, tuples))) + def test_map_chunksize(self): try: self.pool.map_async(sqr, [], chunksize=1).get(timeout=TIMEOUT1) diff -r a7365994f4d6 Misc/ACKS --- a/Misc/ACKS Sun Aug 07 17:13:13 2011 +0200 +++ b/Misc/ACKS Mon Aug 08 12:24:14 2011 +0200 @@ -838,6 +838,7 @@ Andreas Schawo Neil Schemenauer David Scherer +Hynek Schlawack Bob Schmertz Gregor Schmid Ralf Schmitt