apply_asyncコールバック関数が呼び出されていません

Question

私はPythonの初心者です。私は自分のデータのための機能を計算し、処理してファイルに書き込むべきリストを返す関数を持っています。計算してからコールバック関数を使ってファイルに書き出しますが、コールバック関数はコールされていません。いくつかのprint文を記述していますが、呼び出されていません。 私のコードは次のようになります。

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 thinkこれは、マルチプロセスlibが子プロセスと親プロセスの間でオブジェクトを前後に渡す前にオブジェクトをpickleしようとするためです。返すものが「pickleable」ではない（つまりシリアル化できない）場合、コールバックは呼び出されません。

私はこのバグを自分で見つけました。そして、それは私にトラブルを与えていたロガーオブジェクトのインスタンスであることが判明しました。問題を再現するためのサンプルコードは次のとおりです。

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

これはうまくいけば、この方向に向いています。