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.181 by root, Tue Sep 1 10:40:05 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
… …
438 delete $self->{_skip_drain_rbuf}; 450 delete $self->{_skip_drain_rbuf};
439 $self->_start; 451 $self->_start;
440 452
441 $self->{on_connect} 453 $self->{on_connect}
442 and $self->{on_connect}($self, $host, $port, sub { 454 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)}; 455 delete @$self{qw(fh _tw _rtw _wtw _ww _rw _eof _queue rbuf _wbuf tls _tls_rbuf _tls_wbuf)};
444 $self->{_skip_drain_rbuf} = 1; 456 $self->{_skip_drain_rbuf} = 1;
445 &$retry; 457 &$retry;
446 }); 458 });
447 459
448 } else { 460 } else {
… …
474sub _start { 486sub _start {
475 my ($self) = @_; 487 my ($self) = @_;
476 488
477 AnyEvent::Util::fh_nonblocking $self->{fh}, 1; 489 AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
478 490
491 $self->{_activity} =
492 $self->{_ractivity} =
479 $self->{_activity} = AnyEvent->now; 493 $self->{_wactivity} = AE::now;
480 $self->_timeout; 494
495 $self->timeout (delete $self->{timeout} ) if $self->{timeout};
496 $self->rtimeout (delete $self->{rtimeout}) if $self->{rtimeout};
497 $self->wtimeout (delete $self->{wtimeout}) if $self->{wtimeout};
481 498
482 $self->no_delay (delete $self->{no_delay}) if exists $self->{no_delay}; 499 $self->no_delay (delete $self->{no_delay}) if exists $self->{no_delay};
483 500
484 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}) 501 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx})
485 if $self->{tls}; 502 if $self->{tls};
… …
489 $self->start_read 506 $self->start_read
490 if $self->{on_read} || @{ $self->{_queue} }; 507 if $self->{on_read} || @{ $self->{_queue} };
491 508
492 $self->_drain_wbuf; 509 $self->_drain_wbuf;
493} 510}
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 511
504sub _error { 512sub _error {
505 my ($self, $errno, $fatal, $message) = @_; 513 my ($self, $errno, $fatal, $message) = @_;
506 514
507 $! = $errno; 515 $! = $errno;
… …
544 $_[0]{on_eof} = $_[1]; 552 $_[0]{on_eof} = $_[1];
545} 553}
546 554
547=item $handle->on_timeout ($cb) 555=item $handle->on_timeout ($cb)
548 556
549Replace the current C<on_timeout> callback, or disables the callback (but 557=item $handle->on_rtimeout ($cb)
550not the timeout) if C<$cb> = C<undef>. See the C<timeout> constructor
551argument and method.
552 558
553=cut 559=item $handle->on_wtimeout ($cb)
554 560
555sub on_timeout { 561Replace the current C<on_timeout>, C<on_rtimeout> or C<on_wtimeout>
556 $_[0]{on_timeout} = $_[1]; 562callback, or disables the callback (but not the timeout) if C<$cb> =
557} 563C<undef>. See the C<timeout> constructor argument and method.
564
565=cut
566
567# see below
558 568
559=item $handle->autocork ($boolean) 569=item $handle->autocork ($boolean)
560 570
561Enables or disables the current autocork behaviour (see C<autocork> 571Enables or disables the current autocork behaviour (see C<autocork>
562constructor argument). Changes will only take effect on the next write. 572constructor argument). Changes will only take effect on the next write.
… …
616 626
617############################################################################# 627#############################################################################
618 628
619=item $handle->timeout ($seconds) 629=item $handle->timeout ($seconds)
620 630
631=item $handle->rtimeout ($seconds)
632
633=item $handle->wtimeout ($seconds)
634
621Configures (or disables) the inactivity timeout. 635Configures (or disables) the inactivity timeout.
622 636
623=cut 637=item $handle->timeout_reset
624 638
625sub timeout { 639=item $handle->rtimeout_reset
640
641=item $handle->wtimeout_reset
642
643Reset the activity timeout, as if data was received or sent.
644
645These methods are cheap to call.
646
647=cut
648
649for my $dir ("", "r", "w") {
650 my $timeout = "${dir}timeout";
651 my $tw = "_${dir}tw";
652 my $on_timeout = "on_${dir}timeout";
653 my $activity = "_${dir}activity";
654 my $cb;
655
656 *$on_timeout = sub {
657 $_[0]{$on_timeout} = $_[1];
658 };
659
660 *$timeout = sub {
626 my ($self, $timeout) = @_; 661 my ($self, $new_value) = @_;
627 662
628 $self->{timeout} = $timeout; 663 $self->{$timeout} = $new_value;
629 delete $self->{_tw}; 664 delete $self->{$tw}; &$cb;
630 $self->_timeout; 665 };
631}
632 666
667 *{"${dir}timeout_reset"} = sub {
668 $_[0]{$activity} = AE::now;
669 };
670
671 # main workhorse:
633# reset the timeout watcher, as neccessary 672 # reset the timeout watcher, as neccessary
634# also check for time-outs 673 # also check for time-outs
635sub _timeout { 674 $cb = sub {
636 my ($self) = @_; 675 my ($self) = @_;
637 676
638 if ($self->{timeout} && $self->{fh}) { 677 if ($self->{$timeout} && $self->{fh}) {
639 my $NOW = AnyEvent->now; 678 my $NOW = AE::now;
640 679
641 # when would the timeout trigger? 680 # when would the timeout trigger?
642 my $after = $self->{_activity} + $self->{timeout} - $NOW; 681 my $after = $self->{$activity} + $self->{$timeout} - $NOW;
643 682
644 # now or in the past already? 683 # now or in the past already?
645 if ($after <= 0) { 684 if ($after <= 0) {
646 $self->{_activity} = $NOW; 685 $self->{$activity} = $NOW;
647 686
648 if ($self->{on_timeout}) { 687 if ($self->{$on_timeout}) {
649 $self->{on_timeout}($self); 688 $self->{$on_timeout}($self);
650 } else { 689 } else {
651 $self->_error (Errno::ETIMEDOUT); 690 $self->_error (Errno::ETIMEDOUT);
691 }
692
693 # callback could have changed timeout value, optimise
694 return unless $self->{$timeout};
695
696 # calculate new after
697 $after = $self->{$timeout};
652 } 698 }
653 699
654 # callback could have changed timeout value, optimise 700 Scalar::Util::weaken $self;
655 return unless $self->{timeout}; 701 return unless $self; # ->error could have destroyed $self
656 702
657 # calculate new after 703 $self->{$tw} ||= AE::timer $after, 0, sub {
658 $after = $self->{timeout}; 704 delete $self->{$tw};
705 $cb->($self);
706 };
707 } else {
708 delete $self->{$tw};
659 } 709 }
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 } 710 }
671} 711}
672 712
673############################################################################# 713#############################################################################
674 714
… …
722 my $len = syswrite $self->{fh}, $self->{wbuf}; 762 my $len = syswrite $self->{fh}, $self->{wbuf};
723 763
724 if (defined $len) { 764 if (defined $len) {
725 substr $self->{wbuf}, 0, $len, ""; 765 substr $self->{wbuf}, 0, $len, "";
726 766
727 $self->{_activity} = AnyEvent->now; 767 $self->{_activity} = $self->{_wactivity} = AE::now;
728 768
729 $self->{on_drain}($self) 769 $self->{on_drain}($self)
730 if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf}) 770 if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf})
731 && $self->{on_drain}; 771 && $self->{on_drain};
732 772
… …
738 778
739 # try to write data immediately 779 # try to write data immediately
740 $cb->() unless $self->{autocork}; 780 $cb->() unless $self->{autocork};
741 781
742 # if still data left in wbuf, we need to poll 782 # if still data left in wbuf, we need to poll
743 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb) 783 $self->{_ww} = AE::io $self->{fh}, 1, $cb
744 if length $self->{wbuf}; 784 if length $self->{wbuf};
745 }; 785 };
746} 786}
747 787
748our %WH; 788our %WH;
… …
838Other languages could read single lines terminated by a newline and pass 878Other languages could read single lines terminated by a newline and pass
839this line into their JSON decoder of choice. 879this line into their JSON decoder of choice.
840 880
841=cut 881=cut
842 882
883sub json_coder() {
884 eval { require JSON::XS; JSON::XS->new->utf8 }
885 || do { require JSON; JSON->new->utf8 }
886}
887
843register_write_type json => sub { 888register_write_type json => sub {
844 my ($self, $ref) = @_; 889 my ($self, $ref) = @_;
845 890
846 require JSON; 891 my $json = $self->{json} ||= json_coder;
847 892
848 $self->{json} ? $self->{json}->encode ($ref) 893 $json->encode ($ref)
849 : JSON::encode_json ($ref)
850}; 894};
851 895
852=item storable => $reference 896=item storable => $reference
853 897
854Freezes the given reference using L<Storable> and writes it to the 898Freezes the given reference using L<Storable> and writes it to the
… …
1147 1191
1148 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read") 1192 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read")
1149 ->($self, $cb, @_); 1193 ->($self, $cb, @_);
1150 } 1194 }
1151 1195
1152
1153 unshift @{ $self->{_queue} }, $cb; 1196 unshift @{ $self->{_queue} }, $cb;
1154 $self->_drain_rbuf; 1197 $self->_drain_rbuf;
1155} 1198}
1156 1199
1157=item $handle->push_read (type => @args, $cb) 1200=item $handle->push_read (type => @args, $cb)
… …
1408=cut 1451=cut
1409 1452
1410register_read_type json => sub { 1453register_read_type json => sub {
1411 my ($self, $cb) = @_; 1454 my ($self, $cb) = @_;
1412 1455
1413 my $json = $self->{json} ||= 1456 my $json = $self->{json} ||= json_coder;
1414 eval { require JSON::XS; JSON::XS->new->utf8 }
1415 || do { require JSON; JSON->new->utf8 };
1416 1457
1417 my $data; 1458 my $data;
1418 my $rbuf = \$self->{rbuf}; 1459 my $rbuf = \$self->{rbuf};
1419 1460
1420 sub { 1461 sub {
… …
1540 my ($self) = @_; 1581 my ($self) = @_;
1541 1582
1542 unless ($self->{_rw} || $self->{_eof}) { 1583 unless ($self->{_rw} || $self->{_eof}) {
1543 Scalar::Util::weaken $self; 1584 Scalar::Util::weaken $self;
1544 1585
1545 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1586 $self->{_rw} = AE::io $self->{fh}, 0, sub {
1546 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf}); 1587 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf});
1547 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; 1588 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
1548 1589
1549 if ($len > 0) { 1590 if ($len > 0) {
1550 $self->{_activity} = AnyEvent->now; 1591 $self->{_activity} = $self->{_ractivity} = AE::now;
1551 1592
1552 if ($self->{tls}) { 1593 if ($self->{tls}) {
1553 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf); 1594 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);
1554 1595
1555 &_dotls ($self); 1596 &_dotls ($self);
… …
1563 $self->_drain_rbuf; 1604 $self->_drain_rbuf;
1564 1605
1565 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1606 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1566 return $self->_error ($!, 1); 1607 return $self->_error ($!, 1);
1567 } 1608 }
1568 }); 1609 };
1569 } 1610 }
1570} 1611}
1571 1612
1572our $ERROR_SYSCALL; 1613our $ERROR_SYSCALL;
1573our $ERROR_WANT_READ; 1614our $ERROR_WANT_READ;
… …
1694 require Net::SSLeay; 1735 require Net::SSLeay;
1695 1736
1696 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL (); 1737 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL ();
1697 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ (); 1738 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ ();
1698 1739
1699 $tls = $self->{tls}; 1740 $tls = delete $self->{tls};
1700 $ctx = $self->{tls_ctx}; 1741 $ctx = $self->{tls_ctx};
1701 1742
1702 local $Carp::CarpLevel = 1; # skip ourselves when creating a new context or session 1743 local $Carp::CarpLevel = 1; # skip ourselves when creating a new context or session
1703 1744
1704 if ("HASH" eq ref $ctx) { 1745 if ("HASH" eq ref $ctx) {
… …
1791 my $fh = delete $self->{fh}; 1832 my $fh = delete $self->{fh};
1792 my $wbuf = delete $self->{wbuf}; 1833 my $wbuf = delete $self->{wbuf};
1793 1834
1794 my @linger; 1835 my @linger;
1795 1836
1796 push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub { 1837 push @linger, AE::io $fh, 1, sub {
1797 my $len = syswrite $fh, $wbuf, length $wbuf; 1838 my $len = syswrite $fh, $wbuf, length $wbuf;
1798 1839
1799 if ($len > 0) { 1840 if ($len > 0) {
1800 substr $wbuf, 0, $len, ""; 1841 substr $wbuf, 0, $len, "";
1801 } else { 1842 } else {
1802 @linger = (); # end 1843 @linger = (); # end
1803 } 1844 }
1804 }); 1845 };
1805 push @linger, AnyEvent->timer (after => $linger, cb => sub { 1846 push @linger, AE::timer $linger, 0, sub {
1806 @linger = (); 1847 @linger = ();
1807 }); 1848 };
1808 } 1849 }
1809} 1850}
1810 1851
1811=item $handle->destroy 1852=item $handle->destroy
1812 1853

Diff Legend

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