Thread::Pool(3pm)

📦

version 0.36 · 4 versions

0.36 · debian trixie, ubuntu resolute, ubuntu stonking (latest)
0.35 · ubuntu jammy
0.35 · ubuntu noble
0.35 · debian bookworm

, in debian trixie, ubuntu resolute, ubuntu stonking

okay so bestie, you found the Thread::Pool(3pm) manual — no cap, we're making it make sense. 💅✨

🌸 NAME 🌸

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

⭐ VERSION ⭐

This documentation describes version 0.36.

✨ SYNOPSIS ✨

use Thread::Pool;
$pool = Thread::Pool->new(
 {
  optimize => 'cpu', # default: 'memory'

  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",

  stream => sub {shift; print "streamline with @_\n",

  monitor => sub { print "monitor with @_\n",
  pre_post_monitor_only => 0, # default: 0 = also for "do"

  checkpoint => sub { print "checkpointing\n" },
  frequency => 1000,

  autoshutdown => 1, # default: 1 = yes

  workers => 10,     # default: 1
  maxjobs => 50,     # default: 5 * workers
  minjobs => 5,      # default: maxjobs / 2
 },
 qw(a b c)           # parameters to "pre" and "post" routine
);

$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

$jobid = $pool->job( qw(j k l) );
@result = $pool->result_dontwait( $jobid ); # do _not_ wait for result

@result = $pool->waitfor( qw(m n o) ); # submit and wait for result

$pool->add;           # add worker(s)
$pool->remove;        # remove worker(s)
$pool->workers( 10 ); # adapt number of workers
$pool->join;          # wait for all removed worker threads to finish

$workers = $pool->workers;
$todo    = $pool->todo;
$removed = $pool->removed;

$pool->maxjobs( 100 );  # adapt or (de-)activate job throttling
$pool->minjobs( 10 );

$pool->autoshutdown( 1 ); # shutdown when object is destroyed
$pool->shutdown;          # wait until all jobs done
$pool->abort;             # finish current job and remove all workers

$done    = $pool->done;   # simple thread-use statistics
$notused = $pool->notused;

Thread::Pool->remove_me;  # inside "do" only

💖 DESCRIPTION 💖

* A note of CAUTION * This module only functions on Perl versions 5.8.0 and later. And then only when threads are enabled with -Dusethreads. It is of no use with any version of Perl before 5.8.0 or without threads enabled. *

okay so Thread::Pool lets you set up a whole squad of (worker) threads to execute a (large) number of similar jobs that need to run asynchronously. The routine that actually performs the job — the "do" routine — must be specified as a name or a reference to a (anonymous) subroutine. no vibes-only jobs, bestie.

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 Thread::Pool object. The initialization ("pre") routine can be specified as a name or as a reference to a (anonymous) subroutine. 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.

Results of jobs must be obtained separately, unless a "stream" or a "monitor" routine is specified. Then the result of each job will be streamed to the "stream" or "monitor" routine in the order in which the jobs were submitted.

Unless told otherwise, all jobs that are assigned, will be executed before the pool is allowed to be destroyed. If a "stream" or "monitor" routine is specified, then all results will be handled by that routine before the pool is allowed to be destroyed.

👑 CLASS METHODS 👑

The following class methods are available.

💅 new

$pool = Thread::Pool->new(
 {
  optimize => 'cpu',                            # default: memory

  do => sub { print "doing with @_\n" },        # must have
  pre => sub { print "starting with @_\n",      # default: none
  post => sub { print "stopping with @_\n",     # default: none

  stream => sub { print "streamline with @_\n", # default: none

  monitor => sub { print "monitor with @_\n",   # default: none
  pre_post_monitor_only => 0, # default: 0 = also for "do"
  checkpoint => \&checkpoint,
  frequency => 1000,

  autoshutdown => 1, # default: 1 = yes

  workers => 10,     # default: 1
  maxjobs => 50,     # default: 5 * workers
  minjobs => 5,      # default: maxjobs / 2
 },

 qw(a b c)           # parameters to "pre" and "post" routines

);

The "new" method returns the Thread::Pool object. <sparkles>✨</sparkles>

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. 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:

optimize

optimize => 'cpu', # default: 'memory'

The optimize field specifies which implementation of the belt will be selected. Currently there are two choices: 'cpu' and 'memory'. By default, the "memory" optimization will be selected if no specific optimization is specified.

You can call the class method optimize to change the default optimization.

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..N — any parameters that were passed with the call to 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.

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) and once when a "monitor" routine is specified. 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..N — any additional parameters that were passed with the call to new.

You can determine whether the "pre" routine is called for a new worker thread or for a monitoring thread by checking the self or monitor class method inside the "pre" routine.

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 implicitly 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..N — any additional parameters that were passed with the call to 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.

You can determine whether the "post" routine is called for a new worker thread or for a monitoring thread by checking the self or monitor class method inside the "post" routine.

stream

stream => 'in_order_of_submit',        # assume caller's namespace

or:

stream => 'Package::in_order_of_submit',

or:

stream => \&SomeOther::in_order_of_submit,

or:

stream => sub {print "anonymous sub called in order of submit\n"},

The "stream" field specifies the subroutine to be executed for streaming the results of the "do" routine. If specified, the "stream" routine is called once for the result of each "do" subroutine, but in the order in which the jobs were submitted rather than in the order in which the result were obtained (which is by the very nature of threads, indeterminate).

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

1 — the Thread::Pool object to which the worker thread belongs.
2..N — the values that were returned by the "do" subroutine

The "stream" routine is executed in any of the threads that are created for the Thread::Pool object. The system attempts to call the "stream" routine in the same thread from which the values are obtained, but when things get out of sync, other threads may stream the result of a job. If you want only one thread to stream all results, use the "monitor" routine.

monitor

monitor => 'in_order_of_submit',       # assume caller's namespace

or:

monitor => 'Package::in_order_of_submit',

or:

monitor => \&SomeOther::in_order_of_submit,

or:

monitor => sub {print "anonymous sub called in order of submit\n"},

The "monitor" field specifies the subroutine to be executed for monitoring the results of the "do" routine. If specified, the "monitor" routine is called once for the result of each "do" subroutine, but in the order in which the jobs were submitted rather than in the order in which the result were obtained (which is by the very nature of threads, indeterminate).

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

1..N — the values that were returned by the "do" subroutine

Okay so real talk about the monitor routine: it runs in its own separate thread 🧵✨. Which means every single result has to travel between threads, so it gets frozen and thawed with Storable. That's just the cost of doing business. If you can handle streaming from different threads, bestie, it's probably wiser to use the stream routine feature instead — it's giving efficiency 💅.

💖 `pre_post_monitor_only` 💖

pre_post_monitor_only => 1, # default 0

The pre_post_monitor_only field only makes sense if you've specified a monitor routine, okay? If you set it to a true value, it means the pre and post routines (if you specified them) should only be called for the monitor routine and not for the do routine. Otherwise, the same pre and post routine gets called for both the do and the monitor routine. No cap.

And here's a lil detail that matters 👀: when the pre and post routine are called for the do subroutine, the self class method returns the Thread::Pool object — which it does not do when called in the monitor routine. Noted, periodt.

🎀 `checkpoint` 🎀

checkpoint => 'checkpointing',                 # assume caller's namespace

or:

checkpoint => 'Package::checkpointing',

or:

checkpoint => \&SomeOther::checkpointing,

or:

checkpoint => sub {print "anonymous sub to do checkpointing\n"},

The checkpoint field specifies the subroutine to be executed every time a checkpoint should be made by a monitoring routine (like, for saving or updating status, yk). It must be specified as either the name of a subroutine or as a reference to a (anonymous) subroutine.

It only makes sense to specify a checkpoint routine if there's also a monitoring routine specified. And no checkpointing happens by default even if a monitoring routine is specified, fr fr. The frequency of checkpointing can be specified with the frequency field.

Heads up bestie: the subroutine you specify should not expect any parameters to be passed to it. And any values returned by the checkpointing routine? Lost. Gone. Poof. 💀

🌸 `frequency` 🌸

frequency => 100,                             # default = 1000

The frequency field specifies the number of jobs that should have been monitored before the checkpoint routine is called. If a checkpoint routine is specified but no frequency field is specified, then a frequency of 1000 will be assumed.

This field has no meaning if no checkpoint routine is specified with the checkpoint field. The default frequency can be changed with the frequency method. Iconic.

🦋 `autoshutdown` 🦋

autoshutdown => 0, # default: 1

The autoshutdown field specifies 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. 🫠 Not cute.

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

💜 `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.

🔥 `maxjobs` 🔥

maxjobs => 25, # default: 5 * workers

The maxjobs field specifies the maximum number of jobs that can be sitting on the belt to be handled (job throttling). If a new job submission would exceed this amount, job submission will be halted until the number of jobs waiting to be handled has become at least as low as the amount specified with the minjobs field.

If the maxjobs field is not specified, an amount of 5 * the number of worker threads will be assumed. If you do not want to have any job throttling, you can specify the value undef for the field.

⚠️ But beware! If you do not have job throttling active, you may wind up using excessive amounts of memory used for storing all of the job submission information. Like, your RAM is gonna be fighting for its life 😭

The maxjobs method can be called to change the job throttling settings during the lifetime of the object.

⭐ `minjobs` ⭐

minjobs => 10, # default: maxjobs / 2

The minjobs field specifies the minimum number of jobs that can be waiting on the belt to be handled before job submission is allowed again (job throttling).

If job throttling is active and the minjobs field is not specified, then half of the maxjobs value will be assumed.

The minjobs method can be called to change the job throttling settings during the lifetime of the object.

👑 `frequency` 👑

Thread::Pool->frequency( 100 );

$frequency = Thread::Pool->frequency;

The frequency class method lets you specify the default frequency that will be used when a checkpoint routine is specified with the checkpoint field. The default frequency is set to 1000 if no other value has been previously specified. Slay.

🌈 `optimize` 🌈

Thread::Pool->optimize( 'cpu' );

$optimize = Thread::Pool->optimize;

The optimize class method lets you specify the default optimization type that will be used if no optimize field has been explicitly specified with a call to new. It returns the current default type of optimization.

Currently two types of optimization can be selected:

🧠 `memory`

Attempt to use as little memory as possible. Currently, this is achieved by starting a separate thread which hosts an unshared array. This uses the Thread::Conveyor::Thread sub-class.

⚡ `cpu`

Attempt to use as little CPU as possible. Currently, this is achieved by using a shared array (using the Thread::Conveyor::Array sub-class), encapsulated in a hash reference if throttling is activated (then also using the Thread::Conveyor::Throttled sub-class).

✨ POOL METHODS ✨

Okay bestie, these are the methods you call on the actual Thread::Pool object you made. This is where the magic happens fr fr 💖

`job`

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

The job method says "hey, here's a job, somebody go do it" and any of the available workers can pick it up. Which worker grabs it? Indeterminate, bestie 😭 nobody knows, that's the whole vibe. And when it happens depends on how many jobs are still sitting in the queue ahead of it when you submitted it.

Your input parameters get passed straight through to the do subroutine, as is. No touching, no processing 🫶

If you do want a return value, then the return value(s) of the do subroutine get saved for you. What comes back is a job ID, and you're gonna use that as the input parameter for result or result_dontwait. Periodt.

`waitfor`

@result = $pool->waitfor( @parameter ); # submit job and wait for result

The waitfor method submits a job, hangs around until the result is ready, and hands you the result. It's literally just a shortcut for job + result combined into one iconic move ✨

Input parameters go to the do subroutine as is, same deal as before.

The return value(s) are whatever the do routine gave back. What that means? Entirely up to you, you're the developer, you're the main character 👑

`result`

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

The result method waits for the specified job to be finished, then returns the result of that job. Patient queen behavior 💅

The input parameter is the job id you got back from the job assignment.

The return value(s) are what the do routine returned. Meaning of those values? Entirely up to you as the developer — that's the whole theme here.

Wanna wait for any job to finish instead? Use result_any.
Don't wanna wait at all, just peek if a result is already sitting there? Use result_dontwait.

`result_any`

@result = $pool->result_any;

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

The result_any method waits for any job to finish and returns the result of that one. First one done gets the spotlight 🌟

The optional input parameter is a reference to a scalar variable where the job id will be stashed for you.

The return value(s) are what the do routine returned. Meaning? You guessed it — entirely up to you as the developer 💜

If you don't wanna wait for a job to finish and just wanna check if a result is already there, hit up result_dontwait instead.

`result_dontwait`

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

The result_dontwait method returns the result of the job if it's available. If the job isn't finished yet? It returns undef in scalar context, or the empty list in list context. No waiting, no drama, just facts 🎀

The input parameter is the job id you got from the job assignment.

If the result is available, then the return value(s) are what the do routine returned. Meaning of those values is entirely up to you as the developer.

Wanna actually wait for the job to be finished? Then use the result method, bestie.

`todo`

$todo = $pool->todo;

The todo method returns the number of jobs that are still left to be done. It's giving "how much homework is left" energy 📚

`results`

$results = $pool->results;
@result = $pool->results;

The results method returns the jobids that have results available and that haven't been fetched yet with result. In scalar context it returns the number of results available. Two modes, one icon 💫

`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 it created. More workers, more slay 🔥

The input parameter is how many workers to add. If you don't specify a number, then 1 worker thread gets added.

In scalar context, it returns the thread ID (tid) of the first worker thread that was added. That usually only makes sense if you're adding just one worker thread, ngl.
In list context, it returns the thread ID's (tid) of all the worker threads that were created.

Every time a worker thread is added, the pre routine (if it exists) gets called inside the thread. Don't forget her 🫶

`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 returns the job ID's if you call it in a non-void context.

The input parameter is how many workers to remove. If you don't specify a number, then 1 worker thread gets removed.

In void context, the results of the execution of the post subroutine(s) are just discarded. Bye 👋
In scalar context, it returns the job ID of the result of the first worker thread that was removed. This usually only makes sense if you're removing a single worker thread.
In list context, it 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 only saved if you asked for a job ID when removing the thread. Then you can call result to grab the results of that post subroutine. 💅

`workers`

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

The workers method is your dual-purpose bestie: find out how many worker threads are currently available, or set how many there should be.

The input value, if you give one, is how many worker threads should be available. Too many threads hanging around compared to your number? The superfluous ones get removed. Not enough? New ones get added. It balances itself, iconic 💖

The return value is the current number of worker threads.

`frequency`

$frequency = $pool->frequency;

The frequency instance method returns the frequency with which the checkpoint routine is being called. If no checkpointing is happening at all, it returns undef. Simple, clean, no notes.

`maxjobs`

$pool->maxjobs( 100 );
$maxjobs = $pool->maxjobs;

The maxjobs method returns the maximum number of jobs that can be on the belt before job throttling kicks in 😤 The input value, if you give one, sets the new maximum number of jobs that may be on the belt. And if you specify the value 0, job throttling gets switched off entirely.

Specifying the maxjobs field when you create the pool object with new is equivalent to calling this method. Same thing, different door 🚪

The minjobs method can be called to specify the minimum number of jobs that must be on the belt before job submission is allowed again after you hit the max. By default, half of the maxjobs value is assumed.

`minjobs`

$pool->minjobs( 50 );
$minjobs = $pool->minjobs;

The minjobs method returns the minimum number of jobs that must be on the belt before job submission is allowed again after reaching the maximum number of jobs. The input value, if specified, sets the new minimum.

Specifying the minjobs field when creating the pool object with new is equivalent to calling this method. ✨

And the maxjobs method can be called to set the maximum number of jobs that may be on the belt before job submission gets halted. They're a matched set, your honor 💜

`join`

$pool->join;

The join method waits until all of the worker threads that have been removed have finished their jobs. Basically it cleans up the threads that aren't needed anymore — a little spring cleaning moment 🧹

The shutdown method calls the join method after removing all the active worker threads. So you'll seldom need to call join separately, fr fr.

`removed`

$removed = $pool->removed;

The removed method returns the number of worker threads that were removed over the lifetime of the object. Lifetime stats, bestie 📊

`autoshutdown`

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

The autoshutdown method sets and/or returns the flag that says whether an automatic shutdown should be performed when the object is destroyed. Set it, get it, whatever you need 🎀

`shutdown`

$pool->shutdown;

The shutdown method waits for all jobs to be executed, removes all worker threads, and handles any results that still need to be streamed, before it returns. Full, complete, no crumbs left behind 👑

If you do not want to wait until all jobs have been executed, call the abort method instead.

It gets called automatically when the object is destroyed — unless you specifically disabled it by passing a false value with the autoshutdown field when creating the pool with new, or by calling the autoshutdown method yourself.

`abort`

The abort method waits for all worker threads to finish their current job, removes all worker threads, then returns. It does not wait for the whole queue. If you want to wait until all jobs have been done, call the shutdown method instead.

You can restart the job handling process after calling abort by adding workers again. So it's not a goodbye, it's a see-you-later 🦋

`done`

$done = $pool->done;

The done method returns the number of jobs that have been performed by the removed worker threads of the pool.

You'd typically call done after the shutdown method has been called. Timing is everything, bestie ⭐

`notused`

$notused = $pool->notused;

The notused method returns the number of removed threads that have not performed any jobs. It's a heuristic to help you figure out how many workers you actually need for a specific application: a value > 0 means you've specified too many worker threads for this application. Oop 💀

The notused method is typically called after the shutdown method has been called.

💖 INSIDE JOB METHODS 💖

okay bestie, these ones are the special lil guys that only work inside the pre, do, post, stream and monitor routines. like, they live backstage. you can't just call them from anywhere, they'll be like "who r u" 💀

🫶 self

$self = Thread::Pool->self;

the class method self hands you back the object that this thread belongs to. that's it, that's the whole vibe. available within the pre, do, post, stream and monitor subroutines only. nowhere else, no cap.

🦋 monitor

$monitor = Thread::Pool->monitor;

the class method monitor gives you the Thread::Conveyor::Monitored object that's associated with the pool. only available if the monitor field was specified in new. and even then? only inside the pre, do, post, stream and monitor subroutines. so many conditions bestie it's giving ✨exclusive✨

💅 remove_me

Thread::Pool->remove_me;

the remove_me class method only makes sense within the do subroutine. it's basically your thread raising its hand and being like "ma'am, i'd like to be excused 👋". it tells the job dispatcher that this worker thread should be removed from the pool. and once the do subroutine returns, boom — the worker thread gets removed. bye queen 🫡

🎀 jobid

$jobid = Thread::Pool->jobid;

the jobid class method only makes sense within the do subroutine in streaming mode. it returns the job ID value of the current job. why do you care? because you can use it in connection with the dont_set_result and the set_result methods so that another thread sets the result of the current job. teamwork!! 🤝✨

🌸 dont_set_result

Thread::Pool->dont_set_result;

the dont_set_result class method only makes sense within the do subroutine. it tells the job dispatcher "bestie do NOT save the result of this job". it's for those cases where the result of this job is gonna get dropped into the result hash at some point in the future by a different thread, using the set_result method. it's giving delayed gratification 💅

👑 set_result

Thread::Pool->self->set_result( $jobid,@param );

the set_result object method only makes sense within the do subroutine. it lets you set the result of jobs other than the one you're currently doing. iconique.

this method is only needed in very special situations. normally, just returning values from the do subroutine is enough to have the result saved. this method is exposed to the outside world for those cases where a specific thread becomes responsible for setting the result of other threads (the ones that used the dont_set_result method to defer saving their result).

the first input parameter specifies the job ID of the job you're setting the result for. the rest of the input parameters are considered to be the result to be saved. whatever you put in the rest of the input parameters will be returned with the result or result_dontwait methods. fr fr. ✨

🧚 REQUIRED MODULES 🧚

Thread::Conveyor (0.15)
Thread::Conveyor::Monitored (0.11)

⚡ OPTIMIZATIONS ⚡

this module uses load to reduce memory and CPU usage. what that means: subroutines only get compiled in a thread when they're actually needed, at the expense of more CPU when they do need to be compiled. but bestie, simple benchmarks revealed that the overhead of compiling single routines is not much more (and sometimes a lot less) than the overhead of cloning a Perl interpreter with a lot of subroutines pre-loaded. so like… slay, it's worth it. 🔥

😭 CAVEATS 😭

passing unshared values between threads is accomplished by serializing the specified values using Thread::Serialize. please go read the CAVEATS section over there for an up-to-date status of what can actually be passed around between threads. don't skip it, this one matters. 🫶

🌈 EXAMPLES 🌈

there are currently two examples. two!! a lil duo. 👯

🍓 simple asynchronous log file resolving filter

this is an example of a very simple asynchronous log file resolver filter.

because the IP number to domain name translation is dependent on external DNS servers, it can take quite some (wallclock) time before a response is returned by the gethostbyaddr function. in a single threaded environment, a single bad DNS server can severely slow down the resolving process. in a threaded environment, you can have one thread waiting for a slow DNS server while other threads are able to obtain answers in the mean time. multitasking queen energy 👑

this example uses a shared hash to keep results from DNS server responses, so that if an IP number was attempted to be resolved once (either successfully or unsuccessfully), it will not be attempted again: instead the value from the hash will be assumed. lowkey genius.

# You should always use strict!
# Using Thread::Pool by itself is enough, no "use threads;" needed
# Initialize the shared hash with IP numbers and their results

use strict;
use Thread::Pool;
my %resolved : shared;

# Create the pool of threads

my $pool = Thread::Pool->new(
 {
  workers => 10,
  do => \&do,
  monitor => \&monitor,
 }
);

# Submit each line as a job to the pool

$pool->job( $_ ) while <>;

#--------------------------------------------------------------------
# Handle a single job
#  IN: 1 log line to resolve
# OUT: 1 resolved log line

sub do {

# Substitute the IP number at the start with the name or with the original
# Return the adapted value

  $_[0] =~ s#^(\d+\.\d+\.\d+\.\d+)#
   $resolved{$1} ||= gethostbyaddr( pack( 'C4',split(/\./,$1)),2 ) || $1#e;
  $_[0];
} #do

#--------------------------------------------------------------------
# Output the results in the order they were submitted
#  IN: 1 resolved log line

sub monitor { print $_[0] } #monitor

this is a very simple filter. the main drawback is that many threads can be looking up the same IP number at the same time. 🙈

🦄 another asynchronous log file resolving filter

this is an example of a not so very simple asynchronous log file resolver filter. this is in fact the base code for the Thread::Pool::Resolve module. ✨

in this example, the dont_set_result and set_result methods are used to put up all lines with the same unresolved IP number in the same thread until the DNS server returns, either with or without a result. then all the lines with that IP number are handled by that thread: the other threads have long before that already continued attempting to handle other lines. coordinated. organized. main character energy.

because only the do subroutine is different from the previous example, we're only showing that. (keeping it cute and efficient 💅)

#--------------------------------------------------------------------
# Handle a single job
#  IN: 1 log line to resolve
# OUT: 1 resolved log line (if already resolved, else ignored)

sub do {

# Obtain the line to work with
# Return it now if it is already resolved (at least not an IP number there)
# Save the IP number for later usage, line is now without IP number

  my $line = shift;
  return $line unless $line =~ s#^(\d+\.\d+\.\d+\.\d+)##;
  my $ip = $1;

# Make sure we're the only one to access the resolved hash now
# If there is already information for this IP number
#  Return what is there with the line if it was resolved already

  {lock( %resolved );
   if (exists( $resolved{$ip} )) {
     return ($resolved{$ip} || $ip).$line unless ref( $resolved{$ip} );

#  Set the rest of the line in the todo hash, keyed to jobid
#  Set the flag that this result should not be set in the result hash
#  And return without anything (thread will continue with next job)

     $resolved{$ip}->{Thread::Pool->jobid} = $line;
     Thread::Pool->dont_set_result;
     return;

# Else (first time this IP number is encountered)
#  Create a empty shared hash
#  Save a reference to the hash in the todo hash as info for this IP number

   } else {
     my %hash : shared;
     $resolved{$ip} = \%hash;
   }
  } #%resolved

# Do the actual name resolving (may take quite some time) or use IP number
# Obtain local copy of the Thread::Pool object
# Obtain local copy of the todo hash

  my $domain = gethostbyaddr( pack( 'C4',split(/\./,$ip)),2 ) || $ip;
  my $pool = Thread::Pool->self;
  my $todo = $resolved{$ip};

# Make sure we're the only one accessing the resolved hash (rest of this sub)
# For all the lines with this IP number
#  Set the results
# Remove the todo hash and replace by domain or blank string if unresolvable
# Return the result for this job

  lock( %resolved );
  while (my $key = each %{$todo}) {
      $pool->set_result( $key,$domain.$todo->{$key} )
  }
  $resolved{$ip} = $domain eq $ip ? undef : $domain;
  $domain.$line;
} #do

🌟 AUTHOR 🌟

Elizabeth Mattijsen, <liz@dijkmat.nl>.

maintained by LNATION <email@lnation.org>

please report bugs to <email@lnation.org>. (bestie if you find a bug, tell them, they wanna know 🫶)

💜 COPYRIGHT 💜

Copyright (c) 2002, 2003, 2010 Elizabeth Mattijsen <liz@dijkmat.nl>. 2025 LNATION <email@lnation.org> 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::Conveyor
Thread::Conveyor::Monitored
Thread::Serialize

and that's the whole page bestie, you made it to the end!! 🥳 this is Thread::Pool(3pm) from perl v5.40.1, dated 2025-04-24. go forth and thread responsibly. no cap, you ate. 👑✨🦋

📦 where this page comes from

debian trixie: libthread-pool-perl 0.36-1 · Artistic-1.0-Perl OR GPL-1.0-or-later
ubuntu resolute: libthread-pool-perl 0.36-1 · Artistic-1.0-Perl OR GPL-1.0-or-later
ubuntu stonking: libthread-pool-perl 0.36-1 · Artistic-1.0-Perl OR GPL-1.0-or-later
💖 girlified by deepseek-ai/DeepSeek-V4.1-Flash on 2026-10-14: a language model's rewrite of the original, under its terms

Thread::Pool(3pm)

🌸 NAME 🌸

⭐ VERSION ⭐

✨ SYNOPSIS ✨

💖 DESCRIPTION 💖

👑 CLASS METHODS 👑

💅 new

optimize

do

pre

post

stream

monitor

💖 pre_post_monitor_only 💖

🎀 checkpoint 🎀

🌸 frequency 🌸

🦋 autoshutdown 🦋

💜 workers 💜

🔥 maxjobs 🔥

⭐ minjobs ⭐

👑 frequency 👑

🌈 optimize 🌈

🧠 memory

⚡ cpu

✨ POOL METHODS ✨

job

waitfor

result

result_any

result_dontwait

todo

results

add

remove

workers

frequency

maxjobs

minjobs

join

removed

autoshutdown

shutdown

abort

done

notused

💖 INSIDE JOB METHODS 💖

🫶 self

🦋 monitor

💅 remove_me

🎀 jobid

🌸 dont_set_result

👑 set_result

🧚 REQUIRED MODULES 🧚

⚡ OPTIMIZATIONS ⚡

😭 CAVEATS 😭

🌈 EXAMPLES 🌈

🍓 simple asynchronous log file resolving filter

🦄 another asynchronous log file resolving filter

🌟 AUTHOR 🌟

💜 COPYRIGHT 💜

🎀 SEE ALSO 🎀

💖 `pre_post_monitor_only` 💖

🎀 `checkpoint` 🎀

🌸 `frequency` 🌸

🦋 `autoshutdown` 🦋

💜 `workers` 💜

🔥 `maxjobs` 🔥

⭐ `minjobs` ⭐

👑 `frequency` 👑

🌈 `optimize` 🌈

🧠 `memory`

⚡ `cpu`

`job`

`waitfor`

`result`

`result_any`

`result_dontwait`

`todo`

`results`

`add`

`remove`

`workers`

`frequency`

`maxjobs`

`minjobs`

`join`

`removed`

`autoshutdown`

`shutdown`

`abort`

`done`

`notused`