[ViewVC] Diff of: cvs/AnyEvent/lib/AnyEvent/Handle.pm

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents):
Revision 1.163 by root, Mon Jul 27 22:08:52 2009 UTC vs.
Revision 1.182 by root, Thu Sep 3 12:35:01 2009 UTC

-package AnyEvent::Handle;
-use Scalar::Util ();
-use Carp ();
-use Errno qw(EAGAIN EINTR);
-use AnyEvent (); BEGIN { AnyEvent::common_sense }
-use AnyEvent::Util qw(WSAEWOULDBLOCK);
 =head1 NAME
 AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent
-=cut
-our $VERSION = 4.87;
 =head1 SYNOPSIS
    use AnyEvent;
    use AnyEvent::Handle;
 C<on_error> callback.
 All callbacks will be invoked with the handle object as their first
 argument.
+=cut
+package AnyEvent::Handle;
+use Scalar::Util ();
+use List::Util ();
+use Carp ();
+use Errno qw(EAGAIN EINTR);
+use AnyEvent (); BEGIN { AnyEvent::common_sense }
+use AnyEvent::Util qw(WSAEWOULDBLOCK);
+our $VERSION = $AnyEvent::VERSION;
 =head1 METHODS
 =over 4
 =item $handle = B<new> AnyEvent::TLS fh => $filehandle, key => value...
 memory and push it into the queue, but instead only read more data from
 the file when the write queue becomes empty.
 =item timeout => $fractional_seconds
+=item rtimeout => $fractional_seconds
+=item wtimeout => $fractional_seconds
-If non-zero, then this enables an "inactivity" timeout: whenever this many
+If non-zero, then these enables an "inactivity" timeout: whenever this
-seconds pass without a successful read or write on the underlying file
+many seconds pass without a successful read or write on the underlying
-handle, the C<on_timeout> callback will be invoked (and if that one is
+file handle (or a call to C<timeout_reset>), the C<on_timeout> callback
-missing, a non-fatal C<ETIMEDOUT> error will be raised).
+will be invoked (and if that one is missing, a non-fatal C<ETIMEDOUT>
+error will be raised).
+There are three variants of the timeouts that work fully independent
+of each other, for both read and write, just read, and just write:
+C<timeout>, C<rtimeout> and C<wtimeout>, with corresponding callbacks
+C<on_timeout>, C<on_rtimeout> and C<on_wtimeout>, and reset functions
+C<timeout_reset>, C<rtimeout_reset>, and C<wtimeout_reset>.
 Note that timeout processing is also active when you currently do not have
 any outstanding read or write requests: If you plan to keep the connection
 idle then you should disable the timout temporarily or ignore the timeout
 in the C<on_timeout> callback, in which case AnyEvent::Handle will simply
 accomplishd by setting this option to a true value.
 The default is your opertaing system's default behaviour (most likely
 enabled), this option explicitly enables or disables it, if possible.
