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;

BEGIN { eval { require warnings } && warnings->unimport ("uninitialized") }

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

use Coro::Util ();

use base 'Coro::Handle';

$VERSION = 0.96;

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}), $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 <pcg@goof.com>
 http://www.goof.com/pcg/marc/

=cut

Revision:	1.24
Committed:	Thu May 13 16:12:17 2004 UTC (20 years ago) by pcg
Branch:	MAIN
Changes since 1.23:	+1 -1 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			=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	pcg	1.21	BEGIN { eval { require warnings } && warnings->unimport ("uninitialized") }
22	root	1.6
23	root	1.1	use Errno ();
24			use Carp qw(croak);
25			use Socket;
26
27			use Coro::Util ();
28
29			use base 'Coro::Handle';
30
31	pcg	1.24	$VERSION = 0.96;
32	root	1.1
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	root	1.14	$port or $host =~ s/:([^:]+)$// and $port = $1;
56	root	1.1	my $_proto = _proto($proto);
57			my $_port = _port($port, $proto);
58
59			# optimize this a bit for a common case
60			if ($host =~ /^(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			\.(25[0-5]\|2[0-4][0-9]\|1[0-9][0-9]\|[0-9][0-9]?)$/) {
64			return pack_sockaddr_in($_port, inet_aton $host);
65			} else {
66			my (undef, undef, undef, undef, @host) = Coro::Util::gethostbyname $host
67			or croak "unknown host: $host";
68			map pack_sockaddr_in($_port,$_), @host;
69			}
70			}
71
72	root	1.2	=item $fh = new Coro::Socket param => value, ...
73	root	1.1
74			Create a new non-blocking tcp handle and connect to the given host
75			and port. The parameter names and values are mostly the same as in
76			IO::Socket::INET (as ugly as I think they are).
77
78			If the host is unreachable or otherwise cannot be connected to this method
79			returns undef. On all other errors ot croak's.
80
81			Multihomed is always enabled.
82
83	root	1.14	$fh = new Coro::Socket PeerHost => "localhost", PeerPort => 'finger';
84	root	1.1
85			=cut
86
87			sub _prepare_socket {
88			my ($class, $arg) = @_;
89			my $fh;
90
91			socket $fh, PF_INET, $arg->{Type}, _proto($arg->{Proto})
92			or return;
93
94			$fh = bless Coro::Handle->new_from_fh($fh, timeout => $arg{Timeout}), $class
95			or return;
96
97			if ($arg->{ReuseAddr}) {
98			$fh->setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
99			or croak "setsockopt(SO_REUSEADDR): $!";
100			}
101
102			if ($arg->{ReusePort}) {
103			$fh->setsockopt(SOL_SOCKET, SO_REUSEPORT, 1)
104			or croak "setsockopt(SO_REUSEPORT): $!";
105			}
106
107	pcg	1.20	if ($arg->{LocalPort} \|\| $arg->{LocalHost}) {
108			my @sa = _sa($arg->{LocalHost} \|\| "0.0.0.0", $arg->{LocalPort} \|\| 0, $arg->{Proto});
109	root	1.1	$fh->bind($sa[0])
110			or croak "bind($arg->{LocalHost}:$arg->{LocalPort}): $!";
111			}
112
113			$fh;
114			}
115
116			sub new {
117			my $class = shift;
118			my %arg = @_;
119			my $fh;
120
121			$arg{Proto} \|\|= 'tcp';
122			$arg{LocalHost} \|\|= delete $arg{LocalAddr};
123			$arg{PeerHost} \|\|= delete $arg{PeerAddr};
124			defined ($arg{Type}) or $arg{Type} = $arg{Proto} eq "tcp" ? SOCK_STREAM : SOCK_DGRAM;
125
126			if ($arg{PeerHost}) {
127			my @sa = _sa($arg{PeerHost}, $arg{PeerPort}, $arg{Proto});
128
129			for (@sa) {
130			$fh = $class->_prepare_socket(\%arg)
131			or return;
132
133			$! = 0;
134
135			if ($fh->connect($_)) {
136			next unless writable $fh;
137			$! = unpack "i", $fh->getsockopt(SOL_SOCKET, SO_ERROR);
138			}
139
140			$! or last;
141
142			$!{ECONNREFUSED} or $!{ENETUNREACH} or $!{ETIMEDOUT} or $!{EHOSTUNREACH}
143			or return;
144	root	1.12
145			undef $fh;
146	root	1.1	}
147			} else {
148			$fh = $class->_prepare_socket(\%arg)
149			or return;
150			if (exists $arg{Listen}) {
151			$fh->listen($arg{Listen})
152			or return;
153			}
154			}
155
156			$fh;
157			}
158
159			=item connect, listen, bind, getsockopt, setsockopt,
160	root	1.8	send, recv, peername, sockname, shutdown
161	root	1.1
162	root	1.5	Do the same thing as the perl builtins or IO::Socket methods (but return
163			true on EINPROGRESS). Remember that these must be method calls.
164	root	1.1
165			=cut
166
167	root	1.3	sub connect { connect tied(${$_[0]})->[0], $_[1] or $! == Errno::EINPROGRESS }
168			sub bind { bind tied(${$_[0]})->[0], $_[1] }
169			sub listen { listen tied(${$_[0]})->[0], $_[1] }
170			sub getsockopt { getsockopt tied(${$_[0]})->[0], $_[1], $_[2] }
171			sub setsockopt { setsockopt tied(${$_[0]})->[0], $_[1], $_[2], $_[3] }
172			sub send { send tied(${$_[0]})->[0], $_[1], $_[2], @_ > 2 ? $_[3] : () }
173			sub recv { recv tied(${$_[0]})->[0], $_[1], $_[2], @_ > 2 ? $_[3] : () }
174	root	1.5	sub sockname { getsockname tied(${$_[0]})->[0] }
175			sub peername { getpeername tied(${$_[0]})->[0] }
176	root	1.9	sub shutdown { shutdown tied(${$_[0]})->[0], $_[1] }
177	root	1.1
178	root	1.10	=item ($fh, $peername) = $listen_fh->accept
179	root	1.1
180			In scalar context, returns the newly accepted socket (or undef) and in
181	root	1.4	list context return the ($fh, $peername) pair (or nothing).
182	root	1.1
183			=cut
184
185			sub accept {
186			my ($peername, $fh);
187			while () {
188	root	1.3	$peername = accept $fh, tied(${$_[0]})->[0]
189	root	1.4	and return wantarray
190	root	1.11	? ($_[0]->new_from_fh($fh), $peername)
191			: $_[0]->new_from_fh($fh);
192	root	1.1
193			return unless $!{EAGAIN};
194
195			$_[0]->readable or return;
196			}
197			}
198
199			1;
200
201	root	1.15	=back
202
203	root	1.1	=head1 AUTHOR
204
205			Marc Lehmann <pcg@goof.com>
206			http://www.goof.com/pcg/marc/
207
208			=cut
209