/[Sack]/trunk/bin/sack.pl
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/bin/sack.pl

Parent Directory Parent Directory | Revision Log Revision Log


Revision 58 - (hide annotations)
Fri Sep 25 12:24:19 2009 UTC (14 years, 8 months ago) by dpavlin
File MIME type: text/plain
File size: 7876 byte(s)
print send/receive only if smaller than 1k or with debug
1 dpavlin 1 #!/usr/bin/perl
2    
3     use warnings;
4     use strict;
5    
6 dpavlin 50 our $VERSION = '0.04';
7 dpavlin 21
8 dpavlin 1 use Time::HiRes qw(time);
9     use Data::Dump qw(dump);
10     use File::Slurp;
11 dpavlin 4 use Getopt::Long;
12 dpavlin 11 use IO::Socket::INET;
13 dpavlin 30 use Storable qw/freeze thaw store/;
14 dpavlin 1
15 dpavlin 4
16 dpavlin 19 my $debug = 0;
17 dpavlin 4 my $path = '/data/isi/full.txt';
18 dpavlin 11 my $limit = 5000;
19 dpavlin 4 my $offset = 0;
20 dpavlin 6 my @views;
21 dpavlin 30 my $port = 0; # interactive
22 dpavlin 11 my @nodes;
23 dpavlin 4
24    
25     GetOptions(
26     'path=s' => \$path,
27     'offset=i' => \$offset,
28     'limit=i' => \$limit,
29 dpavlin 6 'view=s' => \@views,
30 dpavlin 30 'listen|port=i' => \$port,
31 dpavlin 11 'connect=s' => \@nodes,
32 dpavlin 19 'debug!' => \$debug,
33 dpavlin 5 ) or die $!;
34 dpavlin 4
35 dpavlin 1 my $t = time;
36    
37 dpavlin 12
38 dpavlin 21 sub send_nodes;
39    
40 dpavlin 12 our $prefix;
41 dpavlin 21 sub BEGIN {
42 dpavlin 12 $prefix = $0;
43 dpavlin 30 if ( $prefix !~ m{^/} ) {
44 dpavlin 12 chomp( my $pwd = `pwd` );
45     $prefix = "$pwd/$prefix";
46     }
47 dpavlin 51 $prefix =~ s{^(.*)/srv/Sack/.+$}{$1};
48 dpavlin 12 warn "# prefix $prefix";
49 dpavlin 21
50     $SIG{INT} = sub {
51     my $signame = shift;
52     send_nodes 'exit';
53 dpavlin 33 #clean if $clean; # FIXME
54 dpavlin 21 die "SIG$signame";
55     };
56 dpavlin 12 }
57    
58 dpavlin 33 use lib "$prefix/srv/Sack/lib/";
59     use Sack::Digest;
60 dpavlin 45 our $digest = Sack::Digest->new( port => $port, clean => 1 );
61     sub digest { $digest->to_int($_[0]) }
62 dpavlin 12
63     use lib "$prefix/srv/webpac2/lib/";
64 dpavlin 1 use WebPAC::Input::ISI;
65 dpavlin 24
66     $WebPAC::Input::ISI::subfields = undef; # disable parsing of subfields
67    
68 dpavlin 1 my $input = WebPAC::Input::ISI->new(
69 dpavlin 12 path => "$prefix/$path",
70 dpavlin 4 offset => $offset,
71     limit => $limit,
72 dpavlin 1 );
73    
74 dpavlin 25 our $num_records = $input->size;
75 dpavlin 1
76     sub report {
77 dpavlin 30 my $description = join(' ',@_);
78 dpavlin 1 my $dt = time - $t;
79     printf "%s in %1.4fs %.2f/s\n", $description, $dt, $input->size / $dt;
80 dpavlin 10 $t = time;
81 dpavlin 1 }
82    
83    
84     report $input->size . ' records loaded';
85    
86     mkdir 'out' unless -e 'out';
87    
88 dpavlin 5 our $out;
89    
90 dpavlin 8 our $cache;
91    
92 dpavlin 13 our $connected;
93    
94 dpavlin 11 sub send_nodes {
95 dpavlin 17 my $content = $#_ > 0 ? pop @_ : ''; # no content with just one argument!
96 dpavlin 21 my $header = defined $content ? length($content) : 0;
97 dpavlin 11 $header .= ' ' . join(' ', @_) if @_;
98    
99     foreach my $node ( @nodes ) {
100    
101     my $sock = IO::Socket::INET->new(
102 dpavlin 56 PeerAddr => '127.0.0.1',
103     PeerPort => $node,
104 dpavlin 11 Proto => 'tcp',
105 dpavlin 16 );
106 dpavlin 11
107 dpavlin 16 if ( ! $sock ) {
108     warn "can't connect to $node - $!"; # FIXME die?
109     next;
110     }
111    
112 dpavlin 30 warn "[$port] >>>> $node $header\n";
113 dpavlin 11 print $sock "$header\n$content" || warn "can't send $header to $node: $!";
114    
115 dpavlin 13 $connected->{$node} = $sock;
116 dpavlin 11 }
117     }
118    
119 dpavlin 16 sub get_node {
120     my $node = shift;
121    
122     my $sock = $connected->{$node};
123     if ( ! $sock ) {
124 dpavlin 30 warn "[$port] ERROR lost connection to $node";
125 dpavlin 16 delete $connected->{$node};
126     return;
127     }
128     chomp( my $size = <$sock> );
129 dpavlin 58 warn "[$port] <<<< $node $size bytes\n" if $debug || $size > 1024;
130 dpavlin 16 my $data;
131     read $sock, $data, $size;
132     return $data;
133     }
134    
135     sub send_sock {
136     my ( $sock, $data ) = @_;
137     my $size = length $data;
138 dpavlin 58 warn "[$port] >>>> $size bytes\n" if $debug || $size > 1024;
139 dpavlin 16 print $sock "$size\n$data" || warn "can't send $size bytes to ", $sock->peerhost;
140     }
141    
142 dpavlin 13 sub merge_out {
143 dpavlin 50 my ( $from_node, $new ) = @_;
144 dpavlin 13
145 dpavlin 50 warn "### merge $from_node";
146    
147     my $from_port = $from_node;
148     $from_port =~ s{.+:(\d+)$}{$1};
149    
150     my $remote_digest = Sack::Digest->new( port => $from_port );
151     my ( $local, $remote ) = ( 0, 0 );
152    
153 dpavlin 13 foreach my $k1 ( keys %$new ) {
154    
155     foreach my $k2 ( keys %{ $new->{$k1} } ) {
156    
157 dpavlin 30 my $n = delete $new->{$k1}->{$k2};
158 dpavlin 50
159     if ( $k1 =~ m{#} ) {
160     die "ASSERT $k1 $k2" unless $k2 =~ m{^\d+$};
161     #warn "XXX $k1 $k2";
162     my $md5 = $remote_digest->{nr_md5}->[$k2] || warn "[$port] no2md5 $n not found in $from_port\n";
163     if ( my $local_k2 = $digest->{md5_nr}->{$md5} ) {
164     $k2 = $local_k2;
165     $local++;
166     } else {
167     $k2 = $digest->to_int( $remote_digest->{md5}->{$md5} );
168     $remote++;
169     }
170     }
171    
172 dpavlin 30 my $ref = ref $out->{$k1}->{$k2};
173 dpavlin 50 #warn "XXXX $k1 $k2 $ref";
174 dpavlin 13 if ( ! defined $out->{$k1}->{$k2} ) {
175     $out->{$k1}->{$k2} = $n;
176     } elsif ( $k1 =~ m{\+} ) {
177 dpavlin 19 # warn "## agregate $k1 $k2";
178 dpavlin 13 $out->{$k1}->{$k2} += $n;
179 dpavlin 21 } elsif ( $ref eq 'ARRAY' ) {
180     if ( ref $n eq 'ARRAY' ) {
181     push @{ $out->{$k1}->{$k2} }, $_ foreach @$n;
182     } else {
183     push @{ $out->{$k1}->{$k2} }, $n;
184     }
185 dpavlin 13 } elsif ( $ref eq '' ) {
186     $out->{$k1}->{$k2} = [ $out->{$k1}->{$k2}, $n ];
187     } else {
188     die "can't merge $k2 [$ref] from ",dump($n), " into ", dump($out->{$k1}->{$k2});
189     }
190     }
191     }
192    
193 dpavlin 50 warn "[$port] merge local $local remote $remote from $from_port\n";
194 dpavlin 19 warn "## merge out ", dump $out if $debug;
195 dpavlin 13 }
196    
197 dpavlin 11 sub run_code {
198     my ( $view, $code ) = @_;
199    
200 dpavlin 21 warn "\n#### CODE $view START ####\n$code\n#### CODE $view END ####\n" if $debug;
201 dpavlin 11
202     send_nodes view => $view => $code;
203    
204     undef $out;
205    
206     my $affected = 0;
207     $t = time;
208    
209     foreach my $pos ( $offset + 1 .. $offset + $input->size ) {
210     my $rec = $cache->{$pos} ||= $input->fetch_rec( $pos );
211     if ( ! $rec ) {
212 dpavlin 41 print STDERR "END @ $pos";
213 dpavlin 11 last;
214     }
215    
216     eval "$code";
217     if ( $@ ) {
218 dpavlin 41 warn "ABORT $pos $@\n";
219 dpavlin 20 last;
220 dpavlin 11 } else {
221     $affected++;
222     }
223 dpavlin 30
224 dpavlin 40 $pos % 10000 == 0 ? print STDERR $pos :
225     $pos % 1000 == 0 ? print STDERR "." : 0 ;
226 dpavlin 11 };
227    
228 dpavlin 30 report "\n[$port] RECS $affected $view";
229 dpavlin 11
230     warn "WARN no \$out defined!" unless defined $out;
231 dpavlin 13
232 dpavlin 46 $digest->sync;
233    
234 dpavlin 13 if ( $connected ) {
235 dpavlin 24 foreach my $node ( keys %$connected ) {
236 dpavlin 30 warn "[$port] get_node $node\n";
237 dpavlin 24 my $o = get_node $node;
238     my $s = length $o;
239     $o = thaw $o;
240 dpavlin 30 warn "[$port] merge $node $s bytes\n";
241 dpavlin 50 merge_out $node => $o;
242 dpavlin 24 }
243 dpavlin 13 }
244 dpavlin 11 }
245    
246 dpavlin 1 sub run_views {
247 dpavlin 6 @views = sort glob 'views/*.pl' unless @views;
248 dpavlin 1 warn "# views ", dump @views;
249    
250     foreach my $view ( @views ) {
251    
252     next if system("perl -c $view") != 0;
253    
254     my $code = read_file $view;
255    
256 dpavlin 11 run_code $view => $code;
257 dpavlin 1
258 dpavlin 11 if ( defined $out ) {
259 dpavlin 1
260 dpavlin 11 my $path = $view;
261     $path =~ s{views?/}{out/} || die "no view in $view";
262 dpavlin 30 $path =~ s{\.pl}{.storable};
263 dpavlin 1
264 dpavlin 9 unlink "$path.last" if -e "$path.last";
265     rename $path, "$path.last";
266 dpavlin 30
267     store $out => $path;
268 dpavlin 31 report "[$port] SAVE $path $offset-$limit", -s $path, "bytes";
269 dpavlin 35
270     if ( -s $path < 4096 ) {
271 dpavlin 45 print '$out = ', dump $digest->undigest_out($out);
272 dpavlin 35 }
273 dpavlin 11 }
274    
275     }
276    
277     }
278    
279 dpavlin 35
280     sub info_tabs {
281     "$port\t$offset\t$limit\t$num_records\t$path\t"
282 dpavlin 36 . join("\t", map {
283 dpavlin 35 my $b = $_;
284 dpavlin 46 $b =~ s{^.+\.$port\.([^/]+)$}{$1};
285 dpavlin 35 "$b " . -s $_
286 dpavlin 46 } glob "/dev/shm/sack.$port.*" );
287 dpavlin 35 }
288    
289    
290 dpavlin 30 if ( $port ) {
291 dpavlin 11 my $sock = IO::Socket::INET->new(
292     Listen => SOMAXCONN,
293 dpavlin 21 LocalAddr => '127.0.0.1',
294 dpavlin 30 LocalPort => $port,
295 dpavlin 11 Proto => 'tcp',
296     Reuse => 1,
297     ) or die $!;
298    
299     while (1) {
300    
301 dpavlin 30 warn "[$port] READY path: $path offset: $offset limit: $limit #recs: $num_records\n";
302 dpavlin 11
303     my $client = $sock->accept();
304    
305 dpavlin 30 warn "[$port] <<<< connect from ", $client->peerhost, $/;
306 dpavlin 11
307     my @header = split(/\s/, <$client>);
308 dpavlin 30 warn "[$port] <<<< header ",dump(@header),$/;
309 dpavlin 11
310     my $size = shift @header;
311    
312     my $content;
313     read $client, $content, $size;
314    
315     if ( $header[0] eq 'view' ) {
316     run_code $header[1] => $content;
317 dpavlin 16 send_sock $client => freeze $out;
318     } elsif ( $header[0] eq 'info' ) {
319 dpavlin 35 my $info = info_tabs;
320 dpavlin 30 warn "[$port] info $info\n";
321 dpavlin 16 send_sock $client => $info;
322     } elsif ( $header[0] eq 'exit' ) {
323 dpavlin 30 warn "[$port] exit";
324 dpavlin 16 exit;
325 dpavlin 5 } else {
326 dpavlin 30 warn "[$port] UNKNOWN $header[0]";
327 dpavlin 1 }
328    
329     }
330     }
331    
332 dpavlin 30 sub info {
333     send_nodes 'info' => $2;
334    
335     my @info = (
336 dpavlin 36 "port\toffset\tlimit\t#recs\tpath",
337     "----\t------\t-----\t-----\t----",
338 dpavlin 35 info_tabs,
339 dpavlin 30 );
340    
341     push @info, get_node $_ foreach @nodes;
342    
343     print "[$port] INFO\n"
344     , join("\n", @info)
345     , "\n\n" ;
346    
347     return @info;
348     }
349    
350     info;
351 dpavlin 3 run_views;
352    
353 dpavlin 1 while ( 1 ) {
354    
355     print "sack> ";
356 dpavlin 21 chomp( my $cmd = <STDIN> );
357 dpavlin 1
358 dpavlin 21 if ( $cmd =~ m{^(h|\?)} ) {
359     print << "__HELP__"
360     Sacks Lorry v$VERSION - path: $path offset: $offset limit: $limit
361    
362     View Run run views
363     VI \\e Output show output of last run
364     Info [\$VERSION] instrospect
365     Quit EXit shutdown
366    
367     __HELP__
368     } elsif ( $cmd =~ m{^(vi|\\e|o)}i ) {
369 dpavlin 33 #system "vi out/*";
370 dpavlin 53 $digest->sync;
371 dpavlin 33 system "bin/storableedit.pl", (glob('out/*.storable'))[0];
372 dpavlin 21 } elsif ( $cmd =~ m{^i(?:nfo)?\s?(.+)?$}i ) {
373 dpavlin 30 info;
374 dpavlin 21 } elsif ( $cmd =~ m{^(q|e|x)}i ) {
375 dpavlin 16 warn "# exit";
376     send_nodes 'exit';
377     exit;
378 dpavlin 21 } elsif ( $cmd =~ m{^(v|r)}i ) {
379 dpavlin 3 run_views;
380 dpavlin 55 } elsif ( $cmd =~ m{^n(ode)?\s*(\d+)}i ) {
381     push @nodes, $1;
382     info;
383 dpavlin 21 } elsif ( $cmd ) {
384     warn "UNKNOWN ", dump $cmd;
385 dpavlin 3 }
386    
387 dpavlin 1 }
388    

Properties

Name Value
svn:executable *

  ViewVC Help
Powered by ViewVC 1.1.26