File Coverage

blib/lib/AnyEvent/FDpasser.pm

Criterion	Covered	Total	%
statement	153	200	76.5
branch	64	128	50.0
condition	7	25	28.0
subroutine	24	27	88.8
pod	9	14	64.2
total	257	394	65.2

package AnyEvent::FDpasser;

use common::sense;

our $VERSION = '0.3.0';

require XSLoader;
XSLoader::load('AnyEvent::FDpasser', $VERSION);

use Carp;
use Errno;
use POSIX; ## Uses POSIX::pipe/dup so we get accurate $!
use Socket qw/AF_UNIX SOCK_STREAM SOL_SOCKET AF_UNSPEC SO_REUSEADDR/;

use AnyEvent;
use AnyEvent::Util;


sub new {
  my ($class, %arg) = @_;
  my $self = bless {}, $class;

  $self->{on_error} = $arg{on_error};

  $self->{obuf} = [];
  $self->{ibuf} = [];

  if (ref $arg{fh} eq 'ARRAY') {
    die "too many elements in fh array" if scalar @{$arg{fh}} > 2;
    $self->{fh} = $arg{fh}->[0];
    $self->{fh_pair} = $arg{fh}->[1];
  } else {
    if (!defined $arg{fh}) {
      ($self->{fh}, $self->{fh_pair}) = fdpasser_socketpair();
    } else {
      $self->{fh} = $arg{fh};
    }
  }

  unless ($arg{dont_set_nonblocking}) {
    AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
    AnyEvent::Util::fh_nonblocking $self->{fh_pair}, 1
      if exists $self->{fh_pair};
  }

  $self->setup_fh_duped;

  return $self;
}




sub i_am_parent {
  my ($self) = @_;

  die "i_am_parent only applicable when socketpair used" if !defined $self->{fh_pair};
  die "passer object is in error_state: $self->{error_state}" if exists $self->{error_state};

  close($self->{fh_pair});
  delete $self->{fh_pair};
}

sub i_am_child {
  my ($self) = @_;

  die "i_am_child only applicable when socketpair used" if !defined $self->{fh_pair};
  die "passer object is in error_state: $self->{error_state}" if exists $self->{error_state};

  close($self->{fh});
  $self->{fh} = $self->{fh_pair};
  delete $self->{fh_pair};
}




sub push_send_fh {
  my ($self, $fh_to_send, $cb) = @_;

  die "passer object is in error_state: $self->{error_state}" if exists $self->{error_state};
  die "must call i_am_parent or i_am_child" if exists $self->{fh_pair};

  $cb ||= sub {};

  push @{$self->{obuf}}, [$fh_to_send, $cb];

  $self->try_to_send;
}


sub push_recv_fh {
  my ($self, $cb) = @_;

  die "passer object is in error_state: $self->{error_state}" if exists $self->{error_state};
  die "must call i_am_parent or i_am_child" if exists $self->{fh_pair};

  push @{$self->{ibuf}}, $cb;

  $self->try_to_recv;
}




sub try_to_send {
  my ($self) = @_;

  return unless $self->{fh};
  return unless @{$self->{obuf}};
  return if defined $self->{owatcher};
  return if defined $self->{full_descriptor_table_state};

  $self->{owatcher} = AE::io $self->{fh}, 1, sub {

    my $fh_to_send = shift @{$self->{obuf}};

    my $rv = send_fd(fileno($self->{fh}), fileno($fh_to_send->[0]));

    if ($rv < 0) {
      if ($!{EAGAIN} || $!{EWOULDBLOCK} || $!{EINTR}) {
        ## Spurious ready notification or signal: put fh back on queue
        unshift @{$self->{obuf}}, $fh_to_send;
      } else {
        ## Unknown error
        $self->error($!);
      }
    } elsif ($rv == 0) {
      $self->error('sendmsg wrote 0 bytes');
    } else {
      $fh_to_send->[1]->();
      ## Don't do a close($fh_to_send->[0]) because the program may wish to keep it alive
      undef $fh_to_send;
      $self->{owatcher} = undef;
      $self->try_to_send;
    }

  };
}


sub try_to_recv {
  my ($self) = @_;

  return unless @{$self->{ibuf}};
  return if defined $self->{iwatcher};
  return if defined $self->{full_descriptor_table_state};

  $self->{iwatcher} = AE::io $self->{fh}, 0, sub {

    my $cb = shift @{$self->{ibuf}};

    POSIX::close($self->{fh_duped});
    delete $self->{fh_duped};

    ## Race condition: If another thread or a signal handler creates a new descriptor at this
    ## exact point in time, it could cause the descriptor table to fill up and the following
    ## to error.

    my $rv = recv_fd(fileno($self->{fh}));

    if ($rv == -1) {
      if ($!{EAGAIN} || $!{EWOULDBLOCK} || $!{EINTR}) {
        ## Spurious ready notification or signal: put the cb back on the queue
        unshift @{$self->{ibuf}}, $cb;
      } elsif ($!{EMSGSIZE} || $!{EMFILE} || $!{ENFILE}) {
        ## File descriptor table is full. This should be very unlikely given the close+duping
        ## technique used to detect this. In this case the descriptor stream may be
        ## desynchronised and we must shutdown the passer.

        my $err = $!;

        carp "AnyEvent::FDpasser - file descriptor table full, closing passer: $!";

        $self->error($err);
      } else {
        ## Unknown error
        $self->error($!);
      }
    } elsif ($rv == -2) {
      $self->error("cmsg truncated");
    } elsif ($rv == 0) {
      ## Orderly shutdown
      $self->error(undef);
    } else {
      open(my $new_fh, '+<&=', $rv);
      $self->{iwatcher} = undef;
      $cb->($new_fh);
      $self->try_to_recv;
    }
  };

  $self->setup_fh_duped;
}





