/[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

Diff of /trunk/bin/sack.pl

Parent Directory Parent Directory | Revision Log Revision Log | View Patch Patch

revision 14 by dpavlin, Tue Sep 22 10:40:32 2009 UTC revision 64 by dpavlin, Fri Sep 25 14:01:31 2009 UTC
# Line 3  Line 3 
3  use warnings;  use warnings;
4  use strict;  use strict;
5    
6    our $VERSION = '0.04';
7    
8  use Time::HiRes qw(time);  use Time::HiRes qw(time);
9  use Data::Dump qw(dump);  use Data::Dump qw(dump);
10  use File::Slurp;  use File::Slurp;
11  use Getopt::Long;  use Getopt::Long;
12  use IO::Socket::INET;  use IO::Socket::INET;
13  use Storable qw/freeze thaw/;  use Storable qw/freeze thaw store/;
14    
15    
16    my $debug  = 0;
17  my $path   = '/data/isi/full.txt';  my $path   = '/data/isi/full.txt';
18  my $limit  = 5000;  my $limit  = 5000;
19  my $offset = 0;  my $offset = 0;
20  my @views;  my @views;
21  my $listen;  my $port   = 0; # interactive
22  my @nodes;  my @nodes;
23    
24    
# Line 24  GetOptions( Line 27  GetOptions(
27          'offset=i' => \$offset,          'offset=i' => \$offset,
28          'limit=i'  => \$limit,          'limit=i'  => \$limit,
29          'view=s'   => \@views,          'view=s'   => \@views,
30          'listen|port=i' => \$listen,          'port|listen=i' => \$port,
31          'connect=s'   => \@nodes,          'node|connect=i'   => \@nodes,
32            'debug!'   => \$debug,
33  ) or die $!;  ) or die $!;
34    
35  my $t = time;  my $t = time;
36    
37    
38    sub send_nodes;
39    
40  our $prefix;  our $prefix;
41  BEGIN {  sub BEGIN {
42          $prefix = $0;          $prefix = $0;
43          if ( $prefix =~ s{^./}{} ) {          if ( $prefix !~ m{^/} ) {
44                  chomp( my $pwd = `pwd` );                  chomp( my $pwd = `pwd` );
45                  $prefix = "$pwd/$prefix";                  $prefix = "$pwd/$prefix";
46          }          }
47          $prefix =~ s{^(.*)/srv/Sack/bin.+$}{$1};          $prefix =~ s{^(.*)/srv/Sack/.+$}{$1};
48          warn "# prefix $prefix";          warn "# prefix $prefix";
49    
50            $SIG{INT} = sub {
51                    my $signame = shift;
52                    send_nodes 'exit';
53                    #clean if $clean;       # FIXME
54                    die "SIG$signame";
55            };
56  }  }
57    
58    use lib "$prefix/srv/Sack/lib/";
59    use Sack::Digest;
60    our $digest = Sack::Digest->new( port => $port, clean => 1 );
61    sub digest { $digest->to_int($_[0]) }
62    
63  use lib "$prefix/srv/webpac2/lib/";  use lib "$prefix/srv/webpac2/lib/";
64  use WebPAC::Input::ISI;  use WebPAC::Input::ISI;
65    
66    $WebPAC::Input::ISI::subfields = undef; # disable parsing of subfields
67    
68  my $input = WebPAC::Input::ISI->new(  my $input = WebPAC::Input::ISI->new(
69          path   => "$prefix/$path",          path   => "$prefix/$path",
70          offset => $offset,          offset => $offset,
71          limit  => $limit,          limit  => $limit,
72  );  );
73    
74    our $num_records = $input->size;
75    
76  sub report {  sub report {
77          my $description = shift;          my $description = join(' ',@_);
78          my $dt = time - $t;          my $dt = time - $t;
79          printf "%s in %1.4fs %.2f/s\n", $description, $dt, $input->size / $dt;          printf "%s in %1.4fs %.2f/s\n", $description, $dt, $input->size / $dt;
80          $t = time;          $t = time;
# Line 70  our $cache; Line 91  our $cache;
91    
92  our $connected;  our $connected;
93    
94    sub node_sock {
95            my $node = shift;
96            my $sock = IO::Socket::INET->new(
97                            PeerAddr => '127.0.0.1',
98                            PeerPort => $node,
99                            Proto    => 'tcp',
100            );
101            
102            return $sock if $sock && $sock->connected;
103    
104            warn "[$port] can't connect to $node - $!\n"; # FIXME die?
105            return;
106    }
107    
108  sub send_nodes {  sub send_nodes {
109          my $content = pop @_;          my $content = $#_ > 0 ? pop @_ : '';    # no content with just one argument!
110          my $header = length($content);          my $header = defined $content ? length($content) : 0;
111          $header .= ' ' . join(' ', @_) if @_;          $header .= ' ' . join(' ', @_) if @_;
112    
113          foreach my $node ( @nodes ) {          warn "# send_nodes ", dump @nodes;
114    
115                  my $sock = IO::Socket::INET->new(          foreach my $node ( @nodes ) {
                         PeerAddr => $node,  
                         Proto    => 'tcp',  
                 ) or die "can't connect to $node - $!";  
116    
117                  print ">>>> $node $header\n";                  my $sock = node_sock($node) || next;
118    
119                    warn "[$port] >>>> $node $header\n";
120                  print $sock "$header\n$content" || warn "can't send $header to $node: $!";                  print $sock "$header\n$content" || warn "can't send $header to $node: $!";
121    
122                  $connected->{$node} = $sock;                  $connected->{$node} = $sock;
123          }          }
124  }  }
125    
126    sub get_node {
127            my $node = shift;
128    
129            my $sock = $connected->{$node};
130            if ( ! $sock || ! $sock->connected ) {
131                    warn "[$port] no connection to $node";
132                    delete $connected->{$node};
133                    return;
134            }
135            chomp( my $size = <$sock> );
136            warn "[$port] <<<< $node $size bytes\n" if $debug || $size > 1024;
137            my $data;
138            read $sock, $data, $size;
139            return $data;
140    }
141    
142    sub send_sock {
143            my ( $sock, $data ) = @_;
144            my $size   = length $data;
145            warn "[$port] >>>> $size bytes\n" if $debug || $size > 1024;
146            print $sock "$size\n$data" || warn "can't send $size bytes to ", $sock->peerhost;
147    }
148    
149    sub pull_node_file {
150            my ( $node, $file ) = @_;
151    
152            my $path = "/dev/shm/sack.$node.$file";
153    #       return if -e $path; # FIXME
154    
155            my $sock = node_sock $node || die "not connected to $node";
156    
157            print $sock "0 file $file\n";
158    
159            my $size = <$sock>;
160            chomp($size);
161            warn "[$port] pull_node_file $node $file $size bytes\n";
162    
163            my $block = 4096;
164            my $buff  = ' ' x $block;
165    
166            open(my $fh, '>', $path) || die "can't open $path";
167            while ( read $sock, $buff, $block ) {
168                    print $fh $buff;
169            }
170            close($fh);
171    }
172    
173  sub merge_out {  sub merge_out {
174          my $new = shift;          my ( $from_node, $new ) = @_;
175    
176          warn "## merge $new\n";          pull_node_file $from_node => 'nr_md5';
177            pull_node_file $from_node => 'md5';
178    
179            my $remote_digest = Sack::Digest->new( port => $from_node );
180            my ( $local, $remote ) = ( 0, 0 );
181    
182            my $tick = 0;
183            print STDERR "[$port] merge $from_node";
184    
185            my $missing;
186    
187          foreach my $k1 ( keys %$new ) {          foreach my $k1 ( keys %$new ) {
188    
189                  foreach my $k2 ( keys %{ $new->{$k1} } ) {                  foreach my $k2 ( keys %{ $new->{$k1} } ) {
190    
191                          my $n   =     $new->{$k1}->{$k2};                          my $n   = delete $new->{$k1}->{$k2};
                         my $ref = ref $out->{$k1}->{$k2};  
192    
193                            if ( $k1 =~ m{#} ) {
194                                    die "ASSERT $k1 $k2" unless $k2 =~ m{^\d+$};
195    #warn "XXX $k1 $k2";
196                                    my $md5 = $remote_digest->{nr_md5}->[$k2];
197    
198                                    if ( ! $md5 ) {
199                                            $missing->{$from_node}++; # FIXME die?
200                                            next;
201                                    }
202    
203                                    if ( my $local_k2 = $digest->{md5_nr}->{$md5} ) {
204                                            $k2 = $local_k2;
205                                            $local++;
206                                    } else {
207                                            $k2 = $digest->to_int( $remote_digest->{md5}->{$md5} );
208                                            $remote++;
209                                    }
210                            }
211    
212                            my $ref = ref    $out->{$k1}->{$k2};
213    #warn "XXXX $k1 $k2 $ref";
214                          if ( ! defined $out->{$k1}->{$k2} ) {                          if ( ! defined $out->{$k1}->{$k2} ) {
215                                  $out->{$k1}->{$k2} = $n;                                  $out->{$k1}->{$k2} = $n;
216                          } elsif ( $k1 =~ m{\+} ) {                          } elsif ( $k1 =~ m{\+} ) {
217                                  warn "# agregate $k1 $k2";  #                               warn "## agregate $k1 $k2";
218                                  $out->{$k1}->{$k2} += $n;                                  $out->{$k1}->{$k2} += $n;
219                          } elsif ( $ref eq 'ARRAY' ) {                          } elsif ( $ref  eq 'ARRAY' ) {
220                                  push @{ $out->{$k1}->{$k2} }, $n;                                  if ( ref $n eq 'ARRAY' ) {
221                                            push @{ $out->{$k1}->{$k2} }, $_ foreach @$n;
222                                    } else {
223                                            push @{ $out->{$k1}->{$k2} }, $n;
224                                    }
225                          } elsif ( $ref eq '' ) {                          } elsif ( $ref eq '' ) {
226                                  $out->{$k1}->{$k2} = [ $out->{$k1}->{$k2}, $n ];                                  $out->{$k1}->{$k2} = [ $out->{$k1}->{$k2}, $n ];
227                          } else {                          } else {
228                                  die "can't merge $k2 [$ref] from ",dump($n), " into ", dump($out->{$k1}->{$k2});                                  die "can't merge $k2 [$ref] from ",dump($n), " into ", dump($out->{$k1}->{$k2});
229                          }                          }
230    
231                            if ( $tick++ % 1000 == 0 ) {
232                                    print STDERR ".";
233                            } elsif ( $tick % 10000 == 0 ) {
234                                    print STDERR $tick;
235                            }
236                  }                  }
237          }          }
238    
239          warn "## merge out ", dump $out;          warn "$tick $local/$remote\n";
240    
241            warn "[$port] missing ", dump $missing if $missing;
242    
243            warn "## merge out ", dump $out if $debug;
244  }  }
245    
246  sub run_code {  sub run_code {
247          my ( $view, $code ) = @_;          my ( $view, $code ) = @_;
248    
249          warn "\n#### CODE $view START ####\n$code\n#### CODE $view END ####\n";          warn "\n#### CODE $view START ####\n$code\n#### CODE $view END ####\n" if $debug;
250    
251          send_nodes view => $view => $code;          send_nodes view => $view => $code;
252    
# Line 135  sub run_code { Line 258  sub run_code {
258          foreach my $pos ( $offset + 1 .. $offset + $input->size ) {          foreach my $pos ( $offset + 1 .. $offset + $input->size ) {
259                  my $rec = $cache->{$pos} ||= $input->fetch_rec( $pos );                  my $rec = $cache->{$pos} ||= $input->fetch_rec( $pos );
260                  if ( ! $rec ) {                  if ( ! $rec ) {
261                          warn "END at $pos";                          print STDERR "END @ $pos";
262                          last;                          last;
263                  }                  }
264    
265                  eval "$code";                  eval "$code";
266                  if ( $@ ) {                  if ( $@ ) {
267                          warn "ERROR [$pos] $@\n";                          warn "ABORT $pos $@\n";
268                            last;
269                  } else {                  } else {
270                          $affected++;                          $affected++;
271                  }                  }
272    
273                    $pos % 10000 == 0 ? print STDERR $pos :
274                    $pos % 1000  == 0 ? print STDERR "."  : 0 ;
275          };          };
276    
277          report "$affected affected records $view";          report "\n[$port] RECS $affected $view";
278    
279          warn "WARN no \$out defined!" unless defined $out;          warn "WARN no \$out defined!" unless defined $out;
280    
281          if ( $connected ) {          $digest->sync;
                 warn "# get results from ", join(' ', keys %$connected );  
282    
283            if ( $connected ) {
284                  foreach my $node ( keys %$connected ) {                  foreach my $node ( keys %$connected ) {
285                          my $sock = $connected->{$node};                          warn "[$port] get_node $node\n";
286                          my $size = <$sock>;                          my $o = get_node $node;
287                          warn "<<<< $node $size bytes\n";                          next unless $o;
288                          my $part;                          my $s = length $o;
289                          read $sock, $part, $size;                          $o = thaw $o;
290                          merge_out( thaw $part );                          warn "[$port] merge $node $s bytes\n";
291                            merge_out $node => $o;
292                            report "[$port] merged $node";
293                  }                  }
294          }          }
295  }  }
# Line 178  sub run_views { Line 307  sub run_views {
307                  run_code $view => $code;                  run_code $view => $code;
308    
309                  if ( defined $out ) {                  if ( defined $out ) {
                         my $dump = dump $out;  
                         my $len  = length $dump;  
310    
311                          my $path = $view;                          my $path = $view;
312                          $path =~ s{views?/}{out/} || die "no view in $view";                          $path =~ s{views?/}{out/} || die "no view in $view";
313                          $path =~ s{\.pl}{};                          $path =~ s{\.pl}{.storable};
   
                         print "OUT $view $offset/$limit $len bytes $path"  
                                 , ( $len < 10000 ?  " \$out = $dump" : ' SAVED ONLY' )  
                                 , "\n"  
                                 ;  
314    
315                          unlink "$path.last" if -e "$path.last";                          unlink "$path.last" if -e "$path.last";
316                          rename $path, "$path.last";                          rename $path, "$path.last";
317                          write_file $path, $dump;  
318                          report "SAVE $path";                          store $out => $path;
319                            report "[$port] SAVE $path $offset-$limit", -s $path, "bytes";
320    
321                            if ( -s $path < 4096 ) {
322                                    print '$out = ', dump $digest->undigest_out($out);
323                            }
324                  }                  }
325    
326          }          }
327    
328  }  }
329    
330  if ( $listen ) {  
331    sub info_tabs {
332            "$port\t$offset\t$limit\t$num_records\t$path\t"
333            . join("\t", map {
334                    my $b = $_;
335                    $b =~ s{^.+\.$port\.([^/]+)$}{$1};
336                    "$b " . -s $_
337            } glob "/dev/shm/sack.$port.*" );
338    }
339    
340    
341    if ( $port ) {
342          my $sock = IO::Socket::INET->new(          my $sock = IO::Socket::INET->new(
343                  Listen    => SOMAXCONN,                  Listen    => SOMAXCONN,
344  #               LocalAddr => '0.0.0.0',                  LocalAddr => '127.0.0.1',
345                  LocalPort => $listen,                  LocalPort => $port,
346                  Proto     => 'tcp',                  Proto     => 'tcp',
347                  Reuse     => 1,                  Reuse     => 1,
348          ) or die $!;          ) or die $!;
349    
350          while (1) {          while (1) {
351    
352                  warn "NODE listen on $listen\n";                  warn "[$port] READY path: $path offset: $offset limit: $limit #recs: $num_records\n";
353    
354                  my $client = $sock->accept();                  my $client = $sock->accept();
355    
356                  warn "<<<< connect from ", $client->peerhost, $/;                  warn "[$port] <<<< connect from ", $client->peerhost, $/;
357    
358                  my @header = split(/\s/, <$client>);                  my @header = split(/\s/, <$client>);
359                  warn "# header ",dump @header;                  warn "[$port] <<<< header ",dump(@header),$/;
360    
361                  my $size = shift @header;                  my $size = shift @header;
362    
# Line 227  if ( $listen ) { Line 365  if ( $listen ) {
365    
366                  if ( $header[0] eq 'view' ) {                  if ( $header[0] eq 'view' ) {
367                          run_code $header[1] => $content;                          run_code $header[1] => $content;
368                            send_sock $client => freeze $out;
369                          my $frozen = freeze $out;                  } elsif ( $header[0] eq 'info' ) {
370                          my $size   = length $frozen;                          my $info = info_tabs;
371                          warn ">>>> $size bytes";                          warn "[$port] info $info\n";
372                          print $client "$size\n$frozen";                          send_sock $client => $info;
373                    } elsif ( $header[0] eq 'exit' ) {
374                            warn "[$port] exit";
375                            exit;
376                    } elsif ( $header[0] eq 'file' ) {
377                            $digest->close;
378                            my $path = "/dev/shm/sack.$port.$header[1]";
379                            my $size = -s $path;
380                            warn "[$port] >>>> file $path $size bytes\n";
381                            print $client "$size\n";
382                            my $block = 4096;
383                            my $buff  = ' ' x $block;
384                            open(my $fh, '<', $path) || die "can't open $path";
385                            while ( read $fh, $buff, $block ) {
386                                    print $client $buff;
387                            }
388                  } else {                  } else {
389                          warn "WARN unknown";                          warn "[$port] UNKNOWN $header[0]";
390                  }                  }
391    
392          }          }
393  }  }
394    
395    sub info {
396            send_nodes 'info' => $2;
397    
398            my @info = (
399                    "port\toffset\tlimit\t#recs\tpath",
400                    "----\t------\t-----\t-----\t----",
401                    info_tabs,
402            );
403    
404            push @info, get_node $_ foreach @nodes;
405    
406            print "[$port] INFO\n"
407                    , join("\n", @info)
408                    , "\n\n" ;
409    
410            return @info;
411    }
412    
413    info;
414  run_views;  run_views;
415    
416  while ( 1 ) {  while ( 1 ) {
417    
418          print "sack> ";          print "sack> ";
419          my $cmd = <STDIN>;          chomp( my $cmd = <STDIN> );
420    
421          if ( $cmd =~ m{^(vi?|\\e|o(?:ut)?)}i ) {          if ( $cmd =~ m{^(h|\?)} ) {
422                  system "vi out/*";                  print << "__HELP__"
423          } elsif ( $cmd =~ m{^i(nfo)?}i ) {  Sacks Lorry v$VERSION - path: $path offset: $offset limit: $limit
424                  print "nodes: ", dump @nodes, $/;  
425          } else {          View Run        run views
426            VI \\e Output   show output of last run
427            Info [\$VERSION]        instrospect
428            Quit EXit       shutdown
429    
430    __HELP__
431            } elsif ( $cmd =~ m{^(vi|\\e|o)}i ) {
432                    #system "vi out/*";
433                    $digest->sync;
434                    system "bin/storableedit.pl", (glob('out/*.storable'))[0];
435            } elsif ( $cmd =~ m{^i(?:nfo)?\s?(.+)?$}i ) {
436                    info;
437            } elsif ( $cmd =~ m{^(q|e|x)}i ) {
438                    warn "# exit";
439                    send_nodes 'exit';
440                    exit;
441            } elsif ( $cmd =~ m{^(v|r)}i ) {
442                  run_views;                  run_views;
443            } elsif ( $cmd =~ m{^n(ode)?\s*(\d+)}i ) {
444                    push @nodes, $2;
445                    info;
446            } elsif ( $cmd ) {
447                    warn "UNKNOWN ", dump $cmd;
448          }          }
449    
450  }  }
Legend:
Removed from v.14  
changed lines
  Added in v.64
  ViewVC Help
Powered by ViewVC 1.1.26