mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-11-01 02:38:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 15

Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool.

Browse source 
Patch by Hynek Schlawack.
This commit is contained in:
Antoine Pitrou 2011-12-21 11:03:24 +01:00
parent 12f65d1fef
commit de911b2915
6 changed files with 72 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

18
Lib/test/test_multiprocessing.py
View file

@ -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
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)