view release on metacpan or  search on metacpan

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

#  You may distribute under the terms of either the GNU General Public License
#  or the Artistic License (the same terms as Perl itself)
#
#  (C) Paul Evans, 2007-2015 -- leonerd@leonerd.org.uk

package IO::Async::Loop;

use strict;
use warnings;

our $VERSION = '0.70';

# When editing this value don't forget to update the docs below
use constant NEED_API_VERSION => '0.33';

# Base value but some classes might override
use constant _CAN_ON_HANGUP => 0;

# Most Loop implementations do not accurately handle sub-second timers.
# This only matters for unit tests
use constant _CAN_SUBSECOND_ACCURATELY => 0;

# Does the loop implementation support IO_ASYNC_WATCHDOG?
use constant _CAN_WATCHDOG => 0;

# Watchdog configuration constants
use constant WATCHDOG_ENABLE   => $ENV{IO_ASYNC_WATCHDOG};
use constant WATCHDOG_INTERVAL => $ENV{IO_ASYNC_WATCHDOG_INTERVAL} || 10;
use constant WATCHDOG_SIGABRT  => $ENV{IO_ASYNC_WATCHDOG_SIGABRT};

use Carp;

use IO::Socket (); # empty import
use Time::HiRes qw(); # empty import
use POSIX qw( WNOHANG );
use Scalar::Util qw( refaddr weaken );
use Socket qw( SO_REUSEADDR AF_INET6 IPPROTO_IPV6 IPV6_V6ONLY );

use IO::Async::OS;

use constant HAVE_SIGNALS => IO::Async::OS->HAVE_SIGNALS;
use constant HAVE_POSIX_FORK => IO::Async::OS->HAVE_POSIX_FORK;
use constant HAVE_THREADS => IO::Async::OS->HAVE_THREADS;

# Never sleep for more than 1 second if a signal proxy is registered, to avoid
# a borderline race condition.
# There is a race condition in perl involving signals interacting with XS code
# that implements blocking syscalls. There is a slight chance a signal will
# arrive in the XS function, before the blocking itself. Perl will not run our
# (safe) deferred signal handler in this case. To mitigate this, if we have a
# signal proxy, we'll adjust the maximal timeout. The signal handler will be 
# run when the XS function returns. 
our $MAX_SIGWAIT_TIME = 1;

# Also, never sleep for more than 1 second if the OS does not support signals
# and we have child watches registered (so we must use waitpid() polling)
our $MAX_CHILDWAIT_TIME = 1;

# Maybe our calling program will have a suggested hint of a specific Loop
# class or list of classes to use
our $LOOP;

# Undocumented; used only by the test scripts.
# Setting this value true will avoid the IO::Async::Loop::$^O candidate in the
# magic constructor
our $LOOP_NO_OS;

# SIGALRM handler for watchdog
$SIG{ALRM} = sub {
   # There are two extra frames here; this one and the signal handler itself
   local $Carp::CarpLevel = $Carp::CarpLevel + 2;
   if( WATCHDOG_SIGABRT ) {
      print STDERR Carp::longmess( "Watchdog timeout" );
      kill ABRT => $$;
   }
   else {
      Carp::confess( "Watchdog timeout" );
   }
} if WATCHDOG_ENABLE;

$SIG{PIPE} = "IGNORE" if ( $SIG{PIPE} || "" ) eq "DEFAULT";

=head1 NAME

C<IO::Async::Loop> - core loop of the C<IO::Async> framework

=head1 SYNOPSIS

 use IO::Async::Stream;
 use IO::Async::Timer::Countdown;

 use IO::Async::Loop;

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

 $loop->add( IO::Async::Timer::Countdown->new(

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

{
   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