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 *
#	Content
1	=head1 NAME
2
3	Coro::Socket - non-blocking socket-io
4
5	=head1 SYNOPSIS
6
7	use Coro::Socket;
8
9	# 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	=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	BEGIN { eval { require warnings } && warnings->unimport ("uninitialized") }
33
34	use Errno ();
35	use Carp qw(croak);
36	use Socket;
37
38	use Coro::Util ();
39
40	use base 'Coro::Handle';
41
42	$VERSION = 1.9;
43
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	$port or $host =~ s/:([^:]+)$// and $port = $1;
67	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	=item $fh = new Coro::Socket param => value, ...
84
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	$fh = new Coro::Socket PeerHost => "localhost", PeerPort => 'finger';
95
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	if ($arg->{LocalPort} \|\| $arg->{LocalHost}) {
119	my @sa = _sa($arg->{LocalHost} \|\| "0.0.0.0", $arg->{LocalPort} \|\| 0, $arg->{Proto});
120	$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
156	undef $fh;
157	}
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	send, recv, peername, sockname, shutdown
172
173	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
176	=cut
177
178	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	sub sockname { getsockname tied(${$_[0]})->[0] }
186	sub peername { getpeername tied(${$_[0]})->[0] }
187	sub shutdown { shutdown tied(${$_[0]})->[0], $_[1] }
188
189	=item ($fh, $peername) = $listen_fh->accept
190
191	In scalar context, returns the newly accepted socket (or undef) and in
192	list context return the ($fh, $peername) pair (or nothing).
193
194	=cut
195
196	sub accept {
197	my ($peername, $fh);
198	while () {
199	$peername = accept $fh, tied(${$_[0]})->[0]
200	and return wantarray
201	? ($_[0]->new_from_fh($fh), $peername)
202	: $_[0]->new_from_fh($fh);
203
204	return unless $!{EAGAIN};
205
206	$_[0]->readable or return;
207	}
208	}
209
210	1;
211
212	=back
213
214	=head1 AUTHOR
215
216	Marc Lehmann <schmorp@schmorp.de>
217	http://home.schmorp.de/
218
219	=cut
220