classification
Title: multiprocess.pool.map_async callables not working
Type: behavior Stage: resolved
Components: Library (Lib) Versions: Python 3.4, Python 3.3
process
Status: closed Resolution: fixed
Dependencies: Superseder:
Assigned To: Nosy List: Bbb, Janne.Karila, hynek, python-dev, sbt
Priority: normal Keywords: patch

Created on 2012-10-24 07:14 by Bbb, last changed 2012-10-27 11:00 by hynek. This issue is now closed.

Files
File name Uploaded Description Edit
callback.patch Janne.Karila, 2012-10-24 08:57 review
test_callback.patch Janne.Karila, 2012-10-25 10:21 review
map-async-fix-with-tests.diff hynek, 2012-10-26 10:56
Messages (13)
msg173653 - (view) Author: Bbb (Bbb) Date: 2012-10-24 07:14
When using map_async() my callbacks don't get called back.  When using apply_async() they do get called.

CODE:
from multiprocessing import Pool,TimeoutError
from time import sleep

servers=["s1","s2","s3","s4","s5","s6"]
blah = "no callback"

def f(x):
    print("start f(" + x + ")")
    sleep(5)
    print("end   f(" + x + ")")
    return "did " + x

def mycallback(x):
    global blah
    blah = "called back"
    print("My callback " + str(x))
    


def myerrorcallback(r):
    print("My errorcallback " + str(r))


if __name__ == '__main__':
    pool = Pool(processes=7)
    results = pool.map_async(f, servers,  callback=mycallback, error_callback=myerrorcallback)
    print(results.get(timeout=11))
    pool.close()
    pool.join()
    print(blah)

OUTPUT:
D:\python>f.py
start f(s1)
start f(s2)
start f(s3)
start f(s4)
start f(s6)
start f(s5)
end   f(s1)
end   f(s2)
end   f(s3)
end   f(s4)
end   f(s5)
end   f(s6)
['did s1', 'did s2', 'did s3', 'did s4', 'did s5', 'did s6']
no callback

...whereas replacing the code with this:
   with Pool(processes=7) as pool:         # start 4 worker processes
        for server in servers:
            r = pool.apply_async(f, (server,),  callback=mycallback, error_callback=myerrorcallback)
        pool.close()
        pool.join()
        print (blah)


GIVES:
D:\python\f2.py
start f(s1)
start f(s2)
start f(s3)
start f(s4)
start f(s5)
start f(s6)
end   f(s2)
end   f(s1)
My callback did s2
My callback did s1
end   f(s4)
My callback did s4
end   f(s3)
My callback did s3
end   f(s5)
My callback did s5
end   f(s6)
My callback did s6
called back


This is v3.3 on Windows XP.
msg173655 - (view) Author: Janne Karila (Janne.Karila) Date: 2012-10-24 08:14
Indeed, the current implementation map_async simply ignores the callback arguments:

def map_async(self, func, iterable, chunksize=None, callback=None,
        error_callback=None):
    '''
    Asynchronous version of `map()` method.
    '''
    return self._map_async(func, iterable, mapstar, chunksize)
msg173657 - (view) Author: Janne Karila (Janne.Karila) Date: 2012-10-24 08:57
Here's a patch that adds the missing arguments to the call to _map_async.
The bug seems to originate from enhancement #12708 that added starmap_async to 3.3.
msg173673 - (view) Author: Hynek Schlawack (hynek) * (Python committer) Date: 2012-10-24 12:18
Could you add a test please? Thanks!
msg173738 - (view) Author: Janne Karila (Janne.Karila) Date: 2012-10-25 07:01
I'm working on the test. It keeps failing...
msg173752 - (view) Author: Janne Karila (Janne.Karila) Date: 2012-10-25 10:21
Three test cases added. 

The tricky part in writing the tests was to realize that when the tests are run with Manager, the callback goes through a proxy object.
msg173834 - (view) Author: Hynek Schlawack (hynek) * (Python committer) Date: 2012-10-26 10:56
Thanks for taking the time! I remember my frustrations when trying to grok how the mp test suite works. :)

A small nit-pick first: you have a lot of extra white space in your patches. Just run 'make patchcheck' first, that should warn you about that.

Not sure, but the tests look rather complex to me. I’d humbly propose the the simplified test attached, which also ensures that error_callback get only called on errors and callback on success.

Opinions?
msg173838 - (view) Author: Janne Karila (Janne.Karila) Date: 2012-10-26 12:12
Otherwise I agree, but what if one of the callbacks is not called? The test case would fail on assertEqual(2, len(call_args)) but you wouldn't know which callback is the problem.
msg173841 - (view) Author: Hynek Schlawack (hynek) * (Python committer) Date: 2012-10-26 12:59
True, it makes sense to push this assert to the end.
msg173888 - (view) Author: Janne Karila (Janne.Karila) Date: 2012-10-26 20:22
The length assertion protects the test from breaking with IndexError. How about this? Though, it looks like two test cases to me.

+        self.pool.map_async(int, ['1'],
+                            callback=call_args.append,
+                            error_callback=call_args.append).wait()
+        self.assertEqual(1, len(call_args))
+        self.assertEqual([1], call_args[0])
+        self.pool.map_async(int, ['a'],
+                            callback=call_args.append,
+                            error_callback=call_args.append).wait()
+        self.assertEqual(2, len(call_args))
+        self.assertIsInstance(call_args[1], ValueError)
msg173913 - (view) Author: Hynek Schlawack (hynek) * (Python committer) Date: 2012-10-27 09:28
LGTM.

Presuming you want to submit more patches in future, please take the time to sign a Python contributor agreement: http://www.python.org/psf/contrib/ . You'll get a pretty star next to your name in the bug tracker in return. ;)
msg173922 - (view) Author: Roundup Robot (python-dev) Date: 2012-10-27 10:58
New changeset 1bccec4ff980 by Hynek Schlawack in branch '3.3':
#16307: Fix multiprocessing.Pool.map_async not calling its callbacks
http://hg.python.org/cpython/rev/1bccec4ff980
msg173923 - (view) Author: Hynek Schlawack (hynek) * (Python committer) Date: 2012-10-27 11:00
Applied. Thank you for your contribution!
History
Date User Action Args
2012-10-27 11:00:09hyneksetstatus: open -> closed
resolution: fixed
messages: + msg173923

stage: patch review -> resolved
2012-10-27 10:58:12python-devsetnosy: + python-dev
messages: + msg173922
2012-10-27 09:28:17hyneksetmessages: + msg173913
2012-10-26 20:22:21Janne.Karilasetmessages: + msg173888
2012-10-26 12:59:15hyneksetmessages: + msg173841
2012-10-26 12:12:09Janne.Karilasetmessages: + msg173838
2012-10-26 10:56:38hyneksetfiles: + map-async-fix-with-tests.diff

messages: + msg173834
2012-10-25 10:21:27Janne.Karilasetfiles: + test_callback.patch

messages: + msg173752
2012-10-25 07:01:26Janne.Karilasetmessages: + msg173738
2012-10-24 19:36:34sbtsetnosy: + sbt
2012-10-24 12:18:03hyneksetmessages: + msg173673
2012-10-24 11:42:48pitrousetnosy: + hynek
stage: patch review

versions: + Python 3.4
2012-10-24 08:57:07Janne.Karilasetfiles: + callback.patch
keywords: + patch
messages: + msg173657
2012-10-24 08:14:48Janne.Karilasetnosy: + Janne.Karila
messages: + msg173655
2012-10-24 07:14:11Bbbcreate