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;

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

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

use Coro::Util ();

use base 'Coro::Handle';

$VERSION = 1.5;

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