Coro/Event/Socket.pm

=head1 NAME

Coro::Socket - non-blocking socket-io

=head1 SYNOPSIS

 use Coro::Socket;

=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 qw(uninitialized);

use Errno ();
use Carp qw(croak);
use Socket;

use Coro::Util ();

use base 'Coro::Handle';

$VERSION = 0.6;

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) = @_;
   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_inet 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}), $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}) {
      my @sa = _sa($arg->{LocalHost} || "0.0.0.0", $arg->{LocalPort}, $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;

=head1 AUTHOR

 Marc Lehmann <pcg@goof.com>
 http://www.goof.com/pcg/marc/

=cut

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