Perl IO::Socket/IO::Select - 从 "ready-to-read"套接字读取

Question

这可能不是 Perl 特定的，但我的演示是用 Perl 编写的。

我的主程序打开一个监听套接字，然后派生一个子进程。 child 的首要任务是与主人联系并打招呼。然后它继续初始化，当它准备好时，它向主机发送READY。

主节点在 fork 子节点后，等待 HELLO，然后进行其他初始化(主要是 fork 其他子节点)。一旦它 fork 了所有的 child 并听到每个 child 的回复，它就会继续等待所有的 child 说“准备好”。

它使用 IO::Select->can_read 执行此操作，然后使用 $socket->getline 检索消息。

简而言之，父级无法收到 READY，即使它是由子级发送的。

这是我的程序的一个匆忙精简版本，演示了这个错误(我试图删除不相关的内容，但可能会留下一些)。我仍然对消息边界是否保留、是否需要“\n”以及使用哪种方法从套接字读取等问题感到困惑。我真的不想考虑组装消息片段，我希望 IO::Select 能让我省去这个麻烦。

为了简单起见，该演示仅生成一个子级。

#!/usr/bin/env perl 

use warnings;
use strict;
use Carp;
use File::Basename;
use IO::Socket;
use IO::Select;
use IO::File;                    # for CONSTANTS
use Net::hostent;                # for OO version of gethostbyaddr
use File::Spec qw{rel2abs};      # for getting path to this script
use POSIX qw{WNOHANG setsid};    # for daemonizing

use 5.010;

my $program    = basename $0;
my $progpath = File::Spec->rel2abs(__FILE__);
my $progdir  = dirname $progpath;

$| = 1;                          # flush STDOUT buffer regularly

# Set up a child-reaping subroutine for SIGCHLD.  Prevent zombies.
#
say "setting up sigchld";

$SIG{CHLD} = sub {
    local ( $!, $^E, $@ );
    while ( ( my $kid = waitpid( -1, WNOHANG ) ) > 0 ) {
        say "Reaping child process $kid";
    }
};

# Open a port for incoming connections
#
my $listen_socket = IO::Socket::INET->new(
    Proto     => 'tcp',
    LocalPort => 2000,
    Listen    => SOMAXCONN,
    Reuse     => 1
);
croak "Can't set up listening socket: $!\n" unless $listen_socket;

my $readers = IO::Select->new($listen_socket)
    or croak "Can't create the IO::Select read object";

say "Forking";

my $manager_pid;
if ( !defined( $manager_pid = fork ) ) {
    exit;
}
elsif ( 0 == $manager_pid ) {
    #
    # ------------------ BEGIN CHILD CODE HERE -------------------
    say "Child starting";

    my ($master_addr, $master_port) = split /:/, 'localhost:2000';

    my $master_socket = IO::Socket::INET->new(
        Proto    => "tcp",
        PeerAddr => $master_addr,
        PeerPort => $master_port,
    ) or die "Cannot connect to $master_addr:$master_port";

    say "Child sending HELLO.";

    $master_socket->printflush("HELLO\n");

    # Simulate elapsed time spent initializing...
    #
    say "Child sleeping for 1 second, pretending to be initializing ";

    sleep 1;
    #
    # Finished initializing.

    say "Child sending READY.";

    $master_socket->printflush("READY\n");
    say "Child sleeping indefinitely now.";

    sleep;
    exit;
    # ------------------- END CHILD CODE HERE --------------------
}

# Resume parent code

# The following blocks until we get a connect() from the manager

say "Parent blocking on ready readers";

my @ready = $readers->can_read;

my $handle;

for $handle (@ready) {
    if ( $handle eq $listen_socket ) {    #connect request?

        my $manager_socket = $listen_socket->accept();
        say "Parent accepting connection.";

        # The first message from the manager must be his greeting
        #
        my $greeting = $manager_socket->getline;
        chomp $greeting;
        say "Parent received $greeting";

    }
    else {
        say( $$, "This has to be a bug" );
    }
}

say "Parent will now wait until child sends a READY message.";
say "NOTE: if the bug works, Ill never receive the message!!";

################################################################################
#
# Wait until all managers have sent a 'READY' message to indicate they've
# finished initializing.
#
################################################################################

$readers->add($handle); # add the newly-established socket to the child

do {
    @ready = $readers->can_read;
    say "Parent is ignoring a signal." if !@ready;

} until @ready;

# a lot of overkill for demo

for my $socket (@ready) {
    if ( $socket ne $listen_socket ) {
        my $user_input;
        $user_input = $socket->getline;
        my $bytes = length $user_input;
        if ( $bytes > 0 ) {
            chomp $user_input;
            if ( $user_input eq 'READY' ) {
                say "Parent got $user_input!";
                $readers->remove($socket);
            }
            else {
                say( $$, "$program RECVS $user_input??" );
            }
        }
        else {
            say( $$, "$program RECVs zero length message? EOF?" );
            $readers->remove($socket);
        }
    }
    else {
        say( $$, "$program RECVS a connect on the listen socket??" );
    }
} # end for @ready
say "Parent is ready to sleep now.";

Answer 1

我不知道这是否是您的(唯一)问题，但始终将 sysread 与 select 一起使用。从未使用过像 getline 这样的缓冲 IO。 getline 毫无意义，因为它可能会阻止尚未收到的数据。

您的select循环应如下所示:

1. 永远，
   1. 等待套接字准备好读取。
   2. 对于每个准备读取的套接字，

      1.  sysread($that_socket, $buffer_for_that_socket, 64*1024,
              length($buffer_for_that_socket));
         

      2. 如果 sysread 返回 undef，

         1. 处理错误。

      3. 如果 sysread 返回 false，

         1. 处理关闭的套接字。不要忘记缓冲区中剩余的数据。

      4. 否则，处理读取数据:

         1.  while ($buffer_for_that_socket =~ s/^(.*)\n//) { my $msg = $1; ... }