view release on metacpan or  search on metacpan

lib/AnyEvent.pm  view on Meta::CPAN

to use a condition variable for the whole process.

Every call to C<< ->begin >> will increment a counter, and every call to
C<< ->end >> will decrement it.  If the counter reaches C<0> in C<< ->end
>>, the (last) callback passed to C<begin> will be executed, passing the
condvar as first argument. That callback is I<supposed> to call C<< ->send
>>, but that is not required. If no group callback was set, C<send> will
be called without any arguments.

You can think of C<< $cv->send >> giving you an OR condition (one call
sends), while C<< $cv->begin >> and C<< $cv->end >> giving you an AND
condition (all C<begin> calls must be C<end>'ed before the condvar sends).

Let's start with a simple example: you have two I/O watchers (for example,
STDOUT and STDERR for a program), and you want to wait for both streams to
close before activating a condvar:

   my $cv = AnyEvent->condvar;

   $cv->begin; # first watcher
   my $w1 = AnyEvent->io (fh => $fh1, cb => sub {
      defined sysread $fh1, my $buf, 4096
         or $cv->end;
   });

   $cv->begin; # second watcher
   my $w2 = AnyEvent->io (fh => $fh2, cb => sub {
      defined sysread $fh2, my $buf, 4096
         or $cv->end;
   });

   $cv->recv;

This works because for every event source (EOF on file handle), there is
one call to C<begin>, so the condvar waits for all calls to C<end> before
sending.

The ping example mentioned above is slightly more complicated, as the
there are results to be passed back, and the number of tasks that are
begun can potentially be zero:

   my $cv = AnyEvent->condvar;

   my %result;
   $cv->begin (sub { shift->send (\%result) });

   for my $host (@list_of_hosts) {
      $cv->begin;
      ping_host_then_call_callback $host, sub {
         $result{$host} = ...;
         $cv->end;
      };
   }

   $cv->end;

   ...

   my $results = $cv->recv;

This code fragment supposedly pings a number of hosts and calls
C<send> after results for all then have have been gathered - in any
order. To achieve this, the code issues a call to C<begin> when it starts
each ping request and calls C<end> when it has received some result for
it. Since C<begin> and C<end> only maintain a counter, the order in which
results arrive is not relevant.

There is an additional bracketing call to C<begin> and C<end> outside the
loop, which serves two important purposes: first, it sets the callback
to be called once the counter reaches C<0>, and second, it ensures that
C<send> is called even when C<no> hosts are being pinged (the loop
doesn't execute once).

This is the general pattern when you "fan out" into multiple (but
potentially zero) subrequests: use an outer C<begin>/C<end> pair to set
the callback and ensure C<end> is called at least once, and then, for each
subrequest you start, call C<begin> and for each subrequest you finish,
call C<end>.

=back

=head3 METHODS FOR CONSUMERS

These methods should only be used by the consuming side, i.e. the
code awaits the condition.

=over 4

=item $cv->recv

Wait (blocking if necessary) until the C<< ->send >> or C<< ->croak
>> methods have been called on C<$cv>, while servicing other watchers
normally.

You can only wait once on a condition - additional calls are valid but
will return immediately.

If an error condition has been set by calling C<< ->croak >>, then this
function will call C<croak>.

In list context, all parameters passed to C<send> will be returned,
in scalar context only the first one will be returned.

Note that doing a blocking wait in a callback is not supported by any
event loop, that is, recursive invocation of a blocking C<< ->recv >> is
not allowed and the C<recv> call will C<croak> if such a condition is
detected. This requirement can be dropped by relying on L<Coro::AnyEvent>
, which allows you to do a blocking C<< ->recv >> from any thread
that doesn't run the event loop itself. L<Coro::AnyEvent> is loaded
automatically when L<Coro> is used with L<AnyEvent>, so code does not need
to do anything special to take advantage of that: any code that would
normally block your program because it calls C<recv>, be executed in an
C<async> thread instead without blocking other threads.

Not all event models support a blocking wait - some die in that case
(programs might want to do that to stay interactive), so I<if you are
using this from a module, never require a blocking wait>. Instead, let the
caller decide whether the call will block or not (for example, by coupling
condition variables with some kind of request results and supporting
callbacks so the caller knows that getting the result will not block,
while still supporting blocking waits if the caller so desires).