File Coverage

blib/lib/IO/Pipely.pm

Criterion	Covered	Total	%
statement	73	179	40.7
branch	25	118	21.1
condition	2	19	10.5
subroutine	12	13	92.3
pod	2	2	100.0
total	114	331	34.4

package IO::Pipely;
$IO::Pipely::VERSION = '0.006';
use warnings;
use strict;

use Symbol qw(gensym);
use IO::Socket qw(
  AF_UNIX
  PF_INET
  PF_UNSPEC
  SOCK_STREAM
  SOL_SOCKET
  SOMAXCONN
  SO_ERROR
  SO_REUSEADDR
  inet_aton
  pack_sockaddr_in
  unpack_sockaddr_in
);
use Fcntl qw(F_GETFL F_SETFL O_NONBLOCK);
use Errno qw(EINPROGRESS EWOULDBLOCK);

our @EXPORT_OK = qw(pipely socketpairly);
use base qw(Exporter);

# The order of pipe primitives depends on the platform.

# It's not always safe to assume that a function can be used if it's
# present.

my (@oneway_pipe_types, @twoway_pipe_types);
if ($^O eq "MSWin32" or $^O eq "MacOS") {
  @oneway_pipe_types = qw(inet socketpair pipe);
  @twoway_pipe_types = qw(inet socketpair pipe);
}
elsif ($^O eq "cygwin") {
  @oneway_pipe_types = qw(pipe inet socketpair);
  @twoway_pipe_types = qw(inet pipe socketpair);
}
else {
  @oneway_pipe_types = qw(pipe socketpair inet);
  @twoway_pipe_types = qw(socketpair inet pipe);
}

# Provide dummy constants so things at least compile.  These constants
# aren't used if we're RUNNING_IN_HELL, but Perl needs to see them.

BEGIN {
  # older perls than 5.10 needs a kick in the arse to AUTOLOAD the constant...
  eval "F_GETFL" if $] < 5.010;

  if ( ! defined &Fcntl::F_GETFL ) {
    if ( ! defined prototype "F_GETFL" ) {
      *F_GETFL = sub { 0 };
      *F_SETFL = sub { 0 };
    } else {
      *F_GETFL = sub () { 0 };
      *F_SETFL = sub () { 0 };
    }
  }
}

# Make a socket.  This is a homebrew socketpair() for systems that
# don't support it.  The things I must do to make Windows happy.

sub _make_socket {

  ### Server side.

  my $acceptor = gensym();
  my $accepted = gensym();

  my $tcp = getprotobyname('tcp') or die "getprotobyname: $!";
  socket( $acceptor, PF_INET, SOCK_STREAM, $tcp ) or die "socket: $!";

  setsockopt( $acceptor, SOL_SOCKET, SO_REUSEADDR, 1) or die "reuse: $!";

  my $server_addr = inet_aton('127.0.0.1') or die "inet_aton: $!";
  $server_addr = pack_sockaddr_in(0, $server_addr)
    or die "sockaddr_in: $!";

  bind( $acceptor, $server_addr ) or die "bind: $!";

  $acceptor->blocking(0);

  $server_addr = getsockname($acceptor);

  listen( $acceptor, SOMAXCONN ) or die "listen: $!";

  ### Client side.

  my $connector = gensym();

  socket( $connector, PF_INET, SOCK_STREAM, $tcp ) or die "socket: $!";

  $connector->blocking(0);

  unless (connect( $connector, $server_addr )) {
    die "connect: $!" if $! and ($! != EINPROGRESS) and ($! != EWOULDBLOCK);
  }

  my $connector_address = getsockname($connector);
  my ($connector_port, $connector_addr) =
    unpack_sockaddr_in($connector_address);

  ### Loop around 'til it's all done.  I thought I was done writing
  ### select loops.  Damnit.

  my $in_read  = '';
  my $in_write = '';

  vec( $in_read,  fileno($acceptor),  1 ) = 1;
  vec( $in_write, fileno($connector), 1 ) = 1;

  my $done = 0;
  while ($done != 0x11) {
    my $hits = select( my $out_read   = $in_read,
                       my $out_write  = $in_write,
                       undef,
                       5
                     );
    unless ($hits) {
      next if ($! and ($! == EINPROGRESS) or ($! == EWOULDBLOCK));
      die "select: $!" unless $hits;
    }

    # Accept happened.
    if (vec($out_read, fileno($acceptor), 1)) {
      my $peer = accept($accepted, $acceptor);
      my ($peer_port, $peer_addr) = unpack_sockaddr_in($peer);

      if ( $peer_port == $connector_port and
           $peer_addr eq $connector_addr
         ) {
        vec($in_read, fileno($acceptor), 1) = 0;
        $done |= 0x10;
      }
    }

    # Connect happened.
    if (vec($out_write, fileno($connector), 1)) {
      $! = unpack('i', getsockopt($connector, SOL_SOCKET, SO_ERROR));
      die "connect: $!" if $!;

      vec($in_write, fileno($connector), 1) = 0;
      $done |= 0x01;
    }
  }

  # Turn blocking back on, damnit.
  $accepted->blocking(1);
  $connector->blocking(1);

  return ($accepted, $connector);
}

