[Summary view] [Print] [Text view]

   1  package DBD::Gofer::Transport::Base;
   2  
   3  #   $Id: Base.pm 11427 2008-06-16 15:24:46Z 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 base qw(DBI::Gofer::Transport::Base);
  14  
  15  our $VERSION = sprintf("0.%06d", q$Revision: 11427 $ =~ /(\d+)/o);
  16  
  17  __PACKAGE__->mk_accessors(qw(
  18      trace
  19      go_dsn
  20      go_url
  21      go_policy
  22      go_timeout
  23      go_retry_hook
  24      go_retry_limit
  25      go_cache
  26      cache_hit
  27      cache_miss
  28      cache_store
  29  ));
  30  __PACKAGE__->mk_accessors_using(make_accessor_autoviv_hashref => qw(
  31      meta
  32  ));
  33  
  34  
  35  sub new {
  36      my ($class, $args) = @_;
  37      $args->{$_} = 0 for (qw(cache_hit cache_miss cache_store));
  38      $args->{keep_meta_frozen} ||= 1 if $args->{go_cache};
  39      #warn "args @{[ %$args ]}\n";
  40      return $class->SUPER::new($args);
  41  }   
  42  
  43  
  44  sub _init_trace { $ENV{DBD_GOFER_TRACE} || 0 }
  45  
  46  
  47  sub new_response {
  48      my $self = shift;
  49      return DBI::Gofer::Response->new(@_);
  50  }
  51  
  52  
  53  sub transmit_request {
  54      my ($self, $request) = @_;
  55      my $trace = $self->trace;
  56      my $response;
  57  
  58      my ($go_cache, $request_cache_key);
  59      if ($go_cache = $self->{go_cache}) {
  60          $request_cache_key
  61              = $request->{meta}{request_cache_key}
  62              = $self->get_cache_key_for_request($request);
  63          if ($request_cache_key) {
  64              my $frozen_response = eval { $go_cache->get($request_cache_key) };
  65              if ($frozen_response) {
  66                  $self->_dump("cached response found for ".ref($request), $request)
  67                      if $trace;
  68                  $response = $self->thaw_response($frozen_response);
  69                  $self->trace_msg("transmit_request is returning a response from cache $go_cache\n")
  70                      if $trace;
  71                  ++$self->{cache_hit};
  72                  return $response;
  73              }
  74              warn $@ if $@;
  75              ++$self->{cache_miss};
  76              $self->trace_msg("transmit_request cache miss\n")
  77                  if $trace;
  78          }
  79      }
  80  
  81      my $to = $self->go_timeout;
  82      my $transmit_sub = sub {
  83          $self->trace_msg("transmit_request\n") if $trace;
  84          local $SIG{ALRM} = sub { die "TIMEOUT\n" } if $to;
  85  
  86          my $response = eval {
  87              local $SIG{PIPE} = sub {
  88                  my $extra = ($! eq "Broken pipe") ? "" : " ($!)";
  89                  die "Unable to send request: Broken pipe$extra\n";
  90              };
  91              alarm($to) if $to;
  92              $self->transmit_request_by_transport($request);
  93          };
  94          alarm(0) if $to;
  95  
  96          if ($@) {
  97              return $self->transport_timedout("transmit_request", $to)
  98                  if $@ eq "TIMEOUT\n";
  99              return $self->new_response({ err => 1, errstr => $@ });
 100          }
 101  
 102          return $response;
 103      };
 104  
 105      $response = $self->_transmit_request_with_retries($request, $transmit_sub);
 106  
 107      if ($response) {
 108          my $frozen_response = delete $response->{meta}{frozen};
 109          $self->_store_response_in_cache($frozen_response, $request_cache_key)
 110              if $request_cache_key;
 111      }
 112  
 113      $self->trace_msg("transmit_request is returning a response itself\n")
 114          if $trace && $response;
 115  
 116      return $response unless wantarray;
 117      return ($response, $transmit_sub);
 118  }
 119  
 120  
 121  sub _transmit_request_with_retries {
 122      my ($self, $request, $transmit_sub) = @_;
 123      my $response;
 124      do {
 125          $response = $transmit_sub->();
 126      } while ( $response && $self->response_needs_retransmit($request, $response) );
 127      return $response;
 128  }
 129  
 130  
 131  sub receive_response {
 132      my ($self, $request, $retransmit_sub) = @_;
 133      my $to = $self->go_timeout;
 134  
 135      my $receive_sub = sub {
 136          $self->trace_msg("receive_response\n");
 137          local $SIG{ALRM} = sub { die "TIMEOUT\n" } if $to;
 138  
 139          my $response = eval {
 140              alarm($to) if $to;
 141              $self->receive_response_by_transport($request);
 142          };
 143          alarm(0) if $to;
 144  
 145          if ($@) {
 146              return $self->transport_timedout("receive_response", $to)
 147                  if $@ eq "TIMEOUT\n";
 148              return $self->new_response({ err => 1, errstr => $@ });
 149          }
 150          return $response;
 151      };
 152  
 153      my $response;
 154      do {
 155          $response = $receive_sub->();
 156          if ($self->response_needs_retransmit($request, $response)) {
 157              $response = $self->_transmit_request_with_retries($request, $retransmit_sub);
 158              $response ||= $receive_sub->();
 159          }
 160      } while ( $self->response_needs_retransmit($request, $response) );
 161  
 162      if ($response) {
 163          my $frozen_response = delete $response->{meta}{frozen};
 164          my $request_cache_key = $request->{meta}{request_cache_key};
 165          $self->_store_response_in_cache($frozen_response, $request_cache_key)
 166              if $request_cache_key && $self->{go_cache};
 167      }
 168  
 169      return $response;
 170  }
 171  
 172  
 173  sub response_retry_preference {
 174      my ($self, $request, $response) = @_;
 175  
 176      # give the user a chance to express a preference (or undef for default)
 177      if (my $go_retry_hook = $self->go_retry_hook) {
 178          my $retry = $go_retry_hook->($request, $response, $self);
 179          $self->trace_msg(sprintf "go_retry_hook returned %s\n",
 180              (defined $retry) ? $retry : 'undef');
 181          return $retry if defined $retry;
 182      }
 183  
 184      # This is the main decision point.  We don't retry requests that got
 185      # as far as executing because the error is probably from the database
 186      # (not transport) so retrying is unlikely to help. But note that any
 187      # severe transport error occuring after execute is likely to return
 188      # a new response object that doesn't have the execute flag set. Beware!
 189      return 0 if $response->executed_flag_set;
 190  
 191      return 1 if ($response->errstr || '') =~ m/induced by DBI_GOFER_RANDOM/;
 192  
 193      return 1 if $request->is_idempotent; # i.e. is SELECT or ReadOnly was set
 194  
 195      return undef; # we couldn't make up our mind
 196  }
 197  
 198  
 199  sub response_needs_retransmit {
 200      my ($self, $request, $response) = @_;
 201  
 202      my $err = $response->err
 203          or return 0; # nothing went wrong
 204  
 205      my $retry = $self->response_retry_preference($request, $response);
 206  
 207      if (!$retry) {  # false or undef
 208          $self->trace_msg("response_needs_retransmit: response not suitable for retry\n");
 209          return 0;
 210      }
 211  
 212      # we'd like to retry but have we retried too much already?
 213  
 214      my $retry_limit = $self->go_retry_limit;
 215      if (!$retry_limit) {
 216          $self->trace_msg("response_needs_retransmit: retries disabled (retry_limit not set)\n");
 217          return 0;
 218      }
 219  
 220      my $request_meta = $request->meta;
 221      my $retry_count = $request_meta->{retry_count} || 0;
 222      if ($retry_count >= $retry_limit) {
 223          $self->trace_msg("response_needs_retransmit: $retry_count is too many retries\n");
 224          # XXX should be possible to disable altering the err
 225          $response->errstr(sprintf "%s (after %d retries by gofer)", $response->errstr, $retry_count);
 226          return 0;
 227      }
 228  
 229      # will retry now, do the admin
 230      ++$retry_count;
 231      $self->trace_msg("response_needs_retransmit: retry $retry_count\n");
 232  
 233      # hook so response_retry_preference can defer some code execution
 234      # until we've checked retry_count and retry_limit.
 235      if (ref $retry eq 'CODE') {
 236          $retry->($retry_count, $retry_limit)
 237              and warn "should return false"; # protect future use
 238      }
 239  
 240      ++$request_meta->{retry_count};         # update count for this request object
 241      ++$self->meta->{request_retry_count};   # update cumulative transport stats
 242  
 243      return 1;
 244  }
 245  
 246  
 247  sub transport_timedout {
 248      my ($self, $method, $timeout) = @_;
 249      $timeout ||= $self->go_timeout;
 250      return $self->new_response({ err => 1, errstr => "DBD::Gofer $method timed-out after $timeout seconds" });
 251  }
 252  
 253  
 254  # return undef if we don't want to cache this request
 255  # subclasses may use more specialized rules
 256  sub get_cache_key_for_request {
 257      my ($self, $request) = @_;
 258  
 259      # we only want to cache idempotent requests
 260      # is_idempotent() is true if GOf_REQUEST_IDEMPOTENT or GOf_REQUEST_READONLY set
 261      return undef if not $request->is_idempotent;
 262  
 263      # XXX would be nice to avoid the extra freeze here
 264      my $key = $self->freeze_request($request, undef, 1);
 265  
 266      #use Digest::MD5; warn "get_cache_key_for_request: ".Digest::MD5::md5_base64($key)."\n";
 267  
 268      return $key;
 269  }
 270  
 271  
 272  sub _store_response_in_cache {
 273      my ($self, $frozen_response, $request_cache_key) = @_;
 274      my $go_cache = $self->{go_cache}
 275          or return;
 276  
 277      # new() ensures that enabling go_cache also enables keep_meta_frozen
 278      warn "No meta frozen in response" if !$frozen_response;
 279      warn "No request_cache_key" if !$request_cache_key;
 280  
 281      if ($frozen_response && $request_cache_key) {
 282          $self->trace_msg("receive_response added response to cache $go_cache\n");
 283          eval { $go_cache->set($request_cache_key, $frozen_response) };
 284          warn $@ if $@;
 285          ++$self->{cache_store};
 286      }
 287  }
 288  
 289  1;
 290  
 291  =head1 NAME
 292  
 293  DBD::Gofer::Transport::Base - base class for DBD::Gofer client transports
 294  
 295  =head1 SYNOPSIS
 296  
 297    my $remote_dsn = "..."
 298    DBI->connect("dbi:Gofer:transport=...;url=...;timeout=...;retry_limit=...;dsn=$remote_dsn",...)
 299  
 300  or, enable by setting the DBI_AUTOPROXY environment variable:
 301  
 302    export DBI_AUTOPROXY='dbi:Gofer:transport=...;url=...'
 303  
 304  which will force I<all> DBI connections to be made via that Gofer server.
 305  
 306  =head1 DESCRIPTION
 307  
 308  This is the base class for all DBD::Gofer client transports.
 309  
 310  =head1 ATTRIBUTES
 311  
 312  Gofer transport attributes can be specified either in the attributes parameter
 313  of the connect() method call, or in the DSN string. When used in the DSN
 314  string, attribute names don't have the C<go_> prefix.
 315  
 316  =head2 go_dsn
 317  
 318  The full DBI DSN that the Gofer server should connect to on your behalf.
 319  
 320  When used in the DSN it must be the last element in the DSN string.
 321  
 322  =head2 go_timeout
 323  
 324  A time limit for sending a request and receiving a response. Some drivers may
 325  implement sending and receiving as separate steps, in which case (currently)
 326  the timeout applies to each separately.
 327  
 328  If a request needs to be resent then the timeout is restarted for each sending
 329  of a request and receiving of a response.
 330  
 331  =head2 go_retry_limit
 332  
 333  The maximum number of times an request may be retried. The default is 2.
 334  
 335  =head2 go_retry_hook
 336  
 337  This subroutine reference is called, if defined, for each response received where $response->err is true.
 338  
 339  The subroutine is pass three parameters: the request object, the response object, and the transport object.
 340  
 341  If it returns an undefined value then the default retry behaviour is used. See L</RETRY ON ERROR> below.
 342  
 343  If it returns a defined but false value then the request is not resent.
 344  
 345  If it returns true value then the request is resent, so long as the number of retries does not exceed C<go_retry_limit>.
 346  
 347  =head1 RETRY ON ERROR
 348  
 349  The default retry on error behaviour is:
 350  
 351   - Retry if the error was due to DBI_GOFER_RANDOM. See L<DBI::Gofer::Execute>.
 352  
 353   - Retry if $request->is_idempotent returns true. See L<DBI::Gofer::Request>.
 354  
 355  A retry won't be allowed if the number of previous retries has reached C<go_retry_limit>.
 356  
 357  =head1 TRACING
 358  
 359  Tracing of gofer requests and reponses can be enabled by setting the
 360  C<DBD_GOFER_TRACE> environment variable. A value of 1 gives a reasonably
 361  compact summary of each request and response. A value of 2 or more gives a
 362  detailed, and voluminous, dump.
 363  
 364  The trace is written using DBI->trace_msg() and so is written to the default
 365  DBI trace output, which is usually STDERR.
 366  
 367  =head1 AUTHOR
 368  
 369  Tim Bunce, L<http://www.tim.bunce.name>
 370  
 371  =head1 LICENCE AND COPYRIGHT
 372  
 373  Copyright (c) 2007, Tim Bunce, Ireland. All rights reserved.
 374  
 375  This module is free software; you can redistribute it and/or
 376  modify it under the same terms as Perl itself. See L<perlartistic>.
 377  
 378  =head1 SEE ALSO
 379  
 380  L<DBD::Gofer>, L<DBI::Gofer::Request>, L<DBI::Gofer::Response>, L<DBI::Gofer::Execute>.
 381  
 382  and some example transports:
 383  
 384  L<DBD::Gofer::Transport::stream>
 385  
 386  L<DBD::Gofer::Transport::http>
 387  
 388  L<DBI::Gofer::Transport::mod_perl>
 389  
 390  =cut