JobQueue(3pm)

📦
version 0.5710 · 4 versions
, in debian bookworm, debian trixie, ubuntu noble, ubuntu resolute, ubuntu stonking

💅 JobQueue(3pm) 💅

okay so this is the manual page for JobQueue(3pm), fresh out of the User Contributed Perl Documentation. It's part one of two, so hydrate, sit criss-cross, and let's get into it. ✨

🦋 NAME 🦋

POE::Component::JobQueue — a component to manage queues and worker pools 🫶

🌈 SYNOPSIS 🌈

use POE qw(Component::JobQueue);

# Passive queue waits for enqueue events.
POE::Component::JobQueue->spawn
  ( Alias         => 'passive',         # defaults to 'queuer'
    WorkerLimit   => 16,                # defaults to 8
    Worker        => \&spawn_a_worker,  # code which will start a session
    Passive       =>
    { Prioritizer => \&job_comparer,    # defaults to sub { 1 } # FIFO
    },
  );

# Active queue fetches jobs and spawns workers.
POE::Component::JobQueue->spawn
  ( Alias          => 'active',          # defaults to 'queuer'
    WorkerLimit    => 32,                # defaults to 8
    Worker         => \&fetch_and_spawn, # fetch a job and start a session
    Active         =>
    { PollInterval => 1,                 # defaults to undef (no polling)
      AckAlias     => 'respondee',       # defaults to undef (no respondee)
      AckState     => 'response',        # defaults to undef
    },
  );

# Enqueuing a job in a passive queue.
$kernel->post( 'passive',   # post to 'passive' alias
               'enqueue',   # 'enqueue' a job
               'postback',  # which of our states is notified when it's done
               @job_params, # job parameters
             );

# Passive worker function.
sub spawn_a_worker {
  my ($postback, @job_params) = @_;     # same parameters as posted
  POE::Session->create
    ( inline_states => \%inline_states, # handwaving over details here
      args          => [ $postback,     # $postback->(@results) to return
                         @job_params,   # parameters of this job
                       ],
    );
}

# Active worker function.
sub fetch_and_spawn {
  my $meta_postback = shift;               # called to create a postback
  my @job_params = &fetch_next_job();      # fetch the next job's parameters
  if (@job_params) {                       # if there's a job to do...
    my $postback = $meta_postback->(@job_params); # ... create a postback
    POE::Session->create                          # ... create a session
      ( inline_states => \%inline_states,  # handwaving over details here
        args          => [ $postback,      # $postback->(@results) to return
                           @job_params,    # parameters of this job
                         ],
      );
  }
}

# Invoke a postback to acknowledge that a job is done.
$postback->( @job_results );

# This is the sub which is called when a postback is invoked.
sub postback_handler {
  my ($request_packet, $response_packet) = @_[ARG0, ARG1];

  my @original_job_params = @{$request_packet};  # original post/fetch
  my @job_results         = @{$response_packet}; # passed to the postback

  print "original job parameters: (@original_job_params)\n";
  print "results of finished job: (@job_results)\n";
}

# Stop a running queue
$kernel->call( 'active' => 'stop' );
ngl read the whole thing before you copy-paste any of it. These are real code blocks, bestie, not vibes. 💜

💖 DESCRIPTION 💖

POE::Component::JobQueue manages a finite pool of worker sessions while those workers handle an arbitrarily large number of tasks. So like — endless tasks, limited workers. Balance. 🧘

It's often used as a form of flow control, stopping a big group of tasks from exhausting some sort of resource. Iconic behaviour, honestly.

🎀 the two flavours of queue 🎀

PoCo::JobQueue implements two kinds of queue: active and passive. Both kinds use a Worker coderef to spawn sessions that process jobs — but the way they use that Worker is not the same, and that's the whole plot twist. 😭

  • Active queues' Worker code fetches a new job from a resource that must be polled. For example, it may read a new line from a file.
  • Passive queues, on the other hand, are given jobs with 'enqueue' events. Their Worker functions are passed the next job as parameters.

🌸 they are not objects, they are sessions 🌸

JobQueue components are not proper objects. Instead of being created, as most objects are, they are "spawned" as separate sessions. To avoid confusion (and hopefully not cause other confusion), they must be spawned wich a spawn method — not created anew with a new one. Yes, "wich". She said what she said and we respect her for it. 💀

⭐ the spawn parameters ⭐

POE::Component::JobQueue's spawn method takes different parameters depending on whether it's going to be an active or a passive session. Regardless, there are a few parameters which are the same for both:

  • Alias => $session_alias — Alias sets the name by which the session will be known. If no alias is given, the component defaults to "queuer". The alias lets several sessions interact with job queues without keeping (or even knowing) hard references to them. It's possible to spawn several queues with different aliases. 👑
  • WorkerLimit => $worker_count — WorkerLimit sets the limit on the number of worker sessions which will run in parallel. It defaults arbitrarily to 8. No more than this number of workers will be active at once. 🚦
  • Worker => \&worker — Worker is a coderef which is called whenever it's time to spawn a new session. What it receives as parameters and what it's expected to do are slightly different for active and passive sessions.
    • Active workers receive just one parameter: a meta-postback. This is used to build a postback once the next job's parameters are known. They're expected to actively fetch the next job's parameters and spawn a new session if necessary. See sub fetch_and_spawn in the SYNOPSIS for an example of an active worker function.>
    • Passive workers' arguments include a pre-built postback and the next job's parameters. Since the JobQueue component already knows what the job parameters are, it's done most of the work for the worker. All that's left is to spawn the session that will process the job. See sub spawn_a_worker in the SYNOPSIS for an example of a passive worker function.

