How to assign callback inside own class using python multiprocessing pool apply_async function

Question

I tried to use multithread pool as in this question. But I want pack all logic to my own class as below. The problem occures in apply_async callback function. When I pack all logic in class, callback function seems to never be called. I do not know how to assign callback functions so that it will be called correctly. In source question there is only result in log_result parameters but I must add additional self parameters.

import numpy 
import pandas as pd 
import multiprocessing as mp 
from multiprocessing import freeze_support

class MutliThread() :
    def __init__(self):
        self.result_list = []

    def foo_pool(index, number):
        data = []
        notFound = []
        try :        
            data.append(index + number)
        except Exception:
            notFound.append(index + number)    
        return data

    def log_result(self, result):
        # This is called whenever foo_pool(i) returns a result.
        # result_list is modified only by the main process, not the pool workers.
        self.result_list.append(self, result)

    def apply_async_with_callback(self):
        pool = mp.Pool()
        data = [1,2,3,4,5,6]
        for index, tarrif in enumerate(data) :
            pool.apply_async(self.foo_pool, args = (index, tarrif), callback = self.log_result)
        pool.close()
        pool.join()
        print(self.result_list)

if __name__ == '__main__':
    freeze_support()
    multiThread = MutliThread()
    multiThread.apply_async_with_callback()

score 1 · Accepted Answer · answered Jun 13 '19 at 08:44

The callback in your example is not called, because the tasks fail. An error_callback would be called with a TypeError for each of the tasks: foo_pool() takes 2 positional arguments but 3 were given.

You either have to make foo_pool a normal method by adding self as the first parameter ...

def foo_pool(self, index, number):

... or by decorating it with @staticmethod:

@staticmethod
def foo_pool(index, number):

Fixing this will cause log_result to fail, because you call list.append with two arguments while it only takes one.

Either wrap self and result in a data structure, e.g. a tuple ...

self.result_list.append((self, result))

... or skip appending self alltogether. In the end this is always going to be your MultiThread instance:

self.result_list.append(result)

The name MultiThread is misleading by the way. Your class wraps a process pool, not a thread pool.

How to assign callback inside own class using python multiprocessing pool apply_async function

1 Answers1