/[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 17 by dpavlin, Tue Sep 22 13:40:02 2009 UTC revision 22 by dpavlin, Tue Sep 22 17:19:25 2009 UTC
# Line 3  Line 3 
3  use warnings;  use warnings;
4  use strict;  use strict;
5    
6    our $VERSION = '0.01';
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;
# Line 11  use IO::Socket::INET; Line 13  use IO::Socket::INET;
13  use Storable qw/freeze thaw/;  use Storable qw/freeze thaw/;
14    
15    
16    my $debug  = 0;
17  my $path   = '/data/isi/full.txt';  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;  my $listen = 0; # off
22  my @nodes;  my @nodes;
23    
24    
# Line 26  GetOptions( Line 29  GetOptions(
29          'view=s'   => \@views,          'view=s'   => \@views,
30          'listen|port=i' => \$listen,          'listen|port=i' => \$listen,
31          'connect=s'   => \@nodes,          'connect=s'   => \@nodes,
32            'debug!'   => \$debug,
33  ) or die $!;  ) or die $!;
34    
35  my $t = time;  my $t = time;
36    
37    
38    sub send_nodes;
39    
40  our $prefix;  our $prefix;
41  BEGIN {  sub BEGIN {
42          $prefix = $0;          $prefix = $0;
43          if ( $prefix =~ s{^./}{} ) {          if ( $prefix =~ s{^./}{} ) {
44                  chomp( my $pwd = `pwd` );                  chomp( my $pwd = `pwd` );
# Line 40  BEGIN { Line 46  BEGIN {
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 {
51                    my $signame = shift;
52                    send_nodes 'exit';
53                    die "SIG$signame";
54            };
55  }  }
56    
57    
# Line 72  our $connected; Line 84  our $connected;
84    
85  sub send_nodes {  sub send_nodes {
86          my $content = $#_ > 0 ? pop @_ : '';    # no content with just one argument!          my $content = $#_ > 0 ? pop @_ : '';    # no content with just one argument!
87          my $header = length($content);          my $header = defined $content ? length($content) : 0;
88          $header .= ' ' . join(' ', @_) if @_;          $header .= ' ' . join(' ', @_) if @_;
89    
90          foreach my $node ( @nodes ) {          foreach my $node ( @nodes ) {
# Line 87  sub send_nodes { Line 99  sub send_nodes {
99                          next;                          next;
100                  }                  }
101    
102                  print ">>>> $node $header\n";                  print ">>>> $listen $node $header\n";
103                  print $sock "$header\n$content" || warn "can't send $header to $node: $!";                  print $sock "$header\n$content" || warn "can't send $header to $node: $!";
104    
105                  $connected->{$node} = $sock;                  $connected->{$node} = $sock;
# Line 104  sub get_node { Line 116  sub get_node {
116                  return;                  return;
117          }          }
118          chomp( my $size = <$sock> );          chomp( my $size = <$sock> );
119          warn "<<<< $node $size bytes\n";          warn "<<<< $listen $node $size bytes\n";
120          my $data;          my $data;
121          read $sock, $data, $size;          read $sock, $data, $size;
122          return $data;          return $data;
# Line 113  sub get_node { Line 125  sub get_node {
125  sub send_sock {  sub send_sock {
126          my ( $sock, $data ) = @_;          my ( $sock, $data ) = @_;
127          my $size   = length $data;          my $size   = length $data;
128          warn ">>>> ", $sock->peerhost, " $size bytes";          warn ">>>> $listen ", $sock->peerhost, " $size bytes";
129          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;
130  }  }
131    
132  sub merge_out {  sub merge_out {
133          my $new = shift;          my $new = shift;
134    
         warn "## merge $new\n";  
   
135          foreach my $k1 ( keys %$new ) {          foreach my $k1 ( keys %$new ) {
136    
137                  foreach my $k2 ( keys %{ $new->{$k1} } ) {                  foreach my $k2 ( keys %{ $new->{$k1} } ) {
# Line 132  sub merge_out { Line 142  sub merge_out {
142                          if ( ! defined $out->{$k1}->{$k2} ) {                          if ( ! defined $out->{$k1}->{$k2} ) {
143                                  $out->{$k1}->{$k2} = $n;                                  $out->{$k1}->{$k2} = $n;
144                          } elsif ( $k1 =~ m{\+} ) {                          } elsif ( $k1 =~ m{\+} ) {
145                                  warn "# agregate $k1 $k2";  #                               warn "## agregate $k1 $k2";
146                                  $out->{$k1}->{$k2} += $n;                                  $out->{$k1}->{$k2} += $n;
147                          } elsif ( $ref eq 'ARRAY' ) {                          } elsif ( $ref  eq 'ARRAY' ) {
148                                  push @{ $out->{$k1}->{$k2} }, $n;                                  if ( ref $n eq 'ARRAY' ) {
149                                            push @{ $out->{$k1}->{$k2} }, $_ foreach @$n;
150                                    } else {
151                                            push @{ $out->{$k1}->{$k2} }, $n;
152                                    }
153                          } elsif ( $ref eq '' ) {                          } elsif ( $ref eq '' ) {
154                                  $out->{$k1}->{$k2} = [ $out->{$k1}->{$k2}, $n ];                                  $out->{$k1}->{$k2} = [ $out->{$k1}->{$k2}, $n ];
155                          } else {                          } else {
# Line 144  sub merge_out { Line 158  sub merge_out {
158                  }                  }
159          }          }
160    
161          warn "## merge out ", dump $out;          warn "## merge out ", dump $out if $debug;
162  }  }
163    
164  sub run_code {  sub run_code {
165          my ( $view, $code ) = @_;          my ( $view, $code ) = @_;
166    
167          warn "\n#### CODE $view START ####\n$code\n#### CODE $view END ####\n";          warn "\n#### CODE $view START ####\n$code\n#### CODE $view END ####\n" if $debug;
168    
169          send_nodes view => $view => $code;          send_nodes view => $view => $code;
170    
# Line 168  sub run_code { Line 182  sub run_code {
182    
183                  eval "$code";                  eval "$code";
184                  if ( $@ ) {                  if ( $@ ) {
185                          warn "ERROR [$pos] $@\n";                          warn "ABORT [$pos] $@\n";
186                            last;
187                  } else {                  } else {
188                          $affected++;                          $affected++;
189                  }                  }
# Line 222  sub run_views { Line 237  sub run_views {
237  if ( $listen ) {  if ( $listen ) {
238          my $sock = IO::Socket::INET->new(          my $sock = IO::Socket::INET->new(
239                  Listen    => SOMAXCONN,                  Listen    => SOMAXCONN,
240  #               LocalAddr => '0.0.0.0',                  LocalAddr => '127.0.0.1',
241                  LocalPort => $listen,                  LocalPort => $listen,
242                  Proto     => 'tcp',                  Proto     => 'tcp',
243                  Reuse     => 1,                  Reuse     => 1,
# Line 230  if ( $listen ) { Line 245  if ( $listen ) {
245    
246          while (1) {          while (1) {
247    
248                  warn "NODE listen on $listen\n";                  warn "NODE $listen ready - wating for connection\n";
249    
250                  my $client = $sock->accept();                  my $client = $sock->accept();
251    
252                  warn "<<<< $listen connect from ", $client->peerhost, $/;                  warn "<<<< $listen connect from ", $client->peerhost, $/;
253    
254                  my @header = split(/\s/, <$client>);                  my @header = split(/\s/, <$client>);
255                  warn "# header ",dump @header;                  warn "<<<< $listen header ",dump(@header),$/;
256    
257                  my $size = shift @header;                  my $size = shift @header;
258    
# Line 249  if ( $listen ) { Line 264  if ( $listen ) {
264                          send_sock $client => freeze $out;                          send_sock $client => freeze $out;
265                  } elsif ( $header[0] eq 'info' ) {                  } elsif ( $header[0] eq 'info' ) {
266                          my $info = "$listen\t$offset\t$limit\t$path";                          my $info = "$listen\t$offset\t$limit\t$path";
267                            $info .= "\t" . eval $header[1] if $header[1];
268                          warn "info $info\n";                          warn "info $info\n";
269                          send_sock $client => $info;                          send_sock $client => $info;
270                  } elsif ( $header[0] eq 'exit' ) {                  } elsif ( $header[0] eq 'exit' ) {
# Line 266  run_views; Line 282  run_views;
282  while ( 1 ) {  while ( 1 ) {
283    
284          print "sack> ";          print "sack> ";
285          my $cmd = <STDIN>;          chomp( my $cmd = <STDIN> );
286    
287          if ( $cmd =~ m{^(vi?|\\e|o(?:ut)?)}i ) {          if ( $cmd =~ m{^(h|\?)} ) {
288                    print << "__HELP__"
289    Sacks Lorry v$VERSION - path: $path offset: $offset limit: $limit
290    
291            View Run        run views
292            VI \\e Output   show output of last run
293            Info [\$VERSION]        instrospect
294            Quit EXit       shutdown
295    
296    __HELP__
297            } elsif ( $cmd =~ m{^(vi|\\e|o)}i ) {
298                  system "vi out/*";                  system "vi out/*";
299          } elsif ( $cmd =~ m{^i(nfo)?}i ) {          } elsif ( $cmd =~ m{^i(?:nfo)?\s?(.+)?$}i ) {
300                  print "# nodes: ", join(' ',@nodes), $/;                  print "# nodes: ", join(' ',@nodes), $/;
301    
302                    send_nodes 'info' => $2;
303    
304                  my @info = (                  my @info = (
305                          "node\toffset\tlimit\tpath",                          "node\toffset\tlimit\tpath",
306                          "----\t------\t-----\t----",                          "----\t------\t-----\t----",
307                          "here\t$offset\t$limit\t$path",                          "0\t$offset\t$limit\t$path",
308                  );                  );
309    
                 send_nodes 'info';  
310                  push @info, get_node $_ foreach @nodes;                  push @info, get_node $_ foreach @nodes;
311    
312                  print "$_\n" foreach @info;                  print "$_\n" foreach @info;
313    
314          } elsif ( $cmd =~ m{^(q(uit)|e(xit))}i ) {          } elsif ( $cmd =~ m{^(q|e|x)}i ) {
315                  warn "# exit";                  warn "# exit";
316                  send_nodes 'exit';                  send_nodes 'exit';
317                  exit;                  exit;
318          } else {          } elsif ( $cmd =~ m{^(v|r)}i ) {
319                  run_views;                  run_views;
320            } elsif ( $cmd ) {
321                    warn "UNKNOWN ", dump $cmd;
322          }          }
323    
324  }  }
Legend:
Removed from v.17  
changed lines
  Added in v.22
  ViewVC Help
Powered by ViewVC 1.1.26