lib/Thread/Queue.pm

   1 package Thread::Queue;
   2
   3 our $VERSION = '1.00';
   4
   5 our $ithreads;
   6 our $othreads;
   7
   8 use Thread qw(cond_wait cond_broadcast);
   9
  10 BEGIN {
  11     use Config;
  12     $ithreads = $Config{useithreads};
  13     $othreads = $Config{use5005threads};
  14     if($ithreads) {
  15         require 'threads/shared/queue.pm';
  16         for my $m (qw(new enqueue dequeue dequeue_nb pending)) {
  17             no strict 'refs';
  18             *{"Thread::Queue::$m"} = \&{"threads::shared::queue::${m}"};
  19         }
  20     } else {
  21         for my $m (qw(new enqueue dequeue dequeue_nb pending)) {
  22             no strict 'refs';
  23             *{"Thread::Queue::$m"} = \&{"Thread::Queue::${m}_othread"};
  24         }
  25     }
  26 }
  27
  28
  29 =head1 NAME
  30
  31 Thread::Queue - thread-safe queues
  32
  33 =head1 SYNOPSIS
  34
  35     use Thread::Queue;
  36     my $q = new Thread::Queue;
  37     $q->enqueue("foo", "bar");
  38     my $foo = $q->dequeue;    # The "bar" is still in the queue.
  39     my $foo = $q->dequeue_nb; # returns "bar", or undef if the queue was
  40                               # empty
  41     my $left = $q->pending;   # returns the number of items still in the queue
  42
  43 =head1 DESCRIPTION
  44
  45 A queue, as implemented by C<Thread::Queue> is a thread-safe data structure
  46 much like a list. Any number of threads can safely add elements to the end
  47 of the list, or remove elements from the head of the list. (Queues don't
  48 permit adding or removing elements from the middle of the list)
  49
  50 =head1 FUNCTIONS AND METHODS
  51
  52 =over 8
  53
  54 =item new
  55
  56 The C<new> function creates a new empty queue.
  57
  58 =item enqueue LIST
  59
  60 The C<enqueue> method adds a list of scalars on to the end of the queue.
  61 The queue will grow as needed to accomodate the list.
  62
  63 =item dequeue
  64
  65 The C<dequeue> method removes a scalar from the head of the queue and
  66 returns it. If the queue is currently empty, C<dequeue> will block the
  67 thread until another thread C<enqueue>s a scalar.
  68
  69 =item dequeue_nb
  70
  71 The C<dequeue_nb> method, like the C<dequeue> method, removes a scalar from
  72 the head of the queue and returns it. Unlike C<dequeue>, though,
  73 C<dequeue_nb> won't block if the queue is empty, instead returning
  74 C<undef>.
  75
  76 =item pending
  77
  78 The C<pending> method returns the number of items still in the queue.  (If
  79 there can be multiple readers on the queue it's best to lock the queue
  80 before checking to make sure that it stays in a consistent state)
  81
  82 =back
  83
  84 =head1 SEE ALSO
  85
  86 L<Thread>
  87
  88 =cut
  89
  90 sub new_othread {
  91     my $class = shift;
  92     return bless [@_], $class;
  93 }
  94
  95 sub dequeue_othread : locked : method {
  96     my $q = shift;
  97     cond_wait $q until @$q;
  98     return shift @$q;
  99 }
 100
 101 sub dequeue_nb_othread : locked : method {
 102   my $q = shift;
 103   if (@$q) {
 104     return shift @$q;
 105   } else {
 106     return undef;
 107   }
 108 }
 109
 110 sub enqueue_othread : locked : method {
 111     my $q = shift;
 112     push(@$q, @_) and cond_broadcast $q;
 113 }
 114
 115 sub pending_othread : locked : method {
 116   my $q = shift;
 117   return scalar(@$q);
 118 }
 119
 120 1;