/[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 26 by dpavlin, Tue Sep 22 21:45:04 2009 UTC revision 60 by dpavlin, Fri Sep 25 13:02:32 2009 UTC
# Line 3  Line 3 
3  use warnings;  use warnings;
4  use strict;  use strict;
5    
6  our $VERSION = '0.01';  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;  my $debug  = 0;
# Line 18  my $path   = '/data/isi/full.txt'; Line 18  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 = 0; # off  my $port   = 0; # interactive
22  my @nodes;  my @nodes;
23    
24    
# Line 27  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,          'listen|port=i' => \$port,
31          'connect=s'   => \@nodes,          'connect=s'   => \@nodes,
32          'debug!'   => \$debug,          'debug!'   => \$debug,
33  ) or die $!;  ) or die $!;
# Line 40  sub send_nodes; Line 40  sub send_nodes;
40  our $prefix;  our $prefix;
41  sub 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 {          $SIG{INT} = sub {
51                  my $signame = shift;                  my $signame = shift;
52                  send_nodes 'exit';                  send_nodes 'exit';
53                    #clean if $clean;       # FIXME
54                  die "SIG$signame";                  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;
# Line 69  my $input = WebPAC::Input::ISI->new( Line 74  my $input = WebPAC::Input::ISI->new(
74  our $num_records = $input->size;  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 91  sub send_nodes { Line 96  sub send_nodes {
96          my $header = defined $content ? length($content) : 0;          my $header = defined $content ? length($content) : 0;
97          $header .= ' ' . join(' ', @_) if @_;          $header .= ' ' . join(' ', @_) if @_;
98    
99            warn "# send_nodes ", dump @nodes;
100    
101          foreach my $node ( @nodes ) {          foreach my $node ( @nodes ) {
102    
103                  my $sock = IO::Socket::INET->new(                  my $sock = IO::Socket::INET->new(
104                          PeerAddr => $node,                          PeerAddr => '127.0.0.1',
105                            PeerPort => $node,
106                          Proto    => 'tcp',                          Proto    => 'tcp',
107                  );                  );
108    
109                  if ( ! $sock ) {                  if ( ! $sock ) {
110                          warn "can't connect to $node - $!"; # FIXME die?                          warn "[$port] can't connect to $node - $!\n"; # FIXME die?
111                          next;                          next;
112                  }                  }
113    
114                  print ">>>> $listen $node $header\n";                  warn "[$port] >>>> $node $header\n";
115                  print $sock "$header\n$content" || warn "can't send $header to $node: $!";                  print $sock "$header\n$content" || warn "can't send $header to $node: $!";
116    
117                  $connected->{$node} = $sock;                  $connected->{$node} = $sock;
# Line 115  sub get_node { Line 123  sub get_node {
123    
124          my $sock = $connected->{$node};          my $sock = $connected->{$node};
125          if ( ! $sock ) {          if ( ! $sock ) {
126                  warn "ERROR: lost connection to $node";                  warn "[$port] ERROR lost connection to $node";
127                  delete $connected->{$node};                  delete $connected->{$node};
128                  return;                  return;
129          }          }
130          chomp( my $size = <$sock> );          chomp( my $size = <$sock> );
131          warn "<<<< $listen $node $size bytes\n";          warn "[$port] <<<< $node $size bytes\n" if $debug || $size > 1024;
132          my $data;          my $data;
133          read $sock, $data, $size;          read $sock, $data, $size;
134          return $data;          return $data;
# Line 129  sub get_node { Line 137  sub get_node {
137  sub send_sock {  sub send_sock {
138          my ( $sock, $data ) = @_;          my ( $sock, $data ) = @_;
139          my $size   = length $data;          my $size   = length $data;
140          warn ">>>> $listen ", $sock->peerhost, " $size bytes\n";          warn "[$port] >>>> $size bytes\n" if $debug || $size > 1024;
141          print $sock "$size\n$data" || warn "can't send $size bytes to ", $sock->peerhost;          print $sock "$size\n$data" || warn "can't send $size bytes to ", $sock->peerhost;
142  }  }
143    
144  sub merge_out {  sub merge_out {
145          my $new = shift;          my ( $from_node, $new ) = @_;
146    
147            my $remote_digest = Sack::Digest->new( port => $from_node );
148            my ( $local, $remote ) = ( 0, 0 );
149    
150            my $tick = 0;
151            print STDERR "[$port] merge $from_node";
152    
153          foreach my $k1 ( keys %$new ) {          foreach my $k1 ( keys %$new ) {
154    
155                  foreach my $k2 ( keys %{ $new->{$k1} } ) {                  foreach my $k2 ( keys %{ $new->{$k1} } ) {
156    
157                          my $n   =     $new->{$k1}->{$k2};                          my $n   = delete $new->{$k1}->{$k2};
158                          my $ref = ref $out->{$k1}->{$k2};  
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_node\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                            my $ref = ref    $out->{$k1}->{$k2};
173    #warn "XXXX $k1 $k2 $ref";
174                          if ( ! defined $out->{$k1}->{$k2} ) {                          if ( ! defined $out->{$k1}->{$k2} ) {
175                                  $out->{$k1}->{$k2} = $n;                                  $out->{$k1}->{$k2} = $n;
176                          } elsif ( $k1 =~ m{\+} ) {                          } elsif ( $k1 =~ m{\+} ) {
# Line 159  sub merge_out { Line 187  sub merge_out {
187                          } else {                          } else {
188                                  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});
189                          }                          }
190    
191                            if ( $tick++ % 1000 == 0 ) {
192                                    print STDERR ".";
193                            } elsif ( $tick % 10000 == 0 ) {
194                                    print STDERR $tick;
195                            }
196                  }                  }
197          }          }
198    
199            warn "$tick $local/$remote\n";
200    
201          warn "## merge out ", dump $out if $debug;          warn "## merge out ", dump $out if $debug;
202  }  }
203    
# Line 180  sub run_code { Line 216  sub run_code {
216          foreach my $pos ( $offset + 1 .. $offset + $input->size ) {          foreach my $pos ( $offset + 1 .. $offset + $input->size ) {
217                  my $rec = $cache->{$pos} ||= $input->fetch_rec( $pos );                  my $rec = $cache->{$pos} ||= $input->fetch_rec( $pos );
218                  if ( ! $rec ) {                  if ( ! $rec ) {
219                          warn "END at $pos";                          print STDERR "END @ $pos";
220                          last;                          last;
221                  }                  }
222    
223                  eval "$code";                  eval "$code";
224                  if ( $@ ) {                  if ( $@ ) {
225                          warn "ABORT [$pos] $@\n";                          warn "ABORT $pos $@\n";
226                          last;                          last;
227                  } else {                  } else {
228                          $affected++;                          $affected++;
229                  }                  }
230    
231                    $pos % 10000 == 0 ? print STDERR $pos :
232                    $pos % 1000  == 0 ? print STDERR "."  : 0 ;
233          };          };
234    
235          report "$affected affected records $view";          report "\n[$port] RECS $affected $view";
236    
237          warn "WARN no \$out defined!" unless defined $out;          warn "WARN no \$out defined!" unless defined $out;
238    
239            $digest->sync;
240    
241          if ( $connected ) {          if ( $connected ) {
242                  foreach my $node ( keys %$connected ) {                  foreach my $node ( keys %$connected ) {
243                          warn "# $listen get_node $node\n";                          warn "[$port] get_node $node\n";
244                          my $o = get_node $node;                          my $o = get_node $node;
245                          my $s = length $o;                          my $s = length $o;
246                          $o = thaw $o;                          $o = thaw $o;
247                          warn "# $listen merge $s bytes\n";                          warn "[$port] merge $node $s bytes\n";
248                          merge_out $o;                          merge_out $node => $o;
249                            report "[$port] merged $node";
250                  }                  }
251          }          }
252  }  }
# Line 222  sub run_views { Line 264  sub run_views {
264                  run_code $view => $code;                  run_code $view => $code;
265    
266                  if ( defined $out ) {                  if ( defined $out ) {
                         my $dump = dump $out;  
                         my $len  = length $dump;  
267    
268                          my $path = $view;                          my $path = $view;
269                          $path =~ s{views?/}{out/} || die "no view in $view";                          $path =~ s{views?/}{out/} || die "no view in $view";
270                          $path =~ s{\.pl}{};                          $path =~ s{\.pl}{.storable};
   
                         print "OUT $view $offset/$limit $len bytes $path"  
                                 , ( $len < 10000 ?  " \$out = $dump" : ' SAVED ONLY' )  
                                 , "\n"  
                                 ;  
271    
272                          unlink "$path.last" if -e "$path.last";                          unlink "$path.last" if -e "$path.last";
273                          rename $path, "$path.last";                          rename $path, "$path.last";
274                          write_file $path, $dump;  
275                          report "SAVE $path";                          store $out => $path;
276                            report "[$port] SAVE $path $offset-$limit", -s $path, "bytes";
277    
278                            if ( -s $path < 4096 ) {
279                                    print '$out = ', dump $digest->undigest_out($out);
280                            }
281                  }                  }
282    
283          }          }
284    
285  }  }
286    
287  if ( $listen ) {  
288    sub info_tabs {
289            "$port\t$offset\t$limit\t$num_records\t$path\t"
290            . join("\t", map {
291                    my $b = $_;
292                    $b =~ s{^.+\.$port\.([^/]+)$}{$1};
293                    "$b " . -s $_
294            } glob "/dev/shm/sack.$port.*" );
295    }
296    
297    
298    if ( $port ) {
299          my $sock = IO::Socket::INET->new(          my $sock = IO::Socket::INET->new(
300                  Listen    => SOMAXCONN,                  Listen    => SOMAXCONN,
301                  LocalAddr => '127.0.0.1',                  LocalAddr => '127.0.0.1',
302                  LocalPort => $listen,                  LocalPort => $port,
303                  Proto     => 'tcp',                  Proto     => 'tcp',
304                  Reuse     => 1,                  Reuse     => 1,
305          ) or die $!;          ) or die $!;
306    
307          while (1) {          while (1) {
308    
309                  warn "NODE $listen ready - path: $path offset: $offset limit: $limit #recs: $num_records\n";                  warn "[$port] READY path: $path offset: $offset limit: $limit #recs: $num_records\n";
310    
311                  my $client = $sock->accept();                  my $client = $sock->accept();
312    
313                  warn "<<<< $listen connect from ", $client->peerhost, $/;                  warn "[$port] <<<< connect from ", $client->peerhost, $/;
314    
315                  my @header = split(/\s/, <$client>);                  my @header = split(/\s/, <$client>);
316                  warn "<<<< $listen header ",dump(@header),$/;                  warn "[$port] <<<< header ",dump(@header),$/;
317    
318                  my $size = shift @header;                  my $size = shift @header;
319    
# Line 273  if ( $listen ) { Line 324  if ( $listen ) {
324                          run_code $header[1] => $content;                          run_code $header[1] => $content;
325                          send_sock $client => freeze $out;                          send_sock $client => freeze $out;
326                  } elsif ( $header[0] eq 'info' ) {                  } elsif ( $header[0] eq 'info' ) {
327                          my $info = "$listen\t$offset\t$limit\t$num_records\t$path";                          my $info = info_tabs;
328                          $info .= "\t" . eval $header[1] if $header[1];                          warn "[$port] info $info\n";
                         warn "info $info\n";  
329                          send_sock $client => $info;                          send_sock $client => $info;
330                  } elsif ( $header[0] eq 'exit' ) {                  } elsif ( $header[0] eq 'exit' ) {
331                          warn "exit $listen";                          warn "[$port] exit";
332                          exit;                          exit;
333                  } else {                  } else {
334                          warn "WARN $listen unknown";                          warn "[$port] UNKNOWN $header[0]";
335                  }                  }
336    
337          }          }
338  }  }
339    
340    sub info {
341            send_nodes 'info' => $2;
342    
343            my @info = (
344                    "port\toffset\tlimit\t#recs\tpath",
345                    "----\t------\t-----\t-----\t----",
346                    info_tabs,
347            );
348    
349            push @info, get_node $_ foreach @nodes;
350    
351            print "[$port] INFO\n"
352                    , join("\n", @info)
353                    , "\n\n" ;
354    
355            return @info;
356    }
357    
358    info;
359  run_views;  run_views;
360    
361  while ( 1 ) {  while ( 1 ) {
# Line 305  Sacks Lorry v$VERSION - path: $path offs Line 374  Sacks Lorry v$VERSION - path: $path offs
374    
375  __HELP__  __HELP__
376          } elsif ( $cmd =~ m{^(vi|\\e|o)}i ) {          } elsif ( $cmd =~ m{^(vi|\\e|o)}i ) {
377                  system "vi out/*";                  #system "vi out/*";
378                    $digest->sync;
379                    system "bin/storableedit.pl", (glob('out/*.storable'))[0];
380          } elsif ( $cmd =~ m{^i(?:nfo)?\s?(.+)?$}i ) {          } elsif ( $cmd =~ m{^i(?:nfo)?\s?(.+)?$}i ) {
381                  print "# nodes: ", join(' ',@nodes), $/;                  info;
   
                 send_nodes 'info' => $2;  
   
                 my @info = (  
                         "node\toffset\tlimit\t#recs\tpath",  
                         "----\t------\t-----\t-----\t----",  
                         "0\t$offset\t$limit\t$num_records\t$path",  
                 );  
   
                 push @info, get_node $_ foreach @nodes;  
   
                 print "$_\n" foreach @info;  
   
382          } elsif ( $cmd =~ m{^(q|e|x)}i ) {          } elsif ( $cmd =~ m{^(q|e|x)}i ) {
383                  warn "# exit";                  warn "# exit";
384                  send_nodes 'exit';                  send_nodes 'exit';
385                  exit;                  exit;
386          } elsif ( $cmd =~ m{^(v|r)}i ) {          } elsif ( $cmd =~ m{^(v|r)}i ) {
387                  run_views;                  run_views;
388            } elsif ( $cmd =~ m{^n(ode)?\s*(\d+)}i ) {
389                    push @nodes, $2;
390                    info;
391          } elsif ( $cmd ) {          } elsif ( $cmd ) {
392                  warn "UNKNOWN ", dump $cmd;                  warn "UNKNOWN ", dump $cmd;
393          }          }
Legend:
Removed from v.26  
changed lines
  Added in v.60
  ViewVC Help
Powered by ViewVC 1.1.26