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 *
#	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	root	1.39	no warnings "uninitialized";
33
34			use strict;
35	root	1.6
36	root	1.1	use Errno ();
37			use Carp qw(croak);
38			use Socket;
39
40			use Coro::Util ();
41
42			use base 'Coro::Handle';
43
44	root	1.39	our $VERSION = 1.9;
45
46			our (%_proto, %_port);
47	root	1.1
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	root	1.14	$port or $host =~ s/:([^:]+)$// and $port = $1;
71	root	1.1	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	root	1.2	=item $fh = new Coro::Socket param => value, ...
88	root	1.1
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	root	1.14	$fh = new Coro::Socket PeerHost => "localhost", PeerPort => 'finger';
99	root	1.1
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	root	1.39	$fh = bless Coro::Handle->new_from_fh (
110			$fh,
111			timeout => $arg->{Timeout},
112			forward_class => $arg->{forward_class},
113			), $class
114	root	1.1	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	pcg	1.20	if ($arg->{LocalPort} \|\| $arg->{LocalHost}) {
127			my @sa = _sa($arg->{LocalHost} \|\| "0.0.0.0", $arg->{LocalPort} \|\| 0, $arg->{Proto});
128	root	1.1	$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	root	1.12
164			undef $fh;
165	root	1.1	}
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	root	1.8	send, recv, peername, sockname, shutdown
180	root	1.1
181	root	1.5	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	root	1.1
184			=cut
185
186	root	1.3	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	root	1.5	sub sockname { getsockname tied(${$_[0]})->[0] }
194			sub peername { getpeername tied(${$_[0]})->[0] }
195	root	1.9	sub shutdown { shutdown tied(${$_[0]})->[0], $_[1] }
196	root	1.1
197	root	1.10	=item ($fh, $peername) = $listen_fh->accept
198	root	1.1
199			In scalar context, returns the newly accepted socket (or undef) and in
200	root	1.4	list context return the ($fh, $peername) pair (or nothing).
201	root	1.1
202			=cut
203
204			sub accept {
205			my ($peername, $fh);
206			while () {
207	root	1.3	$peername = accept $fh, tied(${$_[0]})->[0]
208	root	1.4	and return wantarray
209	root	1.11	? ($_[0]->new_from_fh($fh), $peername)
210			: $_[0]->new_from_fh($fh);
211	root	1.1
212			return unless $!{EAGAIN};
213
214			$_[0]->readable or return;
215			}
216			}
217
218			1;
219
220	root	1.15	=back
221
222	root	1.1	=head1 AUTHOR
223
224	root	1.29	Marc Lehmann <schmorp@schmorp.de>
225	root	1.27	http://home.schmorp.de/
226	root	1.1
227			=cut
228