問題タブ [django-celery]

Celery で次のシナリオを実装しようとしています: (同じ) 長時間実行タスクの 2 つのキュー、1 つは「通常」優先度、もう 1 つは「アイドル」優先度。

ワーカーに両方のキューを監視させ、最初に「通常の」優先度キューからタスクを取得し、空の場合は「アイドル」優先度からタスクを取得します。

私の質問は: ワーカーがタスク キューをチェックする順序を保証することは可能ですか? また、これは優先事項を実装するための正しいアプローチですか?

背景: タスクは ffmpeg トランスコーディング ジョブを実行しています。「通常」の優先度は、入ってくる新しいビデオ (できるだけ早くトランスコードする必要があります) であり、「アイドル」の優先度は、古い (40.000 以上のビデオ) アーカイブを更新された形式設定に再トランスコードするタスクです。マルチサーバー タスクのディスパッチに使用できるサーバーがいくつかありません。

topユーティリティプロセスでわかるようにcelery、多くのCPU時間を消費します。だから私はそれをプロファイリングしたいと思います。

私は次のように開発者のマシンで手動でそれを行うことができます：

しかし、正確なタイミングを得るには、本番マシンでプロファイルを作成する必要があります。そのマシン（Fedora 14）では、セロリはinitスクリプトによって起動されます。例えば

私は、これらのスクリプトが最終的に最終的に呼び出すことを理解しましたmanage.py celeryd_multi。celeryd_multiだから私の質問は、プロファイリングを有効にしてセロリを開始するようにどのように伝えることができますか？私の場合、これはに-m cProfile -o out.profオプションを追加することを意味しますpython。

どんな助けでも大歓迎です。

ビューでいくつかの Web サービス呼び出し (私の場合は facebook グラフ API) を行う必要がある場合、django ビューの処理を高速化したいと考えています。これは、処理にかなりの時間がかかります (ビューをレンダリングするだけで約 15 ～ 16 秒かかります)。そのほとんどは Facebook からデータを取得するために費やされます)。だから代わりに

私はやろうと思った：

このようにして、プロフィール情報が取得されるのを待たずに友達データの処理を進めることができます。ただし、セロリ ワーカーがビジー状態の場合、プロファイル データをすぐに取得できない可能性があるため、ビューのレンダリングに以前のアプローチよりもさらに時間がかかる可能性があります。

3 番目のアプローチは、アプローチ 2 と同じことをすべて実行することですが、タスクがまだ開始されていない場合はタスクをキャンセルし、タスクを起動してプロファイル情報を取得する代わりに通常の関数呼び出しを行います。

誰かが Facebook のバッチ リクエストを使用してプロファイルと友人の情報を取得することを提案した場合に備えて: 上記のコードは単なるスニペットであるため、私の場合は不可能です。ユーザーが最初にアプリにアクセスしたときに、実際にミドルウェアでプロファイルを取得しています。

上記の3つのアプローチのうち、どちらが優れているかはわかりません。Web リクエストを並列化する他の方法があるかどうかを提案してください。

失敗したタスクを再試行しようとすると、断続的に (約 20% の確率で) Celery から IOError 例外が発生します。

これが私の仕事です：

例外：

RabbitMQ ログ

なぜこれが起こっているのでしょうか？

ありがとう！

私はCeleryを使用して、作成したCGIアプリケーションからジョブをキューに入れています。私が設定した方法では、Celeryは、またはを設定することにより、各ジョブを一度に1つまたは2つ実行しますCELERYD_CONCURRENCY = 1（= 2したがって、プロセッサが混雑したり、メモリ消費によってスラッシュが発生したりすることはありません）。StackOverflowで得たアドバイスのおかげで、キューはうまく機能します。

これらの各ジョブにはかなりの時間がかかりますが（シリアルで約30分）、恥ずかしい並列化が可能です。このため、私はPool.mapそれを分割して並行して作業を行うために使用していました。コマンドラインからはうまく機能し、新しいメニーコアチップを使用して約5分のランタイムを取得しました。

残念ながら、デーモンプロセスがサブプロセスを持つことを許可しないいくつかの制限があり、CGIキュー内で派手な並列化されたコードを実行すると、次のエラーが発生します。

AssertionError：デーモンプロセスに子を含めることは許可されていません

他の人も同様の質問をしていることに気づきましたがPool.map、完全に放棄して、より複雑なスレッドコードを作成する必要のない答えを見つけることができません。

ここでの適切な設計の選択は何ですか？Celeryキューを使用してシリアルジョブを簡単に実行できます。キューなしで、はるかに高速な並列化ジョブを実行することもできます。これにどのようにアプローチすればよいですか？また、必要なもの（キューとジョブごとの並列化の両方）を取得することは可能ですか？

私が持っていたいくつかのアイデア（いくつかはかなりハッキーです）：

• Celeryキューに送信されたジョブは、コマンドラインプログラムを呼び出すだけです。そのプログラムは、プールを好きなように使用して、結果の数値とデータをファイルに保存できます（現在と同じように）。
  欠点：ジョブのステータスを確認したり、正常に終了したかどうかを確認したりできません。また、CGIからのシステムコールはセキュリティの問題を引き起こす可能性があります。
