apply_async 콜백 함수가 호출되지 않습니다.

Question

저는 파이썬에 초보자입니다. 제 데이터를위한 기능을 계산하고 파일에서 처리하고 작성해야하는 목록을 반환하는 함수가 있습니다. 계산 및 다음 콜백 함수를 사용하여 파일에 쓸 수 있지만 콜백 함수를 호출하지 않는, 나는 그것에 몇 가지 인쇄 문을 넣었지만 그것은 확실히 호출되지 않습니다. 내 코드는 다음과 같습니다

이

def write_arrow_format(results): 
print("writer called") 
results[1].to_csv("../data/model_data/feature-"+results[2],sep='\t',encoding='utf-8') 
with open('../data/model_data/arow-'+results[2],'w') as f: 
    for dic in results[0]: 
     feature_list=[] 
     print(dic) 
     beginLine=True 
     for key,value in dic.items(): 
       if(beginLine): 
       feature_list.append(str(value)) 
       beginLine=False 
       else: 
       feature_list.append(str(key)+":"+str(value)) 
     feature_line=" ".join(feature_list) 
     f.write(feature_line+"\n") 


def generate_features(users,impressions,interactions,items,filename): 
    #some processing 
    return [result1,result2,filename] 





if __name__=="__main__": 
    pool=mp.Pool(mp.cpu_count()-1) 

    for i in range(interval): 
     if i==interval: 
      pool.apply_async(generate_features,(users[begin:],impressions,interactions,items,str(i)),callback=write_arrow_format) 
     else: 
      pool.apply_async(generate_features,(users[begin:begin+interval],impressions,interactions,items,str(i)),callback=write_arrow_format) 
      begin=begin+interval 
    pool.close() 
    pool.join() 

Answer 1

그것은 generate_features에 의해 반환 목록에 포함 된 어떤 게시물에서 분명 아니다. 그러나 result1, result2 또는 filename 중 하나가 직렬화 가능하지 않으면 어떤 이유로 다중 처리 lib가 콜백 함수를 호출하지 않고 자동으로 실패합니다. I 은이라고 생각합니다. 다중 처리 lib가 하위 프로세스와 상위 프로세스간에 개체를 앞뒤로 전달하기 전에 피클 링하려고하기 때문입니다. 반환하는 항목이 "피클 가능"(즉 직렬화되지 않음)이 아닌 경우 콜백이 호출되지 않습니다.

이 버그가 직접 발생하여 문제가되는 로거 객체의 인스턴스로 판명되었습니다.

import multiprocessing as mp 
import logging 

def bad_test_func(ii): 
    print('Calling bad function with arg %i'%ii) 
    name = "file_%i.log"%ii 
    logging.basicConfig(filename=name,level=logging.DEBUG) 
    if ii < 4: 
     log = logging.getLogger() 
    else: 
     log = "Test log %i"%ii 
    return log 

def good_test_func(ii): 
    print('Calling good function with arg %i'%ii) 
    instance = ('hello', 'world', ii) 
    return instance 

def pool_test(func): 
    def callback(item): 
     print('This is the callback') 
     print('I have been given the following item: ') 
     print(item) 
    num_processes = 3 
    pool = mp.Pool(processes = num_processes) 
    results = [] 
    for i in range(5): 
     res = pool.apply_async(func, (i,), callback=callback) 
     results.append(res) 
    pool.close() 
    pool.join() 

def main(): 

    print('#'*30) 
    print('Calling pool test with bad function') 
    print('#'*30) 

    pool_test(bad_test_func) 

    print('#'*30) 
    print('Calling pool test with good function') 
    print('#'*30) 
    pool_test(good_test_func) 

if __name__ == '__main__': 
    main() 

희망이 도움이 올바른 방향을 가리 킵니다 : 여기 내 문제를 재현하기위한 몇 가지 예제 코드입니다.