/[Sack]/trunk/lib/Sack/Server.pm
This is repository of my old source code which isn't updated any more. Go to git.rot13.org for current projects!
ViewVC logotype

Annotation of /trunk/lib/Sack/Server.pm

Parent Directory Parent Directory | Revision Log Revision Log


Revision 247 - (hide annotations)
Thu Nov 26 22:08:36 2009 UTC (14 years, 6 months ago) by dpavlin
File size: 8708 byte(s)
refactor and extend client server configuration

- load shards only if they are not allready loaded
- collect client responses and return it in repl using . (dot)
- push all unknown repl commands to clients
- eval on client now returns result to server
- save will dump current chards on node to /tmp/sack.data and
  reload it on next startup

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     sub info {
45     my $port = shift;
46 dpavlin 211 push @{ $info->{node}->{$port} }, [ @_ ];
47 dpavlin 166 }
48    
49 dpavlin 179 sub fork_ssh {
50 dpavlin 163 my ( $port, $host ) = @_;
51    
52     if ( my $pid = fork ) {
53     # parent
54 dpavlin 166 info $port => 'forked', $pid;
55 dpavlin 163 return $port;
56    
57     } elsif ( ! defined $pid ) {
58     warn "can't fork $host $port";
59     return;
60     } else {
61     # child
62 dpavlin 169 my $cmd = qq|ssh -F $cloud_path.ssh -R $port:127.0.0.1:$listen_port $host $node_path $port|;
63 dpavlin 163 warn "# exec: $cmd\n";
64     exec $cmd;
65     }
66     }
67    
68     my $node_port = 4000;
69    
70 dpavlin 169 foreach my $host ( @cloud ) {
71 dpavlin 176 system "find /srv/Sack/ | cpio --create --dereference | ssh -T -F $cloud_path.ssh $host cpio --extract --make-directories --unconditional";
72 dpavlin 179 fork_ssh( $node_port++, $host );
73 dpavlin 163 }
74    
75     my $session;
76    
77 dpavlin 188 sub to_all {
78     my $data = shift;
79     foreach my $port ( keys %{ $session->{port} } ) {
80 dpavlin 232 warn ">>>> [$port]\n" if $debug;
81 dpavlin 188 Storable::store_fd( $data, $session->{port}->{$port} );
82     }
83     }
84    
85 dpavlin 221 our @shard_load_queue;
86     sub load_shard {
87     my $shard = shift @_ || return;
88 dpavlin 192
89 dpavlin 221 warn "# load_shard $shard\n";
90    
91     my @shards = glob "$shard/*";
92    
93     if ( ! @shards ) {
94     warn "no shards for $shard\n";
95     return;
96     }
97    
98 dpavlin 247 foreach my $s ( @shards ) {
99     next if $info->{shard}->{$s} =~ m{^\d+$};
100     $info->{shard}->{$s} = 'wait';
101     push @shard_load_queue, $s;
102     warn "queued $s for loading\n";
103     }
104 dpavlin 221 to_all { load => $shard };
105     }
106    
107 dpavlin 232 sub run_view {
108     my ( $path ) = @_;
109     if ( ! -r $path ) {
110     warn "ERROR view $path: $!";
111     return;
112     }
113     my $code = read_file $path;
114     Sack::Merge->clean;
115     delete( $info->{view} );
116 dpavlin 235 delete( $info->{merge} );
117 dpavlin 232 to_all { code => $code, view => $path };
118     };
119    
120 dpavlin 247 our @responses;
121    
122 dpavlin 163 while (1) {
123     for my $sock ($sel->can_read(1)) {
124     if ($sock == $lsn) {
125     my $new = $lsn->accept;
126     $sel->add($new);
127 dpavlin 174 $session->{peerport}->{ $new->peerport } = $new;
128 dpavlin 164 warn "[socket] connect\n";
129 dpavlin 163 Storable::store_fd( { ping => 1 }, $new );
130 dpavlin 210 } elsif ( $sock == $www ) {
131     my $client = $www->accept;
132     Sack::Server::HTTP::request( $client, sub {
133     my ( $send, $method, $param ) = @_;
134 dpavlin 212
135     if ( $method =~ m{views} ) {
136 dpavlin 232 run_view $method;
137 dpavlin 212 print $send "HTTP/1.0 302 $method\r\nLocation: /\r\n\r\n";
138 dpavlin 213 return 1;
139 dpavlin 212 } elsif ( $method =~ m{^/tmp/sack} ) {
140 dpavlin 221 load_shard $method;
141 dpavlin 212 print $send "HTTP/1.0 302 $method\r\nLocation: /\r\n\r\n";
142 dpavlin 213 return 1;
143     } elsif ( $method =~ m{^/out/(.+)} ) {
144     print $send "HTTP/1.0 200 OK\r\nContent-Type: text/html\r\n\r\n";
145     Sack::Server::HTML::send_out( $send, Sack::Merge->out, $1, $param );
146     return 1;
147 dpavlin 241 } elsif ( $method =~ m{^/gnuplot} ) {
148     eval {
149 dpavlin 245 my $path = Sack::Server::Gnuplot::date( Sack::Merge->out, $param );
150 dpavlin 241 if ( -e $path ) {
151     print $send "HTTP/1.0 200 OK\r\nContent-Type: image/png\r\n\r\n";
152     open(my $fh, '<', $path) || die $path;
153     my $b;
154     while ( read($fh, $b, 4096) ) {
155     print $send $b;
156     }
157     return 1;
158 dpavlin 245 } else {
159     print $send "HTTP/1.0 404 no graph\r\n\r\n";
160     return 1;
161 dpavlin 241 }
162     };
163     warn "ERROR: $@" if $@;
164 dpavlin 212 }
165    
166     print $send "HTTP/1.0 200 OK\r\nContent-Type: text/html\r\n\r\n";
167    
168 dpavlin 232 print $send qq|
169 dpavlin 212
170 dpavlin 232 <style type="text/css">
171 dpavlin 213
172 dpavlin 232 .running {
173     color: #f00;
174     }
175    
176    
177 dpavlin 219 .forked {
178     color: #f00;
179     }
180 dpavlin 212
181 dpavlin 219 .ping {
182     color: #0f0;
183     }
184 dpavlin 222
185     .shards {
186     font-family: monospace;
187     line-height: 0.8em;
188     color: #0f0;
189     }
190    
191     .send {
192     color: #ff0;
193     }
194    
195     .wait {
196     color: #f00;
197     }
198    
199    
200 dpavlin 219 </style>
201 dpavlin 232
202     |;
203    
204     print $send qq|<h1>Views</h1><ul>|
205     , join("\n", map {
206     my $view = $_;
207     my $html = '';
208     if ( my $s = $info->{view}->{$view} ) {
209     my @nodes = sort keys %$s;
210     my $total = 0;
211 dpavlin 237 $html .= qq|<table><tr><th>node</th><th>rows</th><th>shards</th><th>&sum;</th><th>merge</th></tr>|;
212 dpavlin 232 foreach my $node ( @nodes ) {
213     my $h = '';
214     my $shard = 0;
215     foreach my $path ( sort keys %{ $s->{$node} } ) {
216     my $affected = $s->{$node}->{$path};
217 dpavlin 233 my $n = '?';
218     $n = $1 if $path =~ m{/(\w)\w+/\d+};
219 dpavlin 235 my $class = $affected == 0 ? 'style="color: #888"' : '';
220     $h .= qq|<tt title="$affected $path"$class>$n</tt>|;
221 dpavlin 232 $shard += $affected;
222     $total += $affected;
223     }
224 dpavlin 247 my $merge = $info->{merge}->{$view}->{$node} || '?';
225 dpavlin 237 $html .= qq|<tr><td><tt>$node</tt></td><td>$shard</td><td>$h</td><td>$total</td><td>$merge</td></tr>\n|;
226 dpavlin 232 }
227    
228     $html .= qq|</table>|;
229     };
230     qq|<li><a href="$view">$view</a>$html|
231     } glob '/srv/Sack/views/*/*.pl' )
232     , qq|</ul>|
233     ;
234    
235     my $out = Sack::Merge->out;
236 dpavlin 241 if ( my $li = join("\n", map { qq|<li><a href="/out/$_" target="$_">$_</a>| } keys %$out ) ) {
237     print $send qq|<h1>Results</h1><ul>$li</ul><img src="/gnuplot/">\n|;
238     }
239 dpavlin 232
240     print $send qq|<h1>Nodes</h1>
241 dpavlin 219 |, join("\n", map {
242     my $class = join(' ', map { $_->[0] } @{ $info->{node}->{$_} });
243     qq|<span class="$class">$_</span>|;
244     } sort keys %{ $info->{node} } );
245    
246 dpavlin 221 print $send qq|<h1>Data</h1><ul>|;
247     foreach my $path ( glob '/tmp/sack/*' ) {
248 dpavlin 222 print $send qq|<li><a href="$path">$path</a><div class="shards">|;
249 dpavlin 221 foreach my $s ( sort grep { m/$path/ } keys %{ $info->{shard} } ) {
250 dpavlin 222 my $i = $info->{shard}->{$s};
251     print $send qq|<span class=$i>$i</span> |;
252 dpavlin 221 }
253 dpavlin 222 print $send qq|</div>|;
254 dpavlin 221 }
255     print $send qq|</ul>|;
256 dpavlin 219
257 dpavlin 233 if ( $param->{info} ) {
258     print $send qq|<a href="?info=0">hide info</a>|;
259     print $send '<pre>', dump($info), '</pre>'
260     } else {
261     print $send qq|<a href="?info=1">show info</a>|;
262     }
263 dpavlin 232 return 1;
264 dpavlin 210 } );
265 dpavlin 163 } else {
266 dpavlin 165 my $data = eval { Storable::fd_retrieve( $sock ) };
267     if ( $@ ) {
268     delete $session->{$sock};
269     warn "[socket] disconnect: $@\n";
270     $sel->remove($sock);
271     $sock->close;
272     } else {
273 dpavlin 232 warn "<<<< ", dump($data), $/ if $debug;
274 dpavlin 196
275     if ( my $path = $data->{shard} ) {
276 dpavlin 211 $info->{shard}->{ $path } = $data->{port};
277     # FIXME will need push for multiple copies of shards
278 dpavlin 196 }
279    
280 dpavlin 192 if ( my $repl = $data->{repl} ) {
281     my $response = { repl_pid => $$ };
282     if ( $repl =~ m/ping/ ) {
283 dpavlin 188 to_all { ping => 1 };
284 dpavlin 192 } elsif ( $repl =~ m/info/ ) {
285 dpavlin 175 $response->{info} = $info;
286 dpavlin 192 } elsif ( $repl =~ m{load\s*(\S+)?} ) {
287 dpavlin 221 load_shard $1;
288 dpavlin 196 } elsif ( $repl =~ m{view\s*(\S+)?} ) {
289 dpavlin 232 run_view $1;
290 dpavlin 200 } elsif ( $repl =~ m{out} ) {
291     my $out = Sack::Merge->out;
292     warn "out ",dump( $out );
293     $response->{out} = $out;
294 dpavlin 203 } elsif ( $repl =~ m{clean} ) {
295     delete $info->{shard};
296     to_all { clean => 1 };
297 dpavlin 232 } elsif ( $repl eq 'exit' ) {
298     to_all { exit => 1 };
299     sleep 1;
300     exit;
301 dpavlin 247 } elsif ( $repl eq '.' ) {
302     $response->{'.'} = [ @responses ];
303     @responses = ();
304     } elsif ( $repl =~ m{(\w+)\s*(.+)?} ) {
305     to_all { $1 => $2 };
306 dpavlin 189 } else {
307 dpavlin 247 $response->{error}->{repl} = $repl;
308 dpavlin 174 }
309 dpavlin 166 Storable::store_fd( $response, $sock );
310     } elsif ( $data->{ping} ) {
311 dpavlin 174 my $port = $data->{port};
312     info $port => 'ping', $port;
313     $session->{port}->{ $data->{port} } = $sock;
314 dpavlin 232 } elsif ( defined $data->{load} && $data->{load} eq 'shard' ) {
315 dpavlin 221 if ( my $path = shift @shard_load_queue ) {
316     $info->{shard}->{$path} = 'read';
317 dpavlin 192 my $shard = Storable::retrieve $path;
318 dpavlin 221 $info->{shard}->{$path} = 'send';
319 dpavlin 192 warn ">>>> [", $data->{port}, "] sending shard $path\n";
320     Storable::store_fd( { path => $path, shard => $shard }, $sock );
321     } else {
322     warn "no more shards for [", $data->{port}, "]\n";
323     }
324 dpavlin 235 } elsif ( exists $data->{out} ) {
325     my $added = Sack::Merge->add( $data->{out} ) if defined $data->{out};
326     $info->{merge}->{ $data->{view} }->{ $data->{port} } = $added;
327     $info->{view }->{ $data->{view} }->{ $data->{port} } = $data->{on_shard};
328 dpavlin 247 } elsif ( exists $data->{port} ) {
329     push @responses, $data;
330     warn "# ",dump($data),$/;
331 dpavlin 192 } else {
332     warn "UNKNOWN ",dump($data);
333 dpavlin 164 }
334 dpavlin 163 }
335     }
336     }
337     }

Properties

Name Value
svn:executable *

  ViewVC Help
Powered by ViewVC 1.1.26