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;

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

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

use Coro::Util ();

use base 'Coro::Handle';

$VERSION = 1.9;

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 <schmorp@schmorp.de>
 http://home.schmorp.de/

=cut

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