File Coverage

blib/lib/Async/Queue.pm

Criterion	Covered	Total	%
statement	84	84	100.0
branch	36	36	100.0
condition	26	27	96.3
subroutine	15	15	100.0
pod	5	5	100.0
total	166	167	99.4

package Async::Queue;

use 5.006;
use strict;
use warnings;

use Carp;
use Scalar::Util qw(looks_like_number);

sub new {
    my ($class, %options) = @_;
    my $self = bless {
        concurrency => 1,
        worker => undef,
        drain => undef,
        empty => undef,
        saturated => undef,
        task_queue => [],
        running => 0,
    }, $class;
    $self->$_($options{$_}) foreach qw(concurrency worker drain empty saturated);
    return $self;
}

sub _define_hook_accessors {
    my ($name, %options) = @_;
    my $class = __PACKAGE__;
    my $fullname = "${class}::$name";
    no strict 'refs';
    *{$fullname} = sub {
        my ($self, $v) = @_;
        if(@_ > 1) {
            croak "$name must not be undef." if !defined($v) && !$options{allow_undef};
            croak "$name must be a coderef" if defined($v) && ref($v) ne 'CODE';
            croak "You cannot set $name while there is a running task." if $self->running > 0;
            $self->{$name} = $v;
        }
        return $self->{$name};
    };
}

sub running {
    my ($self) = @_;
    return $self->{running};
}

sub concurrency {
    my ($self, $conc) = @_;
    if(@_ > 1) {
        croak "You cannot set concurrency while there is a running task" if $self->running > 0;
        $conc = 1 if not defined($conc);
        croak "concurrency must be a number" if !looks_like_number($conc);
        $self->{concurrency} = int($conc);
    }
    return $self->{concurrency};
}

sub length {
    my ($self) = @_;
    return int(@{$self->{task_queue}});
}

*waiting = \&length;

_define_hook_accessors 'worker';
_define_hook_accessors $_, allow_undef => 1 foreach qw(drain empty saturated);

sub push {
    my ($self, $task, $cb) = @_;
    if(@_ < 2) {
        croak("You must specify something to push.");
    }
    if(defined($cb) && ref($cb) ne 'CODE') {
        croak("callback for a task must be a coderef");
    }
    push(@{$self->{task_queue}}, [$task, $cb]);
    $self->_shift_run(1);
    return $self;
}

sub _shift_run {
    my ($self, $from_push) = @_;
    return if $self->concurrency > 0 && $self->running >= $self->concurrency;
    my $args_ref = shift(@{$self->{task_queue}});
    return if !defined($args_ref);
    my ($task, $cb) = @$args_ref;
    $self->{running} += 1;
    if($self->running == $self->concurrency && $from_push && defined($self->saturated)) {
        $self->saturated->($self);
    }
    if(@{$self->{task_queue}} == 0 && defined($self->empty)) {
        $self->empty->($self);
    }
    my $sync = 1;
    my $sync_completed = 0;
    $self->worker->($task, sub {
        my (@worker_results) = @_;
        $cb->(@worker_results) if defined($cb);
        $self->{running} -= 1;
        if(@{$self->{task_queue}} == 0 && $self->running == 0 && defined($self->drain)) {
            $self->drain->($self);
        }
        if($sync) {
            $sync_completed = 1;
        }else {
            @_ = ($self);
            goto &_shift_run;
        }
    }, $self);
    $sync = 0;
    if($sync_completed) {
        @_ = ($self);
        goto &_shift_run;
    }
}


=head1 NAME

Async::Queue - control concurrency of asynchronous tasks

=head1 VERSION

Version 0.021

=cut

our $VERSION = '0.021';


