File Coverage

blib/lib/Future/IO/Impl/Ppoll.pm

Criterion	Covered	Total	%
statement	132	137	96.3
branch	22	34	64.7
condition	14	24	58.3
subroutine	23	23	100.0
pod	2	5	40.0
total	193	223	86.5

#  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, 2026 -- [e-mail]

package Future::IO::Impl::Ppoll 0.05;

use v5.20;
use warnings;
use base qw( Future::IO::ImplBase );

use feature qw( postderef signatures );
no warnings qw( experimental::postderef experimental::signatures );

use Carp;
our @CARP_NOT = qw( Future::IO::Impl::Ppoll::_Future );

use IO::Ppoll 0.14 qw( POLLIN POLLOUT POLLPRI POLLHUP POLLERR POLLNVAL );
use POSIX qw( SIG_BLOCK sigprocmask );
use Struct::Dumb qw( readonly_struct );
use Time::HiRes qw( time );

BEGIN {
   if( $^V ge v5.36 ) {
      builtin->import(qw( refaddr ));
      warnings->unimport(qw( experimental::builtin )) if $^V lt v5.40;
   }
   else {
      require Scalar::Util;
      Scalar::Util->import(qw( refaddr ));
   }
}

BEGIN {
   Future::IO->VERSION( '0.20' );
   # Just check for sanity
   IO::Ppoll->$_ == Future::IO->$_ or die "This implementation relies on the Future::IO $_ constant being the same as system\n"
      for qw( POLLIN POLLOUT POLLPRI POLLHUP POLLERR POLLNVAL );
}

__PACKAGE__->APPLY;

=head1 NAME

C - implement C using C

=head1 DESCRIPTION

=for highlighter language=perl

This module provides an implementation for L which uses the
C system call, via L.

There are no additional methods to use in this module; it simply has to be
loaded, and it will provide the C implementation methods:

   use Future::IO;
   use Future::IO::Impl::Ppoll;

   my $f = Future::IO->sleep(5);
   ...

Note that the C method will take over the operation of C<$SIG{CHLD}>,
and will print a warning if it discovers some other code has already set it.
Make sure not to otherwise override this signal handler in your program, or
else the C futures will stop working.

=cut

readonly_struct Poller => [qw( events f )];
my %fh_by_refaddr;
my %pollers_by_refaddr;

readonly_struct Alarm => [qw( time f )];
my @alarms;

my $ppoll;
sub ppoll () { $ppoll //= IO::Ppoll->new }

sub _update_poll ( $fh )
{
   my $refaddr = refaddr $fh;
   defined $fh->fileno or
      carp "Filehandle $fh lost its fileno (was closed?) during poll";

   my $mask = 0;
   $mask |= $_->events for ( $pollers_by_refaddr{$refaddr} // [] )->@*;

   ppoll()->mask( $fh => $mask );
   if( $mask ) {
      $fh_by_refaddr{$refaddr} = $fh;
   }
   else {
      delete $fh_by_refaddr{$refaddr};
      delete $pollers_by_refaddr{$refaddr};
   }
}

our $RECURSION;

sub _tick ( $ )
{
   carp "Future::IO::Impl::Ppoll is not reentrant; strange things may happen" if $RECURSION;
   local $RECURSION = 1;

   my $ppoll = ppoll();

   my $timeout = undef;
   if( @alarms ) {
      # These are sorted by time order, so head is soonest
      $timeout = $alarms[0]->time - time();
      $timeout = 0 if $timeout < 0;
   }

   $ppoll->poll( $timeout );

   foreach my $refaddr ( keys %fh_by_refaddr ) {
      my $fh = $fh_by_refaddr{$refaddr};
      # If the filehandle has no fileno, that means it was closed. It was
      # definitely invalid
      my $revents = ( defined $fh->fileno ) ? $ppoll->events( $fh ) : POLLNVAL
         or next;

      my $pollers = $pollers_by_refaddr{$refaddr} or next;
      # TODO: if nobody cared, maybe we should remove it?

      # Find the next poller which cares about at least one of these events
      foreach my $idx ( 0 .. $#$pollers ) {
         my $want_revents = $revents & ( $pollers->[$idx]->events | POLLHUP|POLLERR|POLLNVAL )
            or next;

         my ( $poller ) = splice @$pollers, $idx, 1, ();

         $poller and $poller->f and $poller->f->done( $want_revents );
         last;
      }

      _update_poll( $fh );
   }

   my $now = time();
   while( @alarms and $alarms[0]->time <= $now ) {
      ( shift @alarms )->f->done;
   }
}

sub alarm ( $class, $time )
{
   my $f = Future::IO::Impl::Ppoll::_Future->new;

   # TODO: Binary search
   my $idx = 0;
   $idx++ while $idx < @alarms and $alarms[$idx]->time < $time;

   splice @alarms, $idx, 0, Alarm( $time, $f );

   $f->on_cancel( sub ( $self ) {
      my $idx = 0;
      $idx++ while $idx < @alarms and $alarms[$idx]->f != $self;

      splice @alarms, $idx, 1, ();
   } );

   return $f;
}

sub sleep ( $class, $secs )
{
   $class->alarm( time() + $secs );
}

sub poll ( $, $fh, $events )
{
   defined $fh or
      croak "Expected a defined filehandle for ->poll";

   my $refaddr = refaddr $fh;

   my $pollers = $pollers_by_refaddr{$refaddr} //= [];

   my $f = Future::IO::Impl::Ppoll::_Future->new;
   my $poller = Poller( $events, $f );

   push @$pollers, $poller;

   _update_poll( $fh );
   return $f;
}

my $captured_sigchld;
my %waitpid_futures;

sub waitpid ( $, $pid )
{
   unless( $captured_sigchld ) {
      $SIG{CHLD} and $SIG{CHLD} ne "IGNORE" and
         warn "Future::IO::Impl::Ppoll is replacing \$SIG{CHLD}";

      $SIG{CHLD} = sub ( $ ) {
         foreach my $pid ( keys %waitpid_futures ) {
            next unless waitpid( $pid, POSIX::WNOHANG ) > 0;
            my $wstatus = $?;

            my $fs = delete $waitpid_futures{$pid};
            $_->done( $wstatus ) for @$fs;
         }
      };

      $captured_sigchld = 1;

      sigprocmask( SIG_BLOCK, POSIX::SigSet->new( POSIX::SIGCHLD ) );
      # SIGCHLD is now blocked but will be atomically unblocked by the ppoll()
      # operation
   }

   my $f = Future::IO::Impl::Ppoll::_Future->new;

   if( waitpid( $pid, POSIX::WNOHANG ) > 0 ) {
      my $wstatus = $?;
      $f->done( $wstatus );
      return $f;
   }

   push $waitpid_futures{$pid}->@*, $f;

   return $f;
}

package Future::IO::Impl::Ppoll::_Future {
   use base qw( Future );

   sub await ( $self )
   {
      Future::IO::Impl::Ppoll->_tick until $self->is_ready;
      return $self;
   }
}

=head1 AUTHOR

Paul Evans

=cut

0x55AA;