PythonでThreadPoolExecutorのコンテキストで軽い永続性

Question

ThreadPoolExecutorを使用して高価なジョブを作成するPythonコードがいくつかあります。どのシステムが完了したのかを把握したいので、このシステムを再起動する必要がある、私はすでに完了したものをやり直す必要はありません。シングルスレッドのコンテキストでは、私が行ったことを棚に記しておくことができます。マルチスレッド環境でのこのアイデアのナイーブなポートは次のとおりです。

from concurrent.futures import ThreadPoolExecutor 
import subprocess 
import shelve 


def do_thing(done, x): 
    # Don't let the command run in the background; we want to be able to tell when it's done 
    _ = subprocess.check_output(["some_expensive_command", x]) 
    done[x] = True 


futs = [] 
with shelve.open("done") as done: 
    with ThreadPoolExecutor(max_workers=18) as executor: 
     for x in things_to_do: 
      if done.get(x, False): 
       continue 
      futs.append(executor.submit(do_thing, done, x)) 
      # Can't run `done[x] = True` here--have to wait until do_thing finishes 
     for future in futs: 
      future.result() 

    # Don't want to wait until here to mark stuff done, as the whole system might be killed at some point 
    # before we get through all of things_to_do 

私はこれをどうして取り除けますか？ documentation for shelveにはスレッドの安全性に関する保証が含まれていないので、私は考えません。

これを処理する簡単な方法は何ですか？おそらくdone[x] = Trueをfuture.add_done_callbackに貼ると思いますが、that will often run in the same thread as the future itselfです。おそらく、ThreadPoolExecutorでうまく動作するロック機構がありますか？それは、眠っているループを書いて、完成した先物をチェックすることが私にとってはよりクリーンなようです。

Answer 1

は普通のpythonは、コンテキストマネージャを閉じたときに、それが唯一のディスクに書き込まれるオブジェクト - あなたは最も外側のwithコンテキストマネージャにまだいる間、done棚があり、それはその__exit__メソッドを実行します。したがって、（CPythonを使用している限り）GILのため、他のPythonオブジェクトと同じようにスレッドセーフです。

具体的には、再割り当てdone[x] = Trueはスレッドセーフであり、アトミックに行われます。

シェルフの__exit__メソッドは、Ctrl-Cの後に実行されますが、Pythonプロセスが突然終了し、シェルフがディスクに保存されない場合は表示されません。

この種の障害から保護するため、私はsqllite3のような軽量なファイルベースのスレッドセーフなデータベースを使用することをお勧めします。