Coro/Event/Socket.pm

=head1 NAME

Coro::Socket - non-blocking socket-io

=head1 SYNOPSIS

 use Coro::Socket;

 # listen on an ipv4 socket
 my $socket = new Coro::Socket PeerHost => "localhost",
                               PeerPort => 'finger';

 # listen on any other type of socket
 my $socket = Coro::Socket->new_from_fh
                 (IO::Socket::UNIX->new
                     Local  => "/tmp/socket",
                     Type   => SOCK_STREAM,
                 );

=head1 DESCRIPTION

This module implements socket-handles in a coroutine-compatible way,
that is, other coroutines can run while reads or writes block on the
handle. L<Coro::Handle>.

=over 4

=cut

package Coro::Socket;

no warnings "uninitialized";

use strict;

use Errno ();
use Carp qw(croak);
use Socket;
use IO::Socket::INET ();

use Coro::Util ();

use base qw(Coro::Handle IO::Socket::INET);

our $VERSION = 1.9;

our (%_proto, %_port);

sub _proto($) {
   $_proto{$_[0]} ||= do {
      ((getprotobyname $_[0])[2] || (getprotobynumber $_[0])[2])
         or croak "unsupported protocol: $_[0]";
   };
}

sub _port($$) {
   $_port{$_[0],$_[1]} ||= do {
      return $_[0] if $_[0] =~ /^\d+$/;

      $_[0] =~ /([^(]+)\s*(?:\((\d+)\))?/x
         or croak "unparsable port number: $_[0]";
      ((getservbyname $1, $_[1])[2]
        || (getservbyport $1, $_[1])[2]
        || $2)
         or croak "unknown port: $_[0]";
   };
}

sub _sa($$$) {
   my ($host, $port, $proto) = @_;
   $port or $host =~ s/:([^:]+)$// and $port = $1;
   my $_proto = _proto($proto);
   my $_port = _port($port, $proto);

   # optimize this a bit for a common case
   if (Coro::Util::dotted_quad $host) {
      return pack_sockaddr_in ($_port, inet_aton $host);
   } else {
      my (undef, undef, undef, undef, @host) = Coro::Util::gethostbyname $host
         or croak "unknown host: $host";
      map pack_sockaddr_in ($_port,$_), @host;
   }
}

=item $fh = new Coro::Socket param => value, ...

Create a new non-blocking tcp handle and connect to the given host
and port. The parameter names and values are mostly the same as in
IO::Socket::INET (as ugly as I think they are).

If the host is unreachable or otherwise cannot be connected to this method
returns undef. On all other errors ot croak's.

Multihomed is always enabled.

   $fh = new Coro::Socket PeerHost => "localhost", PeerPort => 'finger';

=cut

sub _prepare_socket {
   my ($self, $arg) = @_;

   $self
}
   
sub new {
   my ($class, %arg) = @_;

   $arg{Proto}     ||= 'tcp';
   $arg{LocalHost} ||= delete $arg{LocalAddr};
   $arg{PeerHost}  ||= delete $arg{PeerAddr};
   defined ($arg{Type}) or $arg{Type} = $arg{Proto} eq "tcp" ? SOCK_STREAM : SOCK_DGRAM;

   socket my $fh, PF_INET, $arg{Type}, _proto ($arg{Proto})
      or return;

   my $self = bless Coro::Handle->new_from_fh (
      $fh,
      timeout       => $arg{Timeout},
      forward_class => $arg{forward_class},
      partial       => $arg{partial},
   ), $class
      or return;

   $self->configure (\%arg)
}

sub configure {
   my ($self, $arg) = @_;

   if ($arg->{ReuseAddr}) {
      $self->setsockopt (SOL_SOCKET, SO_REUSEADDR, 1)
         or croak "setsockopt(SO_REUSEADDR): $!";
   }

   if ($arg->{ReusePort}) {
      $self->setsockopt (SOL_SOCKET, SO_REUSEPORT, 1)
         or croak "setsockopt(SO_REUSEPORT): $!";
   }

   if ($arg->{LocalPort} || $arg->{LocalHost}) {
      my @sa = _sa($arg->{LocalHost} || "0.0.0.0", $arg->{LocalPort} || 0, $arg->{Proto});
      $self->bind ($sa[0])
         or croak "bind($arg->{LocalHost}:$arg->{LocalPort}): $!";
   }

   if ($arg->{PeerHost}) {
      my @sa = _sa ($arg->{PeerHost}, $arg->{PeerPort}, $arg->{Proto});

      for (@sa) {
         $! = 0;

         if ($self->connect ($_)) {
            next unless writable $self;
            $! = unpack "i", $self->getsockopt (SOL_SOCKET, SO_ERROR);
         }

         $! or last;

         $!{ECONNREFUSED} or $!{ENETUNREACH} or $!{ETIMEDOUT} or $!{EHOSTUNREACH}
            or return;
      }
   } elsif (exists $arg->{Listen}) {
      $self->listen ($arg->{Listen})
         or return;
   }

   1
}

=item connect, listen, bind, getsockopt, setsockopt,
send, recv, peername, sockname, shutdown, peerport, peerhost

Do the same thing as the perl builtins or IO::Socket methods (but return
true on EINPROGRESS). Remember that these must be method calls.

=cut

sub connect     { connect tied(${$_[0]})->[0], $_[1] or $! == Errno::EINPROGRESS }
sub bind        { bind    tied(${$_[0]})->[0], $_[1] }
sub listen      { listen  tied(${$_[0]})->[0], $_[1] }
sub getsockopt  { getsockopt tied(${$_[0]})->[0], $_[1], $_[2] }
sub setsockopt  { setsockopt tied(${$_[0]})->[0], $_[1], $_[2], $_[3] }
sub send        { send tied(${$_[0]})->[0], $_[1], $_[2], @_ > 2 ? $_[3] : () }
sub recv        { recv tied(${$_[0]})->[0], $_[1], $_[2], @_ > 2 ? $_[3] : () }
sub sockname    { getsockname tied(${$_[0]})->[0] }
sub peername    { getpeername tied(${$_[0]})->[0] }
sub shutdown    { shutdown tied(${$_[0]})->[0], $_[1] }

=item ($fh, $peername) = $listen_fh->accept

In scalar context, returns the newly accepted socket (or undef) and in
list context return the ($fh, $peername) pair (or nothing).

=cut

sub accept {
   my ($peername, $fh);
   while () {
      $peername = accept $fh, tied(${$_[0]})->[0]
         and return wantarray 
                    ? ($_[0]->new_from_fh($fh), $peername)
                    :  $_[0]->new_from_fh($fh);

      return unless $!{EAGAIN};

      $_[0]->readable or return;
   }
}

1;

=back

=head1 AUTHOR

 Marc Lehmann <schmorp@schmorp.de>
 http://home.schmorp.de/

=cut

Revision:	1.41
Committed:	Sun Nov 5 19:16:09 2006 UTC (17 years, 7 months ago) by root
Branch:	MAIN
CVS Tags:	rel-2_5
Changes since 1.40:	+10 -10 lines
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	=head1 NAME
2
3			Coro::Socket - non-blocking socket-io
4
5			=head1 SYNOPSIS
6
7			use Coro::Socket;
8
9	root	1.33	# listen on an ipv4 socket
10			my $socket = new Coro::Socket PeerHost => "localhost",
11			PeerPort => 'finger';
12
13			# listen on any other type of socket
14			my $socket = Coro::Socket->new_from_fh
15			(IO::Socket::UNIX->new
16			Local => "/tmp/socket",
17			Type => SOCK_STREAM,
18			);
19
20	root	1.1	=head1 DESCRIPTION
21
22			This module implements socket-handles in a coroutine-compatible way,
23			that is, other coroutines can run while reads or writes block on the
24			handle. L<Coro::Handle>.
25
26			=over 4
27
28			=cut
29
30			package Coro::Socket;
31
32	root	1.39	no warnings "uninitialized";
33
34			use strict;
35	root	1.6
36	root	1.1	use Errno ();
37			use Carp qw(croak);
38			use Socket;
39	root	1.41	use IO::Socket::INET ();
40	root	1.1
41			use Coro::Util ();
42
43	root	1.41	use base qw(Coro::Handle IO::Socket::INET);
44	root	1.1
45	root	1.39	our $VERSION = 1.9;
46
47			our (%_proto, %_port);
48	root	1.1
49			sub _proto($) {
50			$_proto{$_[0]} \|\|= do {
51			((getprotobyname $_[0])[2] \|\| (getprotobynumber $_[0])[2])
52			or croak "unsupported protocol: $_[0]";
53			};
54			}
55
56			sub _port($$) {
57			$_port{$_[0],$_[1]} \|\|= do {
58			return $_[0] if $_[0] =~ /^\d+$/;
59
60			$_[0] =~ /([^(]+)\s*(?:\((\d+)\))?/x
61			or croak "unparsable port number: $_[0]";
62			((getservbyname $1, $_[1])[2]
63			\|\| (getservbyport $1, $_[1])[2]
64			\|\| $2)
65			or croak "unknown port: $_[0]";
66			};
67			}
68
69			sub _sa($$$) {
70			my ($host, $port, $proto) = @_;
71	root	1.14	$port or $host =~ s/:([^:]+)$// and $port = $1;
72	root	1.1	my $_proto = _proto($proto);
73			my $_port = _port($port, $proto);
74
75			# optimize this a bit for a common case
76	root	1.40	if (Coro::Util::dotted_quad $host) {
77			return pack_sockaddr_in ($_port, inet_aton $host);
78	root	1.1	} else {
79			my (undef, undef, undef, undef, @host) = Coro::Util::gethostbyname $host
80			or croak "unknown host: $host";
81	root	1.40	map pack_sockaddr_in ($_port,$_), @host;
82	root	1.1	}
83			}
84
85	root	1.2	=item $fh = new Coro::Socket param => value, ...
86	root	1.1
87			Create a new non-blocking tcp handle and connect to the given host
88			and port. The parameter names and values are mostly the same as in
89			IO::Socket::INET (as ugly as I think they are).
90
91			If the host is unreachable or otherwise cannot be connected to this method
92			returns undef. On all other errors ot croak's.
93
94			Multihomed is always enabled.
95
96	root	1.14	$fh = new Coro::Socket PeerHost => "localhost", PeerPort => 'finger';
97	root	1.1
98			=cut
99
100			sub _prepare_socket {
101	root	1.40	my ($self, $arg) = @_;
102	root	1.1
103	root	1.40	$self
104			}
105
106			sub new {
107			my ($class, %arg) = @_;
108
109			$arg{Proto} \|\|= 'tcp';
110			$arg{LocalHost} \|\|= delete $arg{LocalAddr};
111			$arg{PeerHost} \|\|= delete $arg{PeerAddr};
112			defined ($arg{Type}) or $arg{Type} = $arg{Proto} eq "tcp" ? SOCK_STREAM : SOCK_DGRAM;
113
114			socket my $fh, PF_INET, $arg{Type}, _proto ($arg{Proto})
115	root	1.1	or return;
116
117	root	1.40	my $self = bless Coro::Handle->new_from_fh (
118	root	1.39	$fh,
119	root	1.40	timeout => $arg{Timeout},
120			forward_class => $arg{forward_class},
121	root	1.41	partial => $arg{partial},
122	root	1.39	), $class
123	root	1.1	or return;
124
125	root	1.40	$self->configure (\%arg)
126			}
127
128			sub configure {
129			my ($self, $arg) = @_;
130
131	root	1.1	if ($arg->{ReuseAddr}) {
132	root	1.40	$self->setsockopt (SOL_SOCKET, SO_REUSEADDR, 1)
133	root	1.1	or croak "setsockopt(SO_REUSEADDR): $!";
134			}
135
136			if ($arg->{ReusePort}) {
137	root	1.40	$self->setsockopt (SOL_SOCKET, SO_REUSEPORT, 1)
138	root	1.1	or croak "setsockopt(SO_REUSEPORT): $!";
139			}
140
141	pcg	1.20	if ($arg->{LocalPort} \|\| $arg->{LocalHost}) {
142			my @sa = _sa($arg->{LocalHost} \|\| "0.0.0.0", $arg->{LocalPort} \|\| 0, $arg->{Proto});
143	root	1.40	$self->bind ($sa[0])
144	root	1.1	or croak "bind($arg->{LocalHost}:$arg->{LocalPort}): $!";
145			}
146
147	root	1.40	if ($arg->{PeerHost}) {
148	root	1.41	my @sa = _sa ($arg->{PeerHost}, $arg->{PeerPort}, $arg->{Proto});
149	root	1.1
150			for (@sa) {
151			$! = 0;
152
153	root	1.40	if ($self->connect ($_)) {
154			next unless writable $self;
155	root	1.41	$! = unpack "i", $self->getsockopt (SOL_SOCKET, SO_ERROR);
156	root	1.1	}
157
158			$! or last;
159
160			$!{ECONNREFUSED} or $!{ENETUNREACH} or $!{ETIMEDOUT} or $!{EHOSTUNREACH}
161			or return;
162			}
163	root	1.41	} elsif (exists $arg->{Listen}) {
164			$self->listen ($arg->{Listen})
165			or return;
166	root	1.1	}
167
168	root	1.41	1
169	root	1.1	}
170
171			=item connect, listen, bind, getsockopt, setsockopt,
172	root	1.41	send, recv, peername, sockname, shutdown, peerport, peerhost
173	root	1.1
174	root	1.5	Do the same thing as the perl builtins or IO::Socket methods (but return
175			true on EINPROGRESS). Remember that these must be method calls.
176	root	1.1
177			=cut
178
179	root	1.3	sub connect { connect tied(${$_[0]})->[0], $_[1] or $! == Errno::EINPROGRESS }
180			sub bind { bind tied(${$_[0]})->[0], $_[1] }
181			sub listen { listen tied(${$_[0]})->[0], $_[1] }
182			sub getsockopt { getsockopt tied(${$_[0]})->[0], $_[1], $_[2] }
183			sub setsockopt { setsockopt tied(${$_[0]})->[0], $_[1], $_[2], $_[3] }
184			sub send { send tied(${$_[0]})->[0], $_[1], $_[2], @_ > 2 ? $_[3] : () }
185			sub recv { recv tied(${$_[0]})->[0], $_[1], $_[2], @_ > 2 ? $_[3] : () }
186	root	1.5	sub sockname { getsockname tied(${$_[0]})->[0] }
187			sub peername { getpeername tied(${$_[0]})->[0] }
188	root	1.9	sub shutdown { shutdown tied(${$_[0]})->[0], $_[1] }
189	root	1.1
190	root	1.10	=item ($fh, $peername) = $listen_fh->accept
191	root	1.1
192			In scalar context, returns the newly accepted socket (or undef) and in
193	root	1.4	list context return the ($fh, $peername) pair (or nothing).
194	root	1.1
195			=cut
196
197			sub accept {
198			my ($peername, $fh);
199			while () {
200	root	1.3	$peername = accept $fh, tied(${$_[0]})->[0]
201	root	1.4	and return wantarray
202	root	1.11	? ($_[0]->new_from_fh($fh), $peername)
203			: $_[0]->new_from_fh($fh);
204	root	1.1
205			return unless $!{EAGAIN};
206
207			$_[0]->readable or return;
208			}
209			}
210
211			1;
212
213	root	1.15	=back
214
215	root	1.1	=head1 AUTHOR
216
217	root	1.29	Marc Lehmann <schmorp@schmorp.de>
218	root	1.27	http://home.schmorp.de/
219	root	1.1
220			=cut
221