/[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 60 by dpavlin, Fri Sep 25 13:02:32 2009 UTC revision 76 by dpavlin, Fri Sep 25 21:12:46 2009 UTC
# Line 3  Line 3 
3  use warnings;  use warnings;
4  use strict;  use strict;
5    
6  our $VERSION = '0.04';  our $VERSION = '0.05';
7    
8  use Time::HiRes qw(time);  use Time::HiRes qw(time);
9  use Data::Dump qw(dump);  use Data::Dump qw(dump);
# 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' => \$port,          'port|listen=i' => \$port,
31          'connect=s'   => \@nodes,          'node|connect=i'   => \@nodes,
32          'debug!'   => \$debug,          'debug!'   => \$debug,
33  ) or die $!;  ) or die $!;
34    
# Line 72  my $input = WebPAC::Input::ISI->new( Line 72  my $input = WebPAC::Input::ISI->new(
72  );  );
73    
74  our $num_records = $input->size;  our $num_records = $input->size;
75    our @reports;
76    
77  sub report {  sub report {
78          my $description = join(' ',@_);          my $description = join(' ',@_);
79          my $dt = time - $t;          my $dt = time - $t;
80          printf "%s in %1.4fs %.2f/s\n", $description, $dt, $input->size / $dt;          my $report = [ $description, $dt, $input->size / $dt ];
81            printf "[$port] %s in %1.4fs %.2f/s\n", @$report;
82            push @reports, $report;
83          $t = time;          $t = time;
84  }  }
85    
86    sub show_report {
87            "\n" . join( "\n", map { sprintf "%8.4fs %10.2f/s  %s", $_->[1], $_->[2], $_->[0] } @reports ) . "\n";
88    }
89    
90  report $input->size . ' records loaded';  report $input->size , 'records loaded';
91    
92  mkdir 'out' unless -e 'out';  mkdir 'out' unless -e 'out';
93    
# Line 91  our $cache; Line 97  our $cache;
97    
98  our $connected;  our $connected;
99    
100    sub node_sock {
101            my $node = shift;
102            my $sock = IO::Socket::INET->new(
103                            PeerAddr => '127.0.0.1',
104                            PeerPort => $node,
105                            Proto    => 'tcp',
106            );
107            
108            return $sock if $sock && $sock->connected;
109    
110            warn "[$port] can't connect to $node - $!\n"; # FIXME die?
111            return;
112    }
113    
114  sub send_nodes {  sub send_nodes {
115          my $content = $#_ > 0 ? pop @_ : '';    # no content with just one argument!          my $content = $#_ > 0 ? pop @_ : '';    # no content with just one argument!
116          my $header = defined $content ? length($content) : 0;          my $header = defined $content ? length($content) : 0;
117          $header .= ' ' . join(' ', @_) if @_;          $header .= ' ' . join(' ', @_) if @_;
118    
119          warn "# send_nodes ", dump @nodes;          warn "# send_nodes ", dump(@_), " to ", dump @nodes;
120    
121          foreach my $node ( @nodes ) {          foreach my $node ( @nodes ) {
122    
123                  my $sock = IO::Socket::INET->new(                  my $sock = node_sock($node) || next;
                         PeerAddr => '127.0.0.1',  
                         PeerPort => $node,  
                         Proto    => 'tcp',  
                 );  
124    
125                  if ( ! $sock ) {                  warn "[$port] >>>> [$node] $header\n";
                         warn "[$port] can't connect to $node - $!\n"; # FIXME die?  
                         next;  
                 }  
   
                 warn "[$port] >>>> $node $header\n";  
126                  print $sock "$header\n$content" || warn "can't send $header to $node: $!";                  print $sock "$header\n$content" || warn "can't send $header to $node: $!";
127    
128                  $connected->{$node} = $sock;                  $connected->{$node} = $sock;
# Line 122  sub get_node { Line 133  sub get_node {
133          my $node = shift;          my $node = shift;
134    
135          my $sock = $connected->{$node};          my $sock = $connected->{$node};
136          if ( ! $sock ) {          if ( ! $sock || ! $sock->connected ) {
137                  warn "[$port] ERROR lost connection to $node";                  warn "[$port] no connection to $node";
138                  delete $connected->{$node};                  delete $connected->{$node};
139                  return;                  return;
140          }          }
141          chomp( my $size = <$sock> );          chomp( my $size = <$sock> );
142          warn "[$port] <<<< $node $size bytes\n" if $debug || $size > 1024;          warn "[$port] <<<< [$node] $size bytes\n" if $debug || $size > 1024;
143          my $data;          my $data;
144          read $sock, $data, $size;          read $sock, $data, $size;
145          return $data;          return $data;
# Line 141  sub send_sock { Line 152  sub send_sock {
152          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;
153  }  }
154    
155    sub pull_node_file {
156            my ( $node, $file ) = @_;
157    
158            my $path = "/dev/shm/sack.$node.$file";
159            return if -e $path; # FIXME
160    
161            my $sock = node_sock $node || die "not connected to $node";
162    
163            print $sock "0 file $file\n";
164    
165            my $size = <$sock>;
166            chomp($size);
167            warn "[$port] pull_node_file $node $file $size bytes\n";
168    
169            my $block = 4096;
170            my $buff  = ' ' x $block;
171    
172            open(my $fh, '>', $path) || die "can't open $path";
173            while ( read $sock, $buff, $block ) {
174                    print $fh $buff;
175            }
176            close($fh);
177    }
178    
179  sub merge_out {  sub merge_out {
180          my ( $from_node, $new ) = @_;          my ( $from_node, $new ) = @_;
181    
182            my $t_merge = time();
183    
184            pull_node_file $from_node => 'nr_md5';
185            pull_node_file $from_node => 'md5';
186    
187          my $remote_digest = Sack::Digest->new( port => $from_node );          my $remote_digest = Sack::Digest->new( port => $from_node );
188          my ( $local, $remote ) = ( 0, 0 );          my ( $local, $remote ) = ( 0, 0 );
189    
190          my $tick = 0;          my $tick = 0;
191          print STDERR "[$port] merge $from_node";          print STDERR "[$port] merge [$from_node]";
192    
193            my $missing;
194    
195          foreach my $k1 ( keys %$new ) {          foreach my $k1 ( keys %$new ) {
196    
# Line 159  sub merge_out { Line 201  sub merge_out {
201                          if ( $k1 =~ m{#} ) {                          if ( $k1 =~ m{#} ) {
202                                  die "ASSERT $k1 $k2" unless $k2 =~ m{^\d+$};                                  die "ASSERT $k1 $k2" unless $k2 =~ m{^\d+$};
203  #warn "XXX $k1 $k2";  #warn "XXX $k1 $k2";
204                                  my $md5 = $remote_digest->{nr_md5}->[$k2] || warn "[$port] no2md5 $n not found in $from_node\n";                                  my $md5 = $remote_digest->{nr_md5}->[$k2];
205    
206                                    if ( ! $md5 ) {
207                                            $missing->{nr_md5}->{$from_node}++; # FIXME die?
208                                            next;
209                                    }
210    
211                                  if ( my $local_k2 = $digest->{md5_nr}->{$md5} ) {                                  if ( my $local_k2 = $digest->{md5_nr}->{$md5} ) {
212                                          $k2 = $local_k2;                                          $k2 = $local_k2;
213                                          $local++;                                          $local++;
214                                  } else {                                  } elsif ( my $full = $remote_digest->{md5}->{$md5} ) {
215                                          $k2 = $digest->to_int( $remote_digest->{md5}->{$md5} );                                          $k2 = $digest->to_int( $remote_digest->{md5}->{$md5} );
216                                          $remote++;                                          $remote++;
217                                    } else {
218                                            $missing->{md5}->{$from_node}++;
219                                  }                                  }
220                          }                          }
221    
# Line 196  sub merge_out { Line 246  sub merge_out {
246                  }                  }
247          }          }
248    
249          warn "$tick $local/$remote\n";          $t_merge = time - $t_merge;
250            printf STDERR "%d in %.4fs %.2f/s local %.1f%% %d/%d\n", $tick, $t_merge, $tick / $t_merge, $local * 100 / $tick, $local, $remote;
251            push @reports, [ "$tick merged $from_node", $t_merge, $tick / $t_merge ];
252    
253            warn "[$port] missing ", dump $missing if $missing;
254    
255          warn "## merge out ", dump $out if $debug;          warn "## merge out ", dump $out if $debug;
256  }  }
# Line 232  sub run_code { Line 286  sub run_code {
286                  $pos % 1000  == 0 ? print STDERR "."  : 0 ;                  $pos % 1000  == 0 ? print STDERR "."  : 0 ;
287          };          };
288    
289          report "\n[$port] RECS $affected $view";          report "$affected affected $view";
290    
291          warn "WARN no \$out defined!" unless defined $out;          warn "WARN no \$out defined!" unless defined $out;
292    
# Line 240  sub run_code { Line 294  sub run_code {
294    
295          if ( $connected ) {          if ( $connected ) {
296                  foreach my $node ( keys %$connected ) {                  foreach my $node ( keys %$connected ) {
297                          warn "[$port] get_node $node\n";                          warn "[$port] get_node [$node]\n";
298                          my $o = get_node $node;                          my $o = get_node $node;
299                            next unless $o;
300                          my $s = length $o;                          my $s = length $o;
301                          $o = thaw $o;                          $o = thaw $o;
302                          warn "[$port] merge $node $s bytes\n";                          warn "[$port] got $s bytes from [$node]\n";
303                          merge_out $node => $o;                          merge_out $node => $o;
                         report "[$port] merged $node";  
304                  }                  }
305          }          }
306  }  }
# Line 273  sub run_views { Line 327  sub run_views {
327                          rename $path, "$path.last";                          rename $path, "$path.last";
328    
329                          store $out => $path;                          store $out => $path;
330                          report "[$port] SAVE $path $offset-$limit", -s $path, "bytes";                          report "save $path", -s $path, "bytes";
331    
332                          if ( -s $path < 4096 ) {                          if ( -s $path < 4096 ) {
333                                  print '$out = ', dump $digest->undigest_out($out);                                  print '$out = ', dump $digest->undigest_out($out);
# Line 296  sub info_tabs { Line 350  sub info_tabs {
350    
351    
352  if ( $port ) {  if ( $port ) {
353    
354            my $pid_path = "/tmp/sack.$port.pid";
355            if ( -e $pid_path ) {
356                    my $pid = read_file $pid_path;
357                    kill 9, $pid && warn "[$port] kill old $pid";
358            }
359            write_file $pid_path, $$;
360    
361    
362          my $sock = IO::Socket::INET->new(          my $sock = IO::Socket::INET->new(
363                  Listen    => SOMAXCONN,                  Listen    => SOMAXCONN,
364                  LocalAddr => '127.0.0.1',                  LocalAddr => '127.0.0.1',
# Line 306  if ( $port ) { Line 369  if ( $port ) {
369    
370          while (1) {          while (1) {
371    
372                  warn "[$port] READY path: $path offset: $offset limit: $limit #recs: $num_records\n";                  warn "[$port] accept path: $path offset: $offset limit: $limit #recs: $num_records\n";
373    
374                  my $client = $sock->accept();                  my $client = $sock->accept();
375    
# Line 326  if ( $port ) { Line 389  if ( $port ) {
389                  } elsif ( $header[0] eq 'info' ) {                  } elsif ( $header[0] eq 'info' ) {
390                          my $info = info_tabs;                          my $info = info_tabs;
391                          warn "[$port] info $info\n";                          warn "[$port] info $info\n";
392                            $info .= "\n" . show_report if $content =~ m{r}i;
393                          send_sock $client => $info;                          send_sock $client => $info;
394                  } elsif ( $header[0] eq 'exit' ) {                  } elsif ( $header[0] eq 'exit' ) {
395                          warn "[$port] exit";                          warn "[$port] exit";
396                          exit;                          exit;
397                    } elsif ( $header[0] eq 'file' ) {
398                            $digest->close;
399                            my $path = "/dev/shm/sack.$port.$header[1]";
400                            my $size = -s $path;
401                            warn "[$port] >>>> file $path $size bytes\n";
402                            print $client "$size\n";
403                            my $block = 4096;
404                            my $buff  = ' ' x $block;
405                            open(my $fh, '<', $path) || die "can't open $path";
406                            while ( read $fh, $buff, $block ) {
407                                    print $client $buff;
408                            }
409                            $digest->open;
410                  } else {                  } else {
411                          warn "[$port] UNKNOWN $header[0]";                          warn "[$port] UNKNOWN $header[0]";
412                  }                  }
413    
414                    $client->close;
415    
416          }          }
417  }  }
418    
419  sub info {  sub info {
420          send_nodes 'info' => $2;          my $detail = shift || '';
421    
422            send_nodes 'info' => $detail;
423    
424          my @info = (          my @info = (
425                  "port\toffset\tlimit\t#recs\tpath",                  "port\toffset\tlimit\t#recs\tpath",
# Line 346  sub info { Line 427  sub info {
427                  info_tabs,                  info_tabs,
428          );          );
429    
430            push @info, show_report if $detail =~ m{r}i;
431    
432          push @info, get_node $_ foreach @nodes;          push @info, get_node $_ foreach @nodes;
433    
434          print "[$port] INFO\n"          print "[$port] INFO", $detail ? " $detail" : '', " \n"
435                  , join("\n", @info)                  , join("\n", @info)
436                  , "\n\n" ;                  , "\n" ;
437    
438          return @info;          return @info;
439  }  }
440    
441  info;  info;
442    while ( keys %$connected != @nodes ) {
443            warn "[$port] wait for ", join(' ', grep { ! defined $connected->{$_} } @nodes );
444            sleep 1;
445    }
446  run_views;  run_views;
447    
448  while ( 1 ) {  while ( 1 ) {
# Line 377  __HELP__ Line 464  __HELP__
464                  #system "vi out/*";                  #system "vi out/*";
465                  $digest->sync;                  $digest->sync;
466                  system "bin/storableedit.pl", (glob('out/*.storable'))[0];                  system "bin/storableedit.pl", (glob('out/*.storable'))[0];
467          } elsif ( $cmd =~ m{^i(?:nfo)?\s?(.+)?$}i ) {          } elsif ( $cmd =~ m{^i(?:nfo)?\s?(\S+)?$}i ) {
468                  info;                  info $1;
469          } elsif ( $cmd =~ m{^(q|e|x)}i ) {          } elsif ( $cmd =~ m{^(q|e|x)}i ) {
470                  warn "# exit";                  warn "# exit";
471                  send_nodes 'exit';                  send_nodes 'exit';
Legend:
Removed from v.60  
changed lines
  Added in v.76
  ViewVC Help
Powered by ViewVC 1.1.26