File Coverage

blib/lib/Forks/Queue.pm

Criterion	Covered	Total	%
statement	100	146	68.4
branch	39	62	62.9
condition	18	19	94.7
subroutine	28	44	63.6
pod	22	24	91.6
total	207	295	70.1

package Forks::Queue;
use strict;
use warnings;
use Scalar::Util 'looks_like_number';
use Carp;
use Time::HiRes;
use Config;

our $VERSION = '0.13';
our $DEBUG = $ENV{FORKS_QUEUE_DEBUG} || 0;

our $NOTIFY_OK = $ENV{FORKS_QUEUE_NOTIFY} // do {
    $Config::Config{sig_name} =~ /\bIO\b/;
};

our $SLEEP_INTERVAL = $NOTIFY_OK ? 2 : 1;
our $SLEEP_INTERVALX = 2;

# default values to apply to all new Forks::Queue implementations
our %OPTS = (limit => -1, on_limit => 'fail', style => 'fifo',
             impl => $ENV{FORKS_QUEUE_IMPL} || 'File' );

sub new {
    my $class = shift;
    no warnings 'once';
    my %opts = (%OPTS, @_);

    if ($opts{impl}) {
        my $pkg = delete $opts{impl};
        if ($pkg =~ /[^\w:]/) {
            croak "Forks::Queue cannot be instantiated. Invalid 'impl' $pkg";
        }
        if (eval "require Forks::Queue::$pkg; 1") {
            $pkg = "Forks::Queue::" . $pkg;
            return $pkg->new(%opts);
        } elsif (eval "require $pkg; 1") {
            return $pkg->new(%opts);
        } else {
            croak "Forks::Queue cannot be instantiated. ",
                "Did not recognize 'impl' option '$opts{impl}'";
        }
    }
    croak "Forks::Queue cannot be instantiated. ",
        "Use an implementation or pass the 'impl' option to the constructor";
}

sub import {
    my ($class, @args) = @_;
    for (my $i=0; $i<@args; $i++) {
        foreach my $key (qw(limit on_limit impl style)) {
            if ($args[$i] eq $key) {
                $OPTS{$args[$i]} = $args[$i+1];
                $i++;
                next;
            }
        }
    }
}

sub put {
    my $self = shift;
    return $self->push(@_);
}

sub enqueue { goto &put; }

sub get {
    my $self = shift;
    _validate_input($_[0], 'count', 1) if @_;
    if ($self->{style} eq 'fifo') {
        return @_ ? $self->shift(@_) : $self->shift;
    } else {
        my @gotten = @_ ? reverse($self->pop(@_)) : $self->pop;
        return @_ ? @gotten : $gotten[0];
    }
}

sub dequeue { goto &get; }

sub dequeue_timed {
    my $self = shift;
    my $timeout = shift;
    _validate_input($timeout, 'timeout', 0, 1);
    local $self->{_expire} = Time::HiRes::time + $timeout;
    local $SLEEP_INTERVAL = $Forks::Queue::SLEEP_INTERVAL;
    $SLEEP_INTERVAL /= 2 while $SLEEP_INTERVAL > 0.25 * $timeout;
    return @_ ? $self->dequeue(@_) : $self->dequeue;
}

sub get_timed {
    my $self = shift;
    my $timeout = shift;
    _validate_input($timeout, 'timeout', 0, 1);
    local $self->{_expire} = Time::HiRes::time + $timeout;
    local $SLEEP_INTERVAL = $Forks::Queue::SLEEP_INTERVAL;
    $SLEEP_INTERVAL /= 2 while $SLEEP_INTERVAL > 0.25 * $timeout;
    return @_ ? $self->get(@_) : $self->get;
}

sub shift_timed {
    my $self = shift;
    my $timeout = shift;
    _validate_input($timeout, 'timeout', 0, 1);
    local $self->{_expire} = Time::HiRes::time + $timeout;
    local $SLEEP_INTERVAL = $Forks::Queue::SLEEP_INTERVAL;
    $SLEEP_INTERVAL /= 2 while $SLEEP_INTERVAL > 0.25 * $timeout;
    return @_ ? $self->shift(@_) : $self->shift;
}

