lib/Sack/Server.pm

#!/usr/bin/perl

package Sack::Server;

use warnings;
use strict;

my $debug = 0;

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" if $debug;
                Storable::store_fd( $data, $session->{port}->{$port} );
        }
}

our @shard_load_queue;
sub load_shard {
        my $shard = shift @_ || return;

        warn "# load_shard $shard\n";

        my @shards = glob "$shard/*";

        if ( ! @shards ) {
                warn "no shards for $shard\n";
                return;
        }

        $info->{shard}->{$_} = 'wait' foreach @shards;
        warn "loading shard $shard from ", dump( @shards );

        push @shard_load_queue, @shards;
        to_all { load => $shard };
}

sub run_view {
        my ( $path ) = @_;
        if ( ! -r $path ) {
                warn "ERROR view $path: $!";
                return;
        }
        my $code = read_file $path;
        Sack::Merge->clean;
        delete( $info->{view} );
        to_all { code => $code, view => $path };
};

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} ) {
                                        run_view $method;
                                        print $send "HTTP/1.0 302 $method\r\nLocation: /\r\n\r\n";
                                        return 1;
                                } elsif ( $method =~ m{^/tmp/sack} ) {
                                        load_shard $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|

<style type="text/css">

.running {
        color: #f00;
}


.forked {
        color: #f00;
}

.ping {
        color: #0f0;
}

.shards {
        font-family: monospace;
        line-height: 0.8em;
        color: #0f0;
}

.send {
        color: #ff0;
}

.wait {
        color: #f00;
}


</style>

|;

                                print $send qq|<h1>Views</h1><ul>|
                                        , join("\n", map {
                                                my $view = $_;
                                                my $html = '';
                                                if ( my $s = $info->{view}->{$view} ) {
                                                        my @nodes = sort keys %$s;
                                                        my $total = 0;
                                                        $html .= qq|<table><tr><th>node</th><th>rows</th><th>shards</th><th>&sum;</th></tr>|;
                                                        foreach my $node ( @nodes ) {
                                                                my $h = '';
                                                                my $shard = 0;
                                                                foreach my $path ( sort keys %{ $s->{$node} } ) {
                                                                        my $affected = $s->{$node}->{$path};
                                                                        $h .= qq|<span title="$affected $path">*</span>|;
                                                                        $shard += $affected;
                                                                        $total += $affected;
                                                                }
                                                                $html .= qq|<tr><td><tt>$node</tt></td><td>$shard</td><td>$h</td><td>$total</td></tr>\n|;
                                                        }

                                                        $html .= qq|</table>|;
                                                };
                                                qq|<li><a href="$view">$view</a>$html|
                                        } 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>
                                |, 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>|;
                                foreach my $path ( glob '/tmp/sack/*' ) {
                                        print $send qq|<li><a href="$path">$path</a><div class="shards">|;
                                        foreach my $s ( sort grep { m/$path/ } keys %{ $info->{shard} } ) {
                                                my $i = $info->{shard}->{$s};
                                                print $send qq|<span class=$i>$i</span> |;
                                        }
                                        print $send qq|</div>|;
                                }
                                print $send qq|</ul>|;

                                print $send '<pre>', dump($info), '</pre>' if $param->{debug};

                                return 1;
                        } );
                } 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 $debug;

                                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+)?} ) {
                                                load_shard $1;
                                        } elsif ( $repl =~ m{view\s*(\S+)?} ) {
                                                run_view $1;
                                        } 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 };
                                        } elsif ( $repl eq 'exit' ) {
                                                to_all { exit => 1 };
                                                sleep 1;
                                                exit;
                                        } 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 ( defined $data->{load} && $data->{load} eq 'shard' ) {
                                        if ( my $path = shift @shard_load_queue ) {
                                                $info->{shard}->{$path} = 'read';
                                                my $shard = Storable::retrieve $path;
                                                $info->{shard}->{$path} = 'send';
                                                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} );
                                        $info->{view}->{ $data->{view} }->{ $data->{port} } = $data->{on_shard};
                                } 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	dpavlin	232	my $debug = 0;
