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;
use Sack::Server::Gnuplot;

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 fork_ssh {
        my ( $port, $host ) = @_;

        if ( my $pid = fork ) {
                # parent
                $info->{forked}->{$port} = $pid;
                $info->{pending}->{$port} = 'forked';
                $info->{host}->{$port} = $host;
                return $port;

        } elsif ( ! defined $pid ) {
                warn "can't fork $host $port";
                return;
        } elsif ( $host =~ m/^(127.0.0.1|localhost)$/ ) {
                warn "# localhost, just execute node\n";
                exec $node_path, $port, $listen_port;
        } else {
                # child
                warn "[$port] cpio last version\n";
                system "find /srv/Sack/ | cpio --create --dereference | ssh -T -F $cloud_path.ssh $host cpio --extract --make-directories --unconditional";

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

my $session;

sub all_ports { keys %{ $session->{port} } }

sub to_all {
        my $data = shift;
        foreach my $port ( all_ports ) {
                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";

        $shard .= '/*' if -d $shard;

        my @shards = glob $shard;

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

        foreach my $s ( @shards ) {
                if ( my $node = $info->{shard}->{$s} ) {
                        next if $node =~ m{^\d+$}; # not node number
                }
                $info->{shard}->{$s} = 'wait';
                push @shard_load_queue, $s;
                warn "queued $s for loading\n";
        }

        # XXX depriciated but not removed yet
        if ( 0 ) {
        to_all { load => $shard };
        $info->{pending}->{$_} = 'load' foreach all_ports;
        return;
        }

        my @nodes = all_ports;
        my $chunk = int( ( $#shard_load_queue + 1 ) / $#nodes );

        my $pid_port;

        foreach my $port ( all_ports ) {
                $info->{pending}->{$port} = 'load';
                my @shards = splice @shard_load_queue, 0, $chunk;

                my $sock = $session->{port}->{$port};

                if ( my $pid = fork ) { 
                        $pid_port->{$pid} = $port;
                        # parent
                } elsif ( ! defined $pid ) {
                        die "can't fork $!";
                } else {
                        # child

                        warn ">>>> [$port] bulk_load ", $#shards + 1, " shards\n";
                        Storable::store_fd( { bulk_load => [ @shards ] }, $sock );
                        my $current = 0;
                        my $total = $#shards + 1;
                        foreach my $s ( @shards ) {
                                warn ">>>> [$port] bulk_load $s $current/$total\n";
                                Storable::store_fd( Storable::retrieve( $s ), $sock );
                                $current++;
                        }
                        warn ">>>> [$port] bulk_load finished, $current shards\n";
                        exit;
                }

        }

        foreach my $child ( keys %$pid_port ) {
                warn "waitpid $child\n";
                waitpid($child, 0);
                delete( $info->{pending}->{ $pid_port->{$child} } );
        }

        $info->{pending}->{$_} = 'view' foreach all_ports;
        to_all { code => "# collect back loaded shards\n", view => 'noop' };
}

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

our @responses;

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;
                                } elsif ( $method =~ m{^/gnuplot} ) {
                                        eval {
                                                my $path = Sack::Server::Gnuplot::date( Sack::Merge->out, $param );
                                                if ( -e $path ) {
                                                        print $send "HTTP/1.0 200 OK\r\nContent-Type: image/png\r\n\r\n";
                                                        open(my $fh, '<', $path) || die $path;
                                                        my $b;
                                                        while ( read($fh, $b, 4096) ) {
                                                                print $send $b;
                                                        }
                                                        return 1;
                                                } else {
                                                        print $send "HTTP/1.0 404 no graph\r\n\r\n";
                                                        return 1;
                                                }
                                        };
                                        warn "ERROR: $@" if $@;
                                } elsif ( $method =~ m{^/favicon.ico} ) {
                                        print $send "HTTP/1.0 200 OK\r\nContent-Type: image/vnd.microsoft.icon\r\n\r\n", read_file( 'artwork/favicon.ico' );
                                        return 1;
                                } elsif ( $method =~ m{/nodes} ) {
                                        to_all { ping => 1 };
                                        my @proc = ( 'MemTotal', 'MemFree', 'Buffers', 'Cached', 'VmSize', 'VmPeak' );
                                        print $send "HTTP/1.0 200 OK\r\nContent-Type: text/html\r\n\r\n",
                                                '<table><tr><th>', join('</th><th>', 'node', 'ip', @proc), '</th></tr>';
                                        foreach my $node ( sort {
                                                $info->{host}->{$a} cmp $info->{host}->{$b} || # ip
                                                $a <=> $b # node
                                        } keys %{$info->{proc}} ) {
                                                print $send '<tr><td>', join('</td><td align=right>',
                                                        $node, $info->{host}->{$node}, map {
                                                                my $kb = $info->{proc}->{$node}->{$_};
                                                                $kb =~ s/\s+kb//i; $kb;
                                                        } @proc ), '</td><tr>';
                                        }
                                        print $send '</table>';
                                        return 1;
                                }

                                my $refresh = $param->{refresh};
                                $refresh = 1 if grep { $info->{pending}->{$_} ne 'forked' } keys %{ $info->{pending} };
                                # we don't care about refresh is nodes are stuck in forked state!

                                print $send "HTTP/1.0 200 OK\r\nContent-Type: text/html"
                                        , ( $refresh ? "\r\nRefresh: $refresh" : '' )
                                        , "\r\n\r\n"
                                        , ( $refresh ? qq|<span style="color: #888; float:right">$refresh</span>| : '' )
                                ;

                                print $send qq|

<style type="text/css">

.running {
        color: #f00;
}


.forked {
        color: #888;
}

.ping {
        color: #0f0;
}

.load {
        color: #f00;
}

.view {
        color: #00f;
}

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

.send {
        color: #ff0;
}

.wait {
        color: #f00;
}


</style>

|;

                                my $out = Sack::Merge->out;     
                                if ( my $li = join("\n", map { qq|<li><input type=checkbox name=filter value="$_"><a href="/out/$_" target="$_">$_</a>| } keys %$out ) ) {
                                        print $send qq|<h1>Results</h1><form method="get" target="gnuplot" action="/gnuplot/"><ul>$li</ul><input type=submit value="Show checked"></form><img src="/gnuplot/">\n|;
                                }

                                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><th>merge</th></tr>|;
                                                        foreach my $node ( @nodes ) {
                                                                my $h = '';
                                                                my $shard = 0;
                                                                foreach my $path ( sort keys %{ $s->{$node} } ) {
                                                                        my $affected = $s->{$node}->{$path};
                                                                        my $n = '?';
                                                                        $n = $1 if $path =~ m{/(\w)\w+/\d+};
                                                                        my $class = $affected == 0 ? 'style="color: #888"' : '';
                                                                        $h .= qq|<tt title="$affected $path"$class>$n</tt>|;
                                                                        $shard += $affected;
                                                                        $total += $affected;
                                                                }
                                                                my $merge = $info->{merge}->{$view}->{$node} || '?';
                                                                $html .= qq|<tr><td><tt>$node</tt></td><td>$shard</td><td>$h</td><td>$total</td><td>$merge</td></tr>\n|;
                                                        }

                                                        $html .= qq|</table>|;
                                                };
                                                qq|<li><a href="$view">$view</a>$html|
                                        } glob '/srv/Sack/views/*/*.pl' )
                                        , qq|</ul>|
                                        ;

                                print $send qq|<h1>Nodes</h1>|;
                                foreach my $node ( sort keys %{ $info->{forked} } ) {
                                        my $attr = qq| title="$info->{host}->{$node}"|;
                                        if ( my $pending = $info->{pending}->{$node} ) {
                                                $attr .= qq| class="$pending"|;
                                        }
                                        print $send qq|<tt$attr>$node</tt>\n|;
                                }
                                print $send qq|<a target=details href=/nodes/>details</a>|;

                                print $send qq|<h1>Shards</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" title="$s">$i</span> |;
                                        }
                                        print $send qq|</div>|;
                                }
                                print $send qq|</ul>|;

                                if ( $param->{info} ) {
                                        print $send qq|<a href="?info=0">hide info</a>|;
                                        print $send '<pre>', dump($info), '</pre>'
                                } else {
                                        print $send qq|<a href="?info=1">show info</a>|;
                                }
                                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{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;
                                        } elsif ( $repl eq '.' ) {
                                                $response->{'.'} = [ @responses ];
                                                @responses = ();
                                        } elsif ( $repl =~ m{(\w+)\s*(.+)?} ) {
                                                to_all { $1 => $2 };
                                        } else {
                                                $response->{error}->{repl} = $repl;
                                        }
                                        Storable::store_fd( $response, $sock );
                                } elsif ( $data->{ping} ) {
                                        my $port = $data->{port};
                                        delete( $info->{pending}->{$port} ) if defined $info->{pending}->{$port} && $info->{pending}->{$port} eq 'forked';
                                        $session->{port}->{$port} = $sock;
                                        foreach my $name ( keys %{$data->{proc}} ) {
                                                if ( $name eq 'loadavg' ) {
                                                        $info->{proc}->{$port}->{loadavg} = [ split(/\s+/, $data->{proc}->{$name}) ];
                                                } else {
                                                        foreach my $line ( split(/\n/,$data->{proc}->{$name}) ) {
                                                                my ($n,$v) = split(/:\s+/,$line);
                                                                $info->{proc}->{$port}->{$n} = $v;
                                                        }
                                                }
                                        }
                                } 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";
                                                delete $info->{pending}->{ $data->{port} };
                                        }
                                } elsif ( exists $data->{out} ) {
                                        my $added = Sack::Merge->add( $data->{out} ) if defined $data->{out};
                                        $info->{merge}->{ $data->{view} }->{ $data->{port} } = $added;
                                        $info->{view }->{ $data->{view} }->{ $data->{port} } = $data->{on_shard};
                                        # refresh shard allocation
                                        $info->{shard}->{ $_ } = $data->{port} foreach keys %{ $data->{on_shard} };
                                        delete $info->{pending}->{ $data->{port} };
                                } elsif ( exists $data->{port} ) {
                                        push @responses, $data;
                                        warn "# ",dump($data),$/;
                                } 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	241	use Sack::Server::Gnuplot;
