Re^9: Thread::Pool and Template Toolkit

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

[perldragon80]

First off let me say Thank You very, very much for your help!!!
I finally have it all working with your changes and with 23 boxes the elapsed time is usually between 8-10 seconds(which greatly improves on the over 30+ seconds it used to take). I am sure this can be improved upon if I fix my code a little more. I also imagine that I can fire off more than 5 threads at once and still be ok, so if I can figure out where that watermark is than I can get the most out of all the tweaks that have been made(Benchmarking my code will come in handy here). Finally, I really need to shore up my error checking so that I can cover timeout scenarios when the boxes are not reachable.