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