Gestionarea resurselor

Queuety împarte controlul resurselor în patru straturi de bază, peste care se poate adăuga dimensionarea opțională a grupului.

Primul strat descrie cât de costisitoare este o unitate de lucru. Al doilea limitează presiunea partajată pe care o poate susține simultan o coadă sau un grup de furnizori. Al treilea definește cât poate consuma o singură execuție a unui flux de lucru. Al patrulea decide dacă un worker ar trebui să pornească imediat următoarea sarcină preluată sau să amâne pornirea până când procesul și containerul curent dispun de suficientă rezervă.

Aceste straturi funcționează împreună. Un handler poate declara că aparține unui grup partajat costisitor, o coadă poate rezerva o anumită cantitate de lucru ponderat pentru un furnizor, un flux de lucru poate limita cantitatea de astfel de lucru pe care este dispus să o creeze, iar workerul poate refuza în continuare să pornească mai mult din acest lucru când procesul sau containerul curent este prea aproape de limite.

Descrierea costului unei unități de lucru

Sarcinile, pașii fluxurilor de lucru și acțiunile la intrarea într-o stare pot publica aceleași trei indicii despre resurse:

• concurrency_group
• concurrency_limit
• cost_units

concurrency_group și concurrency_limit descriu presiunea partajată asupra unui sistem din aval. Dacă mai mulți handleri accesează același furnizor, cont API sau client, pot folosi același grup pentru ca workerii să nu pornească prea multe operațiuni simultan.

cost_units descrie ponderea relativă. Reîmprospătarea memoriei cache și un apel amplu către un furnizor pot fi ambele „o sarcină”, dar nu ar trebui să contribuie în mod egal când bugetul fluxului de lucru sau regula de admitere a workerilor decide dacă este sigur să înceapă și mai multă muncă.

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,
        ];
    }
}

Aceeași idee se aplică sarcinilor care pot fi lansate, prin proprietățile clasei, și lansărilor individuale, prin PendingJob::concurrency_group() și PendingJob::cost_units().

Bugetarea cozilor și a grupurilor de furnizori partajate

Presiunea partajată nu se referă întotdeauna la un simplu număr de execuții concurente. Uneori, partea costisitoare este ponderea cumulată a operațiunilor deja în curs.

Queuety oferă această funcționalitate prin bugete ponderate pentru cozi și grupuri. Bugetele cozilor sunt asociate cu numele cozii. Bugetele grupurilor sunt asociate cu concurrency_group, ceea ce permite handlerilor fără legătură să folosească același cadru de resurse partajate.

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

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

Când preia o sarcină, workerul verifică rândurile processing active care aparțin deja acelei cozi sau acelui grup. Dacă costul lucrărilor în curs, adunat cu cost_units al sarcinii preluate, ar depăși bugetul configurat, workerul eliberează sarcina și încearcă din nou mai târziu.

Astfel, obțineți un tip de control diferit de cel oferit de concurrency_limit. Un furnizor poate permite trei solicitări concurente, dar poate aloca în continuare mai puține unități de cost totale solicitărilor mai mari decât celor mai mici.

Încadrarea unui flux de lucru într-un buget

Bugetele fluxurilor de lucru sunt utile când problema nu este o singură sarcină grea, ci o execuție care continuă să descopere mai mult de lucru decât era prevăzut.

Acest lucru este des întâlnit în sistemele cu agenți. Un planificator poate reveni asupra aceleiași ramuri costisitoare, o poate împărți în prea multe sarcini secundare sau poate continua să pornească fluxuri de lucru copil, respectând totuși un model obișnuit de reîncercare.

Queuety oferă limite la nivel de flux de lucru pentru acest caz:

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() împiedică execuția să consume mai mult lucru ponderat decât s-a prevăzut. max_started_workflows() împiedică un părinte să devină un arbore fără limite de execuții copil de nivel superior. Celelalte măsuri de protecție rămân importante, deoarece nu toate fluxurile de lucru scăpate de sub control au același tip de cost.

Starea fluxului de lucru expune limitele configurate și utilizarea curentă prin payloadul public budget, astfel încât execuția poate fi inspectată cât timp aceste limite sunt active.

