lib/Sack/Server.pm

#!/usr/bin/perl

package Sack::Server;

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;
my $cloud_path = $ENV{CLOUD} || die "start with: CLOUD=etc/cloud perl -I/srv/Sack/lib $0\n";
@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}{Client};

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_ssh {
        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_ssh( $node_port++, $host );
}

my $session;

sub to_all {
        my $data = shift;
        foreach my $port ( keys %{ $session->{port} } ) {
                warn ">>>> [$port]\n";
                Storable::store_fd( $data, $session->{port}->{$port} );
        }
}

my @shard_paths;

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 ( my $repl = $data->{repl} ) {
                                        my $response = { repl_pid => $$ };
                                        if ( $repl =~ m/ping/ ) {
                                                to_all { ping => 1 };
                                        } elsif ( $repl =~ m/info/ ) {
                                                $response->{info} = $info;
                                        } elsif ( $repl =~ m{load\s*(\S+)?} ) {
                                                my $name = $1 || 'shard';
                                                @shard_paths = glob "/tmp/sack/$name/*";
                                                warn "loading shards ", dump( @shard_paths );
                                                to_all { load => $name };
                                        } else {
                                                $response->{error}->{unknown} = $data;
                                        }
                                        Storable::store_fd( $response, $sock );
                                } elsif ( $data->{ping} ) {
                                        my $port = $data->{port};
                                        info $port => 'ping', $port;
                                        $session->{port}->{ $data->{port} } = $sock;
                                } elsif ( $data->{load} eq 'shard' ) {
                                        if ( my $path = shift @shard_paths ) {
                                                warn "retrieve $path ", -s $path;
                                                my $shard = Storable::retrieve $path;
                                                warn ">>>> [", $data->{port}, "] sending shard $path\n";
                                                Storable::store_fd( { path => $path, shard => $shard }, $sock );
                                        } else {
                                                warn "no more shards for [", $data->{port}, "]\n";
                                        }
                                } else {
                                        warn "UNKNOWN ",dump($data);
                                }
                        }
                }
        }
}
1	dpavlin	163	#!/usr/bin/perl
2
3	dpavlin	179	package Sack::Server;
4
5	dpavlin	163	use warnings;
6			use strict;
7
8			use IO::Socket::INET;
9			use IO::Select;
10
11			use Data::Dump qw(dump);
12			use Storable qw();
13	dpavlin	169	use File::Slurp;
14	dpavlin	176	use Cwd qw(abs_path);
15	dpavlin	163
16	dpavlin	179	my @cloud;
17			my $cloud_path = $ENV{CLOUD} \|\| die "start with: CLOUD=etc/cloud perl -I/srv/Sack/lib $0\n";
18	dpavlin	169	@cloud = read_file $cloud_path;
19			@cloud = map { chomp $_; $_ } @cloud;
20
21			warn "# cloud ",dump( @cloud );
22
23	dpavlin	163	my $listen_port = 4444;
24
25	dpavlin	176	my $node_path = abs_path $0;
26	dpavlin	181	$node_path =~ s{Server}{Client};
27	dpavlin	163
28	dpavlin	169	my $lsn = IO::Socket::INET->new(Listen => 1, LocalPort => $listen_port, Reuse => 1) or die $!;
29	dpavlin	163	my $sel = IO::Select->new($lsn);
30
31	dpavlin	166	my $info;
32			sub info {
33			my $port = shift;
34			push @{ $info->{$port} }, [ @_ ];
35			}
36
37	dpavlin	179	sub fork_ssh {
38	dpavlin	163	my ( $port, $host ) = @_;
39
40			if ( my $pid = fork ) {
41			# parent
42	dpavlin	166	info $port => 'forked', $pid;
43	dpavlin	163	return $port;
44
45			} elsif ( ! defined $pid ) {
46			warn "can't fork $host $port";
47			return;
48			} else {
49			# child
50	dpavlin	169	my $cmd = qq\|ssh -F $cloud_path.ssh -R $port:127.0.0.1:$listen_port $host $node_path $port\|;
51	dpavlin	163	warn "# exec: $cmd\n";
52			exec $cmd;
53			}
54			}
55
56			my $node_port = 4000;
57
58	dpavlin	169	foreach my $host ( @cloud ) {
59	dpavlin	176	system "find /srv/Sack/ \| cpio --create --dereference \| ssh -T -F $cloud_path.ssh $host cpio --extract --make-directories --unconditional";
60	dpavlin	179	fork_ssh( $node_port++, $host );
61	dpavlin	163	}
62
63			my $session;
64
65	dpavlin	188	sub to_all {
66			my $data = shift;
67			foreach my $port ( keys %{ $session->{port} } ) {
68			warn ">>>> [$port]\n";
69			Storable::store_fd( $data, $session->{port}->{$port} );
70			}
71			}
72
73	dpavlin	192	my @shard_paths;
74
75	dpavlin	163	while (1) {
76			for my $sock ($sel->can_read(1)) {
77			if ($sock == $lsn) {
78			my $new = $lsn->accept;
79			$sel->add($new);
80	dpavlin	174	$session->{peerport}->{ $new->peerport } = $new;
81	dpavlin	164	warn "[socket] connect\n";
82	dpavlin	163	Storable::store_fd( { ping => 1 }, $new );
83	dpavlin	166	info 0 => 'ping', $new->peerport;
84	dpavlin	163	} else {
85	dpavlin	165	my $data = eval { Storable::fd_retrieve( $sock ) };
86			if ( $@ ) {
87			delete $session->{$sock};
88			warn "[socket] disconnect: $@\n";
89			$sel->remove($sock);
90			$sock->close;
91			} else {
92	dpavlin	163	warn "<<<< ", dump($data), $/;
93	dpavlin	192	if ( my $repl = $data->{repl} ) {
94			my $response = { repl_pid => $$ };
95			if ( $repl =~ m/ping/ ) {
96	dpavlin	188	to_all { ping => 1 };
97	dpavlin	192	} elsif ( $repl =~ m/info/ ) {
98	dpavlin	175	$response->{info} = $info;
99	dpavlin	192	} elsif ( $repl =~ m{load\s*(\S+)?} ) {
100			my $name = $1 \|\| 'shard';
101			@shard_paths = glob "/tmp/sack/$name/*";
102			warn "loading shards ", dump( @shard_paths );
103			to_all { load => $name };
104	dpavlin	189	} else {
105			$response->{error}->{unknown} = $data;
106	dpavlin	174	}
107	dpavlin	166	Storable::store_fd( $response, $sock );
108			} elsif ( $data->{ping} ) {
109	dpavlin	174	my $port = $data->{port};
110			info $port => 'ping', $port;
111			$session->{port}->{ $data->{port} } = $sock;
112	dpavlin	192	} elsif ( $data->{load} eq 'shard' ) {
113			if ( my $path = shift @shard_paths ) {
114			warn "retrieve $path ", -s $path;
115			my $shard = Storable::retrieve $path;
116			warn ">>>> [", $data->{port}, "] sending shard $path\n";
117			Storable::store_fd( { path => $path, shard => $shard }, $sock );
118			} else {
119			warn "no more shards for [", $data->{port}, "]\n";
120			}
121			} else {
122			warn "UNKNOWN ",dump($data);
123	dpavlin	164	}
124	dpavlin	163	}
125			}
126			}
127			}