Skip to content

Commit e283490

Browse files
author
Florian Perucki
committed
Fixed subpool_map to register the pool's main greenlet
1 parent 9c4c3d2 commit e283490

1 file changed

Lines changed: 16 additions & 1 deletion

File tree

mrq/context.py

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -211,6 +211,10 @@ def subpool_map(pool_size, func, iterable):
211211
current_job = get_current_job()
212212

213213
def inner_func(*args):
214+
""" As each call to 'func' will be done in a random greenlet of the subpool, we need to
215+
register their IDs with set_current_job() to make get_current_job() calls work properly
216+
inside 'func'.
217+
"""
214218
next(counter)
215219
if current_job:
216220
set_current_job(current_job)
@@ -219,9 +223,20 @@ def inner_func(*args):
219223
set_current_job(None)
220224
return ret
221225

226+
def inner_iterable():
227+
""" This will be called inside the pool's main greenlet, which ID also needs to be registered """
228+
if current_job:
229+
set_current_job(current_job)
230+
231+
for x in iterable:
232+
yield x
233+
234+
if current_job:
235+
set_current_job(None)
236+
222237
start_time = time.time()
223238
pool = gevent.pool.Pool(size=pool_size)
224-
ret = pool.map(inner_func, iterable)
239+
ret = pool.map(inner_func, inner_iterable())
225240
pool.join(raise_error=True)
226241
total_time = time.time() - start_time
227242

0 commit comments

Comments
 (0)