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 Coro::Util ();

use base 'Coro::Handle';

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 ($host =~ /^(25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[0-9][0-9]?)
                \.(25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[0-9][0-9]?)
                \.(25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[0-9][0-9]?)
                \.(25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[0-9][0-9]?)$/) {
      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 ($class, $arg) = @_;
   my $fh;

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

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

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

   if ($arg->{ReusePort}) {
      $fh->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});
      $fh->bind($sa[0])
         or croak "bind($arg->{LocalHost}:$arg->{LocalPort}): $!";
   }

   $fh;
}
   
sub new {
   my $class = shift;
   my %arg = @_;
   my $fh;

   $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;

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

      for (@sa) {
         $fh = $class->_prepare_socket(\%arg)
            or return;

         $! = 0;

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

         $! or last;

         $!{ECONNREFUSED} or $!{ENETUNREACH} or $!{ETIMEDOUT} or $!{EHOSTUNREACH}
            or return;

         undef $fh;
      }
   } else {
      $fh = $class->_prepare_socket(\%arg)
         or return;
      if (exists $arg{Listen}) {
         $fh->listen($arg{Listen})
            or return;
      }
   }

   $fh;
}

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

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