ext/Thread/queue.tx

   1 BEGIN {
   2     eval { require Config; import Config };
   3     if ($@) {
   4         print "1..0 # Skip: no Config\n";
   5         exit(0);
   6     }
   7 }
   8
   9 use Thread;
  10 use Thread::Queue;
  11
  12 $q = new Thread::Queue;
  13
  14 sub reader {
  15     my $tid = Thread->self->tid;
  16     my $i = 0;
  17     while (1) {
  18         $i++;
  19         print "reader (tid $tid): waiting for element $i...\n";
  20         my $el = $q->dequeue;
  21         print "reader (tid $tid): dequeued element $i: value $el\n";
  22         select(undef, undef, undef, rand(2));
  23         if ($el == -1) {
  24             # end marker
  25             print "reader (tid $tid) returning\n";
  26             return;
  27         }
  28     }
  29 }
  30
  31 my $nthreads = 3;
  32
  33 for (my $i = 0; $i < $nthreads; $i++) {
  34     Thread->new(\&reader, $i);
  35 }
  36
  37 for (my $i = 1; $i <= 10; $i++) {
  38     my $el = int(rand(100));
  39     select(undef, undef, undef, rand(2));
  40     print "writer: enqueuing value $el\n";
  41     $q->enqueue($el);
  42 }
  43
  44 $q->enqueue((-1) x $nthreads); # one end marker for each thread