コンテンツへスキップ

リソース管理

Markdownとして表示 GitHubで編集 ドキュメント: 2cc7b2b

Queuetyはリソース制御を4つの中核レイヤーに分け、その上にオプションのプールサイズ調整を加えています。

第1レイヤーでは、作業単位あたりのコストを定義します。第2レイヤーでは、1つのキューまたはプロバイダーグループが同時に負担できる共有負荷を制限します。第3レイヤーでは、1回のワークフロー実行で消費できる量を定めます。第4レイヤーでは、ワーカーが取得した次のジョブをすぐに開始するか、現在のプロセスとコンテナに十分な余力ができるまで延期するかを決めます。

各レイヤーは連携して動作します。ハンドラーは負荷の高い共有プールに属することを宣言でき、キューはプロバイダーごとの重み付き作業量に上限を設けられ、ワークフローは作成する作業量を制限できます。さらにワーカーは、現在のプロセスやコンテナがすでに限界に近い場合、その作業の開始を拒否できます。

作業単位あたりのコストを定義する

ジョブ、ワークフローのステップ、状態遷移時のアクションは、同じ3つのリソースヒントを公開できます。

  • concurrency_group
  • concurrency_limit
  • cost_units

concurrency_groupとconcurrency_limitは、下流の1つのシステムにかかる共有負荷を表します。複数のハンドラーが同じプロバイダー、APIアカウント、またはテナントを利用する場合、同じグループを指定することで、ワーカーが一度に開始する数を抑えられます。

cost_unitsは相対的な重みを表します。キャッシュの更新も大規模なプロバイダー呼び出しも、どちらも「1つのジョブ」ですが、ワークフロー予算やワーカーの実行許可で作業の追加が安全かどうかを判断する際に、同じ重みで数えるべきとは限りません。

PHP
final class CallProviderStep implements \Queuety\Step
{
    public function handle( array $state ): array
    {
        // ...
    }

    public function config(): array
    {
        return [
            'concurrency_group' => 'providers',
            'concurrency_limit' => 3,
            'cost_units' => 5,
        ];
    }
}

同じ仕組みは、クラスプロパティによるディスパッチ可能なジョブの指定や、PendingJob::concurrency_group()とPendingJob::cost_units()による個々のディスパッチにも適用されます。

共有キューとプロバイダーグループの予算を設定する

共有負荷の制御が、単純な並行数の制限とは限りません。すでに実行中の処理の重みの合計が、負荷の大きさを決める場合もあります。

Queuetyは、キューとグループに重み付き予算を設定できます。キューの予算はキュー名で管理されます。グループの予算はconcurrency_groupで管理されるため、無関係なハンドラー同士でも同じ共有枠に参加できます。

PHP
define( 'QUEUETY_RESOURCE_QUEUE_COST_BUDGETS', [
    'providers' => 20,
    'exports' => 8,
] );

define( 'QUEUETY_RESOURCE_GROUP_COST_BUDGETS', [
    'openai' => 12,
    'tenant:acme' => 6,
] );

ワーカーがジョブを取得すると、そのキューまたはグループに属する実行中のprocessing行を確認します。現在実行中のコストに取得したジョブのcost_unitsを加えると設定済みの予算を超える場合、ワーカーはジョブを解放し、後で再試行します。

これはconcurrency_limitとは異なる制御方法です。プロバイダーは同時に3件のリクエストを許可しながら、コスト単位の合計は小さなリクエストより大きなリクエストの方を少なくできる場合があります。

1つのワークフローを予算内に収める

ワークフロー予算は、重いジョブが1つある場合ではなく、実行中に想定以上の作業を発見し続けてしまう場合に役立ちます。

これはエージェント型システムでよく起こります。プランナーが同じ高コストの分岐を何度も再訪したり、サブタスクを作りすぎたり、通常の再試行モデルの範囲内でありながら子ワークフローを次々に開始し続けたりすることがあります。

Queuetyでは、これに対応するワークフロー単位の制限を設定できます。

PHP
Queuety::workflow( 'research_run' )
    ->max_transitions( 20 )
    ->max_for_each_items( 12 )
    ->max_state_bytes( 32768 )
    ->max_cost_units( 40 )
    ->max_started_workflows( 8 )
    ->then( PlanResearchStep::class )
    ->for_each( 'tasks', ExecuteResearchTask::class, 'results' )
    ->dispatch();

