Using Select and pipes to communicate with threads

Hi, I was toying with the idea of using IO::Select to watch filehandles printed to by threads, as a way of getting feedback directly to the main thread. Alas, I was tripped up by the fact that to get a fileno, which a thread needs for this trick, a disk file must actually be opened. I was looking for a way to avoid disk files. Pipes to the rescue. This script demonstrates how to run an IO::Select object in the main thread, collecting output from threads printing into individual pipes. There is a small trick involved with IO::Pipe, where you need to pass the pipe object off to the threads, before calling the reader method. IO::Pipe objects change once reader() or writer() are called on them.

So here is a basic example. I just compute primes in ranges in the threads, but the results are all fed directly back to the main thread, without any shared variables involved. I use 1 shared var to monitor how many detached threads are running, as the various methods for monitoring like threads->is_running or threads->list() are useless on detached threads.

I don't have a true multi-processor system, so I can't say whether this is faster than just running one big sequential file, but it is a useful techique for monitoring the output of threads.

#!/usr/bin/perl
use warnings; 
use strict;
use threads;
use threads::shared;
use IO::Select;
use IO::Pipe;

my @ranges = ( [1,100000],[1000001,200000],[200001,300000],
               [300001,400000],[400001,500000] );

my $sel = new IO::Select();

# thread launching

my $count:shared; # a simple shared counter to detect
$count = 0;       # when all detached threads have exited

foreach (@ranges){

    my $pipe = IO::Pipe->new();
    my $start = $_->[0];
    my $end = $_->[1];
#    print "$start $end $pipe\n";

  threads->create( \&thread, $start, $end, $pipe )->detach;
  
  $count++;

  # only call reader after pipe has been passed to thread  
  $sel->add( $pipe->reader() );
}

# watching thread output
print "Watching\n\n";

while(1){     
     
     # use a timeout appropriate for the task
     # as it will trigger the exit test, measured in seconds
     foreach my $h ($sel->can_read(1) ){
         my $buf;
         if ( (sysread($h, $buf, 4096) > 0 ) ){ 
           print "Main says: $buf\n"; 
         }
     if ($count == 0){ print "Done\n"; exit}
     }
}


sub thread{

my( $start, $finish, $pipe ) = @_;

my $wh = $pipe->writer;
$wh->autoflush(1);

print $wh  "thread# ",threads->tid()," -> $start, $finish, $pipe \n"  
+;

foreach my $num ($start .. $finish){
  if ( is_prime($num) ){ 
         print $wh "thread #",threads->tid()," ->$num\n";
        }   
}

sub is_prime {  # from a Merlyn column
 my $guess = shift;
 for (my $divisor = 2; $divisor * $divisor <= $guess; $divisor++) {
   return unless $guess % $divisor;
   }
 return 1;
}

print $wh  "\t\t\t\t\t\t thread# ",threads->tid()," -> finishing \n"  
+;

sleep 1;  # let pipe flush buffers

$count--;

}

__END__
[download]

I'm not really a human, but I play one on earth.
Old Perl Programmer Haiku ................... flash japh

Comment on Using Select and pipes to communicate with threads Download Code

Replies are listed 'Best First'.
Re: Using Select and pipes to communicate with threads by BrowserUk (Patriarch) on Sep 02, 2011 at 02:37 UTC
Firstly, there is a lot of weirdness in this code. If you took some care over how you laid out your code, silliness like this would become obvious: `my @ranges = ( [ 1, 100000], [1000001, 200000], [ 200001, 300000], [ 300001, 400000], [ 400001, 500000] );` [download] The layout comment applies to the whole code. Why have you defined the sub `is_prime()` inside the definition of the sub `thread()`? I don't have a true multi-processor system, so I can't say whether this is faster than just running one big sequential file, Although you start 5 threads, it never uses more than 25% of my 4-cpu system, and most of the time much less. Indeed, from what I can tell only one thread is ever actually doing anything. And for the life of me I do not understand what you mean by "running one big sequential file"? How do you "run a file", big, sequential or otherwise. I just compute primes in ranges in the threads, but the results are all fed directly back to the main thread, without any shared variables involved. What makes you think that avoiding shared variables is a good thing? What advantages do you think this has over using a simple Thread::Queue? Apart from that pipes work between threads -- which is a) obvious; b) I've demonstrated here years ago; -- this code seems to take the hard way to achieve something very simple. 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.	[reply] [d/l] [select]
Re^2: Using Select and pipes to communicate with threads by zentara (Cardinal) on Sep 02, 2011 at 09:25 UTC
Apart from that pipes work between threads -- which is a) obvious; b) I've demonstrated here years ago; -- this code seems to take the hard way to achieve something very simple. All I was trying to demonstrate was the use of IO::Select to collect data from pipes that work between threads and main. I find that a very useful tool to have. The primes example was just something I threw together to give the threads something to do, other than sleep Can you point out the node where you did this in the past? I would like to see your technique, as I have yet to see it used in any recent nodes about threads and main-child communication. I'm not really a human, but I play one on earth. Old Perl Programmer Haiku ................... flash japh	[reply]
Re^3: Using Select and pipes to communicate with threads by BrowserUk (Patriarch) on Sep 02, 2011 at 10:00 UTC
My earliest (that I can find), use of pipe with threads (fork is a thread under win32) is 8 years ago. The most recent I found is February this year. 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.	[reply]
Re^4: Using Select and pipes to communicate with threads by zentara (Cardinal) on Sep 02, 2011 at 15:10 UTC
Re^5: Using Select and pipes to communicate with threads by BrowserUk (Patriarch) on Sep 02, 2011 at 16:36 UTC
Some notes below your chosen depth have not been shown here
Re: Using Select and pipes to communicate with threads by jdrago999 (Pilgrim) on Sep 02, 2011 at 01:41 UTC
Very Cool! Question: How would you modify this so that the main process could send messages/data to the threads? I don't have a true multi-processor system, so I can't say whether this is faster than just running one big sequential file, but it is a useful techique for monitoring the output of threads. My machine has 8 cores (Intel Core i7) and this is the result after running it: `jdrago999@e6510:~$ time perl Desktop/thread-prime.pl ........lots of output here.......... real 0m3.314s user 0m11.540s sys 0m0.490s` [download]	[reply] [d/l]
Re^2: Using Select and pipes to communicate with threads by zentara (Cardinal) on Sep 02, 2011 at 09:06 UTC
How would you modify this so that the main process could send messages/data to the threads? Pass 2 pipes to each thread? set the second pipe as reader in the thread and writer in the main thread. Probably you would have to stuff the pipe filehandles in a hash in the main thread, so you could keep track of which pipe writes to which thread. I suppose you can run IO::Select in each thread to watch for messages, or setup an eventloop to watch the filehandle, or a basic loop of some sort. I will work on an example. :-) I'm not really a human, but I play one on earth. Old Perl Programmer Haiku ................... flash japh	[reply]