+=item keepalive => <boolean>
+Enables (default disable) the SO_KEEPALIVE option on the stream socket:
+normally, TCP connections have no time-out once established, so TCP
+conenctions, once established, can stay alive forever even when the other
+side has long gone. TCP keepalives are a cheap way to take down long-lived
+TCP connections whent he other side becomes unreachable. While the default
+is OS-dependent, TCP keepalives usually kick in after around two hours,
+and, if the other side doesn't reply, take down the TCP connection some 10
+to 15 minutes later.
+It is harmless to specify this option for file handles that do not support
+keepalives, and enabling it on connections that are potentially long-lived
+is usually a good idea.
+=item oobinline => <boolean>
+BSD majorly fucked up the implementation of TCP urgent data. The result
+is that almost no OS implements TCP according to the specs, and every OS
+implements it slightly differently.
+If you want to handle TCP urgent data, then setting this flag gives you
+the most portable way of getting urgent data, by putting it into the
+stream.
 =item read_size => <bytes>
 The default read block size (the amount of bytes this module will
 try to read during each loop iteration, which affects memory
 requirements). Default: C<8192>.
                      delete $self->{_skip_drain_rbuf};
                      $self->_start;
                      $self->{on_connect}
                         and $self->{on_connect}($self, $host, $port, sub {
-                               delete @$self{qw(fh _tw _ww _rw _eof _queue rbuf _wbuf tls _tls_rbuf _tls_wbuf)};
+                               delete @$self{qw(fh _tw _rtw _wtw _ww _rw _eof _queue rbuf _wbuf tls _tls_rbuf _tls_wbuf)};
                                $self->{_skip_drain_rbuf} = 1;
                                &$retry;
                             });
                   } else {
 sub _start {
    my ($self) = @_;
    AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
+   $self->{_activity}  =
+   $self->{_ractivity} =
-   $self->{_activity} = AnyEvent->now;
+   $self->{_wactivity} = AE::now;
-   $self->_timeout;
+   $self->timeout   (delete $self->{timeout}  ) if $self->{timeout};
+   $self->rtimeout  (delete $self->{rtimeout} ) if $self->{rtimeout};
+   $self->wtimeout  (delete $self->{wtimeout} ) if $self->{wtimeout};
-   $self->no_delay (delete $self->{no_delay}) if exists $self->{no_delay};
+   $self->no_delay  (delete $self->{no_delay} ) if exists $self->{no_delay};
+   $self->keepalive (delete $self->{keepalive}) if exists $self->{keepalive};
+   $self->oobinline (delete $self->{oobinline}) if exists $self->{oobinline};
-   $self->starttls (delete $self->{tls}, delete $self->{tls_ctx})
+   $self->starttls  (delete $self->{tls}, delete $self->{tls_ctx})
       if $self->{tls};
-   $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
+   $self->on_drain  (delete $self->{on_drain}) if $self->{on_drain};
    $self->start_read
       if $self->{on_read} || @{ $self->{_queue} };
    $self->_drain_wbuf;
 }
-#sub _shutdown {
-#   my ($self) = @_;
-#
-#   delete @$self{qw(_tw _rw _ww fh wbuf on_read _queue)};
-#   $self->{_eof} = 1; # tell starttls et. al to stop trying
-#
-#   &_freetls;
-#}
 sub _error {
    my ($self, $errno, $fatal, $message) = @_;
    $! = $errno;
    $_[0]{on_eof} = $_[1];
 }
 =item $handle->on_timeout ($cb)
-Replace the current C<on_timeout> callback, or disables the callback (but
+=item $handle->on_rtimeout ($cb)
-not the timeout) if C<$cb> = C<undef>. See the C<timeout> constructor
-argument and method.
-=cut
+=item $handle->on_wtimeout ($cb)
-sub on_timeout {
+Replace the current C<on_timeout>, C<on_rtimeout> or C<on_wtimeout>
-   $_[0]{on_timeout} = $_[1];
+callback, or disables the callback (but not the timeout) if C<$cb> =
-}
+C<undef>. See the C<timeout> constructor argument and method.
+=cut
+# see below
 =item $handle->autocork ($boolean)
 Enables or disables the current autocork behaviour (see C<autocork>
 constructor argument). Changes will only take effect on the next write.
 sub no_delay {
    $_[0]{no_delay} = $_[1];
    eval {
       local $SIG{__DIE__};
-      setsockopt $_[0]{fh}, &Socket::IPPROTO_TCP, &Socket::TCP_NODELAY, int $_[1]
+      setsockopt $_[0]{fh}, Socket::IPPROTO_TCP (), Socket::TCP_NODELAY (), int $_[1]
          if $_[0]{fh};
    };
 }
+=item $handle->keepalive ($boolean)
+Enables or disables the C<keepalive> setting (see constructor argument of
+the same name for details).
+=cut
+sub keepalive {
+   $_[0]{keepalive} = $_[1];
+   eval {
+      local $SIG{__DIE__};
+      setsockopt $_[0]{fh}, Socket::SOL_SOCKET (), Socket::SO_KEEPALIVE (), int $_[1]
+         if $_[0]{fh};
+   };
+}
+=item $handle->oobinline ($boolean)
+Enables or disables the C<oobinline> setting (see constructor argument of
+the same name for details).
+=cut
+sub oobinline {
+   $_[0]{oobinline} = $_[1];
+   eval {
+      local $SIG{__DIE__};
+      setsockopt $_[0]{fh}, Socket::SOL_SOCKET (), Socket::SO_OOBINLINE (), int $_[1]
+         if $_[0]{fh};
+   };
+}
+=item $handle->keepalive ($boolean)
+Enables or disables the C<keepalive> setting (see constructor argument of
+the same name for details).
+=cut
+sub keepalive {
+   $_[0]{keepalive} = $_[1];
+   eval {
+      local $SIG{__DIE__};
+      setsockopt $_[0]{fh}, Socket::SOL_SOCKET (), Socket::SO_KEEPALIVE (), int $_[1]
+         if $_[0]{fh};
+   };
+}
 =item $handle->on_starttls ($cb)
 Replace the current C<on_starttls> callback (see the C<on_starttls> constructor argument).
 =cut
 sub on_starttls {
    $_[0]{on_stoptls} = $_[1];
 }
+=item $handle->rbuf_max ($max_octets)
+Configures the C<rbuf_max> setting (C<undef> disables it).
+=cut
+sub rbuf_max {
+   $_[0]{rbuf_max} = $_[1];
+}
 #############################################################################
 =item $handle->timeout ($seconds)
+=item $handle->rtimeout ($seconds)
+=item $handle->wtimeout ($seconds)
 Configures (or disables) the inactivity timeout.
-=cut
+=item $handle->timeout_reset
-sub timeout {
+=item $handle->rtimeout_reset
+=item $handle->wtimeout_reset
+Reset the activity timeout, as if data was received or sent.
+These methods are cheap to call.
+=cut
+for my $dir ("", "r", "w") {
+   my $timeout    = "${dir}timeout";
+   my $tw         = "_${dir}tw";
+   my $on_timeout = "on_${dir}timeout";
+   my $activity   = "_${dir}activity";
+   my $cb;
+   *$on_timeout = sub {
+      $_[0]{$on_timeout} = $_[1];
+   };
+   *$timeout = sub {
-   my ($self, $timeout) = @_;
+      my ($self, $new_value) = @_;
-   $self->{timeout} = $timeout;
+      $self->{$timeout} = $new_value;
-   $self->_timeout;
+      delete $self->{$tw}; &$cb;
-}
+   };
+   *{"${dir}timeout_reset"} = sub {
+      $_[0]{$activity} = AE::now;
+   };
+   # main workhorse:
-# reset the timeout watcher, as neccessary
+   # reset the timeout watcher, as neccessary
-# also check for time-outs
+   # also check for time-outs
-sub _timeout {
+   $cb = sub {
-   my ($self) = @_;
+      my ($self) = @_;
-   if ($self->{timeout} && $self->{fh}) {
+      if ($self->{$timeout} && $self->{fh}) {
-      my $NOW = AnyEvent->now;
+         my $NOW = AE::now;
-      # when would the timeout trigger?
+         # when would the timeout trigger?
-      my $after = $self->{_activity} + $self->{timeout} - $NOW;
+         my $after = $self->{$activity} + $self->{$timeout} - $NOW;
-      # now or in the past already?
+         # now or in the past already?
-      if ($after <= 0) {
+         if ($after <= 0) {
-         $self->{_activity} = $NOW;
+            $self->{$activity} = $NOW;
-         if ($self->{on_timeout}) {
+            if ($self->{$on_timeout}) {
-            $self->{on_timeout}($self);
+               $self->{$on_timeout}($self);
-         } else {
+            } else {
-            $self->_error (Errno::ETIMEDOUT);
+               $self->_error (Errno::ETIMEDOUT);
+            }
+            # callback could have changed timeout value, optimise
+            return unless $self->{$timeout};
+            # calculate new after
+            $after = $self->{$timeout};
          }
-         # callback could have changed timeout value, optimise
+         Scalar::Util::weaken $self;
-         return unless $self->{timeout};
+         return unless $self; # ->error could have destroyed $self
-         # calculate new after
+         $self->{$tw} ||= AE::timer $after, 0, sub {
-         $after = $self->{timeout};
+            delete $self->{$tw};
+            $cb->($self);
+         };
+      } else {
+         delete $self->{$tw};
       }
-      Scalar::Util::weaken $self;
-      return unless $self; # ->error could have destroyed $self
-      $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
-         delete $self->{_tw};
-         $self->_timeout;
-      });
-   } else {
-      delete $self->{_tw};
    }
 }
 #############################################################################
          my $len = syswrite $self->{fh}, $self->{wbuf};
          if (defined $len) {
             substr $self->{wbuf}, 0, $len, "";
-            $self->{_activity} = AnyEvent->now;
+            $self->{_activity} = $self->{_wactivity} = AE::now;
             $self->{on_drain}($self)
                if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf})
                   && $self->{on_drain};
       # try to write data immediately
       $cb->() unless $self->{autocork};
       # if still data left in wbuf, we need to poll
-      $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)
+      $self->{_ww} = AE::io $self->{fh}, 1, $cb
          if length $self->{wbuf};
    };
 }
 our %WH;
 Other languages could read single lines terminated by a newline and pass
 this line into their JSON decoder of choice.
 =cut