When a postback is called, it posts its parameters (plus the parameters passed when it was created) to the session it belongs to. Postbacks are discussed in the POE::Session manpage. 🫶

🦋 passive-queue-only parameters 🦋

  • Passive => \%passive_parameters — Passive contains a hashref of passive queue parameters. The Passive parameter block's presence indicates that the queue will be passive, but its contents may be empty, since all its parameters are optional:
Passive => { }, # all passive parameters take default values
heads up: a queue can't be both active and passive at the same time. Pick a lane, bestie. 💅

The Passive block takes up to one parameter.

  • Prioritizer => \&prioritizer_function — Prioritizer holds a function that defines how a job queue will be ordered. The prioritizer function receives references to two jobs, and it returns a value which tells the JobQueue component which job should be dealt with first.

In the Unix tradition, lower priorities go first. This transforms the prioritizer into a simple sort function, which it has been modelled after. Like sort's sorter sub, the prioritizer returns -1 if the first job goes before the second one; 0 if both jobs have the same priority; and 1 if the first job goes after the second. It's easier to write an example than to describe it:

sub low_priorities_first {
  my ($first_job, $second_job) = @_;
  return $first_job->{priority} <=> $second_job->{priority};
}

The first argument always refers to the new job being enqueued. 🌟

The default prioritizer always returns 1. Since the first argument always refers to the new job being enqueued, this effects a FIFO queue. Replacing it with a prioritizer that always returns -1 will turn the JobQueue into a stack (last in, first out). 🥞

🌈 active-queue-only parameters 🌈

  • Active => \%active_parameters — Active contains a hashref of active queue parameters. The Active parameter block's presence indicates that the queue will be active, but its contens may be empty since all its parameters are optional.
Active => { }, # all active parameters take default values
same rule, same drama: a queue can't be both active and passive at the same time.

The Active block takes up to three parameters.

  • PollInterval => $seconds — Active Worker functions indicate that they've run out of jobs by failing to spawn new sessions. When this happens, an active queue may go into "polling" mode. In this mode, the Worker is called periodically to see if new jobs have appeared in whatever it's getting them from.

PollInterval, if present, tells the job queue how often to call Worker in the absence of new sessions. If it's omitted, the active queue stops after the first time it runs out of jobs. 😭

  • AckAlias => $alias
  • AckState => $state — AckAlias and AckState tell the active job queue where to send acknowledgements of jobs which have been completed. If one is specified, then both must be. 👯

🍓 how sessions actually talk 🍓

Sessions communicate asynchronously with passive JobQueue components. They post "enqueue" requests to it, and it posts job results back.

Requests are posted to the component's enqueue state. They include the name of a state to post responses back to, and a list of job parameters. For example:

$kernel->post( 'queue', 'enqueue', # queuer session alias & state
               'job_results',      # my state to receive responses
               @job_parameters,    # parameters of the job
             );

Once the job is completed, the handler for 'job_results' will be called with the job parameters and results. See sub postback_handler in the SYNOPSIS for an example results handler. 💖

Active JobQueue components act as event generators. They don't receive jobs from the outside; instead, they poll for them and post acknowledgements as they're completed.

Running queues can be stopped by posting a stop state to the component. Any currently running workers will be allowed to complete, but no new workers will be started. 🛑

$kernel->call( 'queue' => 'stop' ); # Stop the running queue

👀 SEE ALSO 👀

This component is built upon and POE. Please see its source code and the documentation for its foundation modules to learn more. 📚

Also see the test program, t/01_queues.t, in the POE::Component::JobQueue distribution.

🐛 BUG TRACKER 🐛

https://rt.cpan.org/Dist/Display.html?Status=Active&Queue=POE-Component-JobQueue

🧚 REPOSITORY 🧚

http://thirdlobe.com/svn/poco-jobqueue/

🌟 OTHER RESOURCES 🌟

http://search.cpan.org/dist/POE-Component-JobQueue/

👑 AUTHOR & COPYRIGHTS 💖

Okay bestie, here's the receipts section, and it's giving iconic because this component has been thriving since before you were born. Fr fr.

POE::Component::JobQueue is Copyright 1999-2009 by Rocco Caputo. All rights are reserved. ✨

But wait — it's not just locked up in a vault somewhere. POE::Component::JobQueue is free software 🦋, meaning you may redistribute it and/or modify it under the same terms as Perl itself. So go off, remix it, share it, make it yours — just keep those same terms, periodt. 💅

🎀 And that's the whole page, bestie!! We made it to the end, no cap. This was pulled from perl v5.36.0 and dated 2022-12-07, filed under JobQueue(3pm). Now go queue up some jobs and slay. 💖✨🫶
📦 where this page comes from