=head1 SYNOPSIS


    use Async::Queue;

    ## create a queue object with concurrency 2
    my $q = Async::Queue->new(
        concurrency => 2, worker => sub {
            my ($task, $callback) = @_;
            print "hello $task->{name}\n";
            $callback->();
        }
    );

    ## assign a callback
    $q->drain(sub {
        print "all items have been processed\n";
    });

    ## add some items to the queue
    $q->push({name => 'foo'}, sub {
        print "finished processing foo\n";
    });
    $q->push({name => 'bar'}, sub {
        print "finished processing bar\n";
    });


=head1 DESCRIPTION

L is used to process tasks with the specified concurrency.
The tasks given to L are processed in parallel with its worker routine up to the concurrency level.
If more tasks arrive at the L object, those tasks will wait for currently running tasks to finish.
When a task is finished, one of the waiting tasks starts to be processed in first-in-first-out (FIFO) order.

In short, L is a Perl port of the C object of async.js (L).

The basic usage of L is as follows:

=over

=item 1.

Create L object with C attribute and optional C attribute.
C is a subroutine reference that processes tasks. C is the concurrency level.

=item 2.

Push tasks to the L object via C method with optional callback functions.

The tasks will be processed in FIFO order by the C subroutine.
When a task is finished, the callback function, if any, is called with the results.


=back


=head1 CLASS METHODS

=head2 $queue = Async::Queue->new(%attributes);

Creates an L object.

It takes named arguments to initialize attributes of the L object.
See L for the list of the attributes.

C attribute is mandatory.


=head1 ATTRIBUTES

An L object has the following attributes.

You can initialize the attributes in C method.
You can get and set the attributes of an L object via their accessor methods (See L).

Note that you cannot set any attribute listed here while there is a task running in the L object.
This is because changing the attributes during task execution is very confusing and leads to unpredictable behavior.
So if you want to set an attribute, make sure there is no task running (C method can be useful).

=head2 worker (CODE($task, $callback, $queue), mandatory)

C attribute is a subroutine reference that processes a task. It must not be C.

C subroutine reference takes three arguments, C<$task>, C<$callback> and C<$queue>.

C<$task> is the task object the C is supposed to process.

C<$callback> is a callback subroutine reference that C must call when the task is finished.
C<$callback> can take any list of arguments, which will be passed to the C<$finish_callback> given to the C method
(See L).

C<$queue> is the L object that holds the worker.

So the C attribute is something like:

    my $q = Async::Queue->new(worker => sub {
        my ($task, $callback, $queue) = @_;
        my @results = some_processing($task);
        $callback->(@results);
    });

You can do asynchonous processing by deferring the call to C<$callback>:

    my $q = Async::Queue->new(worker => sub {
        my ($task, $callback, $queue) = @_;
        some_async_processing($task, on_finish => sub {
            my @results = @_;
            $callback->(@results);
        });
    });


=head2 concurrency (INT, optional, default = 1)

C attribute is the maximum number of tasks that can be processed at the same time.
It must be an integer number.

If C is set to 0 or any negative number, the concurrency level becomes infinite,
i.e. pushed tasks are immediately processed no matter how many are already running.

If C is set to C (or omitted in C method), it will be 1.


=head2 saturated (CODE($queue), optional, default = undef)

C attribute is a subroutine reference that is called when the number of running tasks hits C.
This means further tasks will wait in the queue.

C subroutine reference takes one argument (C<$queue>), which is the L object holding it.


=head2 empty (CODE($queue), optional, default = undef)

C attribute is a subroutine reference that is called when the last task from the queue is given to the worker.
This means there is no task waiting in the L object.

If the L object is not saturated, C subroutine is called every time a task is pushed.
This is because every pushed task goes into the queue first even if the L object can process the task immediately.

C subroutine reference takes one argument (C<$queue>), which is the L object holding it.

=head2 drain (CODE($queue), optional, default = undef)

C attribute is a subroutine reference that is called when the last task in the L object has finished.
This means there is no task running or waiting in the L object.

C subroutine reference takes one argument (C<$queue>), which is the C object holding it.

=head1 OBJECT METHODS

