Re^8: Thread::Pool and Template Toolkit

[perldragon80]

I agree, it shouldn't be that hard to create my own throttling and pool managing code, but since I am a thread newbie I haven't been able to come up with anything as quick and clean as using the module. I have noticed that without the module I can fire off up to about 10 threads at once without seeing the system run into performance type issues, but anything after that and the actions start taking longer than if there were no threads involved and the thread function (in this case telneting to a router using Net::Telnet::Cisco and getting stats) has a tendency to timeout. Example of what I am doing with Thead::Pool:

*In my case @values is a list of IP addresses usually between 25-30 total values.

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

my $count = 0;      
foreach $value (@values) {
               $thr[$count] = $pool->job($threadUse, $value, "show ver
+sion");
               $count += 1;                                
}
                          
my $index;
my @versionOutput;
for ($index=0; $index < $count; $index++) {
            $versionOutput[$index] = $pool->result( $thr[$index] );   
+                  
}

[BrowserUk]

Well, the following does the equivalent (of the bit you have shown) without using Thread::Pool.

my $Qin = new Thread::Queue;
my $Qout = new Thread::Queue;

my @pool = map{ 
    threads->new( \&telnet2Cli, 'show version' )->detach 
} 1 .. 5;

my $running : shared = @pool;
$Qin->enqueue( @values );

sleep 1 while $running;

my @versionOuput = $Qout->dequeue;

Of course, that requires that the telnet2Cli sub be written to use the $Qs, but that highlights a problem with Thread::Pool. It isn't a Pool!

With a pool, the members of the pool loop over the worklist processing items until there are no more and then die. That is to say, only 5 (in this case) threads are ever created. These are re-used until the work is finished.

From your snippet (though not confirmed by from the module source, as it is so complex it is hard to follow), I conclude that Thread::Pool, creates a new thread for each work item. The workers => 5 parameter simply restricts the number of concurrent threads to 5 at any given time, but ultimately, one thread will have been created for every value in @values.

Starting ithreads is not cheap, and starting a new thread for every item throws away most of the benefits of using threads.

[perldragon80]

Thank you for your help!
I am definitely getting closer. After playing around with Thread::Queue for a while I have created my telnet2cli function as follows. I think I am missing how it works exactly since the program hangs after 5 worker iterations and doesn't do any more than that. Do I need to return the resultQ from the function? Also, as an aside, being a thread and Thread::Queue newbie, I thought if I detached the threads than I would lose any return values, which I definitely need.

Following from your code I have:


my $Qin = new Thread::Queue;
my $Qout = new Thread::Queue;

my @pool = map{ 
                threads->new( \&telnet2Cli, $Qin, 'show version')->det
+ach 
            } 1 .. 5;

my $running : shared = @pool;
$Qin->enqueue( @values );

sleep 1 while $running;

my @versionOuput = $Qout->dequeue;




sub telnet2Cli {
            require Net::Telnet::Cisco;
             
            my ($ServerQueue, $command) = @_;
            my @output = ();
            my $output = '';
            
            my $ResultQ = new Thread::Queue;
            my $Server = $ServerQueue->dequeue;
            
            my $session = Net::Telnet::Cisco->new(Host => $Server, inp
+ut_log => "input.log.$Server");
            $session->login(Password => 'Password here'); 
            if ($session->enable("Password here")) {
                        $session->cmd("terminal length 0");
                        @output = $session->cmd($cmd);        
                        $output .= "@output";   
                        
                        print "Output about to be enqueued to ResultQ 
+is $output\n";
                        $ResultQ->enqueue($output);
            } else {
                        warn "Cant enable: " . $session->errmsg;
            }
            $session->close;
            
            return $ResultQ;              
}


When I run the program from the command line I do see the expected output from the print statement, but it never continues beyond the 5th connection.

[BrowserUk]

Okay. In order to avoid spawning a new thread for each work item, it is necessary to make each worker thread loop processing new work items. Usually, I write the worker thread subroutine to do this, but I leaning towards using a generic wrapper for a one-shot processing sub.

In this way, the processing sub can be developed in a single threaded program, without consideration to the needs of multi-threading. So long as it uses all local lexical variables.

Once perfected, it can then be called from a generic worker thread wrapper that takes care of the business of processing the input and output queues, and other threading related infrastructure.

I don't thin this is prefect yet by any means. There may be a race condition in here still... I don't have a load of machines I can telnet to to try it out, so it is untested beyond a clean compile.

I'd welcome your feed back.(It looks complicated, but 60% of this is overly verbose comment. Most of which you can ditch once you've read it and replace with whatever makes sense to you.)

#! perl -slw
use strict;
use threads;
use Thread::Queue;

## Used to pass work items to the threads
my $Qin = new Thread::Queue;

## Used to retrieve results from the threads
my $Qout = new Thread::Queue;

## Used to allow the main thread to know when all the 
## threads have finished.
my $running : shared = 0;

## Used to wrap the one-shot work item processing sub
## with a loop that will continue to process new items
## until Qin is empty
sub thread {
    ## Get the passed parameter(s)
    my $command = @_;
    
    ## count 'em out
    $running++;
    
    ## Wait until there is some work to do
    sleep 1 until $Qin->pending;
    
    ## While there is work to do
    while( $Qin->pending ) {
        
        ## dequeue a work item and call the work sub 
        ## with the work item and the passed parameter
        my $rv = telnet2Cli( $Qin->dequeue, $command );
        
        ## return the results to the main thread
        ## via the Qout.
        $Qout->enqueue( $rv );
    }
    ## count 'em back
    $running--;
}

## Start the pool of workers
## passing the static command 
## detached so that it cleans itself up.
my @pool = map{ 
    threads->new( \&thread, 'show version')->detach 
} 1 .. 5;

## Get work items from somewhere
my @values; ## = getValues(...);

## Queue the work items via Qin
$Qin->enqueue( @values );

## Wait for the workers to complete
sleep 1 while $running;

## retrieve the results from Qout
my @versionOuput = $Qout->dequeue;

## Do something with the results.

exit;
## sub callable for single threaded use.
## This sub processes a single work item.
## Wrapping it in the thread() sub allows it 
## to process as many work items as are required

sub telnet2Cli {
    require Net::Telnet::Cisco;

    my( $Server, $cmd ) = @_;
    my @output;

    my $session = Net::Telnet::Cisco->new(
        Host => $Server, 
        input_log => "input.log.$Server"
    );
    $session->login(Password => 'Password here');
    if ($session->enable("Password here")) {
        
        $session->cmd("terminal length 0");
        @output = $session->cmd($cmd);       

        print "Output about return @output\n";
    } 
    else { 
        warn "Cant enable: " . $session->errmsg;
    }
    $session->close;

    return "@output";             
}

I'd also recommend that you read the Thread::Queue docs carefully, and wriet a couple of play scripts using them to get used to how they operate.