TCP server that listens on multiple processes: accept succeeds only on one process for a connection.

paraserv.pl

#!/usr/bin/env perl

use strict;
use warnings;

use Socket;
use IO::Socket;
use IO::Socket::INET;
use IO::Select;
use IO::Handle;
use POSIX ":sys_wait_h";
use Time::HiRes qw(usleep);

our $port = 5001;
our $listen_backlog = 10;

our $concurrency = 5;

if ($0 eq __FILE__) {
    main();
}

sub main {
    my $sock = bind_server();
    my $fd = fileno $sock;
    my %children;
    my $parent_pid = $$;
    my ($child, $parent);
    socketpair $child, $parent, AF_UNIX, SOCK_STREAM, PF_UNSPEC;
    for my $i (0 .. $concurrency - 1) {
        fork_child(\%children, \$parent, \$child, $fd);
    }
    do {
        my $pid = waitpid(-1, WNOHANG);
        if ($pid > 0) {
            warn "Process $pid finished";
            delete $children{$pid};
            fork_child(\%children, \$parent, \$child, $fd);
        }
        usleep(1000);
    } while (%children);
}

sub fork_child {
    my ($children, $parent, $child, $fd) = @_;
    my $child_pid = fork;
    if ($child_pid) {
        close $$parent;
        $children->{$child_pid} = { pid => $child_pid, sock => $$child, };
        return;
    }
    close $$child;
    my $nsocket = prepare_parents_fd($fd);
    worker_main($nsocket, $$parent);
    exit 0;
}

sub prepare_parents_fd {
    my $orig_fd = shift;
    my $socket;
    open $socket, "<&=", $orig_fd or die;
    return $socket;
}

sub worker_main {
    my ($listen_sock, $parent_sock) = @_;
    print STDERR "main $$\n";
    listen $listen_sock, $listen_backlog;
    my $sel = IO::Select->new;
    my @established = ();
    $sel->add($listen_sock);
    while(1) {
        my @ready = $sel->can_read();
        #warn "ready @ready";
        for my $sock (@ready) {
            if ($sock == $listen_sock) {
                my $new_conn;
                accept $new_conn, $listen_sock;
                warn "Accepted $$";
                push @established, $new_conn;
                $sel->add($new_conn);
            } elsif (grep { $_ == $sock } @established) {
                warn "Received $$";
                process_data($sock);
                $sel->add($sock);
            } elsif ($sock == $parent_sock) {
                warn "Currently " . scalar @established . " connections";
                print $sock pack("i", scalar @established);
                $sel->add($sock);
            }
        }
    }
}

sub process_data {
    my $sock = shift;
    my $data = <$sock>;
    warn $data;
    exit(0);  # dummy exit
}

# Bind a socket to the local port and returns the socket
sub bind_server {
    my $addr = "0.0.0.0";
    my $socket = IO::Socket::INET->new(
        LocalAddr => $addr,
        LocalPort => $port,
        Proto => "tcp",
        ReuseAddr => 1,
    ) or die "Failed to bind";
    return $socket;
}