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 (Coro::Util::dotted_quad $host) {
      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 ($self, $arg) = @_;

   $self
}
   
sub new {
   my ($class, %arg) = @_;

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

   socket my $fh, PF_INET, $arg{Type}, _proto ($arg{Proto})
      or return;

   my $self = bless Coro::Handle->new_from_fh (
      $fh,
      timeout       => $arg{Timeout},
      forward_class => $arg{forward_class},
   ), $class
      or return;

   $self->configure (\%arg)
}

sub configure {
   my ($self, $arg) = @_;

   if ($arg->{ReuseAddr}) {
      $self->setsockopt (SOL_SOCKET, SO_REUSEADDR, 1)
         or croak "setsockopt(SO_REUSEADDR): $!";
   }

   if ($arg->{ReusePort}) {
      $self->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});
      $self->bind ($sa[0])
         or croak "bind($arg->{LocalHost}:$arg->{LocalPort}): $!";
   }

   if ($arg->{PeerHost}) {
      my @sa = _sa($arg->{PeerHost}, $arg->{PeerPort}, $arg->{Proto});

      for (@sa) {
         $! = 0;

         if ($self->connect ($_)) {
            next unless writable $self;
            $! = unpack "i", $self->getsockopt(SOL_SOCKET, SO_ERROR);
         }

         $! or last;

         $!{ECONNREFUSED} or $!{ENETUNREACH} or $!{ETIMEDOUT} or $!{EHOSTUNREACH}
            or return;
      }
   } else {
      if (exists $arg->{Listen}) {
         $self->listen ($arg->{Listen})
            or return;
      }
   }

   $self
}

