Залежності робочих процесів

Переглянути як Markdown Редагувати на GitHub Документація для 2cc7b2b

Використовуйте очікування на залежності робочих процесів, коли один основний робочий процес має продовжитися лише після завершення інших. Так ви координуватимете кілька робочих процесів, не зводячи все до одного громіздкого визначення.

Такі очікування особливо добре поєднуються з Асинхронними передаваннями: один робочий процес спочатку запускає незалежні дочірні робочі процеси, а потім очікує на них.

Розгорнуті приклади планувальника й виконавця див. у розділі Оркестрація агентів.

Queuety надає три методи конструктора:

wait_for_workflow() для одного робочого процесу
wait_for_workflows() для кількох робочих процесів із семантикою all, any або quorum
wait_for_workflow_group(), якщо ідентифікатори робочих процесів отримано з раніше запущеної групи

Очікування на один робочий процес

PHP

$research_id = Queuety::workflow( 'research_run' )
    ->then( PlanResearchStep::class )
    ->for_each( 'tasks', ExecuteResearchTask::class, 'results' )
    ->dispatch( [ 'brief_id' => 42 ] );

$editorial_id = Queuety::workflow( 'editorial_run' )
    ->wait_for_workflow( $research_id, 'research' )
    ->then( DraftBriefStep::class )
    ->dispatch();

Коли research_run завершиться, editorial_run відновить роботу. Якщо вказати result_key, стан завершеної залежності буде збережено під цим ключем:

PHP

$state['research']; // public state from the completed research workflow

Якщо не вказати result_key, публічний стан залежного робочого процесу буде об’єднано з верхньорівневим станом робочого процесу, що очікує.

Очікування на кілька робочих процесів

PHP

use Queuety\Enums\WaitMode;

Queuety::workflow( 'synthesis_run' )
    ->wait_for_workflows(
        [ $research_id, $editorial_id, $legal_id ],
        WaitMode::All,
        'dependencies'
    )
    ->then( PublishPacketStep::class )
    ->dispatch();

WaitMode::All відновлює роботу, лише коли завершаться всі перелічені робочі процеси.
WaitMode::Any відновлює роботу, щойно завершиться перший із перелічених робочих процесів.
WaitMode::Quorum відновлює роботу, щойно потрібна кількість робочих процесів успішно завершиться.

PHP

Queuety::workflow( 'synthesis_run' )
    ->wait_for_workflows(
        [ $research_id, $editorial_id, $legal_id ],
        WaitMode::Quorum,
        'dependencies',
        quorum: 2,
    )
    ->then( PublishPacketStep::class )
    ->dispatch();

Якщо вказати result_key, стан завершених залежностей у разі очікування на кілька робочих процесів зберігається за ідентифікатором робочого процесу:

PHP

[
    '42' => [ 'summary' => '...' ],
    '43' => [ 'summary' => '...' ],
]

Якщо не вказати result_key, стан завершених залежностей об’єднується з верхньорівневим станом робочого процесу, що очікує, у порядку залежностей.

Отримання ідентифікаторів залежностей зі стану робочого процесу

Ви можете відкласти формування списку залежностей до виконання, отримавши ідентифікатори робочих процесів із публічного стану:

PHP

Queuety::workflow( 'agent_follow_up' )
    ->wait_for_workflow( 'parent_workflow_id', 'parent' )
    ->then( FinalizeAgentRunStep::class )
    ->dispatch( [ 'parent_workflow_id' => $workflow_id ] );

Для кількох залежностей:

PHP

Queuety::workflow( 'aggregator' )
    ->wait_for_workflows( 'dependency_ids', WaitMode::Any, 'winner' )
    ->dispatch( [ 'dependency_ids' => [ 12, 18, 22 ] ] );

Очікування на запущену групу робочих процесів

Якщо попередній крок start_workflows() або start_agents() зберіг довготривалий ключ групи, можна безпосередньо очікувати на цю групу, не передаючи ідентифікатори робочих процесів через публічний стан.

PHP

Queuety::workflow( 'brief_research' )
    ->then( PlanTopicsStep::class )
    ->start_agents( 'agent_tasks', $agent_run, group_key: 'researchers' )
    ->wait_for_agent_group( 'researchers', WaitMode::Quorum, 2, 'agent_results' )
    ->then( SynthesizeBriefStep::class )
    ->dispatch();

Що відбувається під час очікування

Коли робочий процес доходить до одного з таких кроків, його статус змінюється на waiting_for_workflows. Поки він очікує, до нього не прив’язаний жоден воркер.

PHP

$state = Queuety::workflow_status( $workflow_id );
echo $state->status->value; // 'waiting_for_workflows'
echo $state->wait_type;     // 'workflow'
print_r( $state->waiting_for ); // ['42', '43']

Якщо залежність завершилася ще до того, як робочий процес дійшов до кроку очікування, він продовжується негайно.

Семантика помилок

Очікування на залежності робочих процесів орієнтоване на успішне завершення:

wait_for_workflow() і WaitMode::All завершуються помилкою, якщо помилку має будь-яка обов’язкова залежність або її скасовано.
WaitMode::Any завершується помилкою, лише коли всі залежності досягли кінцевого стану, але жодна з них не завершилася успішно.
WaitMode::Quorum завершується помилкою, коли активних робочих процесів, що залишилися, вже недостатньо для досягнення потрібного кворуму.

Завдяки цьому очікування на робочі процеси корисне для систем планувальника й виконавця: один робочий процес може розподіляти роботу, а інший — очікувати на результати all або на першу успішну гілку, що завершиться.