+sub json_coder() {
+   eval { require JSON::XS; JSON::XS->new->utf8 }
+      || do { require JSON; JSON->new->utf8 }
+}
 register_write_type json => sub {
    my ($self, $ref) = @_;
-   require JSON;
+   my $json = $self->{json} ||= json_coder;
-   $self->{json} ? $self->{json}->encode ($ref)
+   $json->encode ($ref)
-                 : JSON::encode_json ($ref)
 };
 =item storable => $reference
 Freezes the given reference using L<Storable> and writes it to the
 sub _drain_rbuf {
    my ($self) = @_;
    # avoid recursion
-   return if exists $self->{_skip_drain_rbuf};
+   return if $self->{_skip_drain_rbuf};
    local $self->{_skip_drain_rbuf} = 1;
-   if (
-      defined $self->{rbuf_max}
-      && $self->{rbuf_max} < length $self->{rbuf}
-   ) {
-      $self->_error (Errno::ENOSPC, 1), return;
-   }
    while () {
       # we need to use a separate tls read buffer, as we must not receive data while
       # we are draining the buffer, and this can only happen with TLS.
       $self->{rbuf} .= delete $self->{_tls_rbuf}
          : $self->_error (0, 1, "Unexpected end-of-file");
       return;
    }
+   if (
+      defined $self->{rbuf_max}
+      && $self->{rbuf_max} < length $self->{rbuf}
+   ) {
+      $self->_error (Errno::ENOSPC, 1), return;
+   }
    # may need to restart read watcher
    unless ($self->{_rw}) {
       $self->start_read
          if $self->{on_read} || @{ $self->{_queue} };
    }
       my $type = shift;
       $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read")
             ->($self, $cb, @_);
    }
    unshift @{ $self->{_queue} }, $cb;
    $self->_drain_rbuf;
 }
 =cut
 register_read_type json => sub {
    my ($self, $cb) = @_;
-   my $json = $self->{json} ||=
+   my $json = $self->{json} ||= json_coder;
-      eval { require JSON::XS; JSON::XS->new->utf8 }
-         || do { require JSON; JSON->new->utf8 };
    my $data;
    my $rbuf = \$self->{rbuf};
    sub {
    my ($self) = @_;
    unless ($self->{_rw} || $self->{_eof}) {
       Scalar::Util::weaken $self;
-      $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
+      $self->{_rw} = AE::io $self->{fh}, 0, sub {
          my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf});
          my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
          if ($len > 0) {
-            $self->{_activity} = AnyEvent->now;
+            $self->{_activity} = $self->{_ractivity} = AE::now;
             if ($self->{tls}) {
                Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);
                &_dotls ($self);
             $self->_drain_rbuf;
          } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
             return $self->_error ($!, 1);
          }
-      });
+      };
    }
 }
 our $ERROR_SYSCALL;
 our $ERROR_WANT_READ;
    require Net::SSLeay;
    $ERROR_SYSCALL   = Net::SSLeay::ERROR_SYSCALL     ();
    $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ   ();
