問題タブ [python-multiprocessing]

multiprocessing と gevent を1つのプログラムで一緒に使用したいのですが、私のコードは次のとおりです:

2 行目と 3 行目をコメントアウトします。コメントをキャンセルすると、プログラムがうまく動作しなくなります。gevent がマルチプロセッシングに影響するのはなぜですか? multiprocessing と gevent の両方を 1 つのプログラム ファイルで使用できるようにするには、どうすれば問題を解決できますか?

ZeroMQ を使用して、Python でローカルに実装されたある種のマルチエージェント システムを実行しています。マスターはメイン スレッド ( python Admin.py) として開始され、このスレッドはワーカー エージェントをmultiprocessing.Processサブプロセスとして開始します。

Admin.py とワーカーの両方が、単純なprintコマンドを使用して出力を書き込みます。ここで、コンソールに表示されるすべての出力を取得しようとします: $ python Admin.py > file.txt. ただし、このステートメントは、Admin.py によって書き込まれた出力のみをフェッチします。stdout に sterr をパイプするなど、さまざまなリダイレクトを試みましたが、stdout を取得しました ( 1>&2 > file.txt) が、結果はありませんでした。この質問は、Satckoverflow (こことここ) でさまざまな形式で尋ねられており、回答にはかなりの追加の Python コーディングが含まれているようです。この問題に対する Bash ソリューションはありますか? コンソールにすべての出力が表示されるのに、Bash がそれをファイルにリダイレクトできないのはなぜですか?

Python スレッドがバイトコードを一度に 1 つずつしか実行できないことは承知していますが、なぜスレッドライブラリはロックを提供するのでしょうか? 一度に 1 つのスレッドしか実行されていない場合、競合状態は発生しないと想定しています。

ライブラリは、ロック、条件、およびセマフォを提供します。これの唯一の目的は実行を同期することですか?

アップデート：

私は小さな実験を行いました：

基本的に、100k スレッドを開始し、1 ずつインクリメントする必要がありました。返された結果は 99993 でした。

a) GIL の同期と競合状態の回避がある場合、結果が 99999 にならないのはなぜですか? b) 100k OS スレッドを開始することさえ可能ですか?

回答を見た後の更新2：

GIL がアトミックにインクリメントするような単純な操作を実行する方法を実際に提供していない場合、それをそこに置く目的は何ですか? 厄介な並行性の問題には役立たないのに、なぜ導入されたのでしょうか? C拡張のユースケースを聞いたことがありますが、誰かがこれを例証できますか?

rpycサービス内で multiprocessing パッケージを使用しようとしてValueError: pickling is disabledいますが、クライアントから公開された関数を呼び出そうとすると取得されます。multiprocesingパッケージが酸洗いを使用してプロセス間で情報を渡すこと、およびrpyc安全でないプロトコルであるため酸洗いが許可されていないことを理解しています。したがって、rpycでマルチプロセッシングを使用するための最良の方法（またはとにかくある場合）がわかりません。rpyc サービス内でマルチプロセッシングを利用するにはどうすればよいですか? サーバー側のコードは次のとおりです。

エラーを生成するクライアント側のコードは次のとおりです。

私はこの次のコードを持っています。

これは decorator と呼ばれる Python モジュールを使用します。

デコレータ モジュールは、datawarhouse 呼び出し関数を装飾するのに役立ちます。したがって、接続のドロップやさまざまな接続ベースの問題に対処する必要はなく、接続をリセットしてタイムアウト後に再試行できます。データ ウェアハウスの読み取りを行うすべての関数をこのメソッドでデコレートするので、無料で再試行できます。

私は次の方法を持っています。

このようなマルチプロセッシングモジュールを使用して、コードをマルチプロセスしようとしています。

すべてがうまくいっているときは正常です。また、リトライ回数以内で直れば正常です。しかし、rety の数を超えると、メイン プロセスでキャッチされない test_retry メソッドで例外が発生します。プロセスは終了し、メイン プロセスによってフォークされたプロセスは孤立したままになります。ここで何か間違ったことをしているのかもしれません。次の問題を解決するための助けを探しています。例外を親プロセスに伝播して、例外を処理し、子供たちを優雅に死なせるようにします。また、子プロセスに正常に終了するように通知する方法を知りたいです。助けてくれてありがとう。

編集：説明するコードを追加しました。

実際の問題は、ジェネレーターが例外をスローしている場合に発生します。pool.imap_unordered() メソッドにラップされている except 句で、ジェネレーターによってスローされた例外をキャッチできません。したがって、例外がスローされた後、メインプロセスはスタックし、子プロセスは永遠に待機します。ここで何が間違っているのかわかりません。

4 つのワーカーでプールを作成し、( を使用して) 何らかのタスクを実行するように設定するとpool.apply_async(..)、 を使用して 内から各プロセスの名前にアクセスできますがmultiprocessing.current_process().name、親プロセスから名前を設定するにはどうすればよいですか (これは主にログ記録用です)。

私はいくつかの異なるコンピューターでマルチプロセッシングを利用しようとしていますが、これはpathos次のことを目的としているようです。しかし、ドキュメントを見ると、クラスターを起動して実行する方法について途方に暮れています。私は探しています：