• 明らかに、キューが非常にジョブでいっぱいの場合は、CPUリソースを利用できます（CELERYD_CONCURRENCY = 6程度に設定することにより）。これにより、一度に多くの人が「列の先頭に立つ」ことができます。
  欠点：各ジョブはキューの先頭で多くの時間を費やします。キューがいっぱいでない場合、スピードアップはありません。また、部分的に終了したジョブの多くは、より多くのRAMを使用して、同時にメモリに保存されます。
• Celeryの@taskを使用して、サブジョブ内で並列化します。次に、CELERYD_CONCURRENCY = 1に設定する代わりに、6に設定します（または、一度にメモリ内で許可したい多くのサブジョブ）。
  欠点：まず第一に、これが「タスク内タスク」の問題をうまく回避できるかどうかはわかりません。ただし、キューの位置の概念が失われる可能性があり、部分的に終了したジョブの多くが一度にメモリに保存される可能性があります。
• おそらく、Pool.mapを呼び出して、スレッドが非デーモンであることを指定する方法がありますか？または、Pool.mapの代わりに使用できるより軽量なものがありますか？これは、別の開いているStackOverflowの質問で採用されたアプローチに似ています。また、Pool.mapを介して利用する並列化は線形代数に似ており、プロセス間通信はありません（それぞれが独立して実行され、他のユーザーと通信せずに結果を返します）。
• Celeryを破棄し、multiprocessing.Queueを使用します。次に、使用するすべてのスレッドに同じ「スレッドの深さ」を使用する方法があるかもしれません（つまり、すべてのスレッドが同じプールを使用して、ネストを回避できる可能性があります）。

よろしくお願いします。

私は、セロリ、rabbitmq を使用するプロジェクトに取り組んでいます。ワーカー(celeryd)にタスクをキューにプッシュする間隔を制御する権利が必要です。

同じタスクを 5 秒後に実行するようにスケジュールする task:recursive_task がありますが、何らかの理由でこのタスクがクラッシュした場合は、再度実行する必要があります。ほぼすべてのシナリオをキャッチしましたが、将来何が起こるかはわかりません。

最初に繰り返しタスクを作成しました:recursive_task のステータスをチェックし、それが長時間実行されなかったかどうか、および正常に完了したかどうかをチェックしますが、これは正しくありませんでした。では、この問題をどのように解決しますか？

Windowsシステムでのdjangoタスクのスケジューリングにセロリを使用しています。何ヶ月も問題なく順調に進んでいます。しかし、今日、再起動後 (これらはほぼ毎日発生します)、セロリを実行できません。Celerybeat は正常に起動しますが、Celeryd を実行するとロードが開始されますが、エラーが発生することなく Windows コマンド ラインに戻ります。

エラーがあると診断が非常に簡単になりますが、エラーがなければどこから始めればよいかわかりません。セロリのことはほとんど何も知らないのですが、すべての設定をした人はとうの昔に会社を辞めてしまったので、どうしたらいいのか途方に暮れています。

問題を診断するための支援は大歓迎です! 提供する必要がある情報がある場合はお知らせください。

編集

次のようなログ ファイルを見つけました。

ありがとう！

私はセロリを初めて使用し、何か間違ったことをしている可能性がありますが、セロリを正しく構成する方法を理解するためにすでに多くの時間を費やしました。

したがって、私の環境には 2 つのリモート サーバーがあります。1つはメイン（パブリックIPアドレスがあり、データベースサーバー、rabbitmqサーバー、Webアプリケーションを実行するWebサーバーなどのほとんどのものがあります）で、もう1つはセロリを使用してメインサーバーから非同期的に呼び出したい特定のタスクに使用されます。

私は、RabbitMQ をブローカーとして、また結果のバックエンドとして使用することを計画していました。Celery の設定は非常に基本的なものです。

メインサーバーでワーカーを実行している場合、タスクは正常に実行されますが、リモートサーバーで実行している場合、いくつかのタスクしか実行されず、ワーカーがスタックしてタスクを実行できなくなります。ワーカーを再起動すると、さらにいくつかのタスクが実行され、再びスタックします。タスク内に特別なことは何もなく、2 つの数字を追加するだけのテスト タスクも試してみました。ワーカーを別の方法で実行しようとしました (デーモン化するかしないか、異なる同時実行性を設定し、celeryd_multi を使用する)、実際には何も役に立ちませんでした。

その理由は何ですか？私は何か見落としてますか？ブローカー (RabbitMQ) 以外のメイン サーバーで何かを実行する必要がありますか? それとも、セロリのバグですか?

うーん...ローカルワーカーで同じ問題を再現したばかりなので、それが何であるかはよくわかりません.N個のタスクが実行されるたびにセロリワーカーを再起動する必要がありますか?

どんな助けでも大歓迎です:)

このスクリプトをdjangoシェルで実行する：

このエラーが発生します：

奇妙なことに、実行すると、chorus.processors.topics.tasks.testが[タスク]のプリントアウトに確実に含まれます。

なぜエラーが発生するのですか？