++ed by:
KEEDI

1 PAUSE user

Elizabeth Mattijsen

NAME

Thread::Pool - group of threads for performing similar jobs

SYNOPSIS

 use Thread::Pool;
 $pool = Thread::Pool->new(
  {
   autoshutdown => 1, # default: 1 = yes
   workers => 5,      # default: 1
   pre => sub {shift; print "starting worker with @_\n",
   do => sub {shift; print "doing job for @_\n"; reverse @_},
   post => sub {shift; print "stopping worker with @_\n",
  },
  qw(a b c)           # parameters to pre-job subroutine
 );

 $pool->job( qw(d e f) );              # not interested in result

 $jobid = $pool->job( qw(g h i) );
 @result = $pool->result( $jobid );    # wait for result to be ready
 print "Result is @result\n";

 $jobid = $pool->job( qw(j k l) );
 @result = $pool->result_dontwait( $jobid ); # do _not_ wait for result
 print "Result is @result\n";          # may be empty when not ready yet

 $pool->add;           # add worker(s)
 $pool->remove;        # remove worker(s)
 $pool->workers( 10 ); # set number of workers

 $workers = $pool->workers; 
 $removed = $pool->removed;
 print "$workers workers available, $removed workers removed\n";
    
 $todo = $pool->todo;
 $done = $pool->done;
 print "$done jobs done, still $todo jobs todo\n";

 $pool->autoshutdown( 1 ); # shutdown when object is destroyed
 $pool->shutdown;          # wait until all jobs done

 @pre = shift->pre; # inside do() and post() only;

DESCRIPTION

The Thread::Pool allows you to set up a group of (worker) threads to execute a (large) number of similar jobs that need to be executed asynchronously. The routine that actually performs the job (the "do" routine), must be specified as a reference to a (anonymous) subroutine.

Once a pool is created, jobs can be executed at will and will be assigned to the next available worker. If the result of the job is important, a job ID is issued. The job ID can then later be used to obtain the result.

Initialization parameters can be passed during the creation of the threads::farm object. The initialization ("pre") routine can be specified as a reference to a (anonymous) subroutine. Parameters returned from the initialization routine are available to the "do" and "post" routine. The "pre" routine can e.g. be used to create a connection to an external source using a non-threadsafe library.

When a worker is told to finish, the "post" routine is executed if available.

Unless told otherwise, all jobs that are assigned, will be executed before the pool is allowed to be destroyed.

METHODS

new

 $pool = Thread::Pool->new(
  {
   do => sub {shift; print "doing with @_\n"; reverse @_}, # must have
   pre => sub {shift; print "starting with @_\n",      # default: none
   post => sub {shift; print "stopping with @_\n",     # default: none
   workers => 5,      # default: 1
   autoshutdown => 1, # default: 1 = yes
  },
  qw(a b c)           # parameters to "pre" subroutine
 );

The "new" method returns the Thread::Pool object.

The first input parameter is a reference to a hash that should at least contain the "do" key with a subroutine reference.

The other input parameters are optional. If specified, they are passed to the the "pre" subroutine whenever a new worker is added.

Each time a worker thread is added, the "pre" subroutine (if available) will be called inside the thread. Its return values are saved and made available to the "do" and "post" routines by means of the pre method. Each time a worker thread is removed, the "post" routine is called. Its return value(s) are saved only if a job ID was requested when removing the thread. Then the result method can be called to obtain the results of the "post" subroutine.

The following field must be specified in the hash reference:

do
 do => 'do_the_job',            # assume caller's namespace

or:

 do => 'Package::do_the_job',

or:

 do => \&SomeOther::do_the_job,

or:

 do => sub {print "anonymous sub doing the job\n"},

The "do" field specifies the subroutine to be executed for each job. It must be specified as either the name of a subroutine or as a reference to a (anonymous) subroutine.

The specified subroutine should expect the following parameters to be passed:

 1     the Thread::Pool object to which the worker thread belongs.
 2..N  any parameters that were passed with the call to L<job>.

Any values that are returned by this subroutine after finishing each job, are accessible with result if a job ID was requested when assigning the job.

The pre method can be called on the Thread::Pool object to gain access to the values returned by the "pre" subroutine.

The following fields are optional in the hash reference:

pre
 pre => 'prepare_jobs',         # assume caller's namespace

or:

 pre => 'Package::prepare_jobs',

or:

 pre => \&SomeOther::prepare_jobs,

or:

 pre => sub {print "anonymous sub preparing the jobs\n"},

The "pre" field specifies the subroutine to be executed each time a new worker thread is started (either when starting the pool, or when new worker threads are added with a call to either add or workers). It must be specified as either the name of a subroutine or as a reference to a (anonymous) subroutine.

The specified subroutine should expect the following parameters to be passed:

 1     the Thread::Pool object to which the worker thread belongs.
 2..N  any additional parameters that were passed with the call to L<new>.

Any values that are returned by this subroutine after initializing the thread, accessible with the pre method inside the "do" and "post" subroutines.

post
 post => 'cleanup_after_worker',        # assume caller's namespace

or:

 post => 'Package::cleanup_after_worker',

or:

 post => \&SomeOther::cleanup_after_worker,

or:

 post => sub {print "anonymous sub cleaning up after the worker removed\n"},

The "post" field specifies the subroutine to be executed each time a worker thread is removed (either when being specifically removed, or when the pool is shutdown specifically or implicitely when the Thread::Pool object is destroyed. It must be specified as either the name of a subroutine or as a reference to a (anonymous) subroutine.

The specified subroutine should expect the following parameters to be passed:

 1     the Thread::Pool object to which the worker thread belongs.
 2..N  any additional parameters that were passed with the call to L<new>.

Any values that are returned by this subroutine after closing down the thread, are accessible with the result method, but only if the thread was removed and a job ID was requested.

workers
 workers => 5, # default: 1

The "workers" field specifies the number of worker threads that should be created when the pool is created. If no "workers" field is specified, then only one worker thread will be created. The workers method can be used to change the number of workers later.

autoshutdown
 autoshutdown => 0, # default: 1

The "autoshutdown" field specified whether the shutdown method should be called when the object is destroyed. By default, this flag is set to 1 indicating that the shutdown method should be called when the object is being destroyed. Setting the flag to a false value, will cause the shutdown method not to be called, causing potential loss of data and error messages when threads are not finished when the program exits.

The setting of the flag can be later changed by calling the autoshutdown method.

job

 $jobid = $pool->job( @parameter );     # saves result
 $pool->job( @parameter );              # does not save result

The "job" method specifies a job to be executed by any of the available workers. Which worker will execute the job, is indeterminate.

The input parameters are passed to the "do" subroutine as is.

If a return value is requested, then the return value(s) of the "do" subroutine will be saved. The returned value is a job ID that should be used as the input parameter to result or result_dontwait.

result

 @result = $pool->result( $jobid );

The "result" method waits for the specified job to be finished and returns the result of that job.

The input parameter is the job id as returned from the job assignment.

The return value(s) are what was returned by the "do" routine. The meaning of the return value(s) is entirely up to you as the developer.

If you don't want to wait for the job to be finished, but just want to see if there is a result already, use the result_dontwait method.

result_dontwait

 @result = $pool->result_dontwait( $jobid );

The "result_dontwait" method returns the result of the job if it is available. If the job is not finished yet, it will return undef in scalar context or the empty list in list context.

The input parameter is the job id as returned from the job assignment.

If the result of the job is available, then the return value(s) are what was returned by the "do" routine. The meaning of the return value(s) is entirely up to you as the developer.

If you want to wait for the job to be finished, use the result method.

todo

 $todo = $pool->todo;

The "todo" method returns the number of jobs that are still left to be done.

done

 $done = $pool->done;

The "done" method returns the number of jobs that has been performed by current and removed worker threads of the pool.

add

 $tid = $pool->add;             # add 1 worker thread
 @tid = $pool->add( 5 );

The "add" method adds the specified number of worker threads to the pool and returns the thread ID's (tid) of the threads that were created.

The input parameter specifies the number of workers to be added. If no number of workers is specified, then 1 worker thread will be added.

In scalar context, returns the thread ID (tid) of the first worker thread that was added. This usually only makes sense if you're adding only one worker thread.

In list context, returns the thread ID's (tid) of the worker threads that were created.

Each time a worker thread is added, the "pre" subroutine (if available) will be called inside the thread. Its return values are saved and made available to the "do" and "post" routines by means of the pre method.

remove

 $pool->remove;                 # remove 1 worker thread
 $pool->remove( 5 );            # remove 5 worker threads

 $jobid = $pool->remove;        # remove 1 worker thread, save result
 @jobid = $pool->remove( 5 );   # remove 5 worker threads, save results

The "remove" method adds the specified number of special "remove" job to the lists of jobs to be done. It will return the job ID's if called in a non-void context.

The input parameter specifies the number of workers to be removed. If no number of workers is specified, then 1 worker thread will be removed.

In void context, the results of the execution of the "post" subroutine(s) is discarded.

In scalar context, returns the job ID of the result of the first worker thread that was removed. This usually only makes sense if you're removing only one worker thread.

In list context, returns the job ID's of the result of all the worker threads that were removed.

Each time a worker thread is removed, the "post" routine is called. Its return value(s) are saved only if a job ID was requested when removing the thread. Then the result method can be called to obtain the results of the "post" subroutine.

workers

 $workers = $pool->workers;     # find out number of worker threads
 $pool->workers( 10 );          # set number of worker threads

The "workers" method can be used to find out how many worker threads there are currently available, or it can be used to set the number of worker threads.

The input value, if specified, specifies the number of worker threads that should be available. If there are more worker threads available than the number specified, then superfluous worker threads will be removed. If there are not enough worker threads available, new worker threads will be added.

The return value is the current number of worker threads.

removed

 $removed = $pool->removed;

The "removed" method returns the number of worker threads that were removed over the lifetime of the object.

autoshutdown

 $pool->autoshutdown( 1 );
 $autoshutdown = $pool->autoshutdown;

The "autoshutdown" method sets and/or returns the flag indicating whether an automatic shutdown should be performed when the object is destroyed.

shutdown

 $pool->shutdown;

The "shutdown" method waits for all jobs to be executed, then removes all worker threads and then returns. It is called automatically when the object is destroyed, unless specifically disabled by providing a false value with the "autoshutdown" field when creating the pool with new, or by calling the autoshutdown method.

pre

 my $self = shift;      # obtain the pool object
 my @pre = $self->pre;  # obtain result of "pre" subroutine
                        # rest of parameters now in @_

The "pre" method only makes sense within the "do" and "post" subroutine that is specified when the pool is created with new. It returns whatever was returned by the "pre" subroutine of pool, as specified when creating the thread with new.

This allows you to pass values and structures that should live for the lifetime of the worker thread in which it is called.

CAVEATS

Passing unshared values between threads is accomplished by serializing the specified values using Storable. This allows for great flexibility at the expense of more CPU usage. It also limits what can be passed, as e.g. code references can not be serialized and therefor not be passed.

AUTHOR

Elizabeth Mattijsen, <liz@dijkmat.nl>.

Please report bugs to <perlbugs@dijkmat.nl>.

COPYRIGHT

Copyright (c) 2002 Elizabeth Mattijsen <liz@dijkmat.nl>. All rights reserved. This program is free software; you can redistribute it and/or modify it under the same terms as Perl itself.

SEE ALSO

threads, Thread::Queue::Any, Storable.