Acme-Sort-Sleep

 view release on metacpan or  search on metacpan

local/lib/perl5/IO/Async/Loop.pm  view on Meta::CPAN

       }

       return 0;
    },
 ) );

 $loop->run;

=head1 DESCRIPTION

This module provides an abstract class which implements the core loop of the
L<IO::Async> framework. Its primary purpose is to store a set of
L<IO::Async::Notifier> objects or subclasses of them. It handles all of the
lower-level set manipulation actions, and leaves the actual IO readiness 
testing/notification to the concrete class that implements it. It also
provides other functionality such as signal handling, child process managing,
and timers.

See also the two bundled Loop subclasses:

=over 4

=item L<IO::Async::Loop::Select>

=item L<IO::Async::Loop::Poll>

=back

Or other subclasses that may appear on CPAN which are not part of the core
L<IO::Async> distribution.

=head2 Ignoring SIGPIPE

Since version I<0.66> loading this module automatically ignores C<SIGPIPE>, as
it is highly unlikely that the default-terminate action is the best course of
action for an L<IO::Async>-based program to take. If at load time the handler
disposition is still set as C<DEFAULT>, it is set to ignore. If already
another handler has been placed there by the program code, it will be left
undisturbed.

=cut

# Internal constructor used by subclasses
sub __new
{
   my $class = shift;

   # Detect if the API version provided by the subclass is sufficient
   $class->can( "API_VERSION" ) or
      die "$class is too old for IO::Async $VERSION; it does not provide \->API_VERSION\n";

   $class->API_VERSION >= NEED_API_VERSION or
      die "$class is too old for IO::Async $VERSION; we need API version >= ".NEED_API_VERSION.", it provides ".$class->API_VERSION."\n";

   WATCHDOG_ENABLE and !$class->_CAN_WATCHDOG and
      warn "$class cannot implement IO_ASYNC_WATCHDOG\n";

   my $self = bless {
      notifiers     => {}, # {nkey} = notifier
      iowatches     => {}, # {fd} = [ $on_read_ready, $on_write_ready, $on_hangup ]
      sigattaches   => {}, # {sig} => \@callbacks
      childmanager  => undef,
      childwatches  => {}, # {pid} => $code
      threadwatches => {}, # {tid} => $code
      timequeue     => undef,
      deferrals     => [],
      os            => {}, # A generic scratchpad for IO::Async::OS to store whatever it wants
   }, $class;

   # It's possible this is a specific subclass constructor. We still want the
   # magic IO::Async::Loop->new constructor to yield this if it's the first
   # one
   our $ONE_TRUE_LOOP ||= $self;

   # Legacy support - temporary until all CPAN classes are updated; bump NEEDAPI version at that point
   my $old_timer = $self->can( "enqueue_timer" ) != \&enqueue_timer;
   if( $old_timer != ( $self->can( "cancel_timer" ) != \&cancel_timer ) ) {
      die "$class should overload both ->enqueue_timer and ->cancel_timer, or neither";
   }

   if( $old_timer ) {
      warnings::warnif( deprecated => "Enabling old_timer workaround for old loop class " . $class );
   }

   $self->{old_timer} = $old_timer;

   return $self;
}

=head1 MAGIC CONSTRUCTOR

=head2 new

   $loop = IO::Async::Loop->new

This function attempts to find a good subclass to use, then calls its
constructor. It works by making a list of likely candidate classes, then
trying each one in turn, C<require>ing the module then calling its C<new>
method. If either of these operations fails, the next subclass is tried. If
no class was successful, then an exception is thrown.

The constructed object is cached, and will be returned again by a subsequent
call. The cache will also be set by a constructor on a specific subclass. This
behaviour makes it possible to simply use the normal constructor in a module
that wishes to interract with the main program's Loop, such as an integration
module for another event system.

For example, the following two C<$loop> variables will refer to the same
object:

 use IO::Async::Loop;
 use IO::Async::Loop::Poll;

 my $loop_poll = IO::Async::Loop::Poll->new;

 my $loop = IO::Async::Loop->new;

While it is not advised to do so under normal circumstances, if the program
really wishes to construct more than one Loop object, it can call the
constructor C<really_new>, or invoke one of the subclass-specific constructors
directly.

local/lib/perl5/IO/Async/Loop.pm  view on Meta::CPAN

}

