Към съдържанието

Зависимости между работни потоци

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

Използвайте изчакване на зависимости между работни потоци, когато един главен работен поток трябва да продължи едва след завършването на други работни потоци. Така можете да координирате различни работни потоци, без да събирате всичко в една огромна дефиниция.

Тези изчаквания са особено подходящи в комбинация с Асинхронно предаване, при което един работен поток първо стартира независими дъщерни работни потоци и изчаква завършването им по-късно.

За по-подробни примери с планировчик и изпълнител вижте Оркестрация на агенти.

Queuety предоставя три метода на конструктора:

  • wait_for_workflow() за един работен поток
  • wait_for_workflows() за няколко работни потока със семантика all, any или quorum
  • wait_for_workflow_group(), когато ID-тата на работните потоци идват от вече стартирана група

Изчакване на един работен поток

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 или първия завършил успешен клон.