=head2 $queue->push($task, [$finish_callback->(@results)] );

Pushes a task into the L object.
The argument C<$task> is mandatory, while C<$finish_callback> is optional.

C<$task> is a task that the worker will process. It will be given as the C<$task> argument to the C subroutine.

C<$finish_callback> is a subroutine reference that will be called when the worker finishes processing the task.
The arguments for C<$finish_callback> (C<@results>) are the arguments for the C<$callback> subroutine reference in the C subroutine.

C method returns the L object.

=head2 $running_num = $queue->running();

Returns the number of currently running tasks in the L object.

=head2 $waiting_num = $queue->waiting();

Returns the number of waiting tasks in the L object.

=head2 $waiting_num = $queue->length();

Alias for C method. It returns the number of waiting tasks in the L object.

=head2 $worker = $queue->worker([$new_worker]);

Accessor for the C attribute.

=head2 $concurrency = $queue->concurrency([$new_concurrency]);

Accessor for the C attribute.

=head2 $saturated = $queue->saturated([$new_saturated]);

Accessor for the C attribute.

=head2 $empty = $queue->empty([$new_empty]);

Accessor for the C attribute.

=head2 $drain = $queue->drain([$new_drain]);

Accessor for the C attribute.

=head1 EXAMPLE

=head2 Concurrent HTTP downloader

    use strict;
    use warnings;
    use AnyEvent;
    use AnyEvent::HTTP;
    use Async::Queue;

    my $q = Async::Queue->new(concurrency => 3, worker => sub {
        my ($url, $callback) = @_;
        print STDERR "Start $url\n";
        http_get $url, sub {
            my ($data, $headers) = @_;
            print STDERR "End $url\n";
            $callback->($data);
        };
    });

    my @urls = (
        'http://www.debian.org/',
        'http://www.ubuntu.com/',
        'http://fedoraproject.org/',
        'http://www.opensuse.org/',
        'http://www.centos.org/',
        'http://www.slackware.com/',
        'http://www.gentoo.org/',
        'http://www.archlinux.org/',
        'http://trisquel.info/',
    );

    my %results = ();
    my $cv = AnyEvent->condvar;
    foreach my $url (@urls) {
        $cv->begin();
        $q->push($url, sub {
            my ($data) = @_;
            $results{$url} = $data;
            $cv->end();
        });
    }
    $cv->recv;

    foreach my $key (keys %results) {
        print STDERR "$key: " . length($results{$key}) . "bytes\n";
    }

This example uses L to send HTTP GET requests for multiple URLs simultaneously.
While simultaneous requests dramatically improve efficiency, it may overload the client host
and/or the network.

This is where L comes in handy. With L you can control the concurrency level
of the HTTP sessions (in this case, up to three).



=head1 SEE ALSO

=over

=item L

The goal of L is the same as that of L: to control concurrency level of asynchronous tasks.
The big difference is that L is a queue of subroutines while L is a queue of tasks (data).
In L, worker subroutine is registered with the object in advance.
In L, it is workers that are pushed to the queue.

You can emulate L with L by pushing subroutine references to it as tasks.

=back


=head1 AUTHOR

Toshio Ito, C<<  >>

=head1 REPOSITORY

L

=head1 BUGS

Please report any bugs or feature requests to C, or through
the web interface at L.  I will be notified, and then you'll
automatically be notified of progress on your bug as I make changes.



=head1 SUPPORT

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

    perldoc Async::Queue


You can also look for information at:

=over 4

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

L

=item * AnnoCPAN: Annotated CPAN documentation

L

=item * CPAN Ratings

L

=item * Search CPAN

L

=back


=head1 LICENSE AND COPYRIGHT

Copyright 2012 Toshio Ito.

This program is free software; you can redistribute it and/or modify it
under the terms of either: the GNU General Public License as published
by the Free Software Foundation; or the Artistic License.

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


=cut

1; # End of Async::Queue