sub consumer {
        while( my($returned_value) = $queue->dequeue ) {
                printf "I have been returned: %i\n", $returned_value;
        }
}
[download]

I think you're being fooled by printf(). The "%i" turns an undef value into 0.

Liz

I thought that enqueuing an undef would have stopped that while loop and hence didn't much care about how it would interact with printf.

Is the behaviour of enqueuing (or perhaps more specifically dequeuing) undef's under Thread::Queue::Any different than under Thread::Queue for some reason that I'm not thinking of?

Many thanks.

while( my($returned_value) = $queue->dequeue ) {
[download]

You forget that $queue->dequeue always returns an array with at least 1 element. So the condition being checked is always 1 or higher. Something like:

while( my ($returned_value) = $queue->dequeue ) {
    last unless defined $returned_value;
[download]

should do the trick, I think.

Liz

I thought that enqueuing an undef would have stopped that while loop

Nope, that's a list assignment in scalar context, which evaluates to the number of items on the right of the assignment.

Further to this, I've just added the following to the documentation of Thread::Conveyor and uploaded it to CPAN:

---

Why would you use Thread::Conveyor over Thread::Queue::Any? Well, Thread::Conveyor has the following extra features:

It works with Perl 5.8.0

Shared arrays leak memory very badly in Perl 5.8.0. Therefore, you cannot really use Thread::Queue in Perl 5.8.0, and consequently cannot use Thread::Queue::Any in any type of production environment.

It provides throttling

A thread that enqueues very many values quickly, can cause a large amount of memory to be used. With throttling, any thread that enqueues will have to wait until there is "room" on the belt again before continuing. See methods "minboxes" and "maxboxes".

You can check for a new value without removing it from the belt

Sometimes it can be nice to check whether there is a new value on the belt without actually removing it from the belt. See the "peek" and "peek_dontwait" methods.

You can reset the entire belt

Sometimes you want to be able to reset the contents of the belt. See the "clean" and "clean_dontwait" methods for that.

You can get everything from the belt in one go

Sometimes you want everything that's on the belt in one go. That can also ba accomplished with the "clean" and "clean_dontwait" methods.

---

Liz

Remove the brackets from the my statement in the while condition.

   while( my $returned_value = $queue->dequeue   ) {
   #        ^               ^
[download]

The code will then behave as you expect.

Update: Sorry! Wrong answer. I hadn't reinstalled Thread::Queue::Any since ungrading and tried it with Thread::Queue instead.

P:\test>type 319494.pl8
use strict;
use threads;
use Thread::Queue;

my $queue = Thread::Queue->new;

my $prod_thread = new threads(\&producer, $queue);
my $cons_thread = new threads(\&consumer, $queue);

$prod_thread->join;
$cons_thread->join;

sub producer {
        my $queue = $_[0];
        my $value = 1234;
        $queue->enqueue( $value );
        $queue->enqueue( undef );
}

sub consumer {
        my $queue = $_[0];
        while( my $returned_value = $queue->dequeue   ) {
                printf "I have been returned: %i\n", $returned_value;
        }
}

P:\test>319494
I have been returned: 1234
P:\test>
[download]

It worked! ...but not with Thread::Queued::Any :(

The reason for Thread::Queue::Any over Thread::Queue (if it matters) is that I need to be queuing references to data structures.

From a previous post of mine it turns out that I cannot do what I need with Thread::Queue otherwise I'd be using it :)

Hey ho, 'tis a shame :)

Many thanks.

Thanks to Molt and to liz - Thread::Conveyor seems to be working with the following piece of code:

#!/opt/perl-5.8.2/bin/perl -w

use strict;
use threads;
use Thread::Conveyor;

my $queue = Thread::Conveyor->new;

my $prod_thread = new threads(\&producer);
my $cons_thread = new threads(\&consumer);

$prod_thread->join;
$cons_thread->join;

sub producer {
        my $value = 1234;
        $queue->put( $value );
        $queue->put( undef );
}

sub consumer {
        while( my $returned_value = $queue->take ) {
                printf "I have been returned: %i\n", $returned_value;
        }
}
[download]