Pythonは結果を返すときに複数のプロセスを停止しますか？

Question

私は、単純な実証的なノンファインダをPythonで記述しようとしています。

def proof_of_work(b, nBytes): 
    nonce = 0 
    # while the first nBytes of hash(b + nonce) are not 0 
    while sha256(b + uint2bytes(nonce))[:nBytes] != bytes(nBytes): 
     nonce = nonce + 1 
    return nonce 

今私は、この多重処理を行うにしようとしていますので、すべてのCPUコアを使用し、より高速なnonceを見つけることができます。 4コアがある場合は、一人一人がこのようなナンスを計算しますので、

def proof_of_work(b, nBytes, num_of_cpus_running, this_cpu_id): 
    nonce = this_cpu_id 
    while sha256(b + uint2bytes(nonce))[:nBytes] != bytes(nBytes): 
     nonce = nonce + num_of_cpus_running 
    return nonce 

：

core 0: 0, 4, 8, 16, 32 ... 
core 1: 1, 5, 9, 17, 33 ... 
core 2: 2, 6, 10, 18, 34 ... 
core 3: 3, 7, 15, 31, 38 ... 

私の考えはそうのような2つのparams num_of_cpus_runningと this_cpu_idを渡して、複数回proof_of_work multiprocessing.Poolを使用して機能を実行することです

したがって、プロセスの誰かがノンスを見つけたら、見つかったノンスは必要なバイトが0になる可能性のある最も低い値でなければならないという点を考慮して、ナンスの検索を中止します。何らかの理由でCPUがスピードアップし、有効なn最も低い有効ナンスよりも一度高い場合、作業証明は有効ではありません。

唯一のことは、プロセスAがプロセスAによって計算されているノンスよりも低いノンスを見つけた場合にプロセスAが停止する部分だけです。 Bが提供するノンスに到着するまで計算しています。

私は自分自身を正しく説明してくれることを願っています。また、私が書いたものの速い実装があれば、それについて聞いてみたいと思います。どうもありがとうございました！

Answer 1

簡単な方法の1つは、マイクロバッチを使用して回答が見つかったかどうかを確認することです。並列ジョブの開始からオーバーヘッドが発生するバッチが小さすぎると、サイズが大きすぎると、他のプロセスが余分な作業をしてしまい、一方のプロセスがすでに回答を見つけています。各バッチは効率的になるには1〜10秒かかります。

サンプルコード：

from multiprocessing import Pool 
from hashlib import sha256 
from time import time 


def find_solution(args): 
    salt, nBytes, nonce_range = args 
    target = '0' * nBytes 

    for nonce in xrange(nonce_range[0], nonce_range[1]): 
     result = sha256(salt + str(nonce)).hexdigest() 

     #print('%s %s vs %s' % (result, result[:nBytes], target)); sleep(0.1) 

     if result[:nBytes] == target: 
      return (nonce, result) 

    return None 


def proof_of_work(salt, nBytes): 
    n_processes = 8 
    batch_size = int(2.5e5) 
    pool = Pool(n_processes) 

    nonce = 0 

    while True: 
     nonce_ranges = [ 
      (nonce + i * batch_size, nonce + (i+1) * batch_size) 
      for i in range(n_processes) 
     ] 

     params = [ 
      (salt, nBytes, nonce_range) for nonce_range in nonce_ranges 
     ] 

     # Single-process search: 
     #solutions = map(find_solution, params) 

     # Multi-process search: 
     solutions = pool.map(find_solution, params) 

     print('Searched %d to %d' % (nonce_ranges[0][0], nonce_ranges[-1][1]-1)) 

     # Find non-None results 
     solutions = filter(None, solutions) 

     if solutions: 
      return solutions 

     nonce += n_processes * batch_size 


if __name__ == '__main__': 
    start = time() 
    solutions = proof_of_work('abc', 6) 
    print('\n'.join('%d => %s' % s for s in solutions)) 
    print('Solution found in %.3f seconds' % (time() - start)) 

出力（コアi7プロセッサーとラップトップ）：

Searched 0 to 1999999 
Searched 2000000 to 3999999 
Searched 4000000 to 5999999 
Searched 6000000 to 7999999 
Searched 8000000 to 9999999 
Searched 10000000 to 11999999 
Searched 12000000 to 13999999 
Searched 14000000 to 15999999 
Searched 16000000 to 17999999 
Searched 18000000 to 19999999 
Searched 20000000 to 21999999 
Searched 22000000 to 23999999 
Searched 24000000 to 25999999 
Searched 26000000 to 27999999 
Searched 28000000 to 29999999 
Searched 30000000 to 31999999 
Searched 32000000 to 33999999 
Searched 34000000 to 35999999 
Searched 36000000 to 37999999 
37196346 => 000000f4c9aee9d427dc94316fd49192a07f1aeca52f6b7c3bb76be10c5adf4d 
Solution found in 20.536 seconds 

シングルコアでは、それは76.468秒を要しました。とにかく、これは解決策を見つけるのに最も効率的な方法ではありませんが、機能します。例えば、saltが長ければ、塩が吸収された後にSHA-256状態をあらかじめ計算し、そこからブルートフォース検索を続けることができます。バイト配列もhexdigest()より効率的です。

Answer 2

これを行うための一般的な方法はにある：例えば、作業パケットの

1. シンクタンク特定の範囲のための計算を実行するために、範囲は、いくつかのマネージャが締結された作業パケットの後に労働者
2. に作業パケットを配布している第二
3. に0.1秒、長く取ると言うべきではありません、教えて結果を管理して新しい作業パケットを依頼する
4. 作業が完了して結果が見つかった場合は、作業者の結果を受け入れて、それ以上の作業を行わないという信号を与えます - 作業者は今すぐ安全に終了できます

この方法では、マネージャーごとに繰り返しを確認する必要はありません（w hichはすべての処理を遅くします）、またはセッションの途中でスレッドを停止するなどの厄介なことを行います。言うまでもなく、マネージャはスレッドセーフである必要があります。

これは、結果が見つかったとしても、他のワーカーの結果が依然として必要なため、モデルに完全に適合します。

---

ご使用のモデルでは、スレッドが他のスレッドと同期しなくなり、遅れている可能性があります。結果が見つかると、さらに100万回の計算をしたくないです。私はモデルが間違っていると思うので、この質問からこれを繰り返し述べています。実装を修正するのではなく、モデルを修正する必要があります。

Answer 3

マルチ処理.Queue（）を使用できます。 CPU /プロセスごとにキューがある。プロセスがナンスを検出すると、プロセスはそれを他のプロセスのキューに入れます。リストがある

def proof_of_work(b, nBytes, num_of_cpus_running, this_cpu_id, qSelf, qOthers): 
    nonce = this_cpu_id 
    while sha256(b + uint2bytes(nonce))[:nBytes] != bytes(nBytes): 
     nonce = nonce + num_of_cpus_running 
     try: 
      otherNonce = qSelf.get(block=False) 
      if otherNonce < nonce: 
       return 
     except: 
      pass 
    for q in qOthers: 
     q.put(nonce) 
    return nonce 

qOthers：他のプロセスには、whileループの各反復でそのキュー（非ブロッキング）を確認し、その上に何がある場合、彼らは、キュー内の値に基づいて継続するか、または終了することを決定しました他のプロセスに属するキューの数（各キュー=マルチプロセッシング.Queue（））

私が提案したようにキューを使用することに決めた場合は、上記のアプローチのより良い/より良い実装を書くことができます。