/[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 9 by dpavlin, Mon Sep 21 20:05:41 2009 UTC revision 36 by dpavlin, Wed Sep 23 22:22:18 2009 UTC
# Line 3  Line 3 
3  use warnings;  use warnings;
4  use strict;  use strict;
5    
6    our $VERSION = '0.03';
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;
13    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  = 10000;  my $limit  = 5000;
19  my $offset = 0;  my $offset = 0;
20  my @views;  my @views;
21    my $port   = 0; # interactive
22    my @nodes;
23    
24    
25  GetOptions(  GetOptions(
# Line 20  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,
31            'connect=s'   => \@nodes,
32            'debug!'   => \$debug,
33  ) or die $!;  ) or die $!;
34    
35  my $t = time;  my $t = time;
36    
37  use lib '/srv/webpac2/lib/';  
38    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/[\./]+bin.+$}{$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    Sack::Digest->clean;
61    Sack::Digest->open( $port );
62    sub digest { Sack::Digest->to_int($_[0]) }
63    
64    use lib "$prefix/srv/webpac2/lib/";
65  use WebPAC::Input::ISI;  use WebPAC::Input::ISI;
66    
67    $WebPAC::Input::ISI::subfields = undef; # disable parsing of subfields
68    
69  my $input = WebPAC::Input::ISI->new(  my $input = WebPAC::Input::ISI->new(
70          path   => $path,          path   => "$prefix/$path",
71          offset => $offset,          offset => $offset,
72          limit  => $limit,          limit  => $limit,
73  );  );
74    
75    our $num_records = $input->size;
76    
77  sub report {  sub report {
78          my $description = shift;          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;          printf "%s in %1.4fs %.2f/s\n", $description, $dt, $input->size / $dt;
81            $t = time;
82  }  }
83    
84    
# Line 48  our $out; Line 90  our $out;
90    
91  our $cache;  our $cache;
92    
93    our $connected;
94    
95    sub send_nodes {
96            my $content = $#_ > 0 ? pop @_ : '';    # no content with just one argument!
97            my $header = defined $content ? length($content) : 0;
98            $header .= ' ' . join(' ', @_) if @_;
99    
100            foreach my $node ( @nodes ) {
101    
102                    my $sock = IO::Socket::INET->new(
103                            PeerAddr => $node,
104                            Proto    => 'tcp',
105                    );
106    
107                    if ( ! $sock ) {
108                            warn "can't connect to $node - $!"; # FIXME die?
109                            next;
110                    }
111    
112                    warn "[$port] >>>> $node $header\n";
113                    print $sock "$header\n$content" || warn "can't send $header to $node: $!";
114    
115                    $connected->{$node} = $sock;
116            }
117    }
118    
119    sub get_node {
120            my $node = shift;
121    
122            my $sock = $connected->{$node};
123            if ( ! $sock ) {
124                    warn "[$port] ERROR lost connection to $node";
125                    delete $connected->{$node};
126                    return;
127            }
128            chomp( my $size = <$sock> );
129            warn "[$port] <<<< $node $size bytes\n";
130            my $data;
131            read $sock, $data, $size;
132            return $data;
133    }
134    
135    sub send_sock {
136            my ( $sock, $data ) = @_;
137            my $size   = length $data;
138            warn "[$port] >>>> ", $sock->peerhost, " $size bytes\n";
139            print $sock "$size\n$data" || warn "can't send $size bytes to ", $sock->peerhost;
140    }
141    
142    sub merge_out {
143            my $new = shift;
144    
145            foreach my $k1 ( keys %$new ) {
146    
147                    foreach my $k2 ( keys %{ $new->{$k1} } ) {
148    
149                            my $n   = delete $new->{$k1}->{$k2};
150                            my $ref = ref    $out->{$k1}->{$k2};
151    
152                            if ( ! defined $out->{$k1}->{$k2} ) {
153                                    $out->{$k1}->{$k2} = $n;
154                            } elsif ( $k1 =~ m{\+} ) {
155    #                               warn "## agregate $k1 $k2";
156                                    $out->{$k1}->{$k2} += $n;
157                            } elsif ( $ref  eq 'ARRAY' ) {
158                                    if ( ref $n eq 'ARRAY' ) {
159                                            push @{ $out->{$k1}->{$k2} }, $_ foreach @$n;
160                                    } else {
161                                            push @{ $out->{$k1}->{$k2} }, $n;
162                                    }
163                            } elsif ( $ref eq '' ) {
164                                    $out->{$k1}->{$k2} = [ $out->{$k1}->{$k2}, $n ];
165                            } else {
166                                    die "can't merge $k2 [$ref] from ",dump($n), " into ", dump($out->{$k1}->{$k2});
167                            }
168                    }
169            }
170    
171            warn "## merge out ", dump $out if $debug;
172    }
173    
174    sub run_code {
175            my ( $view, $code ) = @_;
176    
177            warn "\n#### CODE $view START ####\n$code\n#### CODE $view END ####\n" if $debug;
178    
179            send_nodes view => $view => $code;
180    
181            undef $out;
182    
183            my $affected = 0;
184            $t = time;
185    
186            foreach my $pos ( $offset + 1 .. $offset + $input->size ) {
187                    my $rec = $cache->{$pos} ||= $input->fetch_rec( $pos );
188                    if ( ! $rec ) {
189                            warn "END at $pos";
190                            last;
191                    }
192    
193                    eval "$code";
194                    if ( $@ ) {
195                            warn "ABORT [$pos] $@\n";
196                            last;
197                    } else {
198                            $affected++;
199                    }
200    
201                    $pos % 10000 == 0 ? print STDERR $pos - $offset :
202                    $pos % 1000  == 0 ? print STDERR "." : 0 ;
203            };
204    
205            report "\n[$port] RECS $affected $view";
206    
207            warn "WARN no \$out defined!" unless defined $out;
208    
209            if ( $connected ) {
210                    foreach my $node ( keys %$connected ) {
211                            warn "[$port] get_node $node\n";
212                            my $o = get_node $node;
213                            my $s = length $o;
214                            $o = thaw $o;
215                            warn "[$port] merge $node $s bytes\n";
216                            merge_out $o;
217                    }
218            }
219    }
220    
221  sub run_views {  sub run_views {
222          @views = sort glob 'views/*.pl' unless @views;          @views = sort glob 'views/*.pl' unless @views;
223          warn "# views ", dump @views;          warn "# views ", dump @views;
224    
225          foreach my $view ( @views ) {          foreach my $view ( @views ) {
226    
                 my ( $nr, $package ) = ( $1, $2 )  
                         if $view =~ m{/(\d+)\.([^/]+(\.pl)?$)};  
   
                 undef $out;  
   
227                  next if system("perl -c $view") != 0;                  next if system("perl -c $view") != 0;
228    
229                  my $code = read_file $view;                  my $code = read_file $view;
                 warn "## CODE\n$code\n## CODE\n";  
230    
231                  my $affected = 0;                  run_code $view => $code;
                 $t = time;  
232    
233                  foreach my $pos ( $offset + 1 .. $offset + $input->size ) {                  if ( defined $out ) {
                         my $rec = $cache->{$pos} ||= $input->fetch_rec( $pos );  
                         if ( ! $rec ) {  
                                 warn "END at $pos";  
                                 last;  
                         }  
234    
235                          eval "$code";                          my $path = $view;
236                          if ( $@ ) {                          $path =~ s{views?/}{out/} || die "no view in $view";
237                                  warn "ERROR [$pos] $@\n";                          $path =~ s{\.pl}{.storable};
238                          } else {  
239                                  $affected++;                          unlink "$path.last" if -e "$path.last";
240                            rename $path, "$path.last";
241    
242                            store $out => $path;
243                            report "[$port] SAVE $path $offset-$limit", -s $path, "bytes";
244    
245                            if ( -s $path < 4096 ) {
246                                    print '$out = ', dump Sack::Digest->undigest_out($out);
247                          }                          }
248                  };                  }
249    
250                  report "$affected affected records $view";          }
251    
252                  if ( defined $out ) {  }
                         my $dump = dump $out;  
                         my $len  = length $dump;  
                         my $path = "out/$nr.$package";  
                         print "out $view $offset/$limit $len bytes $path"  
                                 , ( $len < 10000 ?  " \$out = $dump" : ' SAVED ONLY' )  
                                 , "\n"  
                                 ;  
253    
254                          unlink "$path.last" if -e "$path.last";  
255                          rename $path, "$path.last";  sub info_tabs {
256                          write_file $path, $dump;          "$port\t$offset\t$limit\t$num_records\t$path\t"
257                          report "save $path";          . join("\t", map {
258                    my $b = $_;
259                    $b =~ s{^.+/([^/]+)$}{$1};
260                    "$b " . -s $_
261            } glob '/dev/shm/sack.*' );
262    }
263    
264    
265    if ( $port ) {
266            my $sock = IO::Socket::INET->new(
267                    Listen    => SOMAXCONN,
268                    LocalAddr => '127.0.0.1',
269                    LocalPort => $port,
270                    Proto     => 'tcp',
271                    Reuse     => 1,
272            ) or die $!;
273    
274            while (1) {
275    
276                    warn "[$port] READY path: $path offset: $offset limit: $limit #recs: $num_records\n";
277    
278                    my $client = $sock->accept();
279    
280                    warn "[$port] <<<< connect from ", $client->peerhost, $/;
281    
282                    my @header = split(/\s/, <$client>);
283                    warn "[$port] <<<< header ",dump(@header),$/;
284    
285                    my $size = shift @header;
286    
287                    my $content;
288                    read $client, $content, $size;
289    
290                    if ( $header[0] eq 'view' ) {
291                            run_code $header[1] => $content;
292                            send_sock $client => freeze $out;
293                    } elsif ( $header[0] eq 'info' ) {
294                            my $info = info_tabs;
295                            warn "[$port] info $info\n";
296                            send_sock $client => $info;
297                    } elsif ( $header[0] eq 'exit' ) {
298                            warn "[$port] exit";
299                            exit;
300                  } else {                  } else {
301                          warn "W: no \$out defined!";                          warn "[$port] UNKNOWN $header[0]";
302                  }                  }
303    
304          }          }
305  }  }
306    
307    sub info {
308            send_nodes 'info' => $2;
309    
310            my @info = (
311                    "port\toffset\tlimit\t#recs\tpath",
312                    "----\t------\t-----\t-----\t----",
313                    info_tabs,
314            );
315    
316            push @info, get_node $_ foreach @nodes;
317    
318            print "[$port] INFO\n"
319                    , join("\n", @info)
320                    , "\n\n" ;
321    
322            return @info;
323    }
324    
325    info;
326  run_views;  run_views;
327    
328  while ( 1 ) {  while ( 1 ) {
329    
330          print "sack> ";          print "sack> ";
331          my $cmd = <STDIN>;          chomp( my $cmd = <STDIN> );
332    
333          if ( $cmd =~ m{(vi|\\e|out)}i ) {          if ( $cmd =~ m{^(h|\?)} ) {
334                  system "vi out/*";                  print << "__HELP__"
335          } else {  Sacks Lorry v$VERSION - path: $path offset: $offset limit: $limit
336    
337            View Run        run views
338            VI \\e Output   show output of last run
339            Info [\$VERSION]        instrospect
340            Quit EXit       shutdown
341    
342    __HELP__
343            } elsif ( $cmd =~ m{^(vi|\\e|o)}i ) {
344                    #system "vi out/*";
345                    system "bin/storableedit.pl", (glob('out/*.storable'))[0];
346            } elsif ( $cmd =~ m{^i(?:nfo)?\s?(.+)?$}i ) {
347                    info;
348            } elsif ( $cmd =~ m{^(q|e|x)}i ) {
349                    warn "# exit";
350                    send_nodes 'exit';
351                    exit;
352            } elsif ( $cmd =~ m{^(v|r)}i ) {
353                  run_views;                  run_views;
354            } elsif ( $cmd ) {
355                    warn "UNKNOWN ", dump $cmd;
356          }          }
357    
358  }  }
Legend:
Removed from v.9  
changed lines
  Added in v.36
  ViewVC Help
Powered by ViewVC 1.1.26