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.168 by root, Fri Jul 31 07:34:00 2009 UTC vs.
Revision 1.176 by root, Sun Aug 9 00:20:35 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 4
14=cut 5=cut
15 6
16our $VERSION = 4.881; 7our $VERSION = 4.92;
17 8
18=head1 SYNOPSIS 9=head1 SYNOPSIS
19 10
20 use AnyEvent; 11 use AnyEvent;
21 use AnyEvent::Handle; 12 use AnyEvent::Handle;
… …
59C<on_error> callback. 50C<on_error> callback.
60 51
61All callbacks will be invoked with the handle object as their first 52All callbacks will be invoked with the handle object as their first
62argument. 53argument.
63 54
55=cut
56
57package AnyEvent::Handle;
58
59use Scalar::Util ();
60use List::Util ();
61use Carp ();
62use Errno qw(EAGAIN EINTR);
63
64use AnyEvent (); BEGIN { AnyEvent::common_sense }
65use AnyEvent::Util qw(WSAEWOULDBLOCK);
66
64=head1 METHODS 67=head1 METHODS
65 68
66=over 4 69=over 4
67 70
68=item $handle = B<new> AnyEvent::TLS fh => $filehandle, key => value... 71=item $handle = B<new> AnyEvent::TLS fh => $filehandle, key => value...
… …
216memory and push it into the queue, but instead only read more data from 219memory and push it into the queue, but instead only read more data from
217the file when the write queue becomes empty. 220the file when the write queue becomes empty.
218 221
219=item timeout => $fractional_seconds 222=item timeout => $fractional_seconds
220 223
224=item rtimeout => $fractional_seconds
225
226=item wtimeout => $fractional_seconds
227
221If non-zero, then this enables an "inactivity" timeout: whenever this many 228If non-zero, then these enables an "inactivity" timeout: whenever this
222seconds pass without a successful read or write on the underlying file 229many 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 230file handle (or a call to C<timeout_reset>), the C<on_timeout> callback
224missing, a non-fatal C<ETIMEDOUT> error will be raised). 231will be invoked (and if that one is missing, a non-fatal C<ETIMEDOUT>
232error will be raised).
233
234There are three variants of the timeouts that work fully independent
235of each other, for both read and write, just read, and just write:
236C<timeout>, C<rtimeout> and C<wtimeout>, with corresponding callbacks
237C<on_timeout>, C<on_rtimeout> and C<on_wtimeout>, and reset functions
238C<timeout_reset>, C<rtimeout_reset>, and C<wtimeout_reset>.
225 239
226Note that timeout processing is also active when you currently do not have 240Note 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 241any outstanding read or write requests: If you plan to keep the connection
228idle then you should disable the timout temporarily or ignore the timeout 242idle then you should disable the timout temporarily or ignore the timeout
229in the C<on_timeout> callback, in which case AnyEvent::Handle will simply 243in the C<on_timeout> callback, in which case AnyEvent::Handle will simply
… …
474sub _start { 488sub _start {
475 my ($self) = @_; 489 my ($self) = @_;
476 490
477 AnyEvent::Util::fh_nonblocking $self->{fh}, 1; 491 AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
478 492
493 $self->{_activity} =
494 $self->{_ractivity} =
479 $self->{_activity} = AnyEvent->now; 495 $self->{_wactivity} = AE::now;
480 $self->_timeout; 496
497 $self->timeout (delete $self->{timeout} ) if $self->{timeout};
498 $self->rtimeout (delete $self->{rtimeout}) if $self->{rtimeout};
499 $self->wtimeout (delete $self->{wtimeout}) if $self->{wtimeout};
481 500
482 $self->no_delay (delete $self->{no_delay}) if exists $self->{no_delay}; 501 $self->no_delay (delete $self->{no_delay}) if exists $self->{no_delay};
483 502
484 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}) 503 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx})
485 if $self->{tls}; 504 if $self->{tls};
… …
544 $_[0]{on_eof} = $_[1]; 563 $_[0]{on_eof} = $_[1];
545} 564}
546 565
547=item $handle->on_timeout ($cb) 566=item $handle->on_timeout ($cb)
548 567
549Replace the current C<on_timeout> callback, or disables the callback (but 568=item $handle->on_rtimeout ($cb)
550not the timeout) if C<$cb> = C<undef>. See the C<timeout> constructor
551argument and method.
552 569
553=cut 570=item $handle->on_wtimeout ($cb)
554 571
555sub on_timeout { 572Replace the current C<on_timeout>, C<on_rtimeout> or C<on_wtimeout>
556 $_[0]{on_timeout} = $_[1]; 573callback, or disables the callback (but not the timeout) if C<$cb> =
557} 574C<undef>. See the C<timeout> constructor argument and method.
575
576=cut
577
578# see below
558 579
559=item $handle->autocork ($boolean) 580=item $handle->autocork ($boolean)
560 581
561Enables or disables the current autocork behaviour (see C<autocork> 582Enables or disables the current autocork behaviour (see C<autocork>
562constructor argument). Changes will only take effect on the next write. 583constructor argument). Changes will only take effect on the next write.
… …
616 637
617############################################################################# 638#############################################################################
618 639
619=item $handle->timeout ($seconds) 640=item $handle->timeout ($seconds)
620 641
642=item $handle->rtimeout ($seconds)
643
644=item $handle->wtimeout ($seconds)
645
621Configures (or disables) the inactivity timeout. 646Configures (or disables) the inactivity timeout.
622 647
623=cut 648=item $handle->timeout_reset
624 649
625sub timeout { 650=item $handle->rtimeout_reset
651
652=item $handle->wtimeout_reset
653
654Reset the activity timeout, as if data was received or sent.
655
656These methods are cheap to call.
657
658=cut
659
660for my $dir ("", "r", "w") {
661 my $timeout = "${dir}timeout";
662 my $tw = "_${dir}tw";
663 my $on_timeout = "on_${dir}timeout";
664 my $activity = "_${dir}activity";
665 my $cb;
666
667 *$on_timeout = sub {
668 $_[0]{$on_timeout} = $_[1];
669 };
670
671 *$timeout = sub {
626 my ($self, $timeout) = @_; 672 my ($self, $new_value) = @_;
627 673
628 $self->{timeout} = $timeout; 674 $self->{$timeout} = $new_value;
629 $self->_timeout; 675 delete $self->{$tw}; &$cb;
630} 676 };
631 677
678 *{"${dir}timeout_reset"} = sub {
679 $_[0]{$activity} = AE::now;
680 };
681
682 # main workhorse:
632# reset the timeout watcher, as neccessary 683 # reset the timeout watcher, as neccessary
633# also check for time-outs 684 # also check for time-outs
634sub _timeout { 685 $cb = sub {
635 my ($self) = @_; 686 my ($self) = @_;
636 687
637 if ($self->{timeout} && $self->{fh}) { 688 if ($self->{$timeout} && $self->{fh}) {
638 my $NOW = AnyEvent->now; 689 my $NOW = AE::now;
639 690
640 # when would the timeout trigger? 691 # when would the timeout trigger?
641 my $after = $self->{_activity} + $self->{timeout} - $NOW; 692 my $after = $self->{$activity} + $self->{$timeout} - $NOW;
642 693
643 # now or in the past already? 694 # now or in the past already?
644 if ($after <= 0) { 695 if ($after <= 0) {
645 $self->{_activity} = $NOW; 696 $self->{$activity} = $NOW;
646 697
647 if ($self->{on_timeout}) { 698 if ($self->{$on_timeout}) {
648 $self->{on_timeout}($self); 699 $self->{$on_timeout}($self);
649 } else { 700 } else {
650 $self->_error (Errno::ETIMEDOUT); 701 $self->_error (Errno::ETIMEDOUT);
702 }
703
704 # callback could have changed timeout value, optimise
705 return unless $self->{$timeout};
706
707 # calculate new after
708 $after = $self->{$timeout};
651 } 709 }
652 710
653 # callback could have changed timeout value, optimise 711 Scalar::Util::weaken $self;
654 return unless $self->{timeout}; 712 return unless $self; # ->error could have destroyed $self
655 713
656 # calculate new after 714 $self->{$tw} ||= AE::timer $after, 0, sub {
657 $after = $self->{timeout}; 715 delete $self->{$tw};
716 $cb->($self);
717 };
718 } else {
719 delete $self->{$tw};
658 } 720 }
659
660 Scalar::Util::weaken $self;
661 return unless $self; # ->error could have destroyed $self
662
663 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
664 delete $self->{_tw};
665 $self->_timeout;
666 });
667 } else {
668 delete $self->{_tw};
669 } 721 }
670} 722}
671 723
672############################################################################# 724#############################################################################
673 725
… …
721 my $len = syswrite $self->{fh}, $self->{wbuf}; 773 my $len = syswrite $self->{fh}, $self->{wbuf};
722 774
723 if (defined $len) { 775 if (defined $len) {
724 substr $self->{wbuf}, 0, $len, ""; 776 substr $self->{wbuf}, 0, $len, "";
725 777
726 $self->{_activity} = AnyEvent->now; 778 $self->{_activity} = $self->{_wactivity} = AE::now;
727 779
728 $self->{on_drain}($self) 780 $self->{on_drain}($self)
729 if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf}) 781 if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf})
730 && $self->{on_drain}; 782 && $self->{on_drain};
731 783
… …
737 789
738 # try to write data immediately 790 # try to write data immediately
739 $cb->() unless $self->{autocork}; 791 $cb->() unless $self->{autocork};
740 792
741 # if still data left in wbuf, we need to poll 793 # if still data left in wbuf, we need to poll
742 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb) 794 $self->{_ww} = AE::io $self->{fh}, 1, $cb
743 if length $self->{wbuf}; 795 if length $self->{wbuf};
744 }; 796 };
745} 797}
746 798
747our %WH; 799our %WH;
… …
994 1046
995 # avoid recursion 1047 # avoid recursion
996 return if $self->{_skip_drain_rbuf}; 1048 return if $self->{_skip_drain_rbuf};
997 local $self->{_skip_drain_rbuf} = 1; 1049 local $self->{_skip_drain_rbuf} = 1;
998 1050
999 if (
1000 defined $self->{rbuf_max}
1001 && $self->{rbuf_max} < length $self->{rbuf}
1002 ) {
1003 $self->_error (Errno::ENOSPC, 1), return;
1004 }
1005
1006 while () { 1051 while () {
1007 # we need to use a separate tls read buffer, as we must not receive data while 1052 # we need to use a separate tls read buffer, as we must not receive data while
1008 # we are draining the buffer, and this can only happen with TLS. 1053 # we are draining the buffer, and this can only happen with TLS.
1009 $self->{rbuf} .= delete $self->{_tls_rbuf} 1054 $self->{rbuf} .= delete $self->{_tls_rbuf}
1010 if exists $self->{_tls_rbuf}; 1055 if exists $self->{_tls_rbuf};
… …
1049 $self->{on_eof} 1094 $self->{on_eof}
1050 ? $self->{on_eof}($self) 1095 ? $self->{on_eof}($self)
1051 : $self->_error (0, 1, "Unexpected end-of-file"); 1096 : $self->_error (0, 1, "Unexpected end-of-file");
1052 1097
1053 return; 1098 return;
1099 }
1100
1101 if (
1102 defined $self->{rbuf_max}
1103 && $self->{rbuf_max} < length $self->{rbuf}
1104 ) {
1105 $self->_error (Errno::ENOSPC, 1), return;
1054 } 1106 }
1055 1107
1056 # may need to restart read watcher 1108 # may need to restart read watcher
1057 unless ($self->{_rw}) { 1109 unless ($self->{_rw}) {
1058 $self->start_read 1110 $self->start_read
… …
1539 my ($self) = @_; 1591 my ($self) = @_;
1540 1592
1541 unless ($self->{_rw} || $self->{_eof}) { 1593 unless ($self->{_rw} || $self->{_eof}) {
1542 Scalar::Util::weaken $self; 1594 Scalar::Util::weaken $self;
1543 1595
1544 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1596 $self->{_rw} = AE::io $self->{fh}, 0, sub {
1545 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf}); 1597 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf});
1546 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; 1598 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
1547 1599
1548 if ($len > 0) { 1600 if ($len > 0) {
1549 $self->{_activity} = AnyEvent->now; 1601 $self->{_activity} = $self->{_ractivity} = AE::now;
1550 1602
1551 if ($self->{tls}) { 1603 if ($self->{tls}) {
1552 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf); 1604 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);
1553 1605
1554 &_dotls ($self); 1606 &_dotls ($self);
… …
1562 $self->_drain_rbuf; 1614 $self->_drain_rbuf;
1563 1615
1564 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1616 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1565 return $self->_error ($!, 1); 1617 return $self->_error ($!, 1);
1566 } 1618 }
1567 }); 1619 };
1568 } 1620 }
1569} 1621}
1570 1622
1571our $ERROR_SYSCALL; 1623our $ERROR_SYSCALL;
1572our $ERROR_WANT_READ; 1624our $ERROR_WANT_READ;
… …
1732 Net::SSLeay::CTX_set_mode ($tls, 1|2); 1784 Net::SSLeay::CTX_set_mode ($tls, 1|2);
1733 1785
1734 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1786 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1735 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1787 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1736 1788
1789 Net::SSLeay::BIO_write ($self->{_rbio}, delete $self->{rbuf});
1790
1737 Net::SSLeay::set_bio ($tls, $self->{_rbio}, $self->{_wbio}); 1791 Net::SSLeay::set_bio ($tls, $self->{_rbio}, $self->{_wbio});
1738 1792
1739 $self->{_on_starttls} = sub { $_[0]{on_starttls}(@_) } 1793 $self->{_on_starttls} = sub { $_[0]{on_starttls}(@_) }
1740 if $self->{on_starttls}; 1794 if $self->{on_starttls};
1741 1795
… …
1770 my ($self) = @_; 1824 my ($self) = @_;
1771 1825
1772 return unless $self->{tls}; 1826 return unless $self->{tls};
1773 1827
1774 $self->{tls_ctx}->_put_session (delete $self->{tls}) 1828 $self->{tls_ctx}->_put_session (delete $self->{tls})
1775 if ref $self->{tls}; 1829 if $self->{tls} > 0;
1776 1830
1777 delete @$self{qw(_rbio _wbio _tls_wbuf _on_starttls)}; 1831 delete @$self{qw(_rbio _wbio _tls_wbuf _on_starttls)};
1778} 1832}
1779 1833
1780sub DESTROY { 1834sub DESTROY {
… …
1788 my $fh = delete $self->{fh}; 1842 my $fh = delete $self->{fh};
1789 my $wbuf = delete $self->{wbuf}; 1843 my $wbuf = delete $self->{wbuf};
1790 1844
1791 my @linger; 1845 my @linger;
1792 1846
1793 push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub { 1847 push @linger, AE::io $fh, 1, sub {
1794 my $len = syswrite $fh, $wbuf, length $wbuf; 1848 my $len = syswrite $fh, $wbuf, length $wbuf;
1795 1849
1796 if ($len > 0) { 1850 if ($len > 0) {
1797 substr $wbuf, 0, $len, ""; 1851 substr $wbuf, 0, $len, "";
1798 } else { 1852 } else {
1799 @linger = (); # end 1853 @linger = (); # end
1800 } 1854 }
1801 }); 1855 };
1802 push @linger, AnyEvent->timer (after => $linger, cb => sub { 1856 push @linger, AE::timer $linger, 0, sub {
1803 @linger = (); 1857 @linger = ();
1804 }); 1858 };
1805 } 1859 }
1806} 1860}
1807 1861
1808=item $handle->destroy 1862=item $handle->destroy
1809 1863

Diff Legend

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