=head2 timeout_future

   $loop->timeout_future( %args )->get

Returns a new L<IO::Async::Future> instance which will fail at a given point
in time. The C<%args> should contain an C<at> or C<after> key as per the
C<watch_time> method. The returned future may be cancelled to cancel the
timer. At the alloted time, the future will fail with the string C<"Timeout">.

=cut

sub timeout_future
{
   my $self = shift;
   my %args = @_;

   my $future = $self->new_future;
   my $id = $self->watch_time( %args,
      code => sub { $future->fail( "Timeout" ) },
   );

   $future->on_cancel( sub { shift->loop->unwatch_time( $id ) } );

   return $future;
}

############
# Features #
############

=head1 FEATURES

Most of the following methods are higher-level wrappers around base
functionality provided by the low-level API documented below. They may be
used by L<IO::Async::Notifier> subclasses or called directly by the program.

The following methods documented with a trailing call to C<< ->get >> return
L<Future> instances.

=cut

sub __new_feature
{
   my $self = shift;
   my ( $classname ) = @_;

   ( my $filename = "$classname.pm" ) =~ s{::}{/}g;
   require $filename;

   # These features aren't supposed to be "user visible", so if methods called
   # on it carp or croak, the shortmess line ought to skip IO::Async::Loop and
   # go on report its caller. To make this work, add the feature class to our
   # @CARP_NOT list.
   push our(@CARP_NOT), $classname;

   return $classname->new( loop => $self );
}

=head2 attach_signal

   $id = $loop->attach_signal( $signal, $code )

This method adds a new signal handler to watch the given signal. The same
signal can be attached to multiple times; its callback functions will all be
invoked, in no particular order.

The returned C<$id> value can be used to identify the signal handler in case
it needs to be removed by the C<detach_signal> method. Note that this value
may be an object reference, so if it is stored, it should be released after it
cancelled, so the object itself can be freed.

=over 8

=item $signal

The name of the signal to attach to. This should be a bare name like C<TERM>.

=item $code

A CODE reference to the handling callback.

=back

Attaching to C<SIGCHLD> is not recommended because of the way all child
processes use it to report their termination. Instead, the C<watch_child>
method should be used to watch for termination of a given child process. A
warning will be printed if C<SIGCHLD> is passed here, but in future versions
of L<IO::Async> this behaviour may be disallowed altogether.

See also L<POSIX> for the C<SIGI<name>> constants.

For a more flexible way to use signals from within Notifiers, see instead the
L<IO::Async::Signal> object.

=cut

sub attach_signal
{
   my $self = shift;
   my ( $signal, $code ) = @_;

   HAVE_SIGNALS or croak "This OS cannot ->attach_signal";

   if( $signal eq "CHLD" ) {
      # We make special exception to allow $self->watch_child to do this
      caller eq "IO::Async::Loop" or
         carp "Attaching to SIGCHLD is not advised - use ->watch_child instead";
   }

   if( not $self->{sigattaches}->{$signal} ) {
      my @attaches;
      $self->watch_signal( $signal, sub {
         foreach my $attachment ( @attaches ) {
            $attachment->();
         }
      } );
      $self->{sigattaches}->{$signal} = \@attaches;
   }

   push @{ $self->{sigattaches}->{$signal} }, $code;

   return \$self->{sigattaches}->{$signal}->[-1];
}

=head2 detach_signal

   $loop->detach_signal( $signal, $id )

Removes a previously-attached signal handler.

=over 8

=item $signal

The name of the signal to remove from. This should be a bare name like
C<TERM>.

=item $id

The value returned by the C<attach_signal> method.

=back

=cut

sub detach_signal
{
   my $self = shift;
   my ( $signal, $id ) = @_;

   HAVE_SIGNALS or croak "This OS cannot ->detach_signal";

   # Can't use grep because we have to preserve the addresses
   my $attaches = $self->{sigattaches}->{$signal} or return;

   for (my $i = 0; $i < @$attaches; ) {
      $i++, next unless \$attaches->[$i] == $id;

      splice @$attaches, $i, 1, ();
   }

   if( !@$attaches ) {
      $self->unwatch_signal( $signal );
      delete $self->{sigattaches}->{$signal};
   }
}

=head2 later

   $loop->later( $code )

Schedules a code reference to be invoked as soon as the current round of IO
operations is complete.