max_cost_units()は、実行で消費する重み付き作業量を意図した範囲内に保ちます。max_started_workflows()は、親が無制限に増えるトップレベルの子実行ツリーになるのを防ぎます。暴走するワークフローのコストは一様ではないため、他の保護策も引き続き重要です。

ワークフローのステータスには、設定された制限と現在の使用量が公開budgetペイロードで示されるため、制限の適用中も実行状況を確認できます。

プロセスがすでに逼迫しているときに、ワーカーに作業を延期させる

メタデータとワークフロー予算を適切に設定していても、ジョブを実行しようとしているワーカープロセスが最終的な判断を下します。

ここでリソースを考慮した実行許可が役立ちます。取得したジョブの開始前に、Queuetyは次の項目を確認できます。

  • ジョブの共有並行処理グループ
  • 重み付きキューとプロバイダーグループの予算
  • ハンドラーの最近の実行時間とピークメモリ使用量の履歴
  • 現在のワーカーに残っているメモリ余力
  • 現在のコンテナまたはホストに残っているメモリ余力
  • ワンショット形式の実行で使える残り時間

ワーカーが設定済みの上限に近づきすぎている場合、Queuetyは正常に完了しそうにない作業を開始せず、ジョブを解放します。

この仕組みは意図的に保守的に設計されています。完全なクラスター スケジューラーではなく、ビジー状態のワーカーが実行中のプロセスに過剰な負荷をかけないようにする安全レイヤーです。メモリの計測データが利用可能な場合は、周囲のコンテナやホストについても同様に保護します。

コンテナまたはホストの確認はオプションです。QUEUETY_RESOURCE_SYSTEM_MEMORY_AWARENESSを有効にすると、Queuetyはまずcgroupのメモリデータを読み取り、Linuxホストでは/proc/meminfoにフォールバックします。信頼できるシステムレベルのスナップショットを利用できない場合も、ワーカーはプロセス単位のチェックを続行します。

キャパシティを考慮しながらワーカープールを拡張する

固定サイズのワーカープールは把握しやすい一方、閑散時と急増時の両方に同じ数を選ぶ必要があります。

適応型プールでは、代わりに下限と上限を設定できます。

シェル
wp queuety work --queue=providers --min-workers=2 --max-workers=6

親プロセスは常に最小数の子プロセスを稼働させ、取得可能な滞留ジョブが増えると最大数に向けて拡張し、キューが設定済みのアイドル猶予期間にわたって静かな状態になると縮小します。

拡張時も利用可能なキャパシティが考慮されます。システムメモリの監視が有効で、親プロセスがワーカー1つ分のメモリを確保できない場合、子プロセスを増やさず現在のプールサイズを維持します。

各レイヤーの組み合わせ例

プロバイダーへの呼び出しを計画し、専門のワークフローを開始した後に結果を統合する調査ワークフローを想定してください。

プロバイダーを呼び出すステップでは、cost_unitsと共有concurrency_groupを公開し、複数のワーカーが同じモデルプロバイダーに殺到するのを防げます。キューでは、そのプロバイダーへのトラフィックに重み付き予算を設定できます。親ワークフローではmax_cost_units()とmax_started_workflows()に上限を設け、プランナーが実行をいつまでも拡張しないようにできます。さらにワーカーは、最近のピークメモリデータからプロセスまたはコンテナがすでに上限に近いと判断した場合、新たに取得したプロバイダー呼び出しを延期できます。

これにより、次の4つの制御ができます。

  1. 作業の重みを定義する
  2. キューまたはプロバイダーの共有キャパシティを確保する
  3. 1回の実行で作成できる作業に上限を設ける
  4. 負荷の高いワーカーが過剰に作業を開始しないようにする

詳細なAPIの参照先

このページでは概念を説明しています。

具体的な設定項目については、次を参照してください。

  • Workersは受付動作と実行時定数を説明します
  • CLIは固定および適応型ワーカープールのコマンドを説明します
  • Workflow Guardrailsはワークフローの予算を説明します
  • Jobsはジョブとハンドラーのメタデータを説明します
  • State Machinesは状態アクションのリソースメタデータを説明します
  • APIはメソッドと戻り値の形式を正確に列挙します