|
NAMEPOE::Component::JobQueue - a component to manage queues and worker poolsSYNOPSISuse 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' ); DESCRIPTIONPOE::Component::JobQueue manages a finite pool of worker sessions as they handle an arbitrarily large number of tasks. It often is used as a form of flow control, preventing a large group of tasks from exhausting some sort of resource.PoCo::JobQueue implements two kinds of queue: active and passive. Both kinds of queue use a Worker coderef to spawn sessions that process jobs, but how they use the Worker differs between them. 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. 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. POE::Component::JobQueue's "spawn" method takes different parameters depending whether it's going to be an active or a passive session. Regardless, there are a few parameters which are the same for both:
These parameters are unique to passive queues:
These parameters are unique to active queues:
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 ALSOThis 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 TRACKERhttps://rt.cpan.org/Dist/Display.html?Status=Active&Queue=POE-Component-JobQueueREPOSITORYhttp://thirdlobe.com/svn/poco-jobqueue/OTHER RESOURCEShttp://search.cpan.org/dist/POE-Component-JobQueue/AUTHOR & COPYRIGHTSPOE::Component::JobQueue is Copyright 1999-2009 by Rocco Caputo. All rights are reserved. POE::Component::JobQueue is free software; you may redistribute it and/or modify it under the same terms as Perl itself.POD ERRORSHey! The above document had some coding errors, which are explained below:
Visit the GSP FreeBSD Man Page Interface. |