Re^2: poorly documented behaviour of readline() and IO::Select

[vsespb]

Just wrote a POC solution which seems to fix issue with sysread/syswrite and specifying length of the message in the beginning of it (of course this solution is suitable if you can patch both client and server i.e. modify your protocol).

It works fine (linux ubuntu).

#!/usr/bin/perl -w 

use strict; 
use IO::Select; 
use IO::Pipe;

  my $fromchild = new IO::Pipe;
  my $tochild = new IO::Pipe;
  my $pid;
  my $parent_pid = $$;
  if($pid = fork()) { # Parent
   $fromchild->reader();
   $fromchild->autoflush(1);
   $fromchild->blocking(1);
   binmode $fromchild;
   $tochild->writer();
   $tochild->autoflush(1);
   $tochild->blocking(1);
   binmode $tochild;

   my $read_set = new IO::Select(); # create handle set for reading
   $read_set->add($fromchild);

   while(1) {
    print "before select\n";
    my ($rh_set, undef, $ex_set) = IO::Select->select($read_set, undef
+, $read_set, 30);
    print "after select\n";
    for my $rh (@$rh_set) {
        my $s =  receive_line($rh);
        print "command: $s";
    }
   }
  } elsif (defined ($pid)) { # Child
   $fromchild->writer();
   $fromchild->autoflush(1);
   $fromchild->blocking(1);
   binmode $fromchild;
   $tochild->reader();
   $tochild->autoflush(1);
   $tochild->blocking(1);
   binmode $tochild;
   send_line($fromchild, "abc\n");
   send_line($fromchild, "def\n");

   sleep(86400);
   die;
  }

sub send_line
{
  my ($socket, $line) = @_;
  my $msg = sprintf("%07d %s", length($line), $line);
  syswrite $socket, $msg;
}

sub receive_line
{
  my ($socket) = @_;
  sysread $socket, my $len, 8;
  sysread $socket, my $line, $len, 8;
  return $line;
}


__END__

Output:
before select
after select
command: abc
before select
after select
command: def
before select

[ikegami]

sysread won't always give you the amount of bytes you request when reading from something that isn't a plain file.

See Re: A suicidal parent OR death of a forking server for a newline terminated solution, and here's a length-prefix adaptation:

#!/usr/bin/perl

use strict;
use warnings;

use IO::Socket::INET qw( );
use IO::Select       qw( );


sub process_msg {
    my ($client, $msg) = @_;
    chomp $msg;
    print "$client->{host} said '$msg'\n";
}


sub process_msgs {
    my ($client) = @_;
    our $buf;  local *buf  = \($client->{buf});
    our $want; local *want = \($client->{want});

    for (;;) {
        if ($want) {
            return if length($buf) < $want;
            my $msg = substr($buf, 0, $want, '');
            $want = 0;
            process_msg($client, $msg);
        } else {
            return if length($buf) < 8;
            $want = 0+substr($buf, 0, 8, '');
        }
    }
}


my $server = IO::Socket::INET->new(
    ...
) or die("Couldn't create server socket: $!\n");

my $select = IO::Select->new($server);

my %clients;

while (my @ready = $select->can_read) {
    for my $fh (@ready) {
        if ($fh == $server) {
            my $client_sock = $server->accept;
            my $host = $client_sock->peerhost;
            print "[Accepted connection from $host]\n";
            $select->add($client_sock);
            $clients{fileno($client_sock)} = {
                host => $host,
                buf  => '',
                want => 0,
            };
        }
        else {
            my $client = $clients{fileno($fh)};
            our $buf;  local *buf  = \($client->{buf});
            our $want; local *want = \($client->{want});

            my $rv = sysread($fh, $buf, 64*1024, length($buf));
            if (!$rv) {
                my $host = $client->{host}fh->peerhost;
                if (defined($rv)) {
                    print "[Error reading from host $host]\n";
                } else {
                    print "[Connection from $host terminated]\n";
                }
                process_msgs($client);
                print "Incomplete message received from $host]\n"
                    if $want || length($buf);
                delete $clients{fileno($fh)};
                $sel->remove($fh);
                next;
            }

            process_msgs($client);
        }
    }
}

[ikegami]

Instead of implementing a limited cooperative multitasking system using select, it's simpler to use threads or Coro.

The following is the Coro equivalent of the code in the parent post.

#!/usr/bin/perl

use strict;
use warnings;

use Coro             qw( async );
use IO::Socket::INET qw( );

sub _read {
    my $rv = read($_[0], $_[1], $_[2], $_[3]||0);
    die("Read error\n") if !defined($rv);
    return if !$rv;
    die("Premature end of file\n") if $rv != $_[1];
}

sub client {
    my ($sock) = @_;

    my $host = $client->peerhost;
    print "[Accepted connection from $host]\n";

    if (defined(eval {
        for (;;) {
            _read($sock, 8, my $buf = '')
                or last;
            my $length = 0+$buf
                or next;
            _read($sock, $length, my $msg = '')
                or die("Premature end of file\n");
            print "$host said '$msg'\n";
        }
    })) {
        print "[Connection from $host terminated]\n";
    } else {
        print "[Connection from $host terminated: $@]\n";
    }
}

my $server = IO::Socket::INET->new(
    ...
) or die("Couldn't create server socket: $!\n");

async(\&client, $server->accept) while 1;

[BrowserUk]

Yes. I've always favoured length prefixes to delimiters in protocols I've written--mostly RS-232, RS-432 and some short range wireless hand-held devices (think barcode scanners in supermarkets but 15+years ago).

But that doesn't really fit with the *nix, everything-is-a-file way of working.