sub pop_timed {
    my $self = shift;
    my $timeout = shift;
    _validate_input($timeout, 'timeout', 0, 1);
    local $self->{_expire} = Time::HiRes::time + $timeout;
    local $SLEEP_INTERVAL = $Forks::Queue::SLEEP_INTERVAL;
    $SLEEP_INTERVAL /= 2 while $SLEEP_INTERVAL > 0.25 * $timeout;
    return @_ ? $self->pop(@_) : $self->pop;
}

sub _expired {
    my ($self) = @_;
    $self->{_expire} && Time::HiRes::time >= $self->{_expire};
}

sub get_nb {
    my $self = shift;
    _validate_input($_[0], 'count', 1) if @_;
    if ($self->{style} eq 'fifo') {
        return @_ ? $self->shift_nb(@_) : $self->shift_nb;
    } else {
        return @_ ? @ := reverse($self->pop_nb(@_)) : $self->pop_nb;
    }
}
sub dequeue_nb { goto &get_nb; }

sub peek {
    my ($self,$index) = @_;
    _validate_input($index, 'index') if @_ > 1;
    if ($self->{style} eq 'lifo') {
        return $self->peek_back($index || 0);
    } else {
        return $self->peek_front($index || 0);
    }
}

sub pending {
    my $self = shift;
    my $s = $self->status;
    return $s->{avail} ? $s->{avail} : $s->{end} ? undef : 0;
}

sub _unimpl {
    my $func = shift;
    croak "Forks::Queue: $func not implemented in abstract base class";
}

sub limit :lvalue {
    my $self = shift;
    if (@_) {
        $self->{limit} = shift @_;
        if (@_) {
            $self->{on_limit} = shift @_;
        }
    }
    $self->{limit};
}

sub _validate_input {
    my ($value,$name,$ge,$float_ok) = @_;
    croak "Invalid '$name'"
        if !defined($value) ||
        !looks_like_number($value) ||
        (!$float_ok && $value != int($value)) ||
        (defined($ge) && $value < $ge);
    return $value;
}

sub push       { _unimpl("push/put") }
sub peek_front { _unimpl("peek") }
sub peek_back  { _unimpl("peek") }
sub shift :method { _unimpl("shift/get") }
sub unshift    { _unimpl("unshift") }
sub pop        { _unimpl("pop/get") }
sub shift_nb   { _unimpl("shift/get") }
sub pop_nb     { _unimpl("pop/get") }
sub status     { _unimpl("pending/status") }
sub clear      { _unimpl("clear") }






sub Forks::Queue::Util::__is_nfs {
    my ($dir) = @_;
    if ($^O ne 'linux') {
        return;
    }
    my $pid = fork();
    if ($pid == 0) {
        close STDOUT;
        close STDERR;
        # http://superuser.com/q/422061
        exec("df",$dir,"-t","nfs");
        die;
    }
    local $?;
    waitpid $pid,0;
    return $? == 0;
}

sub Forks::Queue::Util::QID {
    no warnings 'uninitialized';
    if ($Forks::Queue::Util::PID != $$) {
        $Forks::Queue::Util::PID = $$;
        $Forks::Queue::Util::QID = 0;
    }
    $Forks::Queue::Util::QID++;
    join("-", $Forks::Queue::Util::PID, $Forks::Queue::Util::QID);
}


# manage global destruction phase
BEGIN {
    if (defined(${^GLOBAL_PHASE})) {
        eval 'sub __inGD(){%{^GLOBAL_PHASE} eq q{DESTRUCT} && __END()};1'
    } else {
        require B;
        eval 'sub __inGD(){${B::main_cv()}==0 && __END()};1'
    }
}
END { &__END }
sub __END {
    no warnings 'redefine';
    *DB::DB = sub {};
    *__inGD = sub () { 1 };
}



1;

=head1 NAME

Forks::Queue - queue that can be shared across processes

=head1 VERSION

0.13

