lib/Sack/Server.pm

#!/usr/bin/perl

use warnings;
use strict;

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

use Data::Dump qw(dump);
use Storable qw();
use File::Slurp;
use Cwd qw(abs_path);

my @cloud = qw(localhost tab.lan llin.lan);

my $cloud_path = $ENV{CLOUD} || '/srv/Sack/etc/lib';
@cloud = read_file $cloud_path;
@cloud = map { chomp $_; $_ } @cloud;

warn "# cloud ",dump( @cloud );

my $listen_port = 4444;

my $node_path = abs_path $0;
$node_path =~ s{server.pl}{client.pl};

my $lsn = IO::Socket::INET->new(Listen => 1, LocalPort => $listen_port, Reuse => 1) or die $!;
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 -F $cloud_path.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 ( @cloud ) {
        system "find /srv/Sack/ | cpio --create --dereference | ssh -T -F $cloud_path.ssh $host cpio --extract --make-directories --unconditional";
        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->{peerport}->{ $new->peerport } = $new;
                        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 => $$ };
                                        if ( $data->{repl} =~ m/ping/ ) {
                                                foreach my $port ( keys %{ $session->{port} } ) {
                                                        warn ">>>> [$port]\n";
                                                        Storable::store_fd( { ping => 1 }, $session->{port}->{$port} );
                                                }
                                        } elsif ( $data->{repl} =~ m/info/ ) {
                                                $response->{info} = $info;
                                        }
                                        Storable::store_fd( $response, $sock );
                                } elsif ( $data->{ping} ) {
                                        my $port = $data->{port};
                                        info $port => 'ping', $port;
                                        $session->{port}->{ $data->{port} } = $sock;
                                }
                        }
                }
        }
}
1	dpavlin	163	#!/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	dpavlin	169	use File::Slurp;
12	dpavlin	176	use Cwd qw(abs_path);
13	dpavlin	163
14	dpavlin	169	my @cloud = qw(localhost tab.lan llin.lan);
15
16	dpavlin	170	my $cloud_path = $ENV{CLOUD} \|\| '/srv/Sack/etc/lib';
17	dpavlin	169	@cloud = read_file $cloud_path;
18			@cloud = map { chomp $_; $_ } @cloud;
19
20			warn "# cloud ",dump( @cloud );
21
22	dpavlin	163	my $listen_port = 4444;
23
24	dpavlin	176	my $node_path = abs_path $0;
25	dpavlin	173	$node_path =~ s{server.pl}{client.pl};
26	dpavlin	163
27	dpavlin	169	my $lsn = IO::Socket::INET->new(Listen => 1, LocalPort => $listen_port, Reuse => 1) or die $!;
28	dpavlin	163	my $sel = IO::Select->new($lsn);
29
30	dpavlin	166	my $info;
31			sub info {
32			my $port = shift;
33			push @{ $info->{$port} }, [ @_ ];
34			}
35
36	dpavlin	163	sub fork_node {
37			my ( $port, $host ) = @_;
38
39			if ( my $pid = fork ) {
40			# parent
41	dpavlin	166	info $port => 'forked', $pid;
42	dpavlin	163	return $port;
43
44			} elsif ( ! defined $pid ) {
45			warn "can't fork $host $port";
46			return;
47			} else {
48			# child
49	dpavlin	169	my $cmd = qq\|ssh -F $cloud_path.ssh -R $port:127.0.0.1:$listen_port $host $node_path $port\|;
50	dpavlin	163	warn "# exec: $cmd\n";
51			exec $cmd;
52			}
53			}
54
55			my $node_port = 4000;
56
57	dpavlin	169	foreach my $host ( @cloud ) {
58	dpavlin	176	system "find /srv/Sack/ \| cpio --create --dereference \| ssh -T -F $cloud_path.ssh $host cpio --extract --make-directories --unconditional";
59	dpavlin	163	fork_node( $node_port++, $host );
60			}
61
62			my $session;
63
64			while (1) {
65			for my $sock ($sel->can_read(1)) {
66			if ($sock == $lsn) {
67			my $new = $lsn->accept;
68			$sel->add($new);
69	dpavlin	174	$session->{peerport}->{ $new->peerport } = $new;
70	dpavlin	164	warn "[socket] connect\n";
71	dpavlin	163	Storable::store_fd( { ping => 1 }, $new );
72	dpavlin	166	info 0 => 'ping', $new->peerport;
73	dpavlin	163	} else {
74	dpavlin	165	my $data = eval { Storable::fd_retrieve( $sock ) };
75			if ( $@ ) {
76			delete $session->{$sock};
77			warn "[socket] disconnect: $@\n";
78			$sel->remove($sock);
79			$sock->close;
80			} else {
81	dpavlin	163	warn "<<<< ", dump($data), $/;
82	dpavlin	164	if ( $data->{repl} ) {
83	dpavlin	175	my $response = { repl => $$ };
84	dpavlin	174	if ( $data->{repl} =~ m/ping/ ) {
85			foreach my $port ( keys %{ $session->{port} } ) {
86			warn ">>>> [$port]\n";
87			Storable::store_fd( { ping => 1 }, $session->{port}->{$port} );
88			}
89	dpavlin	175	} elsif ( $data->{repl} =~ m/info/ ) {
90			$response->{info} = $info;
91	dpavlin	174	}
92	dpavlin	166	Storable::store_fd( $response, $sock );
93			} elsif ( $data->{ping} ) {
94	dpavlin	174	my $port = $data->{port};
95			info $port => 'ping', $port;
96			$session->{port}->{ $data->{port} } = $sock;
97	dpavlin	164	}
98	dpavlin	163	}
99			}
100			}
101			}