Nonblocking read server

Carbonblack has asked for the wisdom of the Perl Monks concerning the following question:

Dear monks of pearl wisdom,

I would like to program a small server that receives messages, processes them and forwards them to another system.

Because the server should run on OpenWRT, I'm limited to the standard libraries.

After reading what if had found on the net I still have a few questions about using sockets (correctly).

Is there a way to tell how much data ( e.g. bytes ) are available to read for a simple read without blocking?
Is there a reliable way to tell that the client has colsed the connection (and all data ) is read?
What kind of errors can happen? Are there elegant ways to handle them?
$socket->atmark returns (always) '0 but true' and if I read till sysread returns 0 $socket->eof is always (?) true.
Sometimes only the first READ_LENGTH bytes are read from the first client buffer-flush (?)/ packet (?). All subsequent packets(?) are read correctly. This always happens if i omit the while( $rr = sysread($sock,$buf, READ_LENGTH, 0 ) ) loop. How can i prevent data-loss? As far as i know i will not be able to ask the client-application to resend the data :-(

Please ave a look at my code. Any suggestions welcome.

#!/usr/bin/perl
use strict; use warnings;
#tcpserver.pl
use IO::Select;
use IO::Socket;
use Data::Dumper;
use Fcntl;

use constant PORT1 => 5000;
use constant TIMEOUT => 10;
use constant READ_LENGTH => 10; 
# Small to see if "buffer overflow" is handled correctly

my $server_socket =  new IO::Socket::INET(
      Listen => 1, 
      LocalPort => PORT1, 
      ReuseAddr => 1) 
    or die "Can't bind server_socket: $@\n";

my $sel = IO::Select->new;
$sel->add($server_socket);

my %connections; # Infos and Buffers for active connections
my $connID;

while(1)
{
  { # Can be used to limit the number of concurrent connections
    my $n = ( keys %connections ); 
    print "$n acive connections",Dumper(\%connections);
  }
  foreach my $sock (  $sel->can_read(TIMEOUT) ) 
  {
    print "sock:",( defined($connections{$sock}) ? 
      $connections{$sock}{'id'} : 
      ( $sock == $server_socket ? "server" : $sock 
    )),"\n";
    if( $sock == $server_socket ) # New connection
    {
      my $new = $server_socket->accept;
      binmode $new;
      my $flags = fcntl($new,  F_GETFL, 0) or 
          die "[new Err] Can't get flags !$\n";
      fcntl($new , F_SETFL, $flags | O_NONBLOCK ) or 
          die "[new Err] Can't set flags !$\n"; # For nonblocking read

      $sel->add($new);
      $connections{$new} = { 
        ip => $new->peerhost, 
        id => ++$connID, 
        buf => ""
      };
      print "server_socket->new($connections{$new}{'ip'}) Nr:$connecti
+ons{$new}{'id'}\n";
      next;
    }
    my ($rr, $r, $id) = (0, 0, $connections{$sock}{'id'});
    my $buffer = $connections{$sock}{'buf'} ;
    my $buf;
    # $sock needs to be O_NONBLOCK
    while( $rr = sysread($sock,$buf, READ_LENGTH, 0 ) ) 
    { print "[read ", ( $r // "undef" ), " ] '$buf'", "\n";
      if( ! defined( $rr) ) # When does this happen?
      { # Error handler 
        print "[Error $connections{$sock}{'ip'} id: $id]: $!\n";
        # close connections?
        last;
      }
      $buffer .= $buf;
      $r += $rr;
    }
    $connections{$sock}{'buf'} = $buffer;
    print "[Buffer($id)] '$buffer'\n";
    if(0 and $sock->eof ) 
    {
      print "[EOF($id)] ";
      process_message( $sock);
      delete $connections{$sock};
      $sel->remove($sock);
      $sock->close;
      next;
    }
    if ( ! $r )
    {
      print "[empty read($id)]\n";
      process_message( $sock);
      delete $connections{$sock};
      $sel->remove($sock);
      $sock->close;
      next;
    }
  }
}


sub process_message
{
  my ( $sock ) = @_;
  my $filename = "rcv_$connections{$sock}{'id'}".
     "_$connections{$sock}{'ip'}.txt" ;
  open OUT, ">", $filename
    or die "Can not write $filename: $!";
  binmode OUT;
  print OUT $connections{$sock}{'buf'};
  close OUT;
  print "[save process_message] wrote $filename \n";
}
[download]

Comment on Nonblocking read server Select or Download Code

Replies are listed 'Best First'.
Re: Nonblocking read server by tybalt89 (Monsignor) on Aug 29, 2019 at 21:00 UTC
Suggestions: Use an Async package if you can. Do not do non-blocking. Only do one sysread per ->can_read (thus non-blocking not needed) sysread return 0 indicates closed connection. Don't worry about how much data can be read, just sysread some big size, and sysread will give you what it can. Forget ->atmark and ->eof.	[reply]
Re^2: Nonblocking read server by Carbonblack (Novice) on Aug 30, 2019 at 06:55 UTC
Thank You very much for your advice! That was the answer i feared ;-) But it makes it a lot easier. I think, I'll avoid fork or Async etc. this time since i expect only a one way communication and i want to use as less resources as possible.	[reply]
Re^3: Nonblocking read server by tybalt89 (Monsignor) on Aug 30, 2019 at 14:41 UTC
Just to show you why I recommended an Async package, here's your problem done using my own Async::Tiny. It should be similarly short in any of the other Async packages. #!/usr/bin/perl use strict; use warnings; use Async::Tiny; use Path::Tiny; use constant PORT1 => 5000; my $connID; my $t = Async::Tiny->new; $t->addListenCallback( PORT1, sub { my $sock = shift; $t->addReadCallback($sock, \&process_message, $sock->peerhost, ++$co +nnID); $t->changeReadMode($sock, 'full'); }); $t->eventloop; sub process_message { my ($data, $peerhost, $id) = @_; my $filename = "rcv_${id}_$peerhost.txt"; path($filename)->spew_raw($data); print "[save process_message] wrote $filename\n"; } [download] See - nice, simple, clean, short :) Where Async::Tiny is Read more... (20 kB)	[reply] [d/l] [select]
Re^4: Nonblocking read server by choroba (Cardinal) on Aug 30, 2019 at 15:03 UTC
Re^5: Nonblocking read server by tybalt89 (Monsignor) on Aug 30, 2019 at 15:22 UTC
Some notes below your chosen depth have not been shown here
Re: Nonblocking read server by tybalt89 (Monsignor) on Aug 30, 2019 at 12:18 UTC
I was bored, so I decided to try to fix your server. Seems to work fine, I don't see any data-loss in the testing I tried. #!/usr/bin/perl # https://perlmonks.org/?node_id=11105232 #!/usr/bin/perl use strict; use warnings; #tcpserver.pl use IO::Select; use IO::Socket; use Data::Dumper; #use Fcntl; use constant PORT1 => 5000; use constant TIMEOUT => 10; use constant READ_LENGTH => 10; # Small to see if "buffer overflow" is handled correctly my $server_socket = new IO::Socket::INET( Listen => 1, LocalPort => PORT1, ReuseAddr => 1) or die "Can't bind server_socket: $@\n"; my $sel = IO::Select->new; $sel->add($server_socket); my %connections; # Infos and Buffers for active connections my $connID; while(1) { { # Can be used to limit the number of concurrent connections my $n = ( keys %connections ); print "$n acive connections",Dumper(\%connections); } foreach my $sock ( $sel->can_read(TIMEOUT) ) { print "sock:",( defined($connections{$sock}) ? $connections{$sock}{'id'} : ( $sock == $server_socket ? "server" : $sock )),"\n"; if( $sock == $server_socket ) # New connection { my $new = $server_socket->accept; # binmode $new; # my $flags = fcntl($new, F_GETFL, 0) or # die "[new Err] Can't get flags !$\n"; # fcntl($new , F_SETFL, $flags \| O_NONBLOCK ) or # die "[new Err] Can't set flags !$\n"; # For nonblocking rea +d $sel->add($new); $connections{$new} = { ip => $new->peerhost, id => ++$connID, buf => "" }; print "server_socket->new($connections{$new}{'ip'}) Nr:$connecti +ons{$new}{'id'}\n"; next; } my ($rr, $r, $id) = (0, 0, $connections{$sock}{'id'}); my $buffer = $connections{$sock}{'buf'} ; my $buf; # $sock needs to be O_NONBLOCK # while( $rr = sysread($sock,$buf, READ_LENGTH, 0 ) ) if( $rr = sysread($sock,$buf, READ_LENGTH, 0 ) ) { print "[read ", ( $r // "undef" ), " ] '$buf'", "\n"; if( ! defined( $rr) ) # When does this happen? { # Error handler print "[Error $connections{$sock}{'ip'} id: $id]: $!\n"; # close connections? last; } $buffer .= $buf; $r += $rr; } $connections{$sock}{'buf'} = $buffer; print "[Buffer($id)] '$buffer'\n"; # if(0 and $sock->eof ) # { # print "[EOF($id)] "; # process_message( $sock); # delete $connections{$sock}; # $sel->remove($sock); # $sock->close; # next; # } if ( ! $r ) { print "[empty read($id)]\n"; process_message( $sock); delete $connections{$sock}; $sel->remove($sock); $sock->close; next; } } } sub process_message { my ( $sock ) = @_; my $filename = "rcv_$connections{$sock}{'id'}". "_$connections{$sock}{'ip'}.txt" ; open OUT, ">", $filename or die "Can not write $filename: $!"; binmode OUT; print OUT $connections{$sock}{'buf'}; close OUT; print "[save process_message] wrote $filename \n"; } [download]	[reply] [d/l]
Re: Nonblocking read server by Anonymous Monk on Aug 31, 2019 at 12:58 UTC
If you don't block or use async notifications you are "busy-waiting" and that's very bad: sucking up 100% of the CPU doing nothing. If your program's reason for existence is to read from any of one-or-more sockets and to write its output somewhere else, why not block? It becomes a classic "select()" scenario. After all, your program has nothing to do until the next message(s) come in.	[reply]
Re^2: Nonblocking read server by Carbonblack (Novice) on Sep 19, 2019 at 15:30 UTC
Thank You very much for Your efford! I'll give it a try... Because system the script will run is an alix-hardware with openWrt it's quite hard to install fancy modules. Luckily select does not wait too busy if timeout is > 0. `time perl Inet.Server6.pl 0 Active connections 0 Active connections real 1m1,987s user 0m0,023s sys 0m0,013s` [download] Well, if i play with this script and two instances of`cat - \| nc -q 1 localhost 5000` the second line seems to vanish if the other instance is blocking the IO. I have no data-loss if the read-size is large enough and with non-blocking IO.	[reply] [d/l] [select]


good chemistry is complicated, and a little bit messy -LW
	PerlMonks