How to use AnyEvent :: A socket handler that uses port reuse

Question

Recently, I came across a large perl module, "AnyEvent", which allows the user to run an asynchronous / event-driven program.

The following snippet is created, which works great. The problem is that after it opens and closes many sockets, it quickly exhausted all the client ports ("netstat - ant" indicates that 20,000+ sockets are in TIME_WAIT state).

$hdl = new AnyEvent::Handle ( connect => [$ip, $port], on_connect=> sub { my ($handle, $host, $port, $tmp) = @_; #print "connect routine for $handle->{ue}\r\n"; #update states. }, on_read => sub { my $hdl = $_[0]; #read data #send response. }); 

I wonder if it is possible to create a TCP socket with IO :: Socket :: INET, and then use the newly created socket in AnyEvent :: Handle:

 my $sock = IO::Socket::INET->new( Proto => 'tcp', PeerAddr => $ue->{vars}->{ip}, PeerPort => $ue->{vars}->{dstPort}, ReusePort => 1, KeepAlive => 1 ) || die "failed to setup outsock $@ \n"; $hdl = new AnyEvent::Handle ( fh => $sock, on_connect=> sub { my ($handle, $host, $port, $tmp) = @_; #print "connect routine for $handle->{ue}\r\n"; #update states. }, on_read => sub { my $hdl = $_[0]; #read data #send response. }); 

I tried, but it did not work. Please rate any suggestions and comments.

Thanks to ikegami who looked into it and gave an offer. However, it looks like SO_REUSEADDR is not taking effect. Here is the code I used (at his suggestion)

 use strict; use warnings; use AnyEvent qw( ); use AnyEvent::Handle qw( ); use AnyEvent::Impl::EV qw( ); use AnyEvent::Socket qw( tcp_connect ); use Socket qw( SOL_SOCKET SO_REUSEPORT SO_REUSEADDR); my $ts = 0; my $trans = 0; my $currentTS; sub transaction { my ($host, $port) = @_; tcp_connect($host, $port, sub { my ($sock) = @_ or die "Can't connect: $!"; my $handle; $handle = AnyEvent::Handle->new( fh => $sock, on_eof => sub { $handle->destroy(); }, on_read => sub { my ($handle) = @_; #print $handle->rbuf(); $trans ++; $currentTS = time(); if ($currentTS > $ts) { $ts = $currentTS; print "$trans\n"; } #printf "recved %d bytes of data\n", length($handle->rbuf); # This should continue to read until header + # Content-Length bytes have been read instead # of stopping after one read. if (length($handle->rbuf) > 0) { $handle->destroy(); } }, ); $handle->push_write("GET /s HTTP/1.1\r\nHost: $host\r\n\r\n"); #$handle->push_shutdown(); # Done writing. }, sub { my ($sock) = @_; #setsockopt($sock, SOL_SOCKET, SO_REUSEPORT, 1) or die $!; setsockopt($sock, SOL_SOCKET, SO_REUSEADDR, 1) or die $!; # die "failed to set linger $!\n"; return undef; }); } { my $cv = AnyEvent->condvar(); my $t = AnyEvent->timer(after=>0.001, interval=>1, cb=> sub { transaction("10.3.0.6", 80 ); }); $cv->recv(); } 

My system is Ubuntu 11.04. In the / proc / sys / net / ipv 4 directory, here are the contents of the two files:

% more than tcp_tw_recycle

1

% more than tcp_tw_reuse

1

Answer 1

I cannot run the following because Windows does not provide SO_REUSEPORT , but I am very sure that the following does what you requested.

However, I am not sure if this will help. From what I read, SO_REUSEPORT allows you to bind to an already active port, but you don't bind to any port.

 use strict; use warnings; use AnyEvent qw( ); use AnyEvent::Handle qw( ); use AnyEvent::Impl::EV qw( ); use AnyEvent::Socket qw( tcp_connect ); use Socket qw( SOL_SOCKET SO_REUSEPORT ); 

 sub transaction { my ($host, $port) = @_; tcp_connect($host, $port, sub { my ($sock) = @_ or die "Can't connect: $!"; my $handle; $handle = AnyEvent::Handle->new( fh => $sock, on_eof => sub { $handle->destroy(); }, on_read => sub { my ($handle) = @_; print $handle->rbuf(); # This should continue to read until header + # Content-Length bytes have been read instead # of stopping after one read. $handle->destroy(); }, ); $handle->push_write("GET / HTTP/1.1\r\nHost: $host\r\n\r\n"); $handle->push_shutdown(); # Done writing. }, sub { my ($sock) = @_; setsockopt($sock, SOL_SOCKET, SO_REUSEPORT, 1) or die $!; return undef; }); } 

 { my $cv = AnyEvent->condvar(); my $t = AnyEvent->timer(after=>0.001, interval=>0.001, cb=> sub { transaction("localhost", $ARGV[0] // die("usage")); }); $cv->recv(); } 

Server used for testing:

 use strict; use warnings; use 5.010; use IO::Socket::INET qw( ); use Socket qw( inet_ntoa ); my $serv = IO::Socket::INET->new( Listen => 1, ); say inet_ntoa($serv->sockaddr) . ":" . $serv->sockport; while (my $client = $serv->accept()) { say "Connection from ".inet_ntoa($client->peeraddr).":".$client->peerport; while (<$client>) { last if /^(?:\r?\n)?\z/; } say $client "HTTP/1.1 200 OK\r\n" . "Content-Type: text/plain\r\n" . "\r\n" . "Hello\n"; say " done."; } 

Answer 2

What you do cannot be done using TCP / IP - what happens is that you start the protection mechanism (TIME_WAIT state), which ensures that tuples (srcip, srcport, dstip, dstport) are no longer used, neither REUSEADDR, neither REUSEPORT nor IO :: Socket can help here - this is a protocol limitation.

The only legal solution is not to create identical tuples too quickly, for example, using different source IP addresses or reusing existing tcp codes for a larger transaction. Or even use multiple destination ports or IP addresses — anything that makes tuples more “unique” will help.

If you are desperate, you can enable unsafe port reuse in your OS, but this poses a real risk of data corruption, so don’t do this when other people’s data is used :)

And, as far as I can see, the fragments of the ikegamis example are pretty good and show how to "import" sockets from IO :: Socket and how to use tcp_connect to bind a socket. If you decide to use multiple source IP addresses, then its code shows how to bind to them.