-   $tls = $self->{tls};
+   $tls = delete $self->{tls};
    $ctx = $self->{tls_ctx};
    local $Carp::CarpLevel = 1; # skip ourselves when creating a new context or session
    if ("HASH" eq ref $ctx) {
    Net::SSLeay::CTX_set_mode ($tls, 1|2);
    $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
    $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
+   Net::SSLeay::BIO_write ($self->{_rbio}, delete $self->{rbuf});
    Net::SSLeay::set_bio ($tls, $self->{_rbio}, $self->{_wbio});
    $self->{_on_starttls} = sub { $_[0]{on_starttls}(@_) }
       if $self->{on_starttls};
    my ($self) = @_;
    return unless $self->{tls};
    $self->{tls_ctx}->_put_session (delete $self->{tls})
-      if ref $self->{tls};
+      if $self->{tls} > 0;
    delete @$self{qw(_rbio _wbio _tls_wbuf _on_starttls)};
 }
 sub DESTROY {
       my $fh   = delete $self->{fh};
       my $wbuf = delete $self->{wbuf};
       my @linger;
-      push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub {
+      push @linger, AE::io $fh, 1, sub {
          my $len = syswrite $fh, $wbuf, length $wbuf;
          if ($len > 0) {
             substr $wbuf, 0, $len, "";
          } else {
             @linger = (); # end
          }
-      });
+      };
-      push @linger, AnyEvent->timer (after => $linger, cb => sub {
+      push @linger, AE::timer $linger, 0, sub {
          @linger = ();
-      });
+      };
    }
 }
 =item $handle->destroy
 Shuts down the handle object as much as possible - this call ensures that
 no further callbacks will be invoked and as many resources as possible