23	dpavlin	200
24	dpavlin	179	my @cloud;
25	dpavlin	209	my $cloud_path = $ENV{CLOUD} \|\| "etc/cloud";
26			die "start with: CLOUD=etc/cloud perl -I/srv/Sack/lib $0\n" unless -e $cloud_path;
27	dpavlin	169	@cloud = read_file $cloud_path;
28			@cloud = map { chomp $_; $_ } @cloud;
29
30			warn "# cloud ",dump( @cloud );
31
32	dpavlin	163	my $listen_port = 4444;
33	dpavlin	210	my $http_port = 4480;
34	dpavlin	163
35	dpavlin	176	my $node_path = abs_path $0;
36	dpavlin	181	$node_path =~ s{Server}{Client};
37	dpavlin	163
38	dpavlin	210	my $lsn = IO::Socket::INET->new(Listen => 1, LocalPort => $listen_port, Reuse => 1) or die "$listen_port $!";
39			my $www = IO::Socket::INET->new(Listen => 1, LocalPort => $http_port, Reuse => 1) or die "$http_port $!";
40	dpavlin	163	my $sel = IO::Select->new($lsn);
41	dpavlin	210	$sel->add( $www );
42	dpavlin	163
43	dpavlin	166	my $info;
44
45	dpavlin	179	sub fork_ssh {
46	dpavlin	163	my ( $port, $host ) = @_;
47
48			if ( my $pid = fork ) {
49			# parent
50	dpavlin	266	$info->{forked}->{$port} = $pid;
51			$info->{pending}->{$port} = 'forked';
52	dpavlin	271	$info->{host}->{$port} = $host;
53	dpavlin	163	return $port;
54
55			} elsif ( ! defined $pid ) {
56			warn "can't fork $host $port";
57			return;
58	dpavlin	276	} elsif ( $host =~ m/^(127.0.0.1\|localhost)$/ ) {
59			warn "# localhost, just execute node\n";
60			exec $node_path, $port, $listen_port;
61	dpavlin	163	} else {
62			# child
63	dpavlin	266	warn "[$port] cpio last version\n";
64			system "find /srv/Sack/ \| cpio --create --dereference \| ssh -T -F $cloud_path.ssh $host cpio --extract --make-directories --unconditional";
65
66	dpavlin	169	my $cmd = qq\|ssh -F $cloud_path.ssh -R $port:127.0.0.1:$listen_port $host $node_path $port\|;
67	dpavlin	163	warn "# exec: $cmd\n";
68			exec $cmd;
69			}
70			}
71
72			my $node_port = 4000;
73
74	dpavlin	169	foreach my $host ( @cloud ) {
75	dpavlin	179	fork_ssh( $node_port++, $host );
76	dpavlin	163	}
77
78			my $session;
79
80	dpavlin	263	sub all_ports { keys %{ $session->{port} } }
81
82	dpavlin	188	sub to_all {
83			my $data = shift;
84	dpavlin	263	foreach my $port ( all_ports ) {
85	dpavlin	232	warn ">>>> [$port]\n" if $debug;
86	dpavlin	188	Storable::store_fd( $data, $session->{port}->{$port} );
87			}
88			}
89
90	dpavlin	221	our @shard_load_queue;
91			sub load_shard {
92			my $shard = shift @_ \|\| return;
93	dpavlin	192
94	dpavlin	221	warn "# load_shard $shard\n";
95
96	dpavlin	260	$shard .= '/*' if -d $shard;
97	dpavlin	221
98	dpavlin	260	my @shards = glob $shard;
99
100	dpavlin	221	if ( ! @shards ) {
101			warn "no shards for $shard\n";
102			return;
103			}
104
105	dpavlin	247	foreach my $s ( @shards ) {
106	dpavlin	254	if ( my $node = $info->{shard}->{$s} ) {
107			next if $node =~ m{^\d+$}; # not node number
108			}
109	dpavlin	247	$info->{shard}->{$s} = 'wait';
110			push @shard_load_queue, $s;
111			warn "queued $s for loading\n";
112			}
113	dpavlin	268
114			# XXX depriciated but not removed yet
115			if ( 0 ) {
116	dpavlin	221	to_all { load => $shard };
117	dpavlin	263	$info->{pending}->{$_} = 'load' foreach all_ports;
118	dpavlin	268	return;
119			}
120
121			my @nodes = all_ports;
122			my $chunk = int( ( $#shard_load_queue + 1 ) / $#nodes );
123
124			my $pid_port;
125
126			foreach my $port ( all_ports ) {
127			$info->{pending}->{$port} = 'load';
128			my @shards = splice @shard_load_queue, 0, $chunk;
129
130			my $sock = $session->{port}->{$port};
131
132			if ( my $pid = fork ) {
133			$pid_port->{$pid} = $port;
134			# parent
135			} elsif ( ! defined $pid ) {
136			die "can't fork $!";
137			} else {
138			# child
139
140			warn ">>>> [$port] bulk_load ", $#shards + 1, " shards\n";
141			Storable::store_fd( { bulk_load => [ @shards ] }, $sock );
142	dpavlin	277	my $current = 0;
143			my $total = $#shards + 1;
144	dpavlin	268	foreach my $s ( @shards ) {
145	dpavlin	277	warn ">>>> [$port] bulk_load $s $current/$total\n";
146	dpavlin	268	Storable::store_fd( Storable::retrieve( $s ), $sock );
147	dpavlin	277	$current++;
148	dpavlin	268	}
149	dpavlin	277	warn ">>>> [$port] bulk_load finished, $current shards\n";
150	dpavlin	268	exit;
151			}
152
153			}
154
155			foreach my $child ( keys %$pid_port ) {
156			warn "waitpid $child\n";
157			waitpid($child, 0);
158			delete( $info->{pending}->{ $pid_port->{$child} } );
159			}
160
161			$info->{pending}->{$_} = 'view' foreach all_ports;
162			to_all { code => "# collect back loaded shards\n", view => 'noop' };
163	dpavlin	221	}
164
165	dpavlin	232	sub run_view {
166			my ( $path ) = @_;
167			if ( ! -r $path ) {
168			warn "ERROR view $path: $!";
169			return;
170			}
171			my $code = read_file $path;
172			Sack::Merge->clean;
173			delete( $info->{view} );
174	dpavlin	235	delete( $info->{merge} );
175	dpavlin	248	delete( $info->{shard} );
176	dpavlin	263	$info->{pending}->{$_} = 'view' foreach all_ports;
177	dpavlin	232	to_all { code => $code, view => $path };
178			};
179
180	dpavlin	247	our @responses;
181
182	dpavlin	163	while (1) {
183			for my $sock ($sel->can_read(1)) {
184			if ($sock == $lsn) {
185			my $new = $lsn->accept;
186			$sel->add($new);
187	dpavlin	174	$session->{peerport}->{ $new->peerport } = $new;
188	dpavlin	164	warn "[socket] connect\n";
189	dpavlin	163	Storable::store_fd( { ping => 1 }, $new );
190	dpavlin	210	} elsif ( $sock == $www ) {
191			my $client = $www->accept;
192			Sack::Server::HTTP::request( $client, sub {
193			my ( $send, $method, $param ) = @_;
194	dpavlin	212
195			if ( $method =~ m{views} ) {
196	dpavlin	232	run_view $method;
197	dpavlin	212	print $send "HTTP/1.0 302 $method\r\nLocation: /\r\n\r\n";
198	dpavlin	213	return 1;
199	dpavlin	212	} elsif ( $method =~ m{^/tmp/sack} ) {
200	dpavlin	221	load_shard $method;
201	dpavlin	212	print $send "HTTP/1.0 302 $method\r\nLocation: /\r\n\r\n";
202	dpavlin	213	return 1;
203			} elsif ( $method =~ m{^/out/(.+)} ) {
204			print $send "HTTP/1.0 200 OK\r\nContent-Type: text/html\r\n\r\n";
205			Sack::Server::HTML::send_out( $send, Sack::Merge->out, $1, $param );
206			return 1;
207	dpavlin	241	} elsif ( $method =~ m{^/gnuplot} ) {
208			eval {
209	dpavlin	245	my $path = Sack::Server::Gnuplot::date( Sack::Merge->out, $param );
210	dpavlin	241	if ( -e $path ) {
211			print $send "HTTP/1.0 200 OK\r\nContent-Type: image/png\r\n\r\n";
212			open(my $fh, '<', $path) \|\| die $path;
213			my $b;
214			while ( read($fh, $b, 4096) ) {
215			print $send $b;
216			}
217			return 1;
218	dpavlin	245	} else {
219			print $send "HTTP/1.0 404 no graph\r\n\r\n";
220			return 1;
221	dpavlin	241	}
222			};
223			warn "ERROR: $@" if $@;
224	dpavlin	273	} elsif ( $method =~ m{^/favicon.ico} ) {
225			print $send "HTTP/1.0 200 OK\r\nContent-Type: image/vnd.microsoft.icon\r\n\r\n", read_file( 'artwork/favicon.ico' );
226			return 1;
227	dpavlin	275	} elsif ( $method =~ m{/nodes} ) {
228			to_all { ping => 1 };
229			my @proc = ( 'MemTotal', 'MemFree', 'Buffers', 'Cached', 'VmSize', 'VmPeak' );
230			print $send "HTTP/1.0 200 OK\r\nContent-Type: text/html\r\n\r\n",
231			'<table><tr><th>', join('</th><th>', 'node', 'ip', @proc), '</th></tr>';
232			foreach my $node ( sort {
233			$info->{host}->{$a} cmp $info->{host}->{$b} \|\| # ip
234			$a <=> $b # node
235			} keys %{$info->{proc}} ) {
236			print $send '<tr><td>', join('</td><td align=right>',
237			$node, $info->{host}->{$node}, map {
238			my $kb = $info->{proc}->{$node}->{$_};
239			$kb =~ s/\s+kb//i; $kb;
240			} @proc ), '</td><tr>';
241			}
242			print $send '</table>';
243			return 1;
244	dpavlin	212	}
245
246	dpavlin	263	my $refresh = $param->{refresh};
247	dpavlin	266	$refresh = 1 if grep { $info->{pending}->{$_} ne 'forked' } keys %{ $info->{pending} };
248			# we don't care about refresh is nodes are stuck in forked state!
249	dpavlin	263
250	dpavlin	262	print $send "HTTP/1.0 200 OK\r\nContent-Type: text/html"
251	dpavlin	263	, ( $refresh ? "\r\nRefresh: $refresh" : '' )
252	dpavlin	262	, "\r\n\r\n"
253	dpavlin	263	, ( $refresh ? qq\|<span style="color: #888; float:right">$refresh</span>\| : '' )
254	dpavlin	262	;
255	dpavlin	212
256	dpavlin	232	print $send qq\|
257	dpavlin	212
258	dpavlin	232	<style type="text/css">
259	dpavlin	213
260	dpavlin	232	.running {
261			color: #f00;
262			}
263
264
265	dpavlin	219	.forked {
266	dpavlin	265	color: #888;
267	dpavlin	263	}
268
269			.ping {
270	dpavlin	265	color: #0f0;
271	dpavlin	263	}
272
273			.load {
274	dpavlin	219	color: #f00;
275			}
276	dpavlin	212
277	dpavlin	263	.view {
278			color: #00f;
279			}
280
281	dpavlin	222	.shards {
282			font-family: monospace;
283	dpavlin	250	// line-height: 0.8em;
284	dpavlin	222	color: #0f0;
285			}
286
287			.send {
288			color: #ff0;
289			}
290
291			.wait {
292			color: #f00;
293			}
294
295
296	dpavlin	219	</style>
297	dpavlin	232
298			\|;
299
300	dpavlin	264	my $out = Sack::Merge->out;
301			if ( my $li = join("\n", map { qq\|<li><input type=checkbox name=filter value="$_"><a href="/out/$_" target="$_">$_</a>\| } keys %$out ) ) {
302			print $send qq\|<h1>Results</h1><form method="get" target="gnuplot" action="/gnuplot/"><ul>$li</ul><input type=submit value="Show checked"></form><img src="/gnuplot/">\n\|;
303			}
304
305	dpavlin	232	print $send qq\|<h1>Views</h1><ul>\|
306			, join("\n", map {
307			my $view = $_;
308			my $html = '';
309			if ( my $s = $info->{view}->{$view} ) {
310			my @nodes = sort keys %$s;
311			my $total = 0;
312	dpavlin	237	$html .= qq\|<table><tr><th>node</th><th>rows</th><th>shards</th><th>∑</th><th>merge</th></tr>\|;
313	dpavlin	232	foreach my $node ( @nodes ) {
314			my $h = '';
315			my $shard = 0;
316			foreach my $path ( sort keys %{ $s->{$node} } ) {
317			my $affected = $s->{$node}->{$path};
318	dpavlin	233	my $n = '?';
319			$n = $1 if $path =~ m{/(\w)\w+/\d+};
320	dpavlin	235	my $class = $affected == 0 ? 'style="color: #888"' : '';
321			$h .= qq\|<tt title="$affected $path"$class>$n</tt>\|;
322	dpavlin	232	$shard += $affected;
323			$total += $affected;
324			}
325	dpavlin	247	my $merge = $info->{merge}->{$view}->{$node} \|\| '?';
326	dpavlin	237	$html .= qq\|<tr><td><tt>$node</tt></td><td>$shard</td><td>$h</td><td>$total</td><td>$merge</td></tr>\n\|;
327	dpavlin	232	}
328
329			$html .= qq\|</table>\|;
330			};
331			qq\|<li><a href="$view">$view</a>$html\|
332			} glob '/srv/Sack/views//.pl' )
333			, qq\|</ul>\|
334			;
335
336	dpavlin	263	print $send qq\|<h1>Nodes</h1>\|;
337	dpavlin	270	foreach my $node ( sort keys %{ $info->{forked} } ) {
338	dpavlin	271	my $attr = qq\| title="$info->{host}->{$node}"\|;
339	dpavlin	265	if ( my $pending = $info->{pending}->{$node} ) {
340	dpavlin	266	$attr .= qq\| class="$pending"\|;
341	dpavlin	265	}
342	dpavlin	266	print $send qq\|<tt$attr>$node</tt>\n\|;
343	dpavlin	263	}
344	dpavlin	278	print $send qq\|<a target=details href=/nodes/>details</a>\|;
345	dpavlin	219
346	dpavlin	249	print $send qq\|<h1>Shards</h1><ul>\|;
347	dpavlin	221	foreach my $path ( glob '/tmp/sack/*' ) {
348	dpavlin	222	print $send qq\|<li><a href="$path">$path</a><div class="shards">\|;
349	dpavlin	221	foreach my $s ( sort grep { m/$path/ } keys %{ $info->{shard} } ) {
350	dpavlin	222	my $i = $info->{shard}->{$s};
351	dpavlin	249	print $send qq\|<span class="$i" title="$s">$i</span> \|;
352	dpavlin	221	}
353	dpavlin	222	print $send qq\|</div>\|;
354	dpavlin	221	}
355			print $send qq\|</ul>\|;
356	dpavlin	219
357	dpavlin	233	if ( $param->{info} ) {
358			print $send qq\|<a href="?info=0">hide info</a>\|;
359			print $send '<pre>', dump($info), '</pre>'
360			} else {
361			print $send qq\|<a href="?info=1">show info</a>\|;
362			}
363	dpavlin	232	return 1;
364	dpavlin	210	} );
365	dpavlin	163	} else {
366	dpavlin	165	my $data = eval { Storable::fd_retrieve( $sock ) };
367			if ( $@ ) {
368			delete $session->{$sock};
369			warn "[socket] disconnect: $@\n";
370			$sel->remove($sock);
371			$sock->close;
372			} else {
373	dpavlin	232	warn "<<<< ", dump($data), $/ if $debug;
374	dpavlin	196
375			if ( my $path = $data->{shard} ) {
376	dpavlin	211	$info->{shard}->{ $path } = $data->{port};
377			# FIXME will need push for multiple copies of shards
378	dpavlin	196	}
379
380	dpavlin	192	if ( my $repl = $data->{repl} ) {
381			my $response = { repl_pid => $$ };
382			if ( $repl =~ m/ping/ ) {
383	dpavlin	188	to_all { ping => 1 };
384	dpavlin	192	} elsif ( $repl =~ m/info/ ) {
385	dpavlin	175	$response->{info} = $info;
386	dpavlin	192	} elsif ( $repl =~ m{load\s*(\S+)?} ) {
387	dpavlin	221	load_shard $1;
388	dpavlin	196	} elsif ( $repl =~ m{view\s*(\S+)?} ) {
389	dpavlin	232	run_view $1;
390	dpavlin	200	} elsif ( $repl =~ m{out} ) {
391			my $out = Sack::Merge->out;
392			warn "out ",dump( $out );
393			$response->{out} = $out;
394	dpavlin	203	} elsif ( $repl =~ m{clean} ) {
395			delete $info->{shard};
396			to_all { clean => 1 };
397	dpavlin	232	} elsif ( $repl eq 'exit' ) {
398			to_all { exit => 1 };
399			sleep 1;
400			exit;
401	dpavlin	247	} elsif ( $repl eq '.' ) {
402			$response->{'.'} = [ @responses ];
403			@responses = ();
404			} elsif ( $repl =~ m{(\w+)\s*(.+)?} ) {
405			to_all { $1 => $2 };
406	dpavlin	189	} else {
407	dpavlin	247	$response->{error}->{repl} = $repl;
408	dpavlin	174	}
409	dpavlin	166	Storable::store_fd( $response, $sock );
410			} elsif ( $data->{ping} ) {
411	dpavlin	174	my $port = $data->{port};
412	dpavlin	275	delete( $info->{pending}->{$port} ) if defined $info->{pending}->{$port} && $info->{pending}->{$port} eq 'forked';
413	dpavlin	274	$session->{port}->{$port} = $sock;
414			foreach my $name ( keys %{$data->{proc}} ) {
415			if ( $name eq 'loadavg' ) {
416			$info->{proc}->{$port}->{loadavg} = [ split(/\s+/, $data->{proc}->{$name}) ];
417			} else {
418			foreach my $line ( split(/\n/,$data->{proc}->{$name}) ) {
419			my ($n,$v) = split(/:\s+/,$line);
420			$info->{proc}->{$port}->{$n} = $v;
421			}
422			}
423			}
424	dpavlin	232	} elsif ( defined $data->{load} && $data->{load} eq 'shard' ) {
425	dpavlin	221	if ( my $path = shift @shard_load_queue ) {
426			$info->{shard}->{$path} = 'read';
427	dpavlin	192	my $shard = Storable::retrieve $path;
428	dpavlin	221	$info->{shard}->{$path} = 'send';
429	dpavlin	192	warn ">>>> [", $data->{port}, "] sending shard $path\n";
430			Storable::store_fd( { path => $path, shard => $shard }, $sock );
431			} else {
432			warn "no more shards for [", $data->{port}, "]\n";
433	dpavlin	263	delete $info->{pending}->{ $data->{port} };
434	dpavlin	192	}
435	dpavlin	235	} elsif ( exists $data->{out} ) {
436			my $added = Sack::Merge->add( $data->{out} ) if defined $data->{out};
437			$info->{merge}->{ $data->{view} }->{ $data->{port} } = $added;
438			$info->{view }->{ $data->{view} }->{ $data->{port} } = $data->{on_shard};
439	dpavlin	248	# refresh shard allocation
440			$info->{shard}->{ $_ } = $data->{port} foreach keys %{ $data->{on_shard} };
441	dpavlin	263	delete $info->{pending}->{ $data->{port} };
442	dpavlin	247	} elsif ( exists $data->{port} ) {
443			push @responses, $data;
444			warn "# ",dump($data),$/;
445	dpavlin	192	} else {
446			warn "UNKNOWN ",dump($data);
447	dpavlin	164	}
448	dpavlin	163	}
449			}
450			}
451			}