ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent/lib/AnyEvent/Handle.pm
(Generate patch)

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents):
Revision 1.174 by root, Sat Aug 8 20:52:06 2009 UTC vs.
Revision 1.182 by root, Thu Sep 3 12:35:01 2009 UTC

1package AnyEvent::Handle;
2
3use Scalar::Util ();
4use Carp ();
5use Errno qw(EAGAIN EINTR);
6
7use AnyEvent (); BEGIN { AnyEvent::common_sense }
8use AnyEvent::Util qw(WSAEWOULDBLOCK);
9
10=head1 NAME 1=head1 NAME
11 2
12AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent 3AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent
13
14=cut
15
16our $VERSION = 4.91;
17 4
18=head1 SYNOPSIS 5=head1 SYNOPSIS
19 6
20 use AnyEvent; 7 use AnyEvent;
21 use AnyEvent::Handle; 8 use AnyEvent::Handle;
… …
59C<on_error> callback. 46C<on_error> callback.
60 47
61All callbacks will be invoked with the handle object as their first 48All callbacks will be invoked with the handle object as their first
62argument. 49argument.
63 50
51=cut
52
53package AnyEvent::Handle;
54
55use Scalar::Util ();
56use List::Util ();
57use Carp ();
58use Errno qw(EAGAIN EINTR);
59
60use AnyEvent (); BEGIN { AnyEvent::common_sense }
61use AnyEvent::Util qw(WSAEWOULDBLOCK);
62
63our $VERSION = $AnyEvent::VERSION;
64
64=head1 METHODS 65=head1 METHODS
65 66
66=over 4 67=over 4
67 68
68=item $handle = B<new> AnyEvent::TLS fh => $filehandle, key => value... 69=item $handle = B<new> AnyEvent::TLS fh => $filehandle, key => value...
… …
216memory and push it into the queue, but instead only read more data from 217memory and push it into the queue, but instead only read more data from
217the file when the write queue becomes empty. 218the file when the write queue becomes empty.
218 219
219=item timeout => $fractional_seconds 220=item timeout => $fractional_seconds
220 221
222=item rtimeout => $fractional_seconds
223
224=item wtimeout => $fractional_seconds
225
221If non-zero, then this enables an "inactivity" timeout: whenever this many 226If non-zero, then these enables an "inactivity" timeout: whenever this
222seconds pass without a successful read or write on the underlying file 227many seconds pass without a successful read or write on the underlying
223handle, the C<on_timeout> callback will be invoked (and if that one is 228file handle (or a call to C<timeout_reset>), the C<on_timeout> callback
224missing, a non-fatal C<ETIMEDOUT> error will be raised). 229will be invoked (and if that one is missing, a non-fatal C<ETIMEDOUT>
230error will be raised).
231
232There are three variants of the timeouts that work fully independent
233of each other, for both read and write, just read, and just write:
234C<timeout>, C<rtimeout> and C<wtimeout>, with corresponding callbacks
235C<on_timeout>, C<on_rtimeout> and C<on_wtimeout>, and reset functions
236C<timeout_reset>, C<rtimeout_reset>, and C<wtimeout_reset>.
225 237
226Note that timeout processing is also active when you currently do not have 238Note that timeout processing is also active when you currently do not have
227any outstanding read or write requests: If you plan to keep the connection 239any outstanding read or write requests: If you plan to keep the connection
228idle then you should disable the timout temporarily or ignore the timeout 240idle then you should disable the timout temporarily or ignore the timeout
229in the C<on_timeout> callback, in which case AnyEvent::Handle will simply 241in the C<on_timeout> callback, in which case AnyEvent::Handle will simply
… …
273accomplishd by setting this option to a true value. 285accomplishd by setting this option to a true value.
274 286
275The default is your opertaing system's default behaviour (most likely 287The default is your opertaing system's default behaviour (most likely
276enabled), this option explicitly enables or disables it, if possible. 288enabled), this option explicitly enables or disables it, if possible.
277 289
290=item keepalive => <boolean>
291
292Enables (default disable) the SO_KEEPALIVE option on the stream socket:
293normally, TCP connections have no time-out once established, so TCP
294conenctions, once established, can stay alive forever even when the other
295side has long gone. TCP keepalives are a cheap way to take down long-lived
296TCP connections whent he other side becomes unreachable. While the default
297is OS-dependent, TCP keepalives usually kick in after around two hours,
298and, if the other side doesn't reply, take down the TCP connection some 10
299to 15 minutes later.
300
301It is harmless to specify this option for file handles that do not support
302keepalives, and enabling it on connections that are potentially long-lived
303is usually a good idea.
304
305=item oobinline => <boolean>
306
307BSD majorly fucked up the implementation of TCP urgent data. The result
308is that almost no OS implements TCP according to the specs, and every OS
309implements it slightly differently.
310
311If you want to handle TCP urgent data, then setting this flag gives you
312the most portable way of getting urgent data, by putting it into the
313stream.
314
278=item read_size => <bytes> 315=item read_size => <bytes>
279 316
280The default read block size (the amount of bytes this module will 317The default read block size (the amount of bytes this module will
281try to read during each loop iteration, which affects memory 318try to read during each loop iteration, which affects memory
282requirements). Default: C<8192>. 319requirements). Default: C<8192>.
… …
438 delete $self->{_skip_drain_rbuf}; 475 delete $self->{_skip_drain_rbuf};
439 $self->_start; 476 $self->_start;
440 477
441 $self->{on_connect} 478 $self->{on_connect}
442 and $self->{on_connect}($self, $host, $port, sub { 479 and $self->{on_connect}($self, $host, $port, sub {
443 delete @$self{qw(fh _tw _ww _rw _eof _queue rbuf _wbuf tls _tls_rbuf _tls_wbuf)}; 480 delete @$self{qw(fh _tw _rtw _wtw _ww _rw _eof _queue rbuf _wbuf tls _tls_rbuf _tls_wbuf)};
444 $self->{_skip_drain_rbuf} = 1; 481 $self->{_skip_drain_rbuf} = 1;
445 &$retry; 482 &$retry;
446 }); 483 });
447 484
448 } else { 485 } else {
… …
474sub _start { 511sub _start {
475 my ($self) = @_; 512 my ($self) = @_;
476 513
477 AnyEvent::Util::fh_nonblocking $self->{fh}, 1; 514 AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
478 515
516 $self->{_activity} =
517 $self->{_ractivity} =
479 $self->{_activity} = AnyEvent->now; 518 $self->{_wactivity} = AE::now;
480 $self->_timeout;
481 519
520 $self->timeout (delete $self->{timeout} ) if $self->{timeout};
521 $self->rtimeout (delete $self->{rtimeout} ) if $self->{rtimeout};
522 $self->wtimeout (delete $self->{wtimeout} ) if $self->{wtimeout};
523
482 $self->no_delay (delete $self->{no_delay}) if exists $self->{no_delay}; 524 $self->no_delay (delete $self->{no_delay} ) if exists $self->{no_delay};
525 $self->keepalive (delete $self->{keepalive}) if exists $self->{keepalive};
526 $self->oobinline (delete $self->{oobinline}) if exists $self->{oobinline};
483 527
484 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}) 528 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx})
485 if $self->{tls}; 529 if $self->{tls};
486 530
487 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; 531 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
488 532
489 $self->start_read 533 $self->start_read
490 if $self->{on_read} || @{ $self->{_queue} }; 534 if $self->{on_read} || @{ $self->{_queue} };
491 535
492 $self->_drain_wbuf; 536 $self->_drain_wbuf;
493} 537}
494
495#sub _shutdown {
496# my ($self) = @_;
497#
498# delete @$self{qw(_tw _rw _ww fh wbuf on_read _queue)};
499# $self->{_eof} = 1; # tell starttls et. al to stop trying
500#
501# &_freetls;
502#}
503 538
504sub _error { 539sub _error {
505 my ($self, $errno, $fatal, $message) = @_; 540 my ($self, $errno, $fatal, $message) = @_;
506 541
507 $! = $errno; 542 $! = $errno;
… …
544 $_[0]{on_eof} = $_[1]; 579 $_[0]{on_eof} = $_[1];
545} 580}
546 581
547=item $handle->on_timeout ($cb) 582=item $handle->on_timeout ($cb)
548 583
549Replace the current C<on_timeout> callback, or disables the callback (but 584=item $handle->on_rtimeout ($cb)
550not the timeout) if C<$cb> = C<undef>. See the C<timeout> constructor
551argument and method.
552 585
553=cut 586=item $handle->on_wtimeout ($cb)
554 587
555sub on_timeout { 588Replace the current C<on_timeout>, C<on_rtimeout> or C<on_wtimeout>
556 $_[0]{on_timeout} = $_[1]; 589callback, or disables the callback (but not the timeout) if C<$cb> =
557} 590C<undef>. See the C<timeout> constructor argument and method.
591
592=cut
593
594# see below
558 595
559=item $handle->autocork ($boolean) 596=item $handle->autocork ($boolean)
560 597
561Enables or disables the current autocork behaviour (see C<autocork> 598Enables or disables the current autocork behaviour (see C<autocork>
562constructor argument). Changes will only take effect on the next write. 599constructor argument). Changes will only take effect on the next write.
… …
577sub no_delay { 614sub no_delay {
578 $_[0]{no_delay} = $_[1]; 615 $_[0]{no_delay} = $_[1];
579 616
580 eval { 617 eval {
581 local $SIG{__DIE__}; 618 local $SIG{__DIE__};
582 setsockopt $_[0]{fh}, &Socket::IPPROTO_TCP, &Socket::TCP_NODELAY, int $_[1] 619 setsockopt $_[0]{fh}, Socket::IPPROTO_TCP (), Socket::TCP_NODELAY (), int $_[1]
583 if $_[0]{fh}; 620 if $_[0]{fh};
584 }; 621 };
585} 622}
586 623
624=item $handle->keepalive ($boolean)
625
626Enables or disables the C<keepalive> setting (see constructor argument of
627the same name for details).
628
629=cut
630
631sub keepalive {
632 $_[0]{keepalive} = $_[1];
633
634 eval {
635 local $SIG{__DIE__};
636 setsockopt $_[0]{fh}, Socket::SOL_SOCKET (), Socket::SO_KEEPALIVE (), int $_[1]
637 if $_[0]{fh};
638 };
639}
640
641=item $handle->oobinline ($boolean)
642
643Enables or disables the C<oobinline> setting (see constructor argument of
644the same name for details).
645
646=cut
647
648sub oobinline {
649 $_[0]{oobinline} = $_[1];
650
651 eval {
652 local $SIG{__DIE__};
653 setsockopt $_[0]{fh}, Socket::SOL_SOCKET (), Socket::SO_OOBINLINE (), int $_[1]
654 if $_[0]{fh};
655 };
656}
657
658=item $handle->keepalive ($boolean)
659
660Enables or disables the C<keepalive> setting (see constructor argument of
661the same name for details).
662
663=cut
664
665sub keepalive {
666 $_[0]{keepalive} = $_[1];
667
668 eval {
669 local $SIG{__DIE__};
670 setsockopt $_[0]{fh}, Socket::SOL_SOCKET (), Socket::SO_KEEPALIVE (), int $_[1]
671 if $_[0]{fh};
672 };
673}
674
587=item $handle->on_starttls ($cb) 675=item $handle->on_starttls ($cb)
588 676
589Replace the current C<on_starttls> callback (see the C<on_starttls> constructor argument). 677Replace the current C<on_starttls> callback (see the C<on_starttls> constructor argument).
590 678
591=cut 679=cut
… …
616 704
617############################################################################# 705#############################################################################
618 706
619=item $handle->timeout ($seconds) 707=item $handle->timeout ($seconds)
620 708
709=item $handle->rtimeout ($seconds)
710
711=item $handle->wtimeout ($seconds)
712
621Configures (or disables) the inactivity timeout. 713Configures (or disables) the inactivity timeout.
622 714
623=cut 715=item $handle->timeout_reset
624 716
625sub timeout { 717=item $handle->rtimeout_reset
718
719=item $handle->wtimeout_reset
720
721Reset the activity timeout, as if data was received or sent.
722
723These methods are cheap to call.
724
725=cut
726
727for my $dir ("", "r", "w") {
728 my $timeout = "${dir}timeout";
729 my $tw = "_${dir}tw";
730 my $on_timeout = "on_${dir}timeout";
731 my $activity = "_${dir}activity";
732 my $cb;
733
734 *$on_timeout = sub {
735 $_[0]{$on_timeout} = $_[1];
736 };
737
738 *$timeout = sub {
626 my ($self, $timeout) = @_; 739 my ($self, $new_value) = @_;
627 740
628 $self->{timeout} = $timeout; 741 $self->{$timeout} = $new_value;
629 delete $self->{_tw}; 742 delete $self->{$tw}; &$cb;
630 $self->_timeout; 743 };
631}
632 744
745 *{"${dir}timeout_reset"} = sub {
746 $_[0]{$activity} = AE::now;
747 };
748
749 # main workhorse:
633# reset the timeout watcher, as neccessary 750 # reset the timeout watcher, as neccessary
634# also check for time-outs 751 # also check for time-outs
635sub _timeout { 752 $cb = sub {
636 my ($self) = @_; 753 my ($self) = @_;
637 754
638 if ($self->{timeout} && $self->{fh}) { 755 if ($self->{$timeout} && $self->{fh}) {
639 my $NOW = AnyEvent->now; 756 my $NOW = AE::now;
640 757
641 # when would the timeout trigger? 758 # when would the timeout trigger?
642 my $after = $self->{_activity} + $self->{timeout} - $NOW; 759 my $after = $self->{$activity} + $self->{$timeout} - $NOW;
643 760
644 # now or in the past already? 761 # now or in the past already?
645 if ($after <= 0) { 762 if ($after <= 0) {
646 $self->{_activity} = $NOW; 763 $self->{$activity} = $NOW;
647 764
648 if ($self->{on_timeout}) { 765 if ($self->{$on_timeout}) {
649 $self->{on_timeout}($self); 766 $self->{$on_timeout}($self);
650 } else { 767 } else {
651 $self->_error (Errno::ETIMEDOUT); 768 $self->_error (Errno::ETIMEDOUT);
769 }
770
771 # callback could have changed timeout value, optimise
772 return unless $self->{$timeout};
773
774 # calculate new after
775 $after = $self->{$timeout};
652 } 776 }
653 777
654 # callback could have changed timeout value, optimise 778 Scalar::Util::weaken $self;
655 return unless $self->{timeout}; 779 return unless $self; # ->error could have destroyed $self
656 780
657 # calculate new after 781 $self->{$tw} ||= AE::timer $after, 0, sub {
658 $after = $self->{timeout}; 782 delete $self->{$tw};
783 $cb->($self);
784 };
785 } else {
786 delete $self->{$tw};
659 } 787 }
660
661 Scalar::Util::weaken $self;
662 return unless $self; # ->error could have destroyed $self
663
664 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
665 delete $self->{_tw};
666 $self->_timeout;
667 });
668 } else {
669 delete $self->{_tw};
670 } 788 }
671} 789}
672 790
673############################################################################# 791#############################################################################
674 792
… …
722 my $len = syswrite $self->{fh}, $self->{wbuf}; 840 my $len = syswrite $self->{fh}, $self->{wbuf};
723 841
724 if (defined $len) { 842 if (defined $len) {
725 substr $self->{wbuf}, 0, $len, ""; 843 substr $self->{wbuf}, 0, $len, "";
726 844
727 $self->{_activity} = AnyEvent->now; 845 $self->{_activity} = $self->{_wactivity} = AE::now;
728 846
729 $self->{on_drain}($self) 847 $self->{on_drain}($self)
730 if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf}) 848 if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf})
731 && $self->{on_drain}; 849 && $self->{on_drain};
732 850
… …
738 856
739 # try to write data immediately 857 # try to write data immediately
740 $cb->() unless $self->{autocork}; 858 $cb->() unless $self->{autocork};
741 859
742 # if still data left in wbuf, we need to poll 860 # if still data left in wbuf, we need to poll
743 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb) 861 $self->{_ww} = AE::io $self->{fh}, 1, $cb
744 if length $self->{wbuf}; 862 if length $self->{wbuf};
745 }; 863 };
746} 864}
747 865
748our %WH; 866our %WH;
… …
838Other languages could read single lines terminated by a newline and pass 956Other languages could read single lines terminated by a newline and pass
839this line into their JSON decoder of choice. 957this line into their JSON decoder of choice.
840 958
841=cut 959=cut
842 960
961sub json_coder() {
962 eval { require JSON::XS; JSON::XS->new->utf8 }
963 || do { require JSON; JSON->new->utf8 }
964}
965
843register_write_type json => sub { 966register_write_type json => sub {
844 my ($self, $ref) = @_; 967 my ($self, $ref) = @_;
845 968
846 require JSON; 969 my $json = $self->{json} ||= json_coder;
847 970
848 $self->{json} ? $self->{json}->encode ($ref) 971 $json->encode ($ref)
849 : JSON::encode_json ($ref)
850}; 972};
851 973
852=item storable => $reference 974=item storable => $reference
853 975
854Freezes the given reference using L<Storable> and writes it to the 976Freezes the given reference using L<Storable> and writes it to the
… …
1147 1269
1148 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read") 1270 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read")
1149 ->($self, $cb, @_); 1271 ->($self, $cb, @_);
1150 } 1272 }
1151 1273
1152
1153 unshift @{ $self->{_queue} }, $cb; 1274 unshift @{ $self->{_queue} }, $cb;
1154 $self->_drain_rbuf; 1275 $self->_drain_rbuf;
1155} 1276}
1156 1277
1157=item $handle->push_read (type => @args, $cb) 1278=item $handle->push_read (type => @args, $cb)
… …
1408=cut 1529=cut
1409 1530
1410register_read_type json => sub { 1531register_read_type json => sub {
1411 my ($self, $cb) = @_; 1532 my ($self, $cb) = @_;
1412 1533
1413 my $json = $self->{json} ||= 1534 my $json = $self->{json} ||= json_coder;
1414 eval { require JSON::XS; JSON::XS->new->utf8 }
1415 || do { require JSON; JSON->new->utf8 };
1416 1535
1417 my $data; 1536 my $data;
1418 my $rbuf = \$self->{rbuf}; 1537 my $rbuf = \$self->{rbuf};
1419 1538
1420 sub { 1539 sub {
… …
1540 my ($self) = @_; 1659 my ($self) = @_;
1541 1660
1542 unless ($self->{_rw} || $self->{_eof}) { 1661 unless ($self->{_rw} || $self->{_eof}) {
1543 Scalar::Util::weaken $self; 1662 Scalar::Util::weaken $self;
1544 1663
1545 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1664 $self->{_rw} = AE::io $self->{fh}, 0, sub {
1546 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf}); 1665 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf});
1547 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; 1666 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
1548 1667
1549 if ($len > 0) { 1668 if ($len > 0) {
1550 $self->{_activity} = AnyEvent->now; 1669 $self->{_activity} = $self->{_ractivity} = AE::now;
1551 1670
1552 if ($self->{tls}) { 1671 if ($self->{tls}) {
1553 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf); 1672 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);
1554 1673
1555 &_dotls ($self); 1674 &_dotls ($self);
… …
1563 $self->_drain_rbuf; 1682 $self->_drain_rbuf;
1564 1683
1565 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1684 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1566 return $self->_error ($!, 1); 1685 return $self->_error ($!, 1);
1567 } 1686 }
1568 }); 1687 };
1569 } 1688 }
1570} 1689}
1571 1690
1572our $ERROR_SYSCALL; 1691our $ERROR_SYSCALL;
1573our $ERROR_WANT_READ; 1692our $ERROR_WANT_READ;
… …
1694 require Net::SSLeay; 1813 require Net::SSLeay;
1695 1814
1696 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL (); 1815 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL ();
1697 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ (); 1816 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ ();
1698 1817
1699 $tls = $self->{tls}; 1818 $tls = delete $self->{tls};
1700 $ctx = $self->{tls_ctx}; 1819 $ctx = $self->{tls_ctx};
1701 1820
1702 local $Carp::CarpLevel = 1; # skip ourselves when creating a new context or session 1821 local $Carp::CarpLevel = 1; # skip ourselves when creating a new context or session
1703 1822
1704 if ("HASH" eq ref $ctx) { 1823 if ("HASH" eq ref $ctx) {
… …
1791 my $fh = delete $self->{fh}; 1910 my $fh = delete $self->{fh};
1792 my $wbuf = delete $self->{wbuf}; 1911 my $wbuf = delete $self->{wbuf};
1793 1912
1794 my @linger; 1913 my @linger;
1795 1914
1796 push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub { 1915 push @linger, AE::io $fh, 1, sub {
1797 my $len = syswrite $fh, $wbuf, length $wbuf; 1916 my $len = syswrite $fh, $wbuf, length $wbuf;
1798 1917
1799 if ($len > 0) { 1918 if ($len > 0) {
1800 substr $wbuf, 0, $len, ""; 1919 substr $wbuf, 0, $len, "";
1801 } else { 1920 } else {
1802 @linger = (); # end 1921 @linger = (); # end
1803 } 1922 }
1804 }); 1923 };
1805 push @linger, AnyEvent->timer (after => $linger, cb => sub { 1924 push @linger, AE::timer $linger, 0, sub {
1806 @linger = (); 1925 @linger = ();
1807 }); 1926 };
1808 } 1927 }
1809} 1928}
1810 1929
1811=item $handle->destroy 1930=item $handle->destroy
1812 1931

Diff Legend

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