sub _convert_fh_to_fd {
  my $fh = shift;
  $fh = fileno($fh) unless $fh =~ /^\d+$/;
  return $fh;
}

sub fdpasser_socketpair {
  my ($s1, $s2);

  if ($^O eq 'MSWin32') {
    die "AnyEvent::FDpasser does not support windows";
  } elsif (fdpasser_mode() == 2) {
    pipe $s1, $s2;
    die "can't pipe: $!" unless $s1;
  } else {
    socketpair $s1, $s2, AF_UNIX, SOCK_STREAM, AF_UNSPEC;
    die "can't make socketpair: $!" unless $s2;
  }

  return ($s1, $s2);
}

sub fdpasser_server {
  my ($path, $backlog) = @_;

  $backlog ||= 10;

  my $fh;

  if ($^O eq 'MSWin32') {
    die "AnyEvent::FDpasser does not support windows";
  } elsif (fdpasser_mode() == 2) {
    my $fd = _fdpasser_server($path);
    die "unable to _fdpasser_server($path) : $!" if $fd < 0;
    open($fh, '+<&=', $fd) || die "couldn't open";
  } else {
    socket($fh, AF_UNIX, SOCK_STREAM, AF_UNSPEC) || die "Unable to create AF_UNIX socket: $!";
    setsockopt($fh, SOL_SOCKET, SO_REUSEADDR, pack("l", 1)) || die "Unable to setsockopt(SO_REUSEADDR): $!";
    unlink($path);
    bind($fh, Socket::sockaddr_un($path)) || die "Unable to bind AF_UNIX socket to $path : $!";
    listen($fh, $backlog) || die "Unable to listen on $path : $!";
  }

  return $fh;
}

sub fdpasser_accept {
  my ($listener_fh) = @_;

  my $passer_fh;

  if ($^O eq 'MSWin32') {
    die "AnyEvent::FDpasser does not support windows";
  } elsif (fdpasser_mode() == 2) {
    my $fd = _fdpasser_accept(fileno($listener_fh));
    die "unable to _fdpasser_accept($listener_fh) : $!" if $fd < 0;
    open($passer_fh, '+<&=', $fd) || die "couldn't open";
  } else {
    accept($passer_fh, $listener_fh);
  }

  return $passer_fh;
}

sub fdpasser_connect {
  my ($path) = @_;

  my $fh;

  if ($^O eq 'MSWin32') {
    die "AnyEvent::FDpasser does not support windows";
  } elsif (fdpasser_mode() == 2) {
    my $fd = _fdpasser_connect($path);
    die "unable to _fdpasser_connect($path) : $!" if $fd < 0;
    open($fh, '+<&=', $fd) || die "couldn't open";
  } else {
    socket($fh, AF_UNIX, SOCK_STREAM, AF_UNSPEC) || die "Unable to create AF_UNIX socket: $!";
    connect($fh, Socket::sockaddr_un($path)) || die "Unable to connect AF_UNIX socket to $path : $!";
  }

  return $fh;
}


sub error {
  my ($self, $err) = @_;

  my $on_error = $self->{on_error};
  close($self->{fh});
  close($self->{fh_pair}) if exists $self->{fh_pair};

  if (exists $self->{fh_duped}) {
    POSIX::close($self->{fh_duped});
    delete $self->{fh_duped};
  }
  if (exists $self->{fh_duped_orig}) {
    POSIX::close($self->{fh_duped_orig});
    delete $self->{fh_duped_orig};
  }

  delete $self->{$_} foreach (qw/owatcher iwatcher obuf ibuf fh fh_pair fh_duped fh_duped_orig on_error/);

  $self->{error_state} = $err;

  {
    local $@ = $err;
    $on_error->() if $on_error;
  }
}


sub DESTROY {
  my ($self) = @_;

  if (exists $self->{fh_duped}) {
    POSIX::close($self->{fh_duped});
    delete $self->{fh_duped};
  }
  if (exists $self->{fh_duped_orig}) {
    POSIX::close($self->{fh_duped_orig});
    delete $self->{fh_duped_orig};
  }
}


sub setup_fh_duped {
  my ($self) = @_;

  return if exists $self->{fh_duped};

  if (!exists $self->{fh_duped_orig}) {
    my ($r, $w) = POSIX::pipe();
    die "can't call pipe: $!" unless defined $r;
    POSIX::close($w);
    $self->{fh_duped_orig} = $r;
  }

  $self->{fh_duped} = POSIX::dup($self->{fh_duped_orig});

  if (!defined $self->{fh_duped}) {
    delete $self->{fh_duped};
    if ($!{EMFILE} || $!{ENFILE}) {
      ## Descriptor table full: have to make sure not to call recvmsg now
      $self->enter_full_descriptor_table_state;
    } else {
      die "unable to dup descriptor for reason other than full descriptor table: $!";
    }
  }
}

sub enter_full_descriptor_table_state {
  my ($self) = @_;

  return if $self->{full_descriptor_table_state};

  $self->{full_descriptor_table_state} = 1;

  undef $self->{iwatcher};

  my $watcher; $watcher = AE::timer 0.05, 0.5, sub {
    $self->setup_fh_duped;
    if (exists $self->{fh_duped}) {
      undef $watcher;
      delete $self->{full_descriptor_table_state};
      $self->try_to_recv;
    }
  };
}


1;

__END__