experiments/protocol-v3/server.pl

#!/usr/bin/perl

use warnings;
use strict;

use IO::Socket::INET;
use IO::Select;

use Data::Dump qw(dump);
use Storable qw();

my $listen_port = 4444;

my $node_path = '/tmp/client.pl';

my $lsn = IO::Socket::INET->new(Listen => 1, LocalPort => $listen_port, Reuse => 1) or exit;
my $sel = IO::Select->new($lsn);

my $info;
sub info {
        my $port = shift;
        push @{ $info->{$port} }, [ @_ ];
}

sub fork_node {
        my ( $port, $host ) = @_;

        if ( my $pid = fork ) {
                # parent
                info $port => 'forked', $pid;
                return $port;

        } elsif ( ! defined $pid ) {
                warn "can't fork $host $port";
                return;
        } else {
                # child
                my $cmd = qq|ssh -R $port:127.0.0.1:$listen_port $host $node_path $port|;
                warn "# exec: $cmd\n";
                exec $cmd;
        }
}

my $node_port = 4000;

foreach my $host ( qw/localhost tab.lan llin.lan/ ) {
        system "scp client.pl $host:$node_path";
        fork_node( $node_port++, $host );
}

my $session;

while (1) {
        for my $sock ($sel->can_read(1)) {
                if ($sock == $lsn) {
                        my $new = $lsn->accept;
                        $sel->add($new);
                        $session->{$new} = $new->peerport;
                        warn "[socket] connect\n";
                        Storable::store_fd( { ping => 1 }, $new );
                        info 0 => 'ping', $new->peerport;
                } else {
                        my $data = eval { Storable::fd_retrieve( $sock ) };
                        if ( $@ ) {
                                delete $session->{$sock};
                                warn "[socket] disconnect: $@\n";
                                $sel->remove($sock);
                                $sock->close;
                        } else {
                                warn "<<<< ", dump($data), $/;
                                if ( $data->{repl} ) {
                                        my $response = { repl => $$, info => $info };
                                        Storable::store_fd( $response, $sock );
                                } elsif ( $data->{ping} ) {
                                        $info->{_peer_port}->{$sock->peerport} = $data->{port};
                                        info $data->{port} => 'peer port', $sock->peerport;
                                }
                        }
                }
        }
}
1	#!/usr/bin/perl
2
3	use warnings;
4	use strict;
5
6	use IO::Socket::INET;
7	use IO::Select;
8
9	use Data::Dump qw(dump);
10	use Storable qw();
11
12	my $listen_port = 4444;
13
14	my $node_path = '/tmp/client.pl';
15
16	my $lsn = IO::Socket::INET->new(Listen => 1, LocalPort => $listen_port, Reuse => 1) or exit;
17	my $sel = IO::Select->new($lsn);
18
19	my $info;
20	sub info {
21	my $port = shift;
22	push @{ $info->{$port} }, [ @_ ];
23	}
24
25	sub fork_node {
26	my ( $port, $host ) = @_;
27
28	if ( my $pid = fork ) {
29	# parent
30	info $port => 'forked', $pid;
31	return $port;
32
33	} elsif ( ! defined $pid ) {
34	warn "can't fork $host $port";
35	return;
36	} else {
37	# child
38	my $cmd = qq\|ssh -R $port:127.0.0.1:$listen_port $host $node_path $port\|;
39	warn "# exec: $cmd\n";
40	exec $cmd;
41	}
42	}
43
44	my $node_port = 4000;
45
46	foreach my $host ( qw/localhost tab.lan llin.lan/ ) {
47	system "scp client.pl $host:$node_path";
48	fork_node( $node_port++, $host );
49	}
50
51	my $session;
52
53	while (1) {
54	for my $sock ($sel->can_read(1)) {
55	if ($sock == $lsn) {
56	my $new = $lsn->accept;
57	$sel->add($new);
58	$session->{$new} = $new->peerport;
59	warn "[socket] connect\n";
60	Storable::store_fd( { ping => 1 }, $new );
61	info 0 => 'ping', $new->peerport;
62	} else {
63	my $data = eval { Storable::fd_retrieve( $sock ) };
64	if ( $@ ) {
65	delete $session->{$sock};
66	warn "[socket] disconnect: $@\n";
67	$sel->remove($sock);
68	$sock->close;
69	} else {
70	warn "<<<< ", dump($data), $/;
71	if ( $data->{repl} ) {
72	my $response = { repl => $$, info => $info };
73	Storable::store_fd( $response, $sock );
74	} elsif ( $data->{ping} ) {
75	$info->{_peer_port}->{$sock->peerport} = $data->{port};
76	info $data->{port} => 'peer port', $sock->peerport;
77	}
78	}
79	}
80	}
81	}