9
10	dpavlin	163	use IO::Socket::INET;
11			use IO::Select;
12
13			use Data::Dump qw(dump);
14			use Storable qw();
15	dpavlin	169	use File::Slurp;
16	dpavlin	176	use Cwd qw(abs_path);
17	dpavlin	163
18	dpavlin	200	use lib '/srv/Sack/lib';
19			use Sack::Merge;
20	dpavlin	210	use Sack::Server::HTTP;
21	dpavlin	213	use Sack::Server::HTML;
22	dpavlin	200
23	dpavlin	179	my @cloud;
24	dpavlin	209	my $cloud_path = $ENV{CLOUD} \|\| "etc/cloud";
25			die "start with: CLOUD=etc/cloud perl -I/srv/Sack/lib $0\n" unless -e $cloud_path;
26	dpavlin	169	@cloud = read_file $cloud_path;
27			@cloud = map { chomp $_; $_ } @cloud;
28
29			warn "# cloud ",dump( @cloud );
30
31	dpavlin	163	my $listen_port = 4444;
32	dpavlin	210	my $http_port = 4480;
33	dpavlin	163
34	dpavlin	176	my $node_path = abs_path $0;
35	dpavlin	181	$node_path =~ s{Server}{Client};
36	dpavlin	163
37	dpavlin	210	my $lsn = IO::Socket::INET->new(Listen => 1, LocalPort => $listen_port, Reuse => 1) or die "$listen_port $!";
38			my $www = IO::Socket::INET->new(Listen => 1, LocalPort => $http_port, Reuse => 1) or die "$http_port $!";
39	dpavlin	163	my $sel = IO::Select->new($lsn);
40	dpavlin	210	$sel->add( $www );
41	dpavlin	163
42	dpavlin	166	my $info;
43			sub info {
44			my $port = shift;
45	dpavlin	211	push @{ $info->{node}->{$port} }, [ @_ ];
46	dpavlin	166	}
47
48	dpavlin	179	sub fork_ssh {
49	dpavlin	163	my ( $port, $host ) = @_;
50
51			if ( my $pid = fork ) {
52			# parent
53	dpavlin	166	info $port => 'forked', $pid;
54	dpavlin	163	return $port;
55
56			} elsif ( ! defined $pid ) {
57			warn "can't fork $host $port";
58			return;
59			} else {
60			# child
61	dpavlin	169	my $cmd = qq\|ssh -F $cloud_path.ssh -R $port:127.0.0.1:$listen_port $host $node_path $port\|;
62	dpavlin	163	warn "# exec: $cmd\n";
63			exec $cmd;
64			}
65			}
66
67			my $node_port = 4000;
68
69	dpavlin	169	foreach my $host ( @cloud ) {
70	dpavlin	176	system "find /srv/Sack/ \| cpio --create --dereference \| ssh -T -F $cloud_path.ssh $host cpio --extract --make-directories --unconditional";
71	dpavlin	179	fork_ssh( $node_port++, $host );
72	dpavlin	163	}
73
74			my $session;
75
76	dpavlin	188	sub to_all {
77			my $data = shift;
78			foreach my $port ( keys %{ $session->{port} } ) {
79	dpavlin	232	warn ">>>> [$port]\n" if $debug;
80	dpavlin	188	Storable::store_fd( $data, $session->{port}->{$port} );
81			}
82			}
83
84	dpavlin	221	our @shard_load_queue;
85			sub load_shard {
86			my $shard = shift @_ \|\| return;
87	dpavlin	192
88	dpavlin	221	warn "# load_shard $shard\n";
89
90			my @shards = glob "$shard/*";
91
92			if ( ! @shards ) {
93			warn "no shards for $shard\n";
94			return;
95			}
96
97			$info->{shard}->{$_} = 'wait' foreach @shards;
98			warn "loading shard $shard from ", dump( @shards );
99
100			push @shard_load_queue, @shards;
101			to_all { load => $shard };
102			}
103
104	dpavlin	232	sub run_view {
105			my ( $path ) = @_;
106			if ( ! -r $path ) {
107			warn "ERROR view $path: $!";
108			return;
109			}
110			my $code = read_file $path;
111			Sack::Merge->clean;
112			delete( $info->{view} );
113			to_all { code => $code, view => $path };
114			};
115
116	dpavlin	163	while (1) {
117			for my $sock ($sel->can_read(1)) {
118			if ($sock == $lsn) {
119			my $new = $lsn->accept;
120			$sel->add($new);
121	dpavlin	174	$session->{peerport}->{ $new->peerport } = $new;
122	dpavlin	164	warn "[socket] connect\n";
123	dpavlin	163	Storable::store_fd( { ping => 1 }, $new );
124	dpavlin	210	} elsif ( $sock == $www ) {
125			my $client = $www->accept;
126			Sack::Server::HTTP::request( $client, sub {
127			my ( $send, $method, $param ) = @_;
128	dpavlin	212
129			if ( $method =~ m{views} ) {
130	dpavlin	232	run_view $method;
131	dpavlin	212	print $send "HTTP/1.0 302 $method\r\nLocation: /\r\n\r\n";
132	dpavlin	213	return 1;
133	dpavlin	212	} elsif ( $method =~ m{^/tmp/sack} ) {
134	dpavlin	221	load_shard $method;
135	dpavlin	212	print $send "HTTP/1.0 302 $method\r\nLocation: /\r\n\r\n";
136	dpavlin	213	return 1;
137			} elsif ( $method =~ m{^/out/(.+)} ) {
138			print $send "HTTP/1.0 200 OK\r\nContent-Type: text/html\r\n\r\n";
139			Sack::Server::HTML::send_out( $send, Sack::Merge->out, $1, $param );
140			return 1;
141	dpavlin	212	}
142
143			print $send "HTTP/1.0 200 OK\r\nContent-Type: text/html\r\n\r\n";
144
145	dpavlin	232	print $send qq\|
146	dpavlin	212
147	dpavlin	232	<style type="text/css">
148	dpavlin	213
149	dpavlin	232	.running {
150			color: #f00;
151			}
152
153
154	dpavlin	219	.forked {
155			color: #f00;
156			}
157	dpavlin	212
158	dpavlin	219	.ping {
159			color: #0f0;
160			}
161	dpavlin	222
162			.shards {
163			font-family: monospace;
164			line-height: 0.8em;
165			color: #0f0;
166			}
167
168			.send {
169			color: #ff0;
170			}
171
172			.wait {
173			color: #f00;
174			}
175
176
177	dpavlin	219	</style>
178	dpavlin	232
179			\|;
180
181			print $send qq\|<h1>Views</h1><ul>\|
182			, join("\n", map {
183			my $view = $_;
184			my $html = '';
185			if ( my $s = $info->{view}->{$view} ) {
186			my @nodes = sort keys %$s;
187			my $total = 0;
188			$html .= qq\|<table><tr><th>node</th><th>rows</th><th>shards</th><th>∑</th></tr>\|;
189			foreach my $node ( @nodes ) {
190			my $h = '';
191			my $shard = 0;
192			foreach my $path ( sort keys %{ $s->{$node} } ) {
193			my $affected = $s->{$node}->{$path};
194			$h .= qq\|<span title="$affected $path">*</span>\|;
195			$shard += $affected;
196			$total += $affected;
197			}
198			$html .= qq\|<tr><td><tt>$node</tt></td><td>$shard</td><td>$h</td><td>$total</td></tr>\n\|;
199			}
200
201			$html .= qq\|</table>\|;
202			};
203			qq\|<li><a href="$view">$view</a>$html\|
204			} glob '/srv/Sack/views//.pl' )
205			, qq\|</ul>\|
206			;
207
208			my $out = Sack::Merge->out;
209			print $send qq\|<h1>Results</h1><ul>\|
210			, join("\n", map { qq\|<li><a href="/out/$_" target="$_">$_</a>\| } keys %$out )
211			, qq\|</ul>\|
212			;
213
214			print $send qq\|<h1>Nodes</h1>
215	dpavlin	219	\|, join("\n", map {
216			my $class = join(' ', map { $_->[0] } @{ $info->{node}->{$_} });
217			qq\|<span class="$class">$_</span>\|;
218			} sort keys %{ $info->{node} } );
219
220	dpavlin	221	print $send qq\|<h1>Data</h1><ul>\|;
221			foreach my $path ( glob '/tmp/sack/*' ) {
222	dpavlin	222	print $send qq\|<li><a href="$path">$path</a><div class="shards">\|;
223	dpavlin	221	foreach my $s ( sort grep { m/$path/ } keys %{ $info->{shard} } ) {
224	dpavlin	222	my $i = $info->{shard}->{$s};
225			print $send qq\|<span class=$i>$i</span> \|;
226	dpavlin	221	}
227	dpavlin	222	print $send qq\|</div>\|;
228	dpavlin	221	}
229			print $send qq\|</ul>\|;
230	dpavlin	219
231	dpavlin	232	print $send '<pre>', dump($info), '</pre>' if $param->{debug};
232	dpavlin	219
233	dpavlin	232	return 1;
234	dpavlin	210	} );
235	dpavlin	163	} else {
236	dpavlin	165	my $data = eval { Storable::fd_retrieve( $sock ) };
237			if ( $@ ) {
238			delete $session->{$sock};
239			warn "[socket] disconnect: $@\n";
240			$sel->remove($sock);
241			$sock->close;
242			} else {
243	dpavlin	232	warn "<<<< ", dump($data), $/ if $debug;
244	dpavlin	196
245			if ( my $path = $data->{shard} ) {
246	dpavlin	211	$info->{shard}->{ $path } = $data->{port};
247			# FIXME will need push for multiple copies of shards
248	dpavlin	196	}
249
250	dpavlin	192	if ( my $repl = $data->{repl} ) {
251			my $response = { repl_pid => $$ };
252			if ( $repl =~ m/ping/ ) {
253	dpavlin	188	to_all { ping => 1 };
254	dpavlin	192	} elsif ( $repl =~ m/info/ ) {
255	dpavlin	175	$response->{info} = $info;
256	dpavlin	192	} elsif ( $repl =~ m{load\s*(\S+)?} ) {
257	dpavlin	221	load_shard $1;
258	dpavlin	196	} elsif ( $repl =~ m{view\s*(\S+)?} ) {
259	dpavlin	232	run_view $1;
260	dpavlin	196	} elsif ( $repl =~ m{debug\s*(.+)?} ) {
261			to_all { debug => $1 };
262	dpavlin	200	} elsif ( $repl =~ m{out} ) {
263			my $out = Sack::Merge->out;
264			warn "out ",dump( $out );
265			$response->{out} = $out;
266	dpavlin	203	} elsif ( $repl =~ m{clean} ) {
267			delete $info->{shard};
268			to_all { clean => 1 };
269	dpavlin	232	} elsif ( $repl eq 'exit' ) {
270			to_all { exit => 1 };
271			sleep 1;
272			exit;
273	dpavlin	189	} else {
274			$response->{error}->{unknown} = $data;
275	dpavlin	174	}
276	dpavlin	166	Storable::store_fd( $response, $sock );
277			} elsif ( $data->{ping} ) {
278	dpavlin	174	my $port = $data->{port};
279			info $port => 'ping', $port;
280			$session->{port}->{ $data->{port} } = $sock;
281	dpavlin	232	} elsif ( defined $data->{load} && $data->{load} eq 'shard' ) {
282	dpavlin	221	if ( my $path = shift @shard_load_queue ) {
283			$info->{shard}->{$path} = 'read';
284	dpavlin	192	my $shard = Storable::retrieve $path;
285	dpavlin	221	$info->{shard}->{$path} = 'send';
286	dpavlin	192	warn ">>>> [", $data->{port}, "] sending shard $path\n";
287			Storable::store_fd( { path => $path, shard => $shard }, $sock );
288			} else {
289			warn "no more shards for [", $data->{port}, "]\n";
290			}
291	dpavlin	200	} elsif ( defined $data->{out} ) {
292			Sack::Merge->add( $data->{out} );
293	dpavlin	232	$info->{view}->{ $data->{view} }->{ $data->{port} } = $data->{on_shard};
294	dpavlin	192	} else {
295			warn "UNKNOWN ",dump($data);
296	dpavlin	164	}
297	dpavlin	163	}
298			}
299			}
300			}