問題タブ [concurrent.futures]

ライブがありconcurrent.futures.ThreadPoolExecutorます。その状態を確認したい。スレッドの数、処理中のタスクの数とタスクの数、空いているスレッドの数、キューに入っているタスクを知りたいです。どうすればこれらのことを知ることができますか?

Yahoo Finance から複数のスレッドを使用して大量のデータをダウンロードしようとしています。私はconcurrent.futures.ThreadPoolExecutor物事をスピードアップするために使用しています。利用可能なすべてのファイル記述子 (デフォルトでは 1024) を消費するまで、すべてがうまくいきます。

urllib.request.urlopen()例外が発生すると、ファイル記述子は開いたままになります (使用するソケットのタイムアウトに関係なく) 。通常、このファイル記述子は、単一の (メイン) スレッドからのみ実行する場合に再利用されるため、この問題は発生しません。しかし、これらの例外的なurlopen()呼び出しがThreadPoolExecutorスレッドから行われると、これらのファイル記述子は開いたままになります。これまでに思いついた唯一の解決策ProcessPoolExecutorは、非常に面倒で非効率的なプロセス ( ) を使用するか、許可されるファイル記述子の数を非常に大きなものに増やすことです (私のライブラリのすべての潜在的なユーザーがとにかくこれを行うわけではありません)。 ）。この問題に対処するためのよりスマートな方法が必要です。

また、これはPythonライブラリのバグなのか、それとも何か間違っているのでしょうか...

Debian で Python 3.4.1 を実行しています (テスト、カーネル 3.10-3-amd64)。

これは、この動作を示すサンプル コードです。

Python 2.7 を使用しています。

現在、次のように ThreadPoolExecuter を使用しています。

問題は、f実行時間が長すぎる場合があることです。を実行するたびにf、その実行を 100 秒に制限してから、それを殺します。

最終的に、 の各要素についてx、param強制終了する必要があるかどうかf、また、そうでない場合の戻り値を示したいと思います。1 つのパラメータがタイムアウトしてもf、次のパラメータで実行したい。

このexecuter.mapメソッドにはパラメーターがありますが、各スレッドごとに個別にではなく、timeoutへの呼び出し時から実行全体のタイムアウトを設定します。executer.map

目的の動作を取得する最も簡単な方法は何ですか?

ctypes.CDLL を活用する Python コードがいくつかあります。ドキュメントによると、これには gil は関係ありません。そうは言っても、プロファイリング時に不明なボトルネックがいくつか発生しています。time.sleep または ctypes.windll.kernel32.Sleep を使用していくつかの簡単なコードを実行すると、スレッドの数がタスクの数と一致するため、タイム スケールが等しく表示されます。つまり、タスクが 1 秒間スリープし、 1 つのスレッドで 1 つのタスクを送信するか、20 のスレッドで 20 のタスクを送信すると、どちらも完了までに約 1 秒かかります。

私のコードに戻ると、期待どおりにスケールアウトするのではなく、直線的にスケールアウトしています。プロファイリングは、_thread.lock で acquire() からの待機を示します。

これをさらに掘り下げて、問題がどこに現れているかを確認するためのテクニックは何ですか? ここでは ThreadPoolExecutor が最適な選択ではないでしょうか? 私はそれが基本的なスレッドプールを実装し、multiprocessing.pool の ThreadPool と変わらないことを理解しましたか?

複数のプロセスを実行しようとしていますが、同時に を使用しconcurrent.futures.ProcessPoolExecutorて CPU を集中的に使用するジョブを実行しています。最初のいくつかのリクエストは問題なく処理されますが、その後 から aKeyErrorが発生しconcurrent.futures.process、サーバーがハングします。

これは Tornado のバグですか?

これは、コードを取り除いた最も単純な形式です。

サーバ：

クライアント：

エラーのトレースバック

私が持っているファイルの上部にあるpythonスクリプトがあります：

キューはプロセスセーフなデータ構造であることを学びました。私は方法を持っています：

ここでのperform_scan.delay()関数は実際にセロリ ワーカーを呼び出しますが、関係ないと思います (非同期プロセス呼び出しです)。

私も持っています：

最後に、main()関数があります:

result_queueprint ステートメントの結果は 0 です。しかし、 inのサイズの print ステートメントを含めるとenqueue_tasks、プログラムの実行中に、サイズが増加し、キューに物が追加されていることがわかります。

何が起こっているのか？