sub pipely {
  my %arg = @_;

  my $conduit_type = delete($arg{type});
  my $debug        = delete($arg{debug}) || 0;

  # Generate symbols to be used as filehandles for the pipe's ends.
  #
  # Filehandle autovivification isn't used for portability with older
  # versions of Perl.

  my ($a_read, $b_write)  = (gensym(), gensym());

  # Try the specified conduit type only.  No fallback.

  if (defined $conduit_type) {
    return ($a_read, $b_write) if _try_oneway_type(
      $conduit_type, $debug, \$a_read, \$b_write
    );
  }

  # Otherwise try all available conduit types until one works.
  # Conduit types that fail are discarded for speed.

  while (my $try_type = $oneway_pipe_types[0]) {
    return ($a_read, $b_write) if _try_oneway_type(
      $try_type, $debug, \$a_read, \$b_write
    );
    shift @oneway_pipe_types;
  }

  # There's no conduit type left.  Bummer!

  $debug and warn "nothing worked";
  return;
}

sub socketpairly {
  my %arg = @_;

  my $conduit_type = delete($arg{type});
  my $debug        = delete($arg{debug}) || 0;

  # Generate symbols to be used as filehandles for the pipe's ends.
  #
  # Filehandle autovivification isn't used for portability with older
  # versions of Perl.

  my ($a_read, $a_write) = (gensym(), gensym());
  my ($b_read, $b_write) = (gensym(), gensym());

  if (defined $conduit_type) {
    return ($a_read, $a_write, $b_read, $b_write) if _try_twoway_type(
      $conduit_type, $debug,
      \$a_read, \$a_write,
      \$b_read, \$b_write
    );
  }

  while (my $try_type = $twoway_pipe_types[0]) {
    return ($a_read, $a_write, $b_read, $b_write) if _try_twoway_type(
      $try_type, $debug,
      \$a_read, \$a_write,
      \$b_read, \$b_write
    );
    shift @oneway_pipe_types;
  }

  # There's no conduit type left.  Bummer!

  $debug and warn "nothing worked";
  return;
}

# Try a pipe by type.

