/[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 29 by dpavlin, Tue Sep 22 22:28:40 2009 UTC revision 30 by dpavlin, Wed Sep 23 18:44:08 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.02';
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/;
 use Digest::MD5 qw/md5/;  
14    
15    
16  my $debug  = 0;  my $debug  = 0;
# Line 19  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 28  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 41  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/[\./]+bin.+$}{$1};
48          warn "# prefix $prefix";          warn "# prefix $prefix";
49    
50          $SIG{INT} = sub {          $SIG{INT} = sub {
# Line 56  sub BEGIN { Line 55  sub BEGIN {
55  }  }
56    
57    
58    # digest experiment
59    use BerkeleyDB;
60    
61    our $seq = 0;
62    our $btree;
63    my $db_file = "/dev/shm/sack.digest.$port.$offset-$limit";
64    sub digest {
65            my $nr;
66            if ( ! $btree ) {
67                    if ( -e $db_file ) {
68                            warn "[$port] CLEAN removed $db_file ", -s $db_file, " bytes\n";
69                            unlink $db_file;
70                    }
71    
72                    our $btree = BerkeleyDB::Btree->new(
73                            -Filename  => $db_file,
74                            -Cachesize => 700_000_000,
75                            -Flags     => DB_CREATE,
76                    ) || die "$db_file: $!";
77    
78                    warn "[$port] BDB created $db_file\n";
79            }
80            $btree->db_get( $_[0] => $nr    ) == 0 && return $nr;
81            $btree->db_put( $_[0] => ++$seq ) == 0 || die "$_[0] [$seq] $!";
82            $btree->db_put(  $seq => $_[0]  ) == 0 || die "[$seq] $_[0] $!";
83            return $seq;
84    }
85    
86    
87  use lib "$prefix/srv/webpac2/lib/";  use lib "$prefix/srv/webpac2/lib/";
88  use WebPAC::Input::ISI;  use WebPAC::Input::ISI;
89    
# Line 70  my $input = WebPAC::Input::ISI->new( Line 98  my $input = WebPAC::Input::ISI->new(
98  our $num_records = $input->size;  our $num_records = $input->size;
99    
100  sub report {  sub report {
101          my $description = shift;          my $description = join(' ',@_);
102          my $dt = time - $t;          my $dt = time - $t;
103          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;
104          $t = time;          $t = time;
# Line 104  sub send_nodes { Line 132  sub send_nodes {
132                          next;                          next;
133                  }                  }
134    
135                  print ">>>> $listen $node $header\n";                  warn "[$port] >>>> $node $header\n";
136                  print $sock "$header\n$content" || warn "can't send $header to $node: $!";                  print $sock "$header\n$content" || warn "can't send $header to $node: $!";
137    
138                  $connected->{$node} = $sock;                  $connected->{$node} = $sock;
# Line 116  sub get_node { Line 144  sub get_node {
144    
145          my $sock = $connected->{$node};          my $sock = $connected->{$node};
146          if ( ! $sock ) {          if ( ! $sock ) {
147                  warn "ERROR: lost connection to $node";                  warn "[$port] ERROR lost connection to $node";
148                  delete $connected->{$node};                  delete $connected->{$node};
149                  return;                  return;
150          }          }
151          chomp( my $size = <$sock> );          chomp( my $size = <$sock> );
152          warn "<<<< $listen $node $size bytes\n";          warn "[$port] <<<< $node $size bytes\n";
153          my $data;          my $data;
154          read $sock, $data, $size;          read $sock, $data, $size;
155          return $data;          return $data;
# Line 130  sub get_node { Line 158  sub get_node {
158  sub send_sock {  sub send_sock {
159          my ( $sock, $data ) = @_;          my ( $sock, $data ) = @_;
160          my $size   = length $data;          my $size   = length $data;
161          warn ">>>> $listen ", $sock->peerhost, " $size bytes\n";          warn "[$port] >>>> ", $sock->peerhost, " $size bytes\n";
162          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;
163  }  }
164    
# Line 141  sub merge_out { Line 169  sub merge_out {
169    
170                  foreach my $k2 ( keys %{ $new->{$k1} } ) {                  foreach my $k2 ( keys %{ $new->{$k1} } ) {
171    
172                          my $n   =     $new->{$k1}->{$k2};                          my $n   = delete $new->{$k1}->{$k2};
173                          my $ref = ref $out->{$k1}->{$k2};                          my $ref = ref    $out->{$k1}->{$k2};
174    
175                          if ( ! defined $out->{$k1}->{$k2} ) {                          if ( ! defined $out->{$k1}->{$k2} ) {
176                                  $out->{$k1}->{$k2} = $n;                                  $out->{$k1}->{$k2} = $n;
# Line 192  sub run_code { Line 220  sub run_code {
220                  } else {                  } else {
221                          $affected++;                          $affected++;
222                  }                  }
223    
224                    $pos % 10000 == 0 ? print STDERR $pos - $offset :
225                    $pos % 1000  == 0 ? print STDERR "." : 0 ;
226          };          };
227    
228          report "$affected affected records $view";          report "\n[$port] RECS $affected $view";
229    
230          warn "WARN no \$out defined!" unless defined $out;          warn "WARN no \$out defined!" unless defined $out;
231    
232          if ( $connected ) {          if ( $connected ) {
233                  foreach my $node ( keys %$connected ) {                  foreach my $node ( keys %$connected ) {
234                          warn "# $listen get_node $node\n";                          warn "[$port] get_node $node\n";
235                          my $o = get_node $node;                          my $o = get_node $node;
236                          my $s = length $o;                          my $s = length $o;
237                          $o = thaw $o;                          $o = thaw $o;
238                          warn "# $listen merge $s bytes\n";                          warn "[$port] merge $node $s bytes\n";
239                          merge_out $o;                          merge_out $o;
240                  }                  }
241          }          }
# Line 223  sub run_views { Line 254  sub run_views {
254                  run_code $view => $code;                  run_code $view => $code;
255    
256                  if ( defined $out ) {                  if ( defined $out ) {
                         my $dump = dump $out;  
                         my $len  = length $dump;  
257    
258                          my $path = $view;                          my $path = $view;
259                          $path =~ s{views?/}{out/} || die "no view in $view";                          $path =~ s{views?/}{out/} || die "no view in $view";
260                          $path =~ s{\.pl}{};                          $path =~ s{\.pl}{.storable};
   
                         print "OUT $view $offset/$limit $len bytes $path"  
                                 , ( $len < 10000 ?  " \$out = $dump" : ' SAVED ONLY' )  
                                 , "\n"  
                                 ;  
261    
262                          unlink "$path.last" if -e "$path.last";                          unlink "$path.last" if -e "$path.last";
263                          rename $path, "$path.last";                          rename $path, "$path.last";
264                          write_file $path, $dump;  
265                          report "SAVE $path";                          store $out => $path;
266                            report "[$port] SAVE $offset/$limit $view $path", -s $path, " bytes";
267                  }                  }
268    
269          }          }
270    
271  }  }
272    
273  if ( $listen ) {  if ( $port ) {
274          my $sock = IO::Socket::INET->new(          my $sock = IO::Socket::INET->new(
275                  Listen    => SOMAXCONN,                  Listen    => SOMAXCONN,
276                  LocalAddr => '127.0.0.1',                  LocalAddr => '127.0.0.1',
277                  LocalPort => $listen,                  LocalPort => $port,
278                  Proto     => 'tcp',                  Proto     => 'tcp',
279                  Reuse     => 1,                  Reuse     => 1,
280          ) or die $!;          ) or die $!;
281    
282          while (1) {          while (1) {
283    
284                  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";
285    
286                  my $client = $sock->accept();                  my $client = $sock->accept();
287    
288                  warn "<<<< $listen connect from ", $client->peerhost, $/;                  warn "[$port] <<<< connect from ", $client->peerhost, $/;
289    
290                  my @header = split(/\s/, <$client>);                  my @header = split(/\s/, <$client>);
291                  warn "<<<< $listen header ",dump(@header),$/;                  warn "[$port] <<<< header ",dump(@header),$/;
292    
293                  my $size = shift @header;                  my $size = shift @header;
294    
# Line 274  if ( $listen ) { Line 299  if ( $listen ) {
299                          run_code $header[1] => $content;                          run_code $header[1] => $content;
300                          send_sock $client => freeze $out;                          send_sock $client => freeze $out;
301                  } elsif ( $header[0] eq 'info' ) {                  } elsif ( $header[0] eq 'info' ) {
302                          my $info = "$listen\t$offset\t$limit\t$num_records\t$path";                          my $info = "$port\t$offset\t$limit\t$num_records\t$path";
303                          $info .= "\t" . eval $header[1] if $header[1];                          $info .= "\t" . eval $header[1] if $header[1];
304                          warn "info $info\n";                          warn "[$port] info $info\n";
305                          send_sock $client => $info;                          send_sock $client => $info;
306                  } elsif ( $header[0] eq 'exit' ) {                  } elsif ( $header[0] eq 'exit' ) {
307                          warn "exit $listen";                          warn "[$port] exit";
308                          exit;                          exit;
309                  } else {                  } else {
310                          warn "WARN $listen unknown";                          warn "[$port] UNKNOWN $header[0]";
311                  }                  }
312    
313          }          }
314  }  }
315    
316    sub info {
317            send_nodes 'info' => $2;
318    
319            my @info = (
320                    "port\toffset\tlimit\t#recs\tpath",
321                    "----\t------\t-----\t-----\t----",
322                    "$port\t$offset\t$limit\t$num_records\t$path",
323            );
324    
325            push @info, get_node $_ foreach @nodes;
326    
327            print "[$port] INFO\n"
328                    , join("\n", @info)
329                    , "\n\n" ;
330    
331            return @info;
332    }
333    
334    info;
335  run_views;  run_views;
336    
337  while ( 1 ) {  while ( 1 ) {
# Line 308  __HELP__ Line 352  __HELP__
352          } elsif ( $cmd =~ m{^(vi|\\e|o)}i ) {          } elsif ( $cmd =~ m{^(vi|\\e|o)}i ) {
353                  system "vi out/*";                  system "vi out/*";
354          } elsif ( $cmd =~ m{^i(?:nfo)?\s?(.+)?$}i ) {          } elsif ( $cmd =~ m{^i(?:nfo)?\s?(.+)?$}i ) {
355                  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;  
   
356          } elsif ( $cmd =~ m{^(q|e|x)}i ) {          } elsif ( $cmd =~ m{^(q|e|x)}i ) {
357                  warn "# exit";                  warn "# exit";
358                  send_nodes 'exit';                  send_nodes 'exit';
Legend:
Removed from v.29  
changed lines
  Added in v.30
  ViewVC Help
Powered by ViewVC 1.1.26