1. 安全な認証を使用して、リモート サーバーまたはリモート サーバーのセットをセットアップします。
2. リモート サーバーに安全に接続します。
3. pool.map標準のマルチプロセッシング パッケージ (この関連する質問の疑似コードなど) のような単純な API を使用して、リモート サーバーとローカル マシンの両方のすべての CPU にタスクをマップします。

(1) の例が見当たらず、(2) のトンネルの例がわかりません。この例では、実際にはローカルホスト上の既存のサービスに接続していません。また、他の誰かがサーバーに接続するのを防ぐために、この通信に何らかのパスワード/キーを要求できるかどうか/その方法を知りたいです。これが SSH 認証を使用していることは理解していますが、トラフィックがインターネットを通過するときに読み取られないことを保証するだけで、他の誰かがサーバーをハイジャックするのを防ぐために何もしない既存のキーがありません。

マルチプロセッシング プロセスによって更新される PySide GUI を作成しようとしています。たとえば、計算後に更新されるウィンドウにテキストを表示する PySide GUI です。QThread を使用することで、問題なく GUI を更新できます。ただし、QThread の代わりにマルチプロセッシング プロセスを使用して同じことを行おうとすると (sys.exit の直前の 2 行のコードを参照)、エラーが発生します。最小限の例を次に示します。

問題は、明らかに GUI はメイン プロセスからしか操作できないため、新しいプロセスから GUI を操作しようとすると、次のエラーが発生することです。

私の即時の応答は、QThread で計算を実行することでした。しかし、計算自体はかなり重いので、完全に別のプロセス (およびコア) で実行する必要があります。ありがとう！

私は最近、タスクを高速化するためにマルチプロセッシングの実験を始めました。ファジー文字列マッチングを行い、さまざまなアルゴリズムを使用してスコアを計算するスクリプトを作成しました (さまざまなマッチング手法を比較したかったため)。完全なソースはhttps://bitbucket.org/bergonzzi/fuzzy-compare/srcにあります。入力として、ペアに結合された 2 つのファイルが必要です (file1 の各行と file2 の各行)。ペアごとに、あいまい一致スコアが計算されます。

3バージョン作りました。私のレポで提供されているサンプル データ (ペアに結合された後の 697.340 アイテムで構成されています) で実行すると、次のタイミングが得られます。

• シンプルな単一プロセス - 0:00:47
• Pool.map() を使用したマルチプロセス - 0:00:13
• キューを使用したマルチプロセス (プロデューサー/コンシューマー パターン) - 0:01:04

Pool.map() バージョンが Queue バージョンよりもはるかに高速である理由を理解しようとしていますが、これは実際には単純な単一プロセスのものよりも遅いです。

Queues を使用しようとする理由は、Pool.map() バージョンがすべてが終了するまで結果を保持し、最後にファイルに書き込むだけだからです。これは、大きなファイルの場合、最終的に大量のメモリを消費することを意味します。私はこのバージョンについて話しています（ここに貼り付けるコードが多いため、リンクしています）。

これを解決するために、私はそれを生産者/消費者パターンにリファクタリングしました(または少なくとも試みました)。ここでは、最初に両方の入力ファイルを組み合わせてジョブを生成し、コンシューマーが処理するキューに入れます (あいまい一致スコアを計算します)。完了したジョブはアウト キューに入れられます。次に、このキューから完了したアイテムを取得してファイルに書き込む単一のプロセスがあります。このようにすると、理論的には、結果がディスクにフラッシュされるため、それほど多くのメモリは必要ありません。正常に動作しているように見えますが、はるかに遅いです。また、Mac OSX のアクティビティ モニターを見ると、生成している 4 つのプロセスが 100% の CPU を使用していないように見えることにも気付きました (Pool.map() バージョンではそうではありません)。

私が気付いたもう 1 つの点は、私のプロデューサー関数はキューを適切に満たすように見えますが、コンシューマー プロセスは、最初のアイテムが到着するとすぐに動作を開始するのではなく、キューがいっぱいになるまで待機しているように見えることです。私はおそらくそこで何か間違ったことをしている...

参考までに、Queue バージョンに関連するコードの一部を次に示します (ただし、上記のリンク先のレポで完全なコードを確認することをお勧めします)。

ここに私のプロデューサー関数があります:

これはライター関数です。

これは、実際の計算を行うワーカー関数です (ここでは違いがないため、ほとんどのコードを削除しました。完全なソースはリポジトリにあります)。

これは私がプロセスをセットアップする方法です：

マルチプロセッシングが時々遅くなることについてここでかなり読んだことがありますが、これは新しいプロセスの作成と管理のオーバーヘッドに関係していることを知っています。また、実行するジョブが十分に「大きく」ない場合、マルチプロセッシングの効果が見えない場合があります。ただし、この場合、ジョブはかなり大きいと思います。また、Pool.map() バージョンは、はるかに高速であるため、それを証明しているようです。

これらすべてのプロセスを管理し、キュー オブジェクトを渡すときに、何か間違ったことをしているでしょうか? 実行中に必要なメモリの量を最小限に抑えるために、処理中に結果をファイルに書き込むことができるように、これを最適化するにはどうすればよいでしょうか?

ありがとう！