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);

use lib '/srv/Sack/lib';
use Sack::Merge;
use Sack::Server::HTTP;
use Sack::Server::HTML;

my @cloud;
my $cloud_path = $ENV{CLOUD} || "etc/cloud";
die "start with: CLOUD=etc/cloud perl -I/srv/Sack/lib $0\n" unless -e $cloud_path;
@cloud = read_file $cloud_path;
@cloud = map { chomp $_; $_ } @cloud;

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

my $listen_port = 4444;
my $http_port   = 4480;

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 "$listen_port $!";
my $www = IO::Socket::INET->new(Listen => 1, LocalPort => $http_port,   Reuse => 1) or die "$http_port $!";
my $sel = IO::Select->new($lsn);
$sel->add( $www );

my $info;
sub info {
        my $port = shift;
        push @{ $info->{node}->{$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 );
                } elsif ( $sock == $www ) {
                        my $client = $www->accept;
                        Sack::Server::HTTP::request( $client, sub {
                                my ( $send, $method, $param ) = @_;

                                if ( $method =~ m{views} ) {
                                        warn "$method ", -s $method, " bytes";
                                        my $code = read_file $method;
                                        Sack::Merge->clean;
                                        to_all { view => $code, path => $method };
                                        print $send "HTTP/1.0 302 $method\r\nLocation: /\r\n\r\n";
                                        return 1;
                                } elsif ( $method =~ m{^/tmp/sack} ) {
                                        @shard_paths = glob "$method/*";
                                        warn "loading shard $method from ", dump( @shard_paths );
                                        to_all { load => $method };
                                        print $send "HTTP/1.0 302 $method\r\nLocation: /\r\n\r\n";
                                        return 1;
                                } elsif ( $method =~ m{^/out/(.+)} ) {
                                        print $send "HTTP/1.0 200 OK\r\nContent-Type: text/html\r\n\r\n";
                                        Sack::Server::HTML::send_out( $send, Sack::Merge->out, $1, $param );
                                        return 1;
                                }

                                print $send "HTTP/1.0 200 OK\r\nContent-Type: text/html\r\n\r\n";

                                print $send qq|<h1>Views</h1><ul>|
                                        , join("\n", map { qq|<li><a href="$_">$_</a>| } glob '/srv/Sack/views/*/*.pl' )
                                        , qq|</ul>|
                                        ;

                                my $out = Sack::Merge->out;     
                                print $send qq|<h1>Results</h1><ul>|
                                        , join("\n", map { qq|<li><a href="/out/$_" target="$_">$_</a>| } keys %$out )
                                        , qq|</ul>|
                                        ;

                                print $send qq|<h1>Nodes</h1>
<style type="text/css">
.forked {
        color: #f00;
}

.ping {
        color: #0f0;
}
</style>
                                |, join("\n", map {
                                        my $class = join(' ', map { $_->[0] } @{ $info->{node}->{$_} });
                                        qq|<span class="$class">$_</span>|;
                                } sort keys %{ $info->{node} } );

                                print $send qq|<h1>Data</h1><ul>|
                                        , join("\n", map { qq|<li><a href="$_">$_</a>| } glob '/tmp/sack/*' )
                                        , qq|</ul>|
                                        ;

                                print $send '<pre>', dump($info), '</pre>';

                        } );
                } 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 $path = $data->{shard} ) {
                                        $info->{shard}->{ $path } = $data->{port};
                                        # FIXME will need push for multiple copies of shards
                                }

                                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 shard $name from ", dump( @shard_paths );
                                                to_all { load => $name };
                                        } elsif ( $repl =~ m{view\s*(\S+)?} ) {
                                                my $path = $1 || '/srv/Sack/views/00.demo.pl';
                                                my $code = read_file $path;
                                                Sack::Merge->clean;
                                                to_all { view => $code, path => $path };
                                                $response->{view}->{$path}->{running};
                                        } elsif ( $repl =~ m{debug\s*(.+)?} ) {
                                                to_all { debug => $1 };
                                        } elsif ( $repl =~ m{out} ) {
                                                my $out = Sack::Merge->out;     
                                                warn "out ",dump( $out );
                                                $response->{out} = $out;
                                        } elsif ( $repl =~ m{clean} ) {
                                                delete $info->{shard};
                                                to_all { clean => 1 };
                                        } 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";
                                        }
                                } elsif ( defined $data->{out} ) {
                                        Sack::Merge->add( $data->{out} );
                                } else {
                                        warn "UNKNOWN ",dump($data);
                                }
                        }
                }
        }
}
1	#!/usr/bin/perl
2
3	package Sack::Server;
4
5	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	use File::Slurp;
14	use Cwd qw(abs_path);
15
16	use lib '/srv/Sack/lib';
17	use Sack::Merge;
18	use Sack::Server::HTTP;
19	use Sack::Server::HTML;
20
21	my @cloud;
22	my $cloud_path = $ENV{CLOUD} \|\| "etc/cloud";
23	die "start with: CLOUD=etc/cloud perl -I/srv/Sack/lib $0\n" unless -e $cloud_path;
24	@cloud = read_file $cloud_path;
25	@cloud = map { chomp $_; $_ } @cloud;
26
27	warn "# cloud ",dump( @cloud );
28
29	my $listen_port = 4444;
30	my $http_port = 4480;
31
32	my $node_path = abs_path $0;
33	$node_path =~ s{Server}{Client};
34
35	my $lsn = IO::Socket::INET->new(Listen => 1, LocalPort => $listen_port, Reuse => 1) or die "$listen_port $!";
36	my $www = IO::Socket::INET->new(Listen => 1, LocalPort => $http_port, Reuse => 1) or die "$http_port $!";
37	my $sel = IO::Select->new($lsn);
38	$sel->add( $www );
39
40	my $info;
41	sub info {
42	my $port = shift;
43	push @{ $info->{node}->{$port} }, [ @_ ];
44	}
45
46	sub fork_ssh {
47	my ( $port, $host ) = @_;
48
49	if ( my $pid = fork ) {
50	# parent
51	info $port => 'forked', $pid;
52	return $port;
53
54	} elsif ( ! defined $pid ) {
55	warn "can't fork $host $port";
56	return;
57	} else {
58	# child
59	my $cmd = qq\|ssh -F $cloud_path.ssh -R $port:127.0.0.1:$listen_port $host $node_path $port\|;
60	warn "# exec: $cmd\n";
61	exec $cmd;
62	}
63	}
64
65	my $node_port = 4000;
66
67	foreach my $host ( @cloud ) {
68	system "find /srv/Sack/ \| cpio --create --dereference \| ssh -T -F $cloud_path.ssh $host cpio --extract --make-directories --unconditional";
69	fork_ssh( $node_port++, $host );
70	}
71
72	my $session;
73
74	sub to_all {
75	my $data = shift;
76	foreach my $port ( keys %{ $session->{port} } ) {
77	warn ">>>> [$port]\n";
78	Storable::store_fd( $data, $session->{port}->{$port} );
79	}
80	}
81
82	my @shard_paths;
83
84	while (1) {
85	for my $sock ($sel->can_read(1)) {
86	if ($sock == $lsn) {
87	my $new = $lsn->accept;
88	$sel->add($new);
89	$session->{peerport}->{ $new->peerport } = $new;
90	warn "[socket] connect\n";
91	Storable::store_fd( { ping => 1 }, $new );
92	} elsif ( $sock == $www ) {
93	my $client = $www->accept;
94	Sack::Server::HTTP::request( $client, sub {
95	my ( $send, $method, $param ) = @_;
96
97	if ( $method =~ m{views} ) {
98	warn "$method ", -s $method, " bytes";
99	my $code = read_file $method;
100	Sack::Merge->clean;
101	to_all { view => $code, path => $method };
102	print $send "HTTP/1.0 302 $method\r\nLocation: /\r\n\r\n";
103	return 1;
104	} elsif ( $method =~ m{^/tmp/sack} ) {
105	@shard_paths = glob "$method/*";
106	warn "loading shard $method from ", dump( @shard_paths );
107	to_all { load => $method };
108	print $send "HTTP/1.0 302 $method\r\nLocation: /\r\n\r\n";
109	return 1;
110	} elsif ( $method =~ m{^/out/(.+)} ) {
111	print $send "HTTP/1.0 200 OK\r\nContent-Type: text/html\r\n\r\n";
112	Sack::Server::HTML::send_out( $send, Sack::Merge->out, $1, $param );
113	return 1;
114	}
115
116	print $send "HTTP/1.0 200 OK\r\nContent-Type: text/html\r\n\r\n";
117
118	print $send qq\|<h1>Views</h1><ul>\|
119	, join("\n", map { qq\|<li><a href="$_">$_</a>\| } glob '/srv/Sack/views//.pl' )
120	, qq\|</ul>\|
121	;
122
123	my $out = Sack::Merge->out;
124	print $send qq\|<h1>Results</h1><ul>\|
125	, join("\n", map { qq\|<li><a href="/out/$_" target="$_">$_</a>\| } keys %$out )
126	, qq\|</ul>\|
127	;
128
129	print $send qq\|<h1>Nodes</h1>
130	<style type="text/css">
131	.forked {
132	color: #f00;
133	}
134
135	.ping {
136	color: #0f0;
137	}
138	</style>
139	\|, join("\n", map {
140	my $class = join(' ', map { $_->[0] } @{ $info->{node}->{$_} });
141	qq\|<span class="$class">$_</span>\|;
142	} sort keys %{ $info->{node} } );
143
144	print $send qq\|<h1>Data</h1><ul>\|
145	, join("\n", map { qq\|<li><a href="$_">$_</a>\| } glob '/tmp/sack/*' )
146	, qq\|</ul>\|
147	;
148
149	print $send '<pre>', dump($info), '</pre>';
150
151	} );
152	} else {
153	my $data = eval { Storable::fd_retrieve( $sock ) };
154	if ( $@ ) {
155	delete $session->{$sock};
156	warn "[socket] disconnect: $@\n";
157	$sel->remove($sock);
158	$sock->close;
159	} else {
160	warn "<<<< ", dump($data), $/;
161
162	if ( my $path = $data->{shard} ) {
163	$info->{shard}->{ $path } = $data->{port};
164	# FIXME will need push for multiple copies of shards
165	}
166
167	if ( my $repl = $data->{repl} ) {
168	my $response = { repl_pid => $$ };
169	if ( $repl =~ m/ping/ ) {
170	to_all { ping => 1 };
171	} elsif ( $repl =~ m/info/ ) {
172	$response->{info} = $info;
173	} elsif ( $repl =~ m{load\s*(\S+)?} ) {
174	my $name = $1 \|\| 'shard';
175	@shard_paths = glob "/tmp/sack/$name/*";
176	warn "loading shard $name from ", dump( @shard_paths );
177	to_all { load => $name };
178	} elsif ( $repl =~ m{view\s*(\S+)?} ) {
179	my $path = $1 \|\| '/srv/Sack/views/00.demo.pl';
180	my $code = read_file $path;
181	Sack::Merge->clean;
182	to_all { view => $code, path => $path };
183	$response->{view}->{$path}->{running};
184	} elsif ( $repl =~ m{debug\s*(.+)?} ) {
185	to_all { debug => $1 };
186	} elsif ( $repl =~ m{out} ) {
187	my $out = Sack::Merge->out;
188	warn "out ",dump( $out );
189	$response->{out} = $out;
190	} elsif ( $repl =~ m{clean} ) {
191	delete $info->{shard};
192	to_all { clean => 1 };
193	} else {
194	$response->{error}->{unknown} = $data;
195	}
196	Storable::store_fd( $response, $sock );
197	} elsif ( $data->{ping} ) {
198	my $port = $data->{port};
199	info $port => 'ping', $port;
200	$session->{port}->{ $data->{port} } = $sock;
201	} elsif ( $data->{load} eq 'shard' ) {
202	if ( my $path = shift @shard_paths ) {
203	warn "retrieve $path ", -s $path;
204	my $shard = Storable::retrieve $path;
205	warn ">>>> [", $data->{port}, "] sending shard $path\n";
206	Storable::store_fd( { path => $path, shard => $shard }, $sock );
207	} else {
208	warn "no more shards for [", $data->{port}, "]\n";
209	}
210	} elsif ( defined $data->{out} ) {
211	Sack::Merge->add( $data->{out} );
212	} else {
213	warn "UNKNOWN ",dump($data);
214	}
215	}
216	}
217	}
218	}