-will be freed. You must not call any methods on the object afterwards.
+will be freed. Any method you will call on the handle object after
+destroying it in this way will be silently ignored (and it will return the
+empty list).
 Normally, you can just "forget" any references to an AnyEvent::Handle
 object and it will simply shut down. This works in fatal error and EOF
 callbacks, as well as code outside. It does I<NOT> work in a read or write
 callback, so when you want to destroy the AnyEvent::Handle object from
 sub destroy {
    my ($self) = @_;
    $self->DESTROY;
    %$self = ();
+   bless $self, "AnyEvent::Handle::destroyed";
+}
+sub AnyEvent::Handle::destroyed::AUTOLOAD {
+   #nop
 }
 =item AnyEvent::Handle::TLS_CTX
 This function creates and returns the AnyEvent::TLS object used by default

Diff Legend

-–
+Removed lines
-+
+Added lines
-<
+Changed lines
->
+Changed lines

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents): Revision 1.163 by root, Mon Jul 27 22:08:52 2009 UTC vs. Revision 1.182 by root, Thu Sep 3 12:35:01 2009 UTC

Diff Legend

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents):
Revision 1.163 by root, Mon Jul 27 22:08:52 2009 UTC vs.
Revision 1.182 by root, Thu Sep 3 12:35:01 2009 UTC