sub _try_oneway_type {
  my ($type, $debug, $a_read, $b_write) = @_;

  # Try a pipe().
  if ($type eq "pipe") {
    eval {
      pipe($$a_read, $$b_write) or die "pipe failed: $!";
    };

    # Pipe failed.
    if (length $@) {
      warn "pipe failed: $@" if $debug;
      return;
    }

    $debug and do {
      warn "using a pipe";
      warn "ar($$a_read) bw($$b_write)\n";
    };

    # Turn off buffering.  POE::Kernel does this for us, but
    # someone might want to use the pipe class elsewhere.
    select((select($$b_write), $| = 1)[0]);
    return 1;
  }

  # Try a UNIX-domain socketpair.
  if ($type eq "socketpair") {
    eval {
      socketpair($$a_read, $$b_write, AF_UNIX, SOCK_STREAM, PF_UNSPEC)
        or die "socketpair failed: $!";
    };

    if (length $@) {
      warn "socketpair failed: $@" if $debug;
      return;
    }

    $debug and do {
      warn "using a UNIX domain socketpair";
      warn "ar($$a_read) bw($$b_write)\n";
    };

    # It's one-way, so shut down the unused directions.
    shutdown($$a_read,  1);
    shutdown($$b_write, 0);

    # Turn off buffering.  POE::Kernel does this for us, but someone
    # might want to use the pipe class elsewhere.
    select((select($$b_write), $| = 1)[0]);
    return 1;
  }

  # Try a pair of plain INET sockets.
  if ($type eq "inet") {
    eval {
      ($$a_read, $$b_write) = _make_socket();
    };

    if (length $@) {
      warn "make_socket failed: $@" if $debug;
      return;
    }

    $debug and do {
      warn "using a plain INET socket";
      warn "ar($$a_read) bw($$b_write)\n";
    };

    # It's one-way, so shut down the unused directions.
    shutdown($$a_read,  1);
    shutdown($$b_write, 0);

    # Turn off buffering.  POE::Kernel does this for us, but someone
    # might want to use the pipe class elsewhere.
    select((select($$b_write), $| = 1)[0]);
    return 1;
  }

  # There's nothing left to try.
  $debug and warn "unknown pipely() socket type ``$type''";
  return;
}

# Try a pipe by type.

sub _try_twoway_type {
  my ($type, $debug, $a_read, $a_write, $b_read, $b_write) = @_;

  # Try a socketpair().
  if ($type eq "socketpair") {
    eval {
      socketpair($$a_read, $$b_read, AF_UNIX, SOCK_STREAM, PF_UNSPEC)
        or die "socketpair 1 failed: $!";
    };

    # Socketpair failed.
    if (length $@) {
      warn "socketpair failed: $@" if $debug;
      return;
    }

    $debug and do {
      warn "using UNIX domain socketpairs";
      warn "ar($$a_read) aw($$a_write) br($$b_read) bw($$b_write)\n";
    };

    # It's two-way, so each reader is also a writer.
    $$a_write = $$a_read;
    $$b_write = $$b_read;

    # Turn off buffering.  POE::Kernel does this for us, but someone
    # might want to use the pipe class elsewhere.
    select((select($$a_write), $| = 1)[0]);
    select((select($$b_write), $| = 1)[0]);
    return 1;
  }

  # Try a couple pipe() calls.
  if ($type eq "pipe") {
    eval {
      pipe($$a_read, $$b_write) or die "pipe 1 failed: $!";
      pipe($$b_read, $$a_write) or die "pipe 2 failed: $!";
    };

    # Pipe failed.
    if (length $@) {
      warn "pipe failed: $@" if $debug;
      return;
    }

    $debug and do {
      warn "using a pipe";
      warn "ar($$a_read) aw($$a_write) br($$b_read) bw($$b_write)\n";
    };

    # Turn off buffering.  POE::Kernel does this for us, but someone
    # might want to use the pipe class elsewhere.
    select((select($$a_write), $| = 1)[0]);
    select((select($$b_write), $| = 1)[0]);
    return 1;
  }

  # Try a pair of plain INET sockets.
  if ($type eq "inet") {
    eval {
      ($$a_read, $$b_read) = _make_socket();
    };

    # Sockets failed.
    if (length $@) {
      warn "make_socket failed: $@" if $debug;
      return;
    }

    $debug and do {
      warn "using a plain INET socket";
      warn "ar($$a_read) aw($$a_write) br($$b_read) bw($$b_write)\n";
    };

    $$a_write = $$a_read;
    $$b_write = $$b_read;

    # Turn off buffering.  POE::Kernel does this for us, but someone
    # might want to use the pipe class elsewhere.
    select((select($$a_write), $| = 1)[0]);
    select((select($$b_write), $| = 1)[0]);
    return 1;
  }

  $debug and warn "unknown pipely(2) socket type ``$type''";
  return;
}

1;

__END__