view release on metacpan or  search on metacpan

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

    },
 ) );

 $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.

The list of candidates is formed from the following choices, in this order:

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

context inside an C<eval> block. The return value will be used as the
C<exit(2)> code from the child if it returns (or 255 if it returned C<undef> or
thows an exception).

=item on_exit => CODE

A optional continuation to be called when the child processes exits. It will
be invoked in the following way:

 $on_exit->( $pid, $exitcode )

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

This key is optional; if not supplied, the calling code should install a
handler using the C<watch_child> method.

=item keep_signals => BOOL

Optional boolean. If missing or false, any CODE references in the C<%SIG> hash
will be removed and restored back to C<DEFAULT> in the child process. If true,
no adjustment of the C<%SIG> hash will be performed.

=back

=cut

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

   HAVE_POSIX_FORK or croak "POSIX fork() is not available";

   my $code = $params{code} or croak "Expected 'code' as a CODE reference";

   my $kid = fork;
   defined $kid or croak "Cannot fork() - $!";

   if( $kid == 0 ) {
      unless( $params{keep_signals} ) {
         foreach( keys %SIG ) {
            next if m/^__(WARN|DIE)__$/;
            $SIG{$_} = "DEFAULT" if ref $SIG{$_} eq "CODE";
         }
      }

      my $exitvalue = eval { $code->() };

      defined $exitvalue or $exitvalue = -1;

      POSIX::_exit( $exitvalue );
   }

   if( defined $params{on_exit} ) {
      $self->watch_child( $kid => $params{on_exit} );
   }

   return $kid;
}

=head2 create_thread

   $tid = $loop->create_thread( %params )

This method creates a new (non-detached) thread to run the given code block,
returning its thread ID.

=over 8

=item code => CODE

A block of code to execute in the thread. It is called in the context given by
the C<context> argument, and its return value will be available to the
C<on_joined> callback. It is called inside an C<eval> block; if it fails the
exception will be caught.

=item context => "scalar" | "list" | "void"

Optional. Gives the calling context that C<code> is invoked in. Defaults to
C<scalar> if not supplied.

=item on_joined => CODE

Callback to invoke when the thread function returns or throws an exception.
If it returned, this callback will be invoked with its result

 $on_joined->( return => @result )

If it threw an exception the callback is invoked with the value of C<$@>

 $on_joined->( died => $! )

=back

=cut

# It is basically impossible to have any semblance of order on global
# destruction, and even harder again to rely on when threads are going to be
# terminated and joined. Instead of ensuring we join them all, just detach any
# we no longer care about at END time
my %threads_to_detach; # {$tid} = $thread_weakly
END {
   $_ and $_->detach for values %threads_to_detach;
}

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

   HAVE_THREADS or croak "Threads are not available";

   eval { require threads } or croak "This Perl does not support threads";

   my $code = $params{code} or croak "Expected 'code' as a CODE reference";
   my $on_joined = $params{on_joined} or croak "Expected 'on_joined' as a CODE reference";

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

   unless( $self->{thread_join_pipe} ) {
      ( my $rd, $self->{thread_join_pipe} ) = IO::Async::OS->pipepair or
         croak "Cannot pipepair - $!";
      $self->{thread_join_pipe}->autoflush(1);

      $self->watch_io(
         handle => $rd,
         on_read_ready => sub {
            sysread $rd, my $buffer, 8192 or return;

            # There's a race condition here in that we might have read from
            # the pipe after the returning thread has written to it but before
            # it has returned. We'll grab the actual $thread object and
            # forcibly ->join it here to ensure we wait for its result.

            foreach my $tid ( unpack "N*", $buffer ) {
               my ( $thread, $on_joined ) = @{ delete $threadwatches->{$tid} }
                  or die "ARGH: Can't find threadwatch for tid $tid\n";
               $on_joined->( $thread->join );
               delete $threads_to_detach{$tid};
            }
         }
      );
   }

   my $wr = $self->{thread_join_pipe};

   my $context = $params{context} || "scalar";

   my ( $thread ) = threads->create(
      sub {
         my ( @ret, $died );
         eval {
            $context eq "list"   ? ( @ret    = $code->() ) :
            $context eq "scalar" ? ( $ret[0] = $code->() ) :
                                               $code->();
            1;
         } or $died = $@;

         $wr->syswrite( pack "N", threads->tid );

         return died => $died if $died;
         return return => @ret;
      }
   );

   $threadwatches->{$thread->tid} = [ $thread, $on_joined ];
   weaken( $threads_to_detach{$thread->tid} = $thread );

   return $thread->tid;
}

=head1 LOW-LEVEL METHODS

As C<IO::Async::Loop> is an abstract base class, specific subclasses of it are
required to implement certain methods that form the base level of
functionality. They are not recommended for applications to use; see instead
the various event objects or higher level methods listed above.

These methods should be considered as part of the interface contract required
to implement a C<IO::Async::Loop> subclass.

=cut

=head2 API_VERSION

   IO::Async::Loop->API_VERSION

This method will be called by the magic constructor on the class before it is
constructed, to ensure that the specific implementation will support the
required API. This method should return the API version that the loop
implementation supports. The magic constructor will use that class, provided
it declares a version at least as new as the version documented here.

The current API version is C<0.49>.

This method may be implemented using C<constant>; e.g

 use constant API_VERSION => '0.49';

=cut

=head2 watch_io

   $loop->watch_io( %params )

This method installs callback functions which will be invoked when the given
IO handle becomes read- or write-ready.

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

=over 8

=item handle => IO

The IO handle to watch.

=item on_read_ready => CODE

Optional. A CODE reference to call when the handle becomes read-ready.

=item on_write_ready => CODE

Optional. A CODE reference to call when the handle becomes write-ready.

=back

There can only be one filehandle of any given fileno registered at any one
time. For any one filehandle, there can only be one read-readiness and/or one
write-readiness callback at any one time. Registering a new one will remove an