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;

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

use Coro::Util ();

use base 'Coro::Handle';

$VERSION = 0.45;

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

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] }

=item ($peername, $fh) = $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 
                    ? ((new_from_fh Coro::Socket $fh), $peername)
                    : (new_from_fh Coro::Socket $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.6
Committed:	Mon Sep 24 00:51:20 2001 UTC (22 years, 8 months ago) by root
Branch:	MAIN
Changes since 1.5:	+2 -0 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	root	1.6	no warnings;
22
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			$VERSION = 0.45;
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	root	1.2	=item $fh = new Coro::Socket param => value, ...
72	root	1.1
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			} else {
145			$fh = $class->_prepare_socket(\%arg)
146			or return;
147			if (exists $arg{Listen}) {
148			$fh->listen($arg{Listen})
149			or return;
150			}
151			}
152
153			$fh;
154			}
155
156			=item connect, listen, bind, getsockopt, setsockopt,
157	root	1.5	send, recv, peername, sockname
158	root	1.1
159	root	1.5	Do the same thing as the perl builtins or IO::Socket methods (but return
160			true on EINPROGRESS). Remember that these must be method calls.
161	root	1.1
162			=cut
163
164	root	1.3	sub connect { connect tied(${$_[0]})->[0], $_[1] or $! == Errno::EINPROGRESS }
165			sub bind { bind tied(${$_[0]})->[0], $_[1] }
166			sub listen { listen tied(${$_[0]})->[0], $_[1] }
167			sub getsockopt { getsockopt tied(${$_[0]})->[0], $_[1], $_[2] }
168			sub setsockopt { setsockopt tied(${$_[0]})->[0], $_[1], $_[2], $_[3] }
169			sub send { send tied(${$_[0]})->[0], $_[1], $_[2], @_ > 2 ? $_[3] : () }
170			sub recv { recv tied(${$_[0]})->[0], $_[1], $_[2], @_ > 2 ? $_[3] : () }
171	root	1.5	sub sockname { getsockname tied(${$_[0]})->[0] }
172			sub peername { getpeername tied(${$_[0]})->[0] }
173	root	1.1
174			=item ($peername, $fh) = $listen_fh->accept
175
176			In scalar context, returns the newly accepted socket (or undef) and in
177	root	1.4	list context return the ($fh, $peername) pair (or nothing).
178	root	1.1
179			=cut
180
181			sub accept {
182			my ($peername, $fh);
183			while () {
184	root	1.3	$peername = accept $fh, tied(${$_[0]})->[0]
185	root	1.4	and return wantarray
186	root	1.5	? ((new_from_fh Coro::Socket $fh), $peername)
187	root	1.4	: (new_from_fh Coro::Socket $fh);
188	root	1.1
189			return unless $!{EAGAIN};
190
191			$_[0]->readable or return;
192			}
193			}
194
195			1;
196
197			=head1 AUTHOR
198
199			Marc Lehmann <pcg@goof.com>
200			http://www.goof.com/pcg/marc/
201
202			=cut
203