1:11
  0   ˆl–ßi~”êa,jeA3q°ÜTÅ£Á„o²_‹uÐb&=LV[(ÁÀ >;    package Thread::Queue;

use strict;
use warnings;

our $VERSION = '3.02';
$VERSION = eval $VERSION;

use threads::shared 1.21;
use Scalar::Util 1.10 qw(looks_like_number blessed reftype refaddr);

# Carp errors from threads::shared calls should complain about caller
our @CARP_NOT = ("threads::shared");

# Predeclarations for internal functions
my ($validate_count, $validate_index, $validate_timeout);

# Create a new queue possibly pre-populated with items
sub new
{
    my $class = shift;
    my @queue :shared = map { shared_clone($_) } @_;
    my %self :shared = ( 'queue' => \@queue );
    return bless(\%self, $class);
}

# Add items to the tail of a queue
sub enqueue
{
    my $self = shift;
    lock(%$self);
    if ($$self{'ENDED'}) {
        require Carp;
        Carp::croak("'enqueue' method called on queue that has been 'end'ed");
    }
    push(@{$$self{'queue'}}, map { shared_clone($_) } @_)
        and cond_signal(%$self);
}

# Return a count of the number of items on a queue
sub pending
{
    my $self = shift;
    lock(%$self);
    return if ($$self{'ENDED'} && ! @{$$self{'queue'}});
    return scalar(@{$$self{'queue'}});
}

# Indicate that no more data will enter the queue
sub end
{
    my $self = shift;
    lock $self;
    # No more data is coming
    $$self{'ENDED'} = 1;
    # Try to release at least one blocked thread
    cond_signal(%$self);
}

# Return 1 or more items from the head of a queue, blocking if needed
sub dequeue
{
    my $self = shift;
    lock(%$self);
    my $queue = $$self{'queue'};

    my $count = @_ ? $validate_count->(shift) : 1;

    # Wait for requisite number of items
    cond_wait(%$self) while ((@$queue < $count) && ! $$self{'ENDED'});
    cond_signal(%$self) if ((@$queue > $count) || $$self{'ENDED'});

    # If no longer blocking, try getting whatever is left on the queue
    return $self->dequeue_nb($count) if ($$self{'ENDED'});

    # Return single item
    return shift(@$queue) if ($count == 1);

    # Return multiple items
    my @items;
    push(@items, shift(@$queue)) for (1..$count);
    return @items;
}

# Return items from the head of a queue with no blocking
sub dequeue_nb
{
    my $self = shift;
    lock(%$self);
    my $queue = $$self{'queue'};

    my $count = @_ ? $validate_count->(shift) : 1;

    # Return single item
    return shift(@$queue) if ($count == 1);

    # Return multiple items
    my @items;
    for (1..$count) {
        last if (! @$queue);
        push(@items, shift(@$queue));
    }
    return @items;
}

# Return items from the head of a queue, blocking if needed up to a timeout
sub dequeue_timed
{
    my $self = shift;
    lock(%$self);
    my $queue = $$self{'queue'};

    # Timeout may be relative or absolute
    my $timeout = @_ ? $validate_timeout->(shift) : -1;
    # Convert to an absolute time for use with cond_timedwait()
    if ($timeout < 32000000) {   # More than one year
        $timeout += time();
    }

    my $count = @_ ? $validate_count->(shift) : 1;

    # Wait for requisite number of items, or until timeout
    while ((@$queue < $count) && ! $$self{'ENDED'}) {
        last if (! cond_timedwait(%$self, $timeout));
    }
    cond_signal(%$self) if ((@$queue > $count) || $$self{'ENDED'});

    # Get whatever we need off the queue if available
    return $self->dequeue_nb($count);
}

# Return an item without removing it from a queue
sub peek
{
    my $self = shift;
    lock(%$self);
    my $index = @_ ? $validate_index->(shift) : 0;
    return $$self{'queue'}[$index];
}

# Insert items anywhere into a queue
sub insert
{
    my $self = shift;
    lock(%$self);

    if ($$self{'ENDED'}) {
        require Carp;
        Carp::croak("'insert' method called on queue that has been 'end'ed");
    }

    my $queue = $$self{'queue'};

    my $index = $validate_index->(shift);

    return if (! @_);   # Nothing to insert

    # Support negative indices
    if ($index < 0) {
        $index += @$queue;
        if ($index < 0) {
            $index = 0;
        }
    }

    # Dequeue