/[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 3 by dpavlin, Sun Sep 20 18:55:34 2009 UTC revision 55 by dpavlin, Fri Sep 25 09:38:48 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;
12    use IO::Socket::INET;
13    use Storable qw/freeze thaw store/;
14    
15    
16    my $debug  = 0;
17    my $path   = '/data/isi/full.txt';
18    my $limit  = 5000;
19    my $offset = 0;
20    my @views;
21    my $port   = 0; # interactive
22    my @nodes;
23    
24    
25    GetOptions(
26            'path=s'   => \$path,
27            'offset=i' => \$offset,
28            'limit=i'  => \$limit,
29            'view=s'   => \@views,
30            'listen|port=i' => \$port,
31            'connect=s'   => \@nodes,
32            'debug!'   => \$debug,
33    ) or die $!;
34    
35  my $t = time;  my $t = time;
36    
37    
38  use lib '/srv/webpac2/lib/';  sub send_nodes;
39    
40    our $prefix;
41    sub BEGIN {
42            $prefix = $0;
43            if ( $prefix !~ m{^/} ) {
44                    chomp( my $pwd = `pwd` );
45                    $prefix = "$pwd/$prefix";
46            }
47            $prefix =~ s{^(.*)/srv/Sack/.+$}{$1};
48            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/";
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  => '/data/isi/full.txt',          path   => "$prefix/$path",
70          limit => shift @ARGV || 1000,          offset => $offset,
71            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;
81  }  }
82    
83    
# Line 29  report $input->size . ' records loaded'; Line 85  report $input->size . ' records loaded';
85    
86  mkdir 'out' unless -e 'out';  mkdir 'out' unless -e 'out';
87    
88    our $out;
89    
90    our $cache;
91    
92    our $connected;
93    
94    sub send_nodes {
95            my $content = $#_ > 0 ? pop @_ : '';    # no content with just one argument!
96            my $header = defined $content ? length($content) : 0;
97            $header .= ' ' . join(' ', @_) if @_;
98    
99            foreach my $node ( @nodes ) {
100    
101                    my $sock = IO::Socket::INET->new(
102                            PeerAddr => $node,
103                            Proto    => 'tcp',
104                    );
105    
106                    if ( ! $sock ) {
107                            warn "can't connect to $node - $!"; # FIXME die?
108                            next;
109                    }
110    
111                    warn "[$port] >>>> $node $header\n";
112                    print $sock "$header\n$content" || warn "can't send $header to $node: $!";
113    
114                    $connected->{$node} = $sock;
115            }
116    }
117    
118    sub get_node {
119            my $node = shift;
120    
121            my $sock = $connected->{$node};
122            if ( ! $sock ) {
123                    warn "[$port] ERROR lost connection to $node";
124                    delete $connected->{$node};
125                    return;
126            }
127            chomp( my $size = <$sock> );
128            warn "[$port] <<<< $node $size bytes\n";
129            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            warn "[$port] >>>> ", $sock->peerhost, " $size bytes\n";
138            print $sock "$size\n$data" || warn "can't send $size bytes to ", $sock->peerhost;
139    }
140    
141    sub merge_out {
142            my ( $from_node, $new ) = @_;
143    
144            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            foreach my $k1 ( keys %$new ) {
153    
154                    foreach my $k2 ( keys %{ $new->{$k1} } ) {
155    
156                            my $n   = delete $new->{$k1}->{$k2};
157    
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                            my $ref = ref    $out->{$k1}->{$k2};
172    #warn "XXXX $k1 $k2 $ref";
173                            if ( ! defined $out->{$k1}->{$k2} ) {
174                                    $out->{$k1}->{$k2} = $n;
175                            } elsif ( $k1 =~ m{\+} ) {
176    #                               warn "## agregate $k1 $k2";
177                                    $out->{$k1}->{$k2} += $n;
178                            } 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                            } 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            warn "[$port] merge local $local remote $remote from $from_port\n";
193            warn "## merge out ", dump $out if $debug;
194    }
195    
196    sub run_code {
197            my ( $view, $code ) = @_;
198    
199            warn "\n#### CODE $view START ####\n$code\n#### CODE $view END ####\n" if $debug;
200    
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                            print STDERR "END @ $pos";
212                            last;
213                    }
214    
215                    eval "$code";
216                    if ( $@ ) {
217                            warn "ABORT $pos $@\n";
218                            last;
219                    } else {
220                            $affected++;
221                    }
222    
223                    $pos % 10000 == 0 ? print STDERR $pos :
224                    $pos % 1000  == 0 ? print STDERR "."  : 0 ;
225            };
226    
227            report "\n[$port] RECS $affected $view";
228    
229            warn "WARN no \$out defined!" unless defined $out;
230    
231            $digest->sync;
232    
233            if ( $connected ) {
234                    foreach my $node ( keys %$connected ) {
235                            warn "[$port] get_node $node\n";
236                            my $o = get_node $node;
237                            my $s = length $o;
238                            $o = thaw $o;
239                            warn "[$port] merge $node $s bytes\n";
240                            merge_out $node => $o;
241                    }
242            }
243    }
244    
245  sub run_views {  sub run_views {
246          my @views = glob 'views/*.pl';          @views = sort glob 'views/*.pl' unless @views;
247          warn "# views ", dump @views;          warn "# views ", dump @views;
248    
249          foreach my $view ( @views ) {          foreach my $view ( @views ) {
250    
                 my ( $nr, $package ) = ( $1, $2 )  
                         if $view =~ m{/(\d+)\.([^/]+(\.pl)?$)};  
   
                 my $out;  
   
251                  next if system("perl -c $view") != 0;                  next if system("perl -c $view") != 0;
252    
253                  my $code = read_file $view;                  my $code = read_file $view;
                 warn $code;  
254    
255                  $t = time;                  run_code $view => $code;
256    
257                    if ( defined $out ) {
258    
259                  foreach my $pos ( 1 .. $input->size ) {                          my $path = $view;
260                          my $rec = $input->fetch_rec($pos);                                $path =~ s{views?/}{out/} || die "no view in $view";
261                            $path =~ s{\.pl}{.storable};
262    
263                          eval $code;                          unlink "$path.last" if -e "$path.last";
264                          die "ERROR [$pos] $@" if $@;                          rename $path, "$path.last";
265                  };  
266                            store $out => $path;
267                  report $view;                          report "[$port] SAVE $path $offset-$limit", -s $path, "bytes";
268    
269                  if ( defined $out ) {                          if ( -s $path < 4096 ) {
270                          my $dump = dump $out;                                  print '$out = ', dump $digest->undigest_out($out);
271                          my $len  = length $dump;                          }
272                          my $path = "out/$nr.$package";                  }
273                          print "# $view $len bytes $path"  
274                                  , $len < 10000 ?  " \$out = $dump" : ' SAVED ONLY'          }
275                                  , "\n"  
276                                  ;  }
277                          write_file $path, $dump;  
278                          report "save $path";  
279    sub info_tabs {
280            "$port\t$offset\t$limit\t$num_records\t$path\t"
281            . join("\t", map {
282                    my $b = $_;
283                    $b =~ s{^.+\.$port\.([^/]+)$}{$1};
284                    "$b " . -s $_
285            } glob "/dev/shm/sack.$port.*" );
286    }
287    
288    
289    if ( $port ) {
290            my $sock = IO::Socket::INET->new(
291                    Listen    => SOMAXCONN,
292                    LocalAddr => '127.0.0.1',
293                    LocalPort => $port,
294                    Proto     => 'tcp',
295                    Reuse     => 1,
296            ) or die $!;
297    
298            while (1) {
299    
300                    warn "[$port] READY path: $path offset: $offset limit: $limit #recs: $num_records\n";
301    
302                    my $client = $sock->accept();
303    
304                    warn "[$port] <<<< connect from ", $client->peerhost, $/;
305    
306                    my @header = split(/\s/, <$client>);
307                    warn "[$port] <<<< header ",dump(@header),$/;
308    
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                            send_sock $client => freeze $out;
317                    } elsif ( $header[0] eq 'info' ) {
318                            my $info = info_tabs;
319                            warn "[$port] info $info\n";
320                            send_sock $client => $info;
321                    } elsif ( $header[0] eq 'exit' ) {
322                            warn "[$port] exit";
323                            exit;
324                    } else {
325                            warn "[$port] UNKNOWN $header[0]";
326                  }                  }
327    
328          }          }
329  }  }
330    
331    sub info {
332            send_nodes 'info' => $2;
333    
334            my @info = (
335                    "port\toffset\tlimit\t#recs\tpath",
336                    "----\t------\t-----\t-----\t----",
337                    info_tabs,
338            );
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  run_views;  run_views;
351    
352  while ( 1 ) {  while ( 1 ) {
353    
354          print "sack> ";          print "sack> ";
355          my $cmd = <STDIN>;          chomp( my $cmd = <STDIN> );
356    
357          if ( $cmd =~ m{(vi|\\e|out)}i ) {          if ( $cmd =~ m{^(h|\?)} ) {
358                  system "vi out/*";                  print << "__HELP__"
359          } else {  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                    #system "vi out/*";
369                    $digest->sync;
370                    system "bin/storableedit.pl", (glob('out/*.storable'))[0];
371            } elsif ( $cmd =~ m{^i(?:nfo)?\s?(.+)?$}i ) {
372                    info;
373            } elsif ( $cmd =~ m{^(q|e|x)}i ) {
374                    warn "# exit";
375                    send_nodes 'exit';
376                    exit;
377            } elsif ( $cmd =~ m{^(v|r)}i ) {
378                  run_views;                  run_views;
379            } elsif ( $cmd =~ m{^n(ode)?\s*(\d+)}i ) {
380                    push @nodes, $1;
381                    info;
382            } elsif ( $cmd ) {
383                    warn "UNKNOWN ", dump $cmd;
384          }          }
385    
386  }  }
Legend:
Removed from v.3  
changed lines
  Added in v.55
  ViewVC Help
Powered by ViewVC 1.1.26