=head1 SYNOPSIS

  use Forks::Queue;
  $q = Forks::Queue->new( impl => ..., style => 'lifo' );

  # put items on queue
  $q->put("a scalar item");
  $q->put(["an","arrayref","item"]);
  $q->put({"a"=>"hash","reference"=>"item"});
  $q->put("list","of","multiple",["items"]);
  $q->end;        # no more jobs will be added to queue

  # retrieve items from queue, possibly after a fork
  $item = $q->get;
  $item = $q->peek;      # get item without removing it
  @up_to_10_items = $q->get(10);
  $remaining_items = $q->pending;

=head1 DESCRIPTION

Interface for a queue object that can be shared across processes
and threads.
Available implementations are L,
L,
L.

=head1 METHODS

Many of these methods pass or return "items". For this distribution,
an "item" is any scalar or reference that can be serialized and
shared across processes.

This will include scalars and most unblessed references

  "42"
  [1,2,3,"forty-two"]
  { name=>"a job", timestamp=>time, input=>{foo=>[19],bar=>\%bardata} }

but will generally preclude data with blessed references and code references

  { name => "bad job", callback => \&my_callback_routine }
  [ 46, $url13, File::Temp->new ]

=head2 new

    $queue = Forks::Queue->new( %opts )

Instantiates a new queue object with the given configuration.

If one of the options is C, the constructor from that
C subclass will be invoked.

Other options that should be supported on all implementations include

=over 4

=item * style

=item * C<< style => 'fifo' | 'lifo' >>

Indicates whether the L<"get"> method will return items in
first-in-first-out order or last-in-first-out order (and which
end of the queue the L<"peek"> method will examine)

=item * limit

=item * C<< limit => int >>

A maximum size for the queue. Set to a non-positive value to
specify an unlimited size queue.

=item * on_limit

=item * C<< on_limit => 'block' | 'fail' >>

Dictates what the queue should do when an attempt is made to
add items beyond the queue's limit. If C, the queue
will block and wait until items are removed from the queue.
If C, the queue will warn and return immediately without
changing the queue.

See the L<"enqueue">, L<"put">, L<"push">, L<"unshift">,
and L<"insert"> methods, which are used to increase the length
of the queue and may be affected by this setting.

=item * join

=item * C<< join => bool >>

