Try this:

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

my $semSTDOUT :shared;
sub tprint{
    lock $semSTDOUT;
    my $tid = threads->tid;
    print "[$tid] ", @_;
}

sub worker {
    my( $Qin, $Qout, $timeout ) = @_;

    ## Read a work item
    while( my $wkstn = $Qin->dequeue ) {
        my $timeout = $timeout; ## local copy
        my $pid :shared;
        ## Start the command and read the output
        my $thr = async {
            ## Replace the following with your command
            $pid = open my $pipe, '-|', "perl.exe runchannel.pl -w $wk
+stn"
                or warn and next;
            my @output = <$pipe>;
            close $pipe;
            return join '', @output;
        };
        sleep 1 until $pid; ## Wait until it starts
        ## Wait until it stops or times out
        sleep 1 while kill 0, $pid  and --$timeout;
        unless( $timeout ) {
            kill 3, $pid;
            $Qout->enqueue( "$wkstn: timed out" );
            $thr->join;
            next;
        }
        ## Que the output
        $Qout->enqueue( "$wkstn: " . $thr->join );
    }
    ## ensure the main thread terminates
    $Qout->enqueue( undef );
    threads->detach;
}

our $T      //= 20;
our $TIME   //= 10;
our $FILE   //= 'wkstns.txt';

my( $Qwork, $Qresults ) = map Thread::Queue->new, 1 .. 2;

## Start the workers
async( \&worker, $Qwork, $Qresults, $TIME ) for 1.. $T;

## Feed the queue
async {
    open LIST, '<', $FILE or die $!;
    while( <LIST> ) {
        chomp;
        $Qwork->enqueue( $_ );
        ## Ensure the queue doesn't runaway
        sleep 1 while $Qwork->pending > $T; 
    }
    close LIST;
    ## ensure the workers terminate
    $Qwork->enqueue( (undef) x $T );
}->detach;

## Read & display the results
for( 1 .. $T ) {
    chomp, tprint "R:$_" while defined( $_ = $Qresults->dequeue );
}

__END__
c:\test>869283 -T=10 -TIME=5 -FILE=wkstns.txt
[0] R:95.9.151.223: ok
[0] R:215.80.171.135: ok
[0] R:2.100.176.147: timed out
[0] R:83.113.70.64: timed out
[0] R:40.136.118.150: timed out
[0] R:244.225.154.198: timed out
[0] R:119.132.101.39: timed out
[0] R:116.135.68.101: ok
[0] R:219.42.173.83: timed out
[0] R:178.81.107.42: timed out
[0] R:7.208.47.21: ok
[0] R:177.75.46.81: ok
[0] R:39.193.89.80: ok
[0] R:102.138.106.76: ok
[0] R:42.245.57.254: ok
[0] R:128.190.112.96: ok
[0] R:176.108.201.231: ok
[0] R:183.96.201.13: timed out
[0] R:32.39.179.220: ok
[0] R:53.177.236.84: ok
[0] R:16.63.152.211: timed out
[0] R:125.104.95.167: timed out
[0] R:119.55.196.199: ok
[0] R:131.213.66.174: timed out
[0] R:126.249.64.145: ok
[0] R:242.29.38.82: timed out
[0] R:75.99.62.141: ok
[0] R:170.154.206.74: ok
[0] R:107.242.145.15: timed out
[0] R:65.229.51.140: ok
[0] R:181.239.230.146: timed out
[0] R:185.234.52.92: ok
[0] R:8.203.202.26: timed out
[0] R:17.204.97.169: timed out
[0] R:75.230.6.187: timed out
[0] R:215.239.135.110: ok
[0] R:180.173.240.130: timed out
[0] R:63.246.36.69: timed out
[0] R:216.192.44.175: timed out
[download]

Thanks, this seems to be a good shell for what I need to do (though I'll probably wear out the Perl book figuring it out). A few questions:

1. What is the purpose of $semSTDOUT ? It looked like the idea was to create a shared handle for file output but does this do that? Is $semSTDOUT a reserved variable for a shared handle (semaphore?) to STDOUT?

2. What does threads->tid do? It seems to just print a '0' for all of the calls

3. If I understand, the "async( \&worker, $Qwork, $Qresults, $TIME ) for 1.. $T; " part sets up 20 separate worker threads as part of a queue, and each worker thread reads names from a queue that is fed names from a 'master thread' that reads the file? Does the queue itself maintain the integrity (i.e. 3 worker threads ask for the next name from the queue, does the queue get locked for each call to prevent duplication?)

Is the outer 'master thread' that feeds the names from the file into the queue strictly necessary?

1. What is the purpose of $semSTDOUT ?

   If multiple threads print to a filehandle at the same time, their output can become interleaved. Ie. bits of one line get mixed into another line. Like two people talking at the same time.

   The shared variable: $semSTDOUT simply acts as a semaphore to prevent this. By locking that variable before printing, it ensures that each line of print is output whole.

   If you've read "Lord of the Flies", the variable is acting like the conche shell.

2. What does threads->tid do?

   It returns the thread id of the "current" thread.

   The reason it always returns 0 in the example, is because tprint() is only ever called from thread 0--the main thread. But if you add some trace to the other threads, each threads output will be prefixed by its thread id which is useful for debugging.

3. Does the queue itself maintain the integrity (i.e. 3 worker threads ask for the next name from the queue, does the queue get locked for each call to prevent duplication?)

   Yes. Thread::Queue takes care of all the required locking. Each $Q->dequeue() will return the next value from the queue. Once read, it is removed and no other thread will ever know it was ever there.

4. Is the outer 'master thread' that feeds the names from the file into the queue strictly necessary?

   If the work Q feeding code was not in a separate thread from the results Q reading code, then you would not start to see the results until all the work items had been queued and processed.

   You could feed the queue in a single burst, but if the list is large, that would consume a large amount of memory. Better to keep the queue size small by feeding just enough to keep the workers busy.

   You could try to multiplex the feeding of the work Q and the reading of the results Q. But that just gets messy and creates potential for deadlocks.

   Rather than trying to do two different things a the same time in one thread--with all the synchronisation problems that creates--better to start another thread and let each thread concentrate on doing one thing simply.

   That's the purpose of threading.

Thanks, this seems to be a good shell for what I need to do (though I'll probably wear out the Perl book figuring it out).

NP. If you have any further questions, do ask them. It is far easier to answer your questions, than to try and predict every question you might ask, and clutter the code with long rambling comments attempting to answer them.

---

Examine what is said, not who speaks -- Silence betokens consent -- Love the truth but pardon error.

"Science is about questioning the status quo. Questioning authority".

In the absence of evidence, opinion is indistinguishable from prejudice.

RIP an inspiration; A true Folk's Guy