apply_async (..) callback에 대한 인수로 Python namedtuple

Question

호출자를 차단하지 않도록 함수를 비동기 적으로 호출하려는 간단한 프로그램을 작성하고 있습니다. 이렇게하려면 파이썬의 multiprocessing 모듈에있는 Pool을 사용하고 있습니다.

비동기 적으로 호출되는 함수에서 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']) 

을 그리고 그에 doSomething()에 return 문을 업데이트 : 당신은 일치하도록이 필요합니다.

전체 코드 :

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

가 (., 클래스 정의를 참조 namedtuple()에 verbose=True를 추가하려면)

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 방법이 아니다, 그러나 당신은 결과 대신 namedtuple로 dict을 보낼 수 있습니다.

Dag Høidahl이 수정됨에 따라 namedtuple이 전달 될 수 있습니다. 다음 줄이 작동합니다.

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