[Summary view] [Print] [Text view]

   1  package DBD::Gofer::Transport::pipeone;
   2  
   3  #   $Id: pipeone.pm 10087 2007-10-16 12:42:37Z timbo $
   4  #
   5  #   Copyright (c) 2007, Tim Bunce, Ireland
   6  #
   7  #   You may distribute under the terms of either the GNU General Public
   8  #   License or the Artistic License, as specified in the Perl README file.
   9  
  10  use strict;
  11  use warnings;
  12  
  13  use Carp;
  14  use Fcntl;
  15  use IO::Select;
  16  use IPC::Open3 qw(open3);
  17  use Symbol qw(gensym);
  18  
  19  use base qw(DBD::Gofer::Transport::Base);
  20  
  21  our $VERSION = sprintf("0.%06d", q$Revision: 10087 $ =~ /(\d+)/o);
  22  
  23  __PACKAGE__->mk_accessors(qw(
  24      connection_info
  25      go_perl
  26  )); 
  27  
  28  
  29  sub new {
  30      my ($self, $args) = @_;
  31      $args->{go_perl} ||= do {
  32          ($INC{"blib.pm"}) ? [ $^X, '-Mblib' ] : [ $^X ];
  33      };
  34      if (not ref $args->{go_perl}) {
  35          # user can override the perl to be used, either with an array ref
  36          # containing the command name and args to use, or with a string
  37          # (ie via the DSN) in which case, to enable args to be passed,
  38          # we split on two or more consecutive spaces (otherwise the path
  39          # to perl couldn't contain a space itself).
  40          $args->{go_perl} = [ split /\s{2,}/, $args->{go_perl} ];
  41      }
  42      return $self->SUPER::new($args);
  43  }
  44  
  45  
  46  # nonblock($fh) puts filehandle into nonblocking mode
  47  sub nonblock { 
  48    my $fh = shift;
  49    my $flags = fcntl($fh, F_GETFL, 0)
  50          or croak "Can't get flags for filehandle $fh: $!";
  51    fcntl($fh, F_SETFL, $flags | O_NONBLOCK)
  52          or croak "Can't make filehandle $fh nonblocking: $!";
  53  }
  54  
  55  
  56  sub start_pipe_command {
  57      my ($self, $cmd) = @_;
  58      $cmd = [ $cmd ] unless ref $cmd eq 'ARRAY';
  59  
  60      # if it's important that the subprocess uses the same
  61      # (versions of) modules as us then the caller should
  62      # set PERL5LIB itself.
  63  
  64      # limit various forms of insanity, for now
  65      local $ENV{DBI_TRACE}; # use DBI_GOFER_TRACE instead
  66      local $ENV{DBI_AUTOPROXY};
  67      local $ENV{DBI_PROFILE};
  68  
  69      my ($wfh, $rfh, $efh) = (gensym, gensym, gensym);
  70      my $pid = open3($wfh, $rfh, $efh, @$cmd)
  71          or die "error starting @$cmd: $!\n";
  72      if ($self->trace) {
  73          $self->trace_msg(sprintf("Started pid $pid: @$cmd {fd: w%d r%d e%d, ppid=$$}\n", fileno $wfh, fileno $rfh, fileno $efh),0);
  74      }
  75      nonblock($rfh);
  76      nonblock($efh);
  77      my $ios = IO::Select->new($rfh, $efh);
  78  
  79      return {
  80          cmd=>$cmd,
  81          pid=>$pid,
  82          wfh=>$wfh, rfh=>$rfh, efh=>$efh,
  83          ios=>$ios,
  84      };
  85  }
  86  
  87  
  88  sub cmd_as_string {
  89      my $self = shift;
  90      # XXX meant to return a properly shell-escaped string suitable for system
  91      # but its only for debugging so that can wait
  92      my $connection_info = $self->connection_info;
  93      return join " ", map { (m/^[-:\w]*$/) ? $_ : "'$_'" } @{$connection_info->{cmd}};
  94  }
  95  
  96  
  97  sub transmit_request_by_transport {
  98      my ($self, $request) = @_;
  99  
 100      my $frozen_request = $self->freeze_request($request);
 101  
 102      my $cmd = [ @{$self->go_perl}, qw(-MDBI::Gofer::Transport::pipeone -e run_one_stdio)];
 103      my $info = $self->start_pipe_command($cmd);
 104  
 105      my $wfh = delete $info->{wfh};
 106      # send frozen request
 107      local $\;
 108      print $wfh $frozen_request
 109          or warn "error writing to @$cmd: $!\n";
 110      # indicate that there's no more
 111      close $wfh
 112          or die "error closing pipe to @$cmd: $!\n";
 113  
 114      $self->connection_info( $info );
 115      return;
 116  }
 117  
 118  
 119  sub read_response_from_fh {
 120      my ($self, $fh_actions) = @_;
 121      my $trace = $self->trace;
 122  
 123      my $info = $self->connection_info || die;
 124      my ($ios) = @{$info}{qw(ios)};
 125      my $errors = 0;
 126      my $complete;
 127  
 128      die "No handles to read response from" unless $ios->count;
 129  
 130      while ($ios->count) {
 131          my @readable = $ios->can_read();
 132          for my $fh (@readable) {
 133              local $_;
 134              my $actions = $fh_actions->{$fh} || die "panic: no action for $fh";
 135              my $rv = sysread($fh, $_='', 1024*31);  # to fit in 32KB slab
 136              unless ($rv) {              # error (undef) or end of file (0)
 137                  my $action;
 138                  unless (defined $rv) {  # was an error
 139                      $self->trace_msg("error on handle $fh: $!\n") if $trace >= 4;
 140                      $action = $actions->{error} || $actions->{eof};
 141                      ++$errors;
 142                      # XXX an error may be a permenent condition of the handle
 143                      # if so we'll loop here - not good
 144                  }
 145                  else {
 146                      $action = $actions->{eof};
 147                      $self->trace_msg("eof on handle $fh\n") if $trace >= 4;
 148                  }
 149                  if ($action->($fh)) {
 150                      $self->trace_msg("removing $fh from handle set\n") if $trace >= 4;
 151                      $ios->remove($fh);
 152                  }
 153                  next;
 154              }
 155              # action returns true if the response is now complete
 156              # (we finish all handles
 157              $actions->{read}->($fh) && ++$complete;
 158          }
 159          last if $complete;
 160      }
 161      return $errors;
 162  }
 163  
 164  
 165  sub receive_response_by_transport {
 166      my $self = shift;
 167  
 168      my $info = $self->connection_info || die;
 169      my ($pid, $rfh, $efh, $ios, $cmd) = @{$info}{qw(pid rfh efh ios cmd)};
 170  
 171      my $frozen_response;
 172      my $stderr_msg;
 173  
 174      $self->read_response_from_fh( {
 175          $efh => {
 176              error => sub { warn "error reading response stderr: $!"; 1 },
 177              eof   => sub { warn "eof on stderr" if 0; 1 },
 178              read  => sub { $stderr_msg .= $_; 0 },
 179          },
 180          $rfh => {
 181              error => sub { warn "error reading response: $!"; 1 },
 182              eof   => sub { warn "eof on stdout" if 0; 1 },
 183              read  => sub { $frozen_response .= $_; 0 },
 184          },
 185      });
 186  
 187      waitpid $info->{pid}, 0
 188          or warn "waitpid: $!"; # XXX do something more useful?
 189  
 190      die ref($self)." command (@$cmd) failed: $stderr_msg"
 191          if not $frozen_response; # no output on stdout at all
 192  
 193      # XXX need to be able to detect and deal with corruption
 194      my $response = $self->thaw_response($frozen_response);
 195  
 196      if ($stderr_msg) {
 197          # add stderr messages as warnings (for PrintWarn)
 198          $response->add_err(0, $stderr_msg, undef, $self->trace)
 199              # but ignore warning from old version of blib
 200              unless $stderr_msg =~ /^Using .*blib/ && "@$cmd" =~ /-Mblib/;
 201      }
 202  
 203      return $response;
 204  }
 205  
 206  
 207  1;
 208  
 209  __END__
 210  
 211  =head1 NAME
 212  
 213  DBD::Gofer::Transport::pipeone - DBD::Gofer client transport for testing
 214  
 215  =head1 SYNOPSIS
 216  
 217    $original_dsn = "...";
 218    DBI->connect("dbi:Gofer:transport=pipeone;dsn=$original_dsn",...)
 219  
 220  or, enable by setting the DBI_AUTOPROXY environment variable:
 221  
 222    export DBI_AUTOPROXY="dbi:Gofer:transport=pipeone"
 223  
 224  =head1 DESCRIPTION
 225  
 226  Connect via DBD::Gofer and execute each request by starting executing a subprocess.
 227  
 228  This is, as you might imagine, spectacularly inefficient!
 229  
 230  It's only intended for testing. Specifically it demonstrates that the server
 231  side is completely stateless.
 232  
 233  It also provides a base class for the much more useful L<DBD::Gofer::Transport::stream>
 234  transport.
 235  
 236  =head1 AUTHOR
 237  
 238  Tim Bunce, L<http://www.tim.bunce.name>
 239  
 240  =head1 LICENCE AND COPYRIGHT
 241  
 242  Copyright (c) 2007, Tim Bunce, Ireland. All rights reserved.
 243  
 244  This module is free software; you can redistribute it and/or
 245  modify it under the same terms as Perl itself. See L<perlartistic>.
 246  
 247  =head1 SEE ALSO
 248  
 249  L<DBD::Gofer::Transport::Base>
 250  
 251  L<DBD::Gofer>
 252  
 253  =cut