If true, expects that the queue referred to by this constructor
has already been created in another process, and that the current
process should access the existing queue. This allows a queue to
be shared across unrelated processes (i.e., processes that do not
have a parent-child relationship).

  # my_daemon.pl - may run "all the time" in the background
  $q = Forks::Queue::File->new(file=>'/var/spool/foo/q17');
  # creates new queue object
  ...

  # worker.pl - may run periodically for a short time, launched from
  #             cron or from command line, but not from the daemon
  $q = Forks::Queue->new( impl => 'File', join => 1,
                          file => '/var/spool/foo/q17',
  # the new queue attaches to existing file at /var/spool/foo/q17
  ...

C is not necessary for child processes forked from a process with
an existing queue

  $q = Forks::Queue->new(...)
  ...
  if (fork() == 0) {
      # $q already exists and the child process can begin using it,
      # no need for a  Forks::Queue  constructor with  join
      ...
  }

=item * persist

=item * C<< persist => bool >>

Active C objects affect your system, writing to disk or
writing to memory, and in general they clean themselves up when they
detect that no more processes are using the queue. The C option,
if set to true, instructs the queue object to leave its state intact
after destruction.

An obvious use case for this option is debugging, to examine the
state of the queue after abnormal termination of your program.

A second use case is to create persistent queues -- queues that are
shared not only among different processes, but among different
processes that are running at different times. The persistent queue
can be used by supplying both the C and the C options
to the C constructor.

    $queue_file = "/tmp/persistent.job.queue";
    $join = -f $queue_file;
    $q = Forks::Queue->new( impl => 'File', file => $queue_file,
                            join => $join, persist => 1 );
    ... work with the queue ...
    # the queue remains intact if this program exits or aborts

=item * C<< list => ARRAYREF >>

Initializes the contents of the queue with the argument to the
C option. The argument must be an array reference.

If the C option is specified, the contents of the list
could be added to an already existing queue.

=back

See the global L<"%OPTS"> variable for information about
default values for many of these settings.

=head2 put

=head2 enqueue

    $count = $queue->put(@items);
    $count = $queue->enqueue(@items)

Place one or more "items" on the queue, and returns the number of
items successfully added to the queue.

Adding items to the queue will fail if the L<"end"> method of
the queue had previously been called from any process.

The C method name is provided for compatibility with
L.

See the L<"limit"> method to see how the C method behaves
when adding items would cause the queue to exceed its maximum size.


=head2 push

    $count = $queue->push(@items)

Equivalent to L<"put">, adding items to the end of the queue and
returning the number of items successfully added. The most recent
items appended to the queue by C or C will be the first
items taken from the queue by L<"pop"> or by L<"get"> with LIFO
style queues, and the last items removed by L<"shift"> or L<"get">
with FIFO style queues.

If the items added to the queue would cause the queue to exceed
its queue size limit (as determined by the L<"limit"> attribute),
this method will either block until queue capacity is available,
or issue a warning about the uninserted items and return the
number of items added, depending on the queue's setting for
L<"on_limit"|Forks::Queue/"new">.


=head2 unshift

    $count = $queue->unshift(@items)

Equivalent to C, adding items to the front
of the queue, and returning the number of items successfully
added. In FIFO queues, items added to the queue with C
will be the last items taken from the queue by L<"get">,
and in LIFO queues, they will be the first items taken from the
queue by L<"get">.

This method is inefficient for some queue implementations.

=head2 end

    $queue->end

Indicates that no more items are to be put on the queue,
so that when a process tries to retrieve an item from an empty queue,
it will not block and wait for new items to be added. Causes any
processes blocking on a L<"get">/L<"dequeue">/L<"shift">/L<"pop">
call to become unblocked and return C.
This method may be called from any process that has access to the queue.

Calling C on a queue more than once will generate a warning
message, even if the caller is not the same process/thread that
made the original C call.


=head2 get

=head2 dequeue

    $item = $queue->get;
    $item = $queue->dequeue;

    @items = $queue->get($count);
    @items = $queue->dequeue($count);

Attempt to retrieve one or more "items" on the queue. If the
queue is empty, and if L<"end"> has not been called on the queue,
this call blocks until an item is available or until the L<"end">
method has been called from some other process. If the queue is
empty and L<"end"> has been called, this method returns an
empty list in list context or C in scalar context.

If a C<$count> argument is supplied, returns up to C<$count> items
or however many items are currently availble on the queue, whichever
is fewer. But the call still blocks if L<"end"> has not been called
until there is at least one item available. See L<"get_nb"> for a
non-blocking version of this method. The return value of this
function when a C<$count> argument is supplied is always a list,
so if you evaluate it in scalar context you will get the number of items
retrieved from the queue, not the items themselves.

    $job = $q->get;         # $job is an item from the queue
    $job = $q->get(1);      # returns # of items retrieved, not an actual item!
    ($job) = $q->get(1);    # $job is an item from the queue

The only important difference between C and C is what
happens when there is a C<$count> argument, and the queue currently has
more than zero but fewer than C<$count> items available. In this case,
the C call will return all of the available items. The C
method will block until at least C<$count> items are available on the
queue, or until the L<"end"> method has been called on the queue.
This C behavior is consistent with the behavior of the
L<"dequeue" method in Thread::Queue|Thread::Queue/"dequeue">.


=head2 pop

    $item = $queue->pop
    @items = $queue->pop($count)

Retrieves one or more items from the "back" of the queue.
For LIFO style queues, the L<"get"> method is equivalent to this method.
Like C<"get">, this method blocks while the queue is empty and the
L<"end"> method has not been called on the queue.

If a C<$count> argument is supplied, returns up to C<$count> items or however
many items are currently available on the queue, whichever is fewer.
(Like the L<"get"> call, this method blocks when waiting for input. See
L<"pop_nb"> for a non-blocking version of the method. Also like
L<"get">, you should be wary of using this method in scalar context
if you provide a C<$count> argument).

=head2 shift

    $item = $queue->shift
    @items = $queue->shift($count)

Retrieves one or more items from the "front" of the queue.
For FIFO style queues, the L<"get"> method is equivalent to this method.
Like C<"get">, this method blocks while the queue is empty and the
L<"end"> method has not been called on the queue.

If a C<$count> argument is supplied, returns up to C<$count> items or however
many items are currently available on the queue, whichever is fewer. (Like the
L<"get"> call, this method blocks when waiting for input. See
L<"shift_nb"> for a non-blocking version of the method. Also like
L<"get">, you should be wary of using this method in scalar context
if you provide a C<$count> argument).


=head2 get_nb

=head2 dequeue_nb

=head2 pop_nb

=head2 shift_nb

    $item = $queue->XXX_nb
    @items = $queue->XXX_nb($count)

Non-blocking versions of the L<"get">, L<"dequeue">, L<"pop">,
and L<"shift"> methods. These functions return immediately if
there are no items in the queue to retrieve, returning C
in the case with no arguments and an empty list when a
C<$count> argument is supplied.

=head2 get_timed

=head2 dequeue_timed

=head2 shift_timed

=head2 pop_timed

    $item = $queue->XXX_timed($timeout)
    @item = $queue->XXX_timed($timeout,$count)

Timed versions of L<"get">, L<"dequeue">, L<"shift">, and L<"pop">
that take a C<$timeout> argument and will stop blocking after
C<$timeout> seconds have elapsed.

If a C<$count> argument is supplied to C, the function
will wait up to C<$timeout> seconds for at least C<$count> items to
be available on the queue. After C<$timeout> seconds have passed,
the function will return up to C<$count> available items.

For other timed methods, supplying a C<$count> argument for a
queue with more than zero but fewer than C<$count> items available
will return all available items without blocking.


=head2 peek

    $item = $queue->peek
    $item = $queue->peek($index)
    $item = $queue->peek_front
    $item = $queue->peek_back

Returns an item from the queue without removing it. The C
and C methods inspect the item at the front and the back of
the queue, respectively. The generic C method is equivalent to
C for FIFO style queues and C for LIFO style
queues. If an index is specified, returns the item at that position
in the queue (where position 0 is the head of the queue). Negative
indices are supported, so a call to  C<< $queue->peek(-2) >>,
for example, would return the second to last item in the queue.

If the queue is empty or if the specified index is larger than the
number of elements currently in the queue, these methods will
return C without blocking.

Note that unlike the
L<< "peek" method in C|Thread::Queue/"peek" >>,
C returns a copy of the item on the queue,
so manipulating a reference returned from C while B
affect the item on the queue.

=head2 extract

    $item = $queue->extract
    $item = $queue->extract($index)
    @items = $queue->extract($index,$count)

Removes and returns the specified number of items from the queue
at the specified index position, to provide random access to the
queue. The method is non-blocking and may return fewer than the
number of items requested (or zero items) if there are not enough
items in the queue to satisfy the request.

If the C<$count> argument is not provided, the method will return
(if available) a single item. If the C<$index> argument is also
not provided, it will return the first item on the queue exactly
like the L<"get_nb"> method with no arguments.

Negative C<$index> values are supported, in which case this
method will extract the corresponding items at the back of the
queue.

Like C vs. C, the return value is always a
scalar when no C<$count> argument is provided, and always a list
when it is.


=head2 insert

    $count = $queue->insert($index, @list)

Provides random access to the queue, inserting the items specified
in C<@list> into the queue after index position C<$index>.
Negative C<$index> values are supported, which indicate that the
items should be inserted after that position relative to the
back of the queue.

Returns the number of items that were inserted into the queue.
If the queue has a L<"limit"> set, and inserting all the items on
the list would cause the queue size to exceed the limit, this
method will either block until capacity to insert the whole list
becomes available, or it will insert items up to the queue size
limit and issue a warning about the uninserted items, depending
on the queue's L<"on_limit"|Forks::Queue/"new"> setting.

This method is inefficient for some queue implementations.


=head2 pending

    $num_items_avail = $queue->pending

Returns the total number of items available on the queue. There is no
guarantee that the number of available items will not change between a
call to C and a subsequent call to L<"get">

=head2 clear

    $queue->clear

Removes all items from the queue.


=head2 status

    $status = $queue->status

Returns a hash reference with meta information about the queue.
The information should at least include the number of items remaining in
the queue. Other implementations may provide additional information
in this return value.


=head2 limit

    $max = $queue->limit
    $queue->limit($new_limit)
    $queue->limit($new_limit,$on_limit)
    $queue->limit = $new_limit    # may not work on some configurations

Returns or updates the maximum size of the queue. With no args, returns
the existing maximum queue size, with a non-positive value indicating
that the queue does not have a maximum size.

The return value also acts as an lvalue through which the maximum
queue size can be set, and allows the C method to be used
in the same way as
L<< the C method in Thread::Queue|Thread::Queue/"limit" >>.
I<< Note: lvalue feature may not work with Perl v<5.14. >>

If arguments are provided, the first argument is used to set the
maximum queue size. A non-positive queue size can be specified to
indicate that the queue does not have a maximum size.
The second argument, if provided, updates the behavior of the queue
when an attempt is made to add items beyond the maximum size.
The acceptable values for the second argument are C, which causes
an insertion operation to block until there is capacity on the queue,
or C, which returns immediately from an insertion operation with
a warning about items that were not added to the queue.


=head1 VARIABLES

=head2 %OPTS

Global hash containing the set of default options for all
C constructors. Initially this hash contains the
key-value pairs

        impl            "File"
        style           "fifo"
        limit           -1
        on_limit        "fail"

but they may be changed at any time to affect all subsequently
constructed C objects. The global options can also
be set at import time with additional arguments for the C
statement.

    use Forks::Queue impl => 'SQLite';    # use SQLite queues by default
    $Forks::Queue::OPTS{impl} = 'SQLite'; # equivalent run-time call

    use Forks::Queue
        on_limit => 'block', limit => 10; # finite, blocking queues by default
    $Forks::Queue::OPTS{limit} = 10;
    $Forks::Queue::OPTS{on_limit} = 'block';  # equivalent run-time calls

=head1 ENVIRONMENT

Some environment variable settings that can affect this module:

=over 4

=item * FORKS_QUEUE_IMPL

Specifies a default implementation to use, overriding the initial setting
of C<$Forks::Queue::OPTS{"impl"}>, in cases where the C
constructor is invoked without passing an C option.

=item * FORKS_QUEUE_DEBUG

If set to a true value, outputs information about the activity of
the queues to standard error.

=item * FORKS_QUEUE_NOTIFY

If set to a false value, disables use of signals on POSIX-y platforms
that may help improve queue performance

=item * FORKS_QUEUE_DIR

Specifies a directory to use for temporary queue files in the
L and
L implementations.
If this directory is not specified, the implementations will try to make
a reasonable choice based on your platform and other environment settings.

=back

=head1 DEPENDENCIES

The C module and all its current implementations require
the L module.

=head1 SEE ALSO

L, L,
L, L,
L, L.

=head1 SUPPORT

You can find documentation for this module with the perldoc command.

    perldoc Forks::Queue


You can also look for information at:

=over 4

=item * RT: CPAN's request tracker (report bugs here)

L

=item * CPAN Ratings

L

=item * Search CPAN

L

=back


=head1 LICENSE AND COPYRIGHT

Copyright (c) 2017-2019, Marty O'Brien.

This library is free software; you can redistribute it and/or modify
it under the same terms as Perl itself, either Perl version 5.10.1 or,
at your option, any later version of Perl 5 you may have available.

See http://dev.perl.org/licenses/ for more information.

=cut

# TODO:
#
#     priorities
#     Directory implementation (see Queue::Dir)
#