Forks::Queue - queue that can be shared across processes
0.03
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;
Interface for a queue object that can be shared across processes. Available implementations are Forks::Queue::File, Forks::Queue::Shmem, Forks::Queue::SQLite.
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 prohibit data with blessed references and code references
{ name => "bad job", callback => \&my_callback_routine } [ 46, $url13, File::Temp->new ]
Instantiates a new queue object with the given configuration.
If one of the options is impl, the constructor from that Forks::Queue subclass will be invoked.
impl
Forks::Queue
If the impl option is not provided, this call will fail.
Options that should be supported on all implementations include
style => 'fifo' | 'lifo'
Indicates whether the "get" method will return items in first-in-first-out order or last-in-first-out order (and which end of the queue the "peek" method will examine)
limit => int
A maximum size for the queue. Set to a non-positive value to specify an unlimited size queue.
on_limit => 'block' | 'fail'
Dictates what the queue should do when an attempt is made to add items beyond the queue's limit. If block, the queue will block and wait until items are removed from the queue. If fail, the queue will warn and return immediately without changing the queue.
block
fail
See the "enqueue", "put", "push", "unshift", and "insert" methods, which are used to increase the length of the queue and may be affected by this setting.
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 ...
join is not necessary for child processes forked from a process with an existing queue
join
$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 ... }
persist => bool
Active Forks::Queue 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 persist option, if set to true, instructs the queue object to leave its state intact after destruction.
persist
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 persist and the join options to the Forks::Queue 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
list => ARRAYREF
Initializes the contents of the queue with the argument to the list option. The argument must be an array reference.
list
If the join option is specified, the contents of the list could be added to an already existing queue.
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 "end" method of the queue had previously been called from any process.
The enqueue method name is provided for compatibility with Thread::Queue.
enqueue
See the "limit" method to see how the put method behaves when adding items would cause the queue to exceed its maximum size.
put
Equivalent to "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 push or put will be the first items taken from the queue by "pop" or by "get" with LIFO style queues, and the last items removed by "shift" or "get" with FIFO style queues.
push
If the items added to the queue would cause the queue to exceed its queue size limit (as determined by the "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 "on_limit".
Equivalent to insert(0,@items), 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 unshift will be the last items taken from the queue by "get", and in LIFO queues, they will be the first items taken from the queue by "get".
insert(0,@items)
unshift
This method is inefficient for some queue implementations.
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 until a new item is added. Causes any processes blocking on a "get"/"dequeue"/"shift"/"pop" call to become unblocked and return undef. This method may be called from any process that has access to the queue.
undef
Attempt to retrieve one or more "items" on the queue. If the queue is empty, and if "end" has not been called on the queue, this call blocks until an item is available or until the "end" method has been called from some other process. If the queue is empty and "end" has been called, this method returns an empty list in list context or undef in scalar context.
If a $count argument is supplied, returns up to $count items or however many items are currently availble on the queue, whichever is fewer. But the call still blocks if "end" has not been called until there is at least one item available. See "get_nb" for a non-blocking version of this method. The return value of this function when a $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.
$count
$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 get and dequeue is what happens when there is a $count argument, and the queue currently has more than zero but less than $count items available. In this case, the get call will return all of the available items. The dequeue method will block until at least $count items are available on the queue, or until the "end" method has been called on the queue. This dequeue behavior is consistent with the behavior of the "dequeue" method in Thread::Queue.
get
dequeue
Retrieves one or more items from the "back" of the queue. For LIFO style queues, the "get" method is equivalent to this method. Like "get", this method blocks while the queue is empty and the "end" method has not been called on the queue.
"get"
If a $count argument is supplied, returns up to $count items or however many items are currently available on the queue, whichever is fewer. (Like the "get" call, this method blocks when waiting for input. See "pop_nb" for a non-blocking version of the method. Also like "get", you should be wary of using this method in scalar context if you provide a $count argument).
Retrieves one or more items from the "front" of the queue. For FIFO style queues, the "get" method is equivalent to this method. Like "get", this method blocks while the queue is empty and the "end" method has not been called on the queue.
If a $count argument is supplied, returns up to $count items or however many items are currently available on the queue, whichever is fewer. (Like the "get" call, this method blocks when waiting for input. See "shift_nb" for a non-blocking version of the method. Also like "get", you should be wary of using this method in scalar context if you provide a $count argument).
Non-blocking versions of the "get", "dequeue", "pop", and "shift" methods. These functions return immediately if there are no items in the queue to retrieve, returning undef in the case with no arguments and an empty list when a $count argument is supplied.
Timed versions of "get", "dequeue", "shift", and "pop" that take a $timeout argument and will stop blocking after $timeout seconds have elapsed.
$timeout
If a $count argument is supplied to dequeue_timed, the function will wait up to $timeout seconds for at least $count items to be available on the queue. After $timeout seconds have passed, the function will return up to $count available items.
dequeue_timed
For other timed methods, supplying a $count argument for a queue with more than zero but less than $count items available will return all available items without blocking.
Returns an item from the queue without removing it. The peek_front and peek_back methods inspect the item at the front and the back of the queue, respectively. The generic peek method is equivalent to peek_front for FIFO style queues and peek_back 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 $queue->peek(-2), for example, would return the second to last item in the queue.
peek_front
peek_back
peek
$queue->peek(-2)
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 undef without blocking.
Note that unlike the <"peek" method in Thread::Queue>, Forks::Queue::peek returns a copy of the item on the queue, so manipulating a reference returned from peek while not affect the item on the queue.
Thread::Queue
Forks::Queue::peek
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 $count argument is not provided, the method will return (if available) a single item. If the $index argument is also not provided, it will return the first item on the queue exactly like the "get_nb" method with no arguments.
$index
Negative $index values are supported, in which case this method will extract the corresponding items at the back of the queue.
Like get() vs. get($count), the return value is always a scalar when no $count argument is provided, and always a list when it is.
get()
get($count)
Provides random access to the queue, inserting the items specified in @list into the queue after index position $index. Negative $index values are supported, which indicate that the items should be inserted after that position relative to the back of the queue.
@list
Returns the number of items that were inserted into the queue. If the queue has a "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 "on_limit" setting.
Returns the total number of items available on the queue. There is no guarentee that the number of available items will not change between a call to pending and a subsequent call to "get"
pending
Removes all items from the queue.
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.
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 limit method to be used in the same way as "limit" in Thread::Queue.
limit
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 block, which causes an insertion operation to block until there is capacity on the queue, or fail, which returns immediately from an insertion operation with a warning about items that were not added to the queue.
The Forks::Queue module and all its current implementations require the JSON module.
Thread::Queue, File::Queue, Queue::Q, MCE::Queue, Queue::DBI, Directory::Queue.
You can find documentation for this module with the perldoc command.
perldoc Forks::Queue
You can also look for information at:
RT: CPAN's request tracker (report bugs here)
http://rt.cpan.org/NoAuth/Bugs.html?Dist=Forks-Queue
AnnoCPAN: Annotated CPAN documentation
http://annocpan.org/dist/Forks-Queue
CPAN Ratings
http://cpanratings.perl.org/d/Forks-Queue
Search CPAN
http://search.cpan.org/dist/Forks-Queue/
Copyright (c) 2017, 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.
To install Forks::Queue, copy and paste the appropriate command in to your terminal.
cpanm
cpanm Forks::Queue
CPAN shell
perl -MCPAN -e shell install Forks::Queue
For more information on module installation, please visit the detailed CPAN module installation guide.