The code reference is never invoked immediately, though the loop will not
perform any blocking operations between when it is installed and when it is
invoked. It may call C<select>, C<poll> or equivalent with a zero-second
timeout, and process any currently-pending IO conditions before the code is
invoked, but it will not block for a non-zero amount of time.

This method is implemented using the C<watch_idle> method, with the C<when>
parameter set to C<later>. It will return an ID value that can be passed to
C<unwatch_idle> if required.

=cut

sub later
{
   my $self = shift;
   my ( $code ) = @_;

   return $self->watch_idle( when => 'later', code => $code );
}

=head2 spawn_child

   $loop->spawn_child( %params )

This method creates a new child process to run a given code block or command.
For more detail, see the C<spawn_child> method on the
L<IO::Async::ChildManager> class.

=cut

sub spawn_child
{
   my $self = shift;
   my %params = @_;

   my $childmanager = $self->{childmanager} ||=
      $self->__new_feature( "IO::Async::ChildManager" );

   $childmanager->spawn_child( %params );
}

=head2 open_child

   $pid = $loop->open_child( %params )

This creates a new child process to run the given code block or command, and
attaches filehandles to it that the parent will watch. This method is a light
wrapper around constructing a new L<IO::Async::Process> object, provided
largely for backward compatibility. New code ought to construct such an object
directly, as it may provide more features than are available here.

The C<%params> hash takes the following keys:

=over 8

=item command => ARRAY or STRING

=item code => CODE

The command or code to run in the child process (as per the C<spawn> method)

=item on_finish => CODE

A continuation to be called when the child process exits and has closed all of
the filehandles that were set up for it. It will be invoked in the following
way:

 $on_finish->( $pid, $exitcode )

The second argument is passed the plain perl C<$?> value.

=item on_error => CODE

Optional continuation to be called when the child code block throws an
exception, or the command could not be C<exec(2)>ed. It will be invoked in the
following way (as per C<spawn>)

 $on_error->( $pid, $exitcode, $dollarbang, $dollarat )

If this continuation is not supplied, then C<on_finish> is used instead. The
value of C<$!> and C<$@> will not be reported.

=item setup => ARRAY

Optional reference to an array to pass to the underlying C<spawn> method.

=back

In addition, the hash takes keys that define how to set up file descriptors in
the child process. (If the C<setup> array is also given, these operations will
be performed after those specified by C<setup>.)

=over 8

=item fdI<n> => HASH

A hash describing how to set up file descriptor I<n>. The hash may contain one
of the following sets of keys:

=over 4

=item on_read => CODE

The child will be given the writing end of a pipe. The reading end will be
wrapped by an L<IO::Async::Stream> using this C<on_read> callback function.

=item from => STRING

local/lib/perl5/IO/Async/Loop.pm  view on Meta::CPAN

=back

Either or both callbacks may be removed at once. It is not an error to attempt
to remove a callback that is not present. If both callbacks were provided to
the C<watch_io> method and only one is removed by this method, the other shall
remain.

=cut

sub __unwatch_io
{
   my $self = shift;
   my %params = @_;

   my $handle = delete $params{handle} or croak "Expected 'handle'";

   my $watch = $self->{iowatches}->{$handle->fileno} or return;

   if( delete $params{on_read_ready} ) {
      undef $watch->[1];
   }

   if( delete $params{on_write_ready} ) {
      undef $watch->[2];
   }

   if( delete $params{on_hangup} ) {
      $self->_CAN_ON_HANGUP or croak "Cannot watch_io for 'on_hangup' in ".ref($self);
      undef $watch->[3];
   }

   if( not $watch->[1] and not $watch->[2] and not $watch->[3] ) {
      delete $self->{iowatches}->{$handle->fileno};
   }

   keys %params and croak "Unrecognised keys for ->unwatch_io - " . join( ", ", keys %params );
}

=head2 watch_signal

   $loop->watch_signal( $signal, $code )

This method adds a new signal handler to watch the given signal.

=over 8

=item $signal

The name of the signal to watch to. This should be a bare name like C<TERM>.

=item $code

A CODE reference to the handling callback.

=back

There can only be one callback per signal name. Registering a new one will
remove an existing one.

Applications should use a L<IO::Async::Signal> object, or call
C<attach_signal> instead of using this method.

