/[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 51 - (hide annotations)
Thu Sep 24 19:45:08 2009 UTC (14 years, 8 months ago) by dpavlin
File MIME type: text/plain
File size: 7730 byte(s)
simplify and fix prefix extraction
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     PeerAddr => $node,
103     Proto => 'tcp',
104 dpavlin 16 );
105 dpavlin 11
106 dpavlin 16 if ( ! $sock ) {
107     warn "can't connect to $node - $!"; # FIXME die?
108     next;
109     }
110    
111 dpavlin 30 warn "[$port] >>>> $node $header\n";
112 dpavlin 11 print $sock "$header\n$content" || warn "can't send $header to $node: $!";
113    
114 dpavlin 13 $connected->{$node} = $sock;
115 dpavlin 11 }
116     }
117    
118 dpavlin 16 sub get_node {
119     my $node = shift;
120    
121     my $sock = $connected->{$node};
122     if ( ! $sock ) {
123 dpavlin 30 warn "[$port] ERROR lost connection to $node";
124 dpavlin 16 delete $connected->{$node};
125     return;
126     }
127     chomp( my $size = <$sock> );
128 dpavlin 30 warn "[$port] <<<< $node $size bytes\n";
129 dpavlin 16 my $data;
130     read $sock, $data, $size;
131     return $data;
132     }
133    
134     sub send_sock {
135     my ( $sock, $data ) = @_;
136     my $size = length $data;
137 dpavlin 30 warn "[$port] >>>> ", $sock->peerhost, " $size bytes\n";
138 dpavlin 16 print $sock "$size\n$data" || warn "can't send $size bytes to ", $sock->peerhost;
139     }
140    
141 dpavlin 13 sub merge_out {
142 dpavlin 50 my ( $from_node, $new ) = @_;
143 dpavlin 13
144 dpavlin 50 warn "### merge $from_node";
145    
146     my $from_port = $from_node;
147     $from_port =~ s{.+:(\d+)$}{$1};
148    
149     my $remote_digest = Sack::Digest->new( port => $from_port );
150     my ( $local, $remote ) = ( 0, 0 );
151    
152 dpavlin 13 foreach my $k1 ( keys %$new ) {
153    
154     foreach my $k2 ( keys %{ $new->{$k1} } ) {
155    
156 dpavlin 30 my $n = delete $new->{$k1}->{$k2};
157 dpavlin 50
158     if ( $k1 =~ m{#} ) {
159     die "ASSERT $k1 $k2" unless $k2 =~ m{^\d+$};
160     #warn "XXX $k1 $k2";
161     my $md5 = $remote_digest->{nr_md5}->[$k2] || warn "[$port] no2md5 $n not found in $from_port\n";
162     if ( my $local_k2 = $digest->{md5_nr}->{$md5} ) {
163     $k2 = $local_k2;
164     $local++;
165     } else {
166     $k2 = $digest->to_int( $remote_digest->{md5}->{$md5} );
167     $remote++;
168     }
169     }
170    
171 dpavlin 30 my $ref = ref $out->{$k1}->{$k2};
172 dpavlin 50 #warn "XXXX $k1 $k2 $ref";
173 dpavlin 13 if ( ! defined $out->{$k1}->{$k2} ) {
174     $out->{$k1}->{$k2} = $n;
175     } elsif ( $k1 =~ m{\+} ) {
176 dpavlin 19 # warn "## agregate $k1 $k2";
177 dpavlin 13 $out->{$k1}->{$k2} += $n;
178 dpavlin 21 } elsif ( $ref eq 'ARRAY' ) {
179     if ( ref $n eq 'ARRAY' ) {
180     push @{ $out->{$k1}->{$k2} }, $_ foreach @$n;
181     } else {
182     push @{ $out->{$k1}->{$k2} }, $n;
183     }
184 dpavlin 13 } elsif ( $ref eq '' ) {
185     $out->{$k1}->{$k2} = [ $out->{$k1}->{$k2}, $n ];
186     } else {
187     die "can't merge $k2 [$ref] from ",dump($n), " into ", dump($out->{$k1}->{$k2});
188     }
189     }
190     }
191    
192 dpavlin 50 warn "[$port] merge local $local remote $remote from $from_port\n";
193 dpavlin 19 warn "## merge out ", dump $out if $debug;
194 dpavlin 13 }
195    
196 dpavlin 11 sub run_code {
197     my ( $view, $code ) = @_;
198    
199 dpavlin 21 warn "\n#### CODE $view START ####\n$code\n#### CODE $view END ####\n" if $debug;
200 dpavlin 11
201     send_nodes view => $view => $code;
202    
203     undef $out;
204    
205     my $affected = 0;
206     $t = time;
207    
208     foreach my $pos ( $offset + 1 .. $offset + $input->size ) {
209     my $rec = $cache->{$pos} ||= $input->fetch_rec( $pos );
210     if ( ! $rec ) {
211 dpavlin 41 print STDERR "END @ $pos";
212 dpavlin 11 last;
213     }
214    
215     eval "$code";
216     if ( $@ ) {
217 dpavlin 41 warn "ABORT $pos $@\n";
218 dpavlin 20 last;
219 dpavlin 11 } else {
220     $affected++;
221     }
222 dpavlin 30
223 dpavlin 40 $pos % 10000 == 0 ? print STDERR $pos :
224     $pos % 1000 == 0 ? print STDERR "." : 0 ;
225 dpavlin 11 };
226    
227 dpavlin 30 report "\n[$port] RECS $affected $view";
228 dpavlin 11
229     warn "WARN no \$out defined!" unless defined $out;
230 dpavlin 13
231 dpavlin 46 $digest->sync;
232    
233 dpavlin 13 if ( $connected ) {
234 dpavlin 24 foreach my $node ( keys %$connected ) {
235 dpavlin 30 warn "[$port] get_node $node\n";
236 dpavlin 24 my $o = get_node $node;
237     my $s = length $o;
238     $o = thaw $o;
239 dpavlin 30 warn "[$port] merge $node $s bytes\n";
240 dpavlin 50 merge_out $node => $o;
241 dpavlin 24 }
242 dpavlin 13 }
243 dpavlin 11 }
244    
245 dpavlin 1 sub run_views {
246 dpavlin 6 @views = sort glob 'views/*.pl' unless @views;
247 dpavlin 1 warn "# views ", dump @views;
248    
249     foreach my $view ( @views ) {
250    
251     next if system("perl -c $view") != 0;
252    
253     my $code = read_file $view;
254    
255 dpavlin 11 run_code $view => $code;
256 dpavlin 1
257 dpavlin 11 if ( defined $out ) {
258 dpavlin 1
259 dpavlin 11 my $path = $view;
260     $path =~ s{views?/}{out/} || die "no view in $view";
261 dpavlin 30 $path =~ s{\.pl}{.storable};
262 dpavlin 1
263 dpavlin 9 unlink "$path.last" if -e "$path.last";
264     rename $path, "$path.last";
265 dpavlin 30
266     store $out => $path;
267 dpavlin 31 report "[$port] SAVE $path $offset-$limit", -s $path, "bytes";
268 dpavlin 35
269     if ( -s $path < 4096 ) {
270 dpavlin 45 print '$out = ', dump $digest->undigest_out($out);
271 dpavlin 35 }
272 dpavlin 11 }
273    
274     }
275    
276     }
277    
278 dpavlin 35
279     sub info_tabs {
280     "$port\t$offset\t$limit\t$num_records\t$path\t"
281 dpavlin 36 . join("\t", map {
282 dpavlin 35 my $b = $_;
283 dpavlin 46 $b =~ s{^.+\.$port\.([^/]+)$}{$1};
284 dpavlin 35 "$b " . -s $_
285 dpavlin 46 } glob "/dev/shm/sack.$port.*" );
286 dpavlin 35 }
287    
288    
289 dpavlin 30 if ( $port ) {
290 dpavlin 11 my $sock = IO::Socket::INET->new(
291     Listen => SOMAXCONN,
292 dpavlin 21 LocalAddr => '127.0.0.1',
293 dpavlin 30 LocalPort => $port,
294 dpavlin 11 Proto => 'tcp',
295     Reuse => 1,
296     ) or die $!;
297    
298     while (1) {
299    
300 dpavlin 30 warn "[$port] READY path: $path offset: $offset limit: $limit #recs: $num_records\n";
301 dpavlin 11
302     my $client = $sock->accept();
303    
304 dpavlin 30 warn "[$port] <<<< connect from ", $client->peerhost, $/;
305 dpavlin 11
306     my @header = split(/\s/, <$client>);
307 dpavlin 30 warn "[$port] <<<< header ",dump(@header),$/;
308 dpavlin 11
309     my $size = shift @header;
310    
311     my $content;
312     read $client, $content, $size;
313    
314     if ( $header[0] eq 'view' ) {
315     run_code $header[1] => $content;
316 dpavlin 16 send_sock $client => freeze $out;
317     } elsif ( $header[0] eq 'info' ) {
318 dpavlin 35 my $info = info_tabs;
319 dpavlin 30 warn "[$port] info $info\n";
320 dpavlin 16 send_sock $client => $info;
321     } elsif ( $header[0] eq 'exit' ) {
322 dpavlin 30 warn "[$port] exit";
323 dpavlin 16 exit;
324 dpavlin 5 } else {
325 dpavlin 30 warn "[$port] UNKNOWN $header[0]";
326 dpavlin 1 }
327    
328     }
329     }
330    
331 dpavlin 30 sub info {
332     send_nodes 'info' => $2;
333    
334     my @info = (
335 dpavlin 36 "port\toffset\tlimit\t#recs\tpath",
336     "----\t------\t-----\t-----\t----",
337 dpavlin 35 info_tabs,
338 dpavlin 30 );
339    
340     push @info, get_node $_ foreach @nodes;
341    
342     print "[$port] INFO\n"
343     , join("\n", @info)
344     , "\n\n" ;
345    
346     return @info;
347     }
348    
349     info;
350 dpavlin 3 run_views;
351    
352 dpavlin 1 while ( 1 ) {
353    
354     print "sack> ";
355 dpavlin 21 chomp( my $cmd = <STDIN> );
356 dpavlin 1
357 dpavlin 21 if ( $cmd =~ m{^(h|\?)} ) {
358     print << "__HELP__"
359     Sacks Lorry v$VERSION - path: $path offset: $offset limit: $limit
360    
361     View Run run views
362     VI \\e Output show output of last run
363     Info [\$VERSION] instrospect
364     Quit EXit shutdown
365    
366     __HELP__
367     } elsif ( $cmd =~ m{^(vi|\\e|o)}i ) {
368 dpavlin 33 #system "vi out/*";
369     system "bin/storableedit.pl", (glob('out/*.storable'))[0];
370 dpavlin 21 } elsif ( $cmd =~ m{^i(?:nfo)?\s?(.+)?$}i ) {
371 dpavlin 30 info;
372 dpavlin 21 } elsif ( $cmd =~ m{^(q|e|x)}i ) {
373 dpavlin 16 warn "# exit";
374     send_nodes 'exit';
375     exit;
376 dpavlin 21 } elsif ( $cmd =~ m{^(v|r)}i ) {
377 dpavlin 3 run_views;
378 dpavlin 21 } elsif ( $cmd ) {
379     warn "UNKNOWN ", dump $cmd;
380 dpavlin 3 }
381    
382 dpavlin 1 }
383    

Properties

Name Value
svn:executable *

  ViewVC Help
Powered by ViewVC 1.1.26