=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.40
Committed:	Sun Nov 5 02:01:24 2006 UTC (17 years, 7 months ago) by root
Branch:	MAIN
Changes since 1.39:	+36 -43 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	root	1.40	if (Coro::Util::dotted_quad $host) {
76			return pack_sockaddr_in ($_port, inet_aton $host);
77	root	1.1	} else {
78			my (undef, undef, undef, undef, @host) = Coro::Util::gethostbyname $host
79			or croak "unknown host: $host";
80	root	1.40	map pack_sockaddr_in ($_port,$_), @host;
81	root	1.1	}
82			}
83
84	root	1.2	=item $fh = new Coro::Socket param => value, ...
85	root	1.1
86			Create a new non-blocking tcp handle and connect to the given host
87			and port. The parameter names and values are mostly the same as in
88			IO::Socket::INET (as ugly as I think they are).
89
90			If the host is unreachable or otherwise cannot be connected to this method
91			returns undef. On all other errors ot croak's.
92
93			Multihomed is always enabled.
94
95	root	1.14	$fh = new Coro::Socket PeerHost => "localhost", PeerPort => 'finger';
96	root	1.1
97			=cut
98
99			sub _prepare_socket {
100	root	1.40	my ($self, $arg) = @_;
101	root	1.1
102	root	1.40	$self
103			}
104
105			sub new {
106			my ($class, %arg) = @_;
107
108			$arg{Proto} \|\|= 'tcp';
109			$arg{LocalHost} \|\|= delete $arg{LocalAddr};
110			$arg{PeerHost} \|\|= delete $arg{PeerAddr};
111			defined ($arg{Type}) or $arg{Type} = $arg{Proto} eq "tcp" ? SOCK_STREAM : SOCK_DGRAM;
112
113			socket my $fh, PF_INET, $arg{Type}, _proto ($arg{Proto})
114	root	1.1	or return;
115
116	root	1.40	my $self = bless Coro::Handle->new_from_fh (
117	root	1.39	$fh,
118	root	1.40	timeout => $arg{Timeout},
119			forward_class => $arg{forward_class},
120	root	1.39	), $class
121	root	1.1	or return;
122
123	root	1.40	$self->configure (\%arg)
124			}
125
126			sub configure {
127			my ($self, $arg) = @_;
128
129	root	1.1	if ($arg->{ReuseAddr}) {
130	root	1.40	$self->setsockopt (SOL_SOCKET, SO_REUSEADDR, 1)
131	root	1.1	or croak "setsockopt(SO_REUSEADDR): $!";
132			}
133
134			if ($arg->{ReusePort}) {
135	root	1.40	$self->setsockopt (SOL_SOCKET, SO_REUSEPORT, 1)
136	root	1.1	or croak "setsockopt(SO_REUSEPORT): $!";
137			}
138
139	pcg	1.20	if ($arg->{LocalPort} \|\| $arg->{LocalHost}) {
140			my @sa = _sa($arg->{LocalHost} \|\| "0.0.0.0", $arg->{LocalPort} \|\| 0, $arg->{Proto});
141	root	1.40	$self->bind ($sa[0])
142	root	1.1	or croak "bind($arg->{LocalHost}:$arg->{LocalPort}): $!";
143			}
144
145	root	1.40	if ($arg->{PeerHost}) {
146			my @sa = _sa($arg->{PeerHost}, $arg->{PeerPort}, $arg->{Proto});
147	root	1.1
148			for (@sa) {
149			$! = 0;
150
151	root	1.40	if ($self->connect ($_)) {
152			next unless writable $self;
153			$! = unpack "i", $self->getsockopt(SOL_SOCKET, SO_ERROR);
154	root	1.1	}
155
156			$! or last;
157
158			$!{ECONNREFUSED} or $!{ENETUNREACH} or $!{ETIMEDOUT} or $!{EHOSTUNREACH}
159			or return;
160			}
161			} else {
162	root	1.40	if (exists $arg->{Listen}) {
163			$self->listen ($arg->{Listen})
164	root	1.1	or return;
165			}
166			}
167
168	root	1.40	$self
169	root	1.1	}
170
171			=item connect, listen, bind, getsockopt, setsockopt,
172	root	1.8	send, recv, peername, sockname, shutdown
173	root	1.1
174	root	1.5	Do the same thing as the perl builtins or IO::Socket methods (but return
175			true on EINPROGRESS). Remember that these must be method calls.
176	root	1.1
177			=cut
178
179	root	1.3	sub connect { connect tied(${$_[0]})->[0], $_[1] or $! == Errno::EINPROGRESS }
180			sub bind { bind tied(${$_[0]})->[0], $_[1] }
181			sub listen { listen tied(${$_[0]})->[0], $_[1] }
182			sub getsockopt { getsockopt tied(${$_[0]})->[0], $_[1], $_[2] }
183			sub setsockopt { setsockopt tied(${$_[0]})->[0], $_[1], $_[2], $_[3] }
184			sub send { send tied(${$_[0]})->[0], $_[1], $_[2], @_ > 2 ? $_[3] : () }
185			sub recv { recv tied(${$_[0]})->[0], $_[1], $_[2], @_ > 2 ? $_[3] : () }
186	root	1.5	sub sockname { getsockname tied(${$_[0]})->[0] }
187			sub peername { getpeername tied(${$_[0]})->[0] }
188	root	1.9	sub shutdown { shutdown tied(${$_[0]})->[0], $_[1] }
189	root	1.1
190	root	1.10	=item ($fh, $peername) = $listen_fh->accept
191	root	1.1
192			In scalar context, returns the newly accepted socket (or undef) and in
193	root	1.4	list context return the ($fh, $peername) pair (or nothing).
194	root	1.1
195			=cut
196
197			sub accept {
198			my ($peername, $fh);
199			while () {
200	root	1.3	$peername = accept $fh, tied(${$_[0]})->[0]
201	root	1.4	and return wantarray
202	root	1.11	? ($_[0]->new_from_fh($fh), $peername)
203			: $_[0]->new_from_fh($fh);
204	root	1.1
205			return unless $!{EAGAIN};
206
207			$_[0]->readable or return;
208			}
209			}
210
211			1;
212
213	root	1.15	=back
214
215	root	1.1	=head1 AUTHOR
216
217	root	1.29	Marc Lehmann <schmorp@schmorp.de>
218	root	1.27	http://home.schmorp.de/
219	root	1.1
220			=cut
221