Python namedtuple как аргумент обратного вызова apply_async (..)

Question

Я пишу короткую программу, где я хочу вызвать функцию асинхронно, чтобы она не блокировала вызывающего. Для этого я использую Pool из модуля multiprocessing python.

В вызываемой функции асинхронно Я хочу вернуть namedtuple в соответствии с логикой остальной частью моей программы, но я нахожу, что namedtuple не кажется, поддерживаемый типа, чтобы перейти от порожденного процесса к обратному вызову (вероятно, потому, что он не может быть маринован). Вот минимальный реестр проблемы.

from multiprocessing import Pool 
from collections import namedtuple 

logEntry = namedtuple("LogEntry", ['logLev', 'msg']) 

def doSomething(x): 
    # Do actual work here 
    logCode = 1 
    statusStr = "Message Here" 
    return logEntry(logLev=logCode, msg=statusStr) 

def callbackFunc(result): 
    print(result.logLev) 
    print(result.msg) 

def userAsyncCall(): 
    pool = Pool() 
    pool.apply_async(doSomething, [1,2], callback=callbackFunc) 

if __name__ == "__main__": 
    userAsyncCall() # Nothing is printed 

    # If this is uncommented, the logLev and status are printed as expected: 
    # y = logEntry(logLev=2, msg="Hello World") 
    # callbackFunc(y) 

Кто-нибудь знает, если есть способ передать значение namedtuple возвращения из процесса асинхронного обратного вызова? Есть ли лучший/более питонический подход к тому, что я делаю?

Answer 1

Проблема заключается в том, что дело обстоит иначе для возвращаемого значения namedtuple() и его параметра typename. То есть, существует несоответствие между определением класса именованного кортежа и именем переменной, которое вы ему дали. Вам нужно два, чтобы соответствовать:

LogEntry = namedtuple("LogEntry", ['logLev', 'msg']) 

И обновить return заявление в doSomething() соответственно.

Полный код:

from multiprocessing import Pool 
from collections import namedtuple 

LogEntry = namedtuple("LogEntry", ['logLev', 'msg']) 

def doSomething(x): 
    # Do actual work here 
    logCode = 1 
    statusStr = "Message Here" 
    return LogEntry(logLev=logCode, msg=statusStr) 

def callbackFunc(result): 
    print(result.logLev) 
    print(result.msg) 

def userAsyncCall(): 
    pool = Pool() 
    return pool.apply_async(doSomething, [1], callback=callbackFunc) 

if __name__ == "__main__": 
    c = userAsyncCall() 

    # To see whether there was an exception, you can attempt to get() the AsyncResult object. 
    # print c.get() 

(. Чтобы увидеть определение класса, добавьте verbose=True к namedtuple())

Answer 2

Причина, по которой ничего не печатается, заключается в том, что apply_async не удалось выполнить молча. Кстати, я думаю, что это плохое поведение, которое просто заставляет людей смущаться. Вы можете передать error_callback для обработки ошибки.

def errorCallback(exception): 
    print(exception) 

def userAsyncCall(): 
    pool = Pool() 
    pool.apply_async(doSomething, [1], callback=callbackFunc, error_callback=errorCallback) 
    # You passed wrong arguments. doSomething() takes 1 positional argument. 
    # I replace [1,2] with [1]. 

if __name__ == "__main__": 
    userAsyncCall() 
    import time 
    time.sleep(3) # You need this, otherwise you will never see the output. 

Когда вы пришли сюда, выход

Error sending result: 'LogEntry(logLev=1, msg='Message Here')'. Reason: 'PicklingError("Can't pickle <class '__mp_main__.LogEntry'>: attribute lookup LogEntry on __mp_main__ failed",)' 

PicklingError! Вы правы, namedtuple не может быть передан из порожденного процесса в обратный вызов.

Может быть, это не более accpetable путь, но вы можете отправить dict как результат вместо namedtuple.

Как исправлено Dag Høidahl, namedtuple может быть передан. Работает следующая строка.

LogEntry = namedtuple("LogEntry", ['logLev', 'msg'])