Permiteți workerilor să amâne lucrul când procesul este deja prea încărcat

Chiar și atunci când bugetele pentru metadate și fluxuri de lucru sunt configurate corect, decizia finală îi aparține tot procesului worker care urmează să execute jobul.

Aici intervine admiterea ținând cont de resurse. Înainte să înceapă un job preluat, Queuety poate verifica:

• grupul de concurență partajat al jobului
• bugetele ponderate pentru coadă și grupul de furnizori
• istoricul recent al duratei și al consumului maxim de memorie pentru handler
• memoria rămasă disponibilă în workerul curent
• memoria rămasă disponibilă în containerul sau gazda curentă
• intervalul de timp rămas pentru execuțiile de tip one-shot

Dacă workerul este deja prea aproape de limita configurată, Queuety eliberează jobul în loc să înceapă o activitate care probabil nu se va încheia fără probleme.

Această abordare este intenționat conservatoare. Nu este un planificator complet pentru cluster. Este un nivel de protecție care ajută un worker ocupat să evite suprasolicitarea procesului în care rulează și, când sunt disponibile date despre memorie, a containerului sau gazdei din jurul lui.

Componenta pentru container sau gazdă este opțională. Când QUEUETY_RESOURCE_SYSTEM_MEMORY_AWARENESS este activată, Queuety încearcă mai întâi să citească datele de memorie cgroup și, pe gazdele Linux, revine la /proc/meminfo. Dacă nu este disponibilă nicio instantanee fiabilă la nivel de sistem, workerul continuă să folosească verificările locale procesului.

Scalați grupurile de workeri fără a ignora capacitatea

Grupurile fixe de workeri sunt ușor de înțeles, dar vă obligă să alegeți o singură valoare atât pentru perioadele liniștite, cât și pentru vârfurile de activitate.

Grupurile adaptive vă permit să setați în schimb o limită minimă și una maximă:

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

Procesul părinte menține activ cel puțin numărul minim de procese copil, crește numărul spre maxim când se acumulează joburi eligibile în coadă și îl reduce după ce coada a rămas liniștită pe durata perioadei de grație pentru inactivitate configurate.

Scalarea în sus ține cont în continuare de capacitatea disponibilă. Dacă monitorizarea memoriei sistemului este activată și procesul părinte nu vede suficientă memorie rămasă pentru încă un worker complet, păstrează grupul la dimensiunea curentă în loc să pornească mai multe procese copil.

O modalitate realistă de a combina straturile

Imaginați-vă un flux de cercetare care planifică apeluri către furnizori, pornește fluxuri de lucru specializate și apoi sintetizează rezultatul.

Pasul care comunică cu furnizorul poate publica cost_units și un concurrency_group partajat, astfel încât mai mulți workeri să nu trimită simultan cereri către același furnizor de modele. Coada poate rezerva un buget ponderat pentru traficul către acel furnizor. Fluxul de lucru părinte poate limita max_cost_units() și max_started_workflows(), astfel încât un planificator să nu poată extinde execuția la nesfârșit. Apoi, workerul poate amâna totuși un apel către furnizor tocmai preluat, dacă datele recente despre consumul maxim de memorie arată că procesul sau containerul este deja prea aproape de limită.

Astfel, aveți patru controale separate:

1. descrieți ponderea activității
2. rezervați capacitate partajată pentru coadă sau furnizor
3. limitați ce poate crea o singură execuție
4. împiedicați un worker solicitat să pornească mai mult decât ar trebui

Unde găsiți API-urile detaliate

Această pagină oferă o prezentare conceptuală.

Pentru opțiunile concrete:

• Workers descrie comportamentul de admitere și constantele de execuție
• CLI descrie comenzile pentru grupurile fixe și adaptive de workeri
• Workflow Guardrails descrie bugetele fluxurilor de lucru
• Jobs descrie metadatele joburilor și handlerelor
• State Machines descrie metadatele resurselor acțiunilor de stare
• API enumeră metodele exacte și structurile valorilor returnate