This and C<unwatch_signal> are optional; a subclass may implement neither, or
both. If it implements neither then signal handling will be performed by the
base class using a self-connected pipe to interrupt the main IO blocking.

=cut

sub watch_signal
{
   my $self = shift;
   my ( $signal, $code ) = @_;

   HAVE_SIGNALS or croak "This OS cannot ->watch_signal";

   IO::Async::OS->loop_watch_signal( $self, $signal, $code );
}

=head2 unwatch_signal

   $loop->unwatch_signal( $signal )

This method removes the signal callback for the given signal.

=over 8

=item $signal

The name of the signal to watch to. This should be a bare name like C<TERM>.

=back

=cut

sub unwatch_signal
{
   my $self = shift;
   my ( $signal ) = @_;

   HAVE_SIGNALS or croak "This OS cannot ->unwatch_signal";

   IO::Async::OS->loop_unwatch_signal( $self, $signal );
}

=head2 watch_time

   $id = $loop->watch_time( %args )

This method installs a callback which will be called at the specified time.
The time may either be specified as an absolute value (the C<at> key), or
as a delay from the time it is installed (the C<after> key).

The returned C<$id> value can be used to identify the timer in case it needs
to be cancelled by the C<unwatch_time> method. Note that this value may be
an object reference, so if it is stored, it should be released after it has
been fired or cancelled, so the object itself can be freed.

The C<%params> hash takes the following keys:

=over 8

local/lib/perl5/IO/Async/Loop.pm  view on Meta::CPAN

      if( defined $childwatches->{$zid} ) {
         $childwatches->{$zid}->( $zid, $status );
         delete $childwatches->{$zid};
      }

      if( defined $childwatches->{0} ) {
         $childwatches->{0}->( $zid, $status );
         # Don't delete it
      }
   }
}

=head2 watch_child

   $loop->watch_child( $pid, $code )

This method adds a new handler for the termination of the given child process
PID, or all child processes.

=over 8

=item $pid

The PID to watch. Will report on all child processes if this is 0.

=item $code

A CODE reference to the exit handler. It will be invoked as

 $code->( $pid, $? )

The second argument is passed the plain perl C<$?> value.

=back

After invocation, the handler for a PID-specific watch is automatically
removed. The all-child watch will remain until it is removed by
C<unwatch_child>.

This and C<unwatch_child> are optional; a subclass may implement neither, or
both. If it implements neither then child watching will be performed by using
C<watch_signal> to install a C<SIGCHLD> handler, which will use C<waitpid> to
look for exited child processes.

If both a PID-specific and an all-process watch are installed, there is no
ordering guarantee as to which will be called first.

=cut

sub watch_child
{
   my $self = shift;
   my ( $pid, $code ) = @_;

   my $childwatches = $self->{childwatches};

   croak "Already have a handler for $pid" if exists $childwatches->{$pid};

   if( HAVE_SIGNALS and !$self->{childwatch_sigid} ) {
      $self->{childwatch_sigid} = $self->attach_signal(
         CHLD => sub { _reap_children( $childwatches ) }
      );

      # There's a chance the child has already exited
      my $zid = waitpid( $pid, WNOHANG );
      if( defined $zid and $zid > 0 ) {
         my $exitstatus = $?;
         $self->later( sub { $code->( $pid, $exitstatus ) } );
         return;
      }
   }

   $childwatches->{$pid} = $code;
}

=head2 unwatch_child

   $loop->unwatch_child( $pid )

This method removes a watch on an existing child process PID.

=cut

sub unwatch_child
{
   my $self = shift;
   my ( $pid ) = @_;

   my $childwatches = $self->{childwatches};

   delete $childwatches->{$pid};

   if( HAVE_SIGNALS and !keys %$childwatches ) {
      $self->detach_signal( CHLD => delete $self->{childwatch_sigid} );
   }
}

=head1 METHODS FOR SUBCLASSES

The following methods are provided to access internal features which are
required by specific subclasses to implement the loop functionality. The use
cases of each will be documented in the above section.

=cut

=head2 _adjust_timeout

   $loop->_adjust_timeout( \$timeout )

Shortens the timeout value passed in the scalar reference if it is longer in
seconds than the time until the next queued event on the timer queue. If there
are pending idle handlers, the timeout is reduced to zero.

=cut

sub _adjust_timeout
{
   my $self = shift;
   my ( $timeref, %params ) = @_;