/[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 46 by dpavlin, Thu Sep 24 13:46:16 2009 UTC revision 64 by dpavlin, Fri Sep 25 14:01:31 2009 UTC
# Line 3  Line 3 
3  use warnings;  use warnings;
4  use strict;  use strict;
5    
6  our $VERSION = '0.03';  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);
# 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 44  sub BEGIN { Line 44  sub BEGIN {
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/.+$}{$1};
48          warn "# prefix $prefix";          warn "# prefix $prefix";
49    
50          $SIG{INT} = sub {          $SIG{INT} = sub {
# Line 91  our $cache; Line 91  our $cache;
91    
92  our $connected;  our $connected;
93    
94    sub node_sock {
95            my $node = shift;
96            my $sock = IO::Socket::INET->new(
97                            PeerAddr => '127.0.0.1',
98                            PeerPort => $node,
99                            Proto    => 'tcp',
100            );
101            
102            return $sock if $sock && $sock->connected;
103    
104            warn "[$port] can't connect to $node - $!\n"; # FIXME die?
105            return;
106    }
107    
108  sub send_nodes {  sub send_nodes {
109          my $content = $#_ > 0 ? pop @_ : '';    # no content with just one argument!          my $content = $#_ > 0 ? pop @_ : '';    # no content with just one argument!
110          my $header = defined $content ? length($content) : 0;          my $header = defined $content ? length($content) : 0;
111          $header .= ' ' . join(' ', @_) if @_;          $header .= ' ' . join(' ', @_) if @_;
112    
113            warn "# send_nodes ", dump @nodes;
114    
115          foreach my $node ( @nodes ) {          foreach my $node ( @nodes ) {
116    
117                  my $sock = IO::Socket::INET->new(                  my $sock = node_sock($node) || next;
                         PeerAddr => $node,  
                         Proto    => 'tcp',  
                 );  
   
                 if ( ! $sock ) {  
                         warn "can't connect to $node - $!"; # FIXME die?  
                         next;  
                 }  
118    
119                  warn "[$port] >>>> $node $header\n";                  warn "[$port] >>>> $node $header\n";
120                  print $sock "$header\n$content" || warn "can't send $header to $node: $!";                  print $sock "$header\n$content" || warn "can't send $header to $node: $!";
# Line 119  sub get_node { Line 127  sub get_node {
127          my $node = shift;          my $node = shift;
128    
129          my $sock = $connected->{$node};          my $sock = $connected->{$node};
130          if ( ! $sock ) {          if ( ! $sock || ! $sock->connected ) {
131                  warn "[$port] ERROR lost connection to $node";                  warn "[$port] no connection to $node";
132                  delete $connected->{$node};                  delete $connected->{$node};
133                  return;                  return;
134          }          }
135          chomp( my $size = <$sock> );          chomp( my $size = <$sock> );
136          warn "[$port] <<<< $node $size bytes\n";          warn "[$port] <<<< $node $size bytes\n" if $debug || $size > 1024;
137          my $data;          my $data;
138          read $sock, $data, $size;          read $sock, $data, $size;
139          return $data;          return $data;
# Line 134  sub get_node { Line 142  sub get_node {
142  sub send_sock {  sub send_sock {
143          my ( $sock, $data ) = @_;          my ( $sock, $data ) = @_;
144          my $size   = length $data;          my $size   = length $data;
145          warn "[$port] >>>> ", $sock->peerhost, " $size bytes\n";          warn "[$port] >>>> $size bytes\n" if $debug || $size > 1024;
146          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;
147  }  }
148    
149    sub pull_node_file {
150            my ( $node, $file ) = @_;
151    
152            my $path = "/dev/shm/sack.$node.$file";
153    #       return if -e $path; # FIXME
154    
155            my $sock = node_sock $node || die "not connected to $node";
156    
157            print $sock "0 file $file\n";
158    
159            my $size = <$sock>;
160            chomp($size);
161            warn "[$port] pull_node_file $node $file $size bytes\n";
162    
163            my $block = 4096;
164            my $buff  = ' ' x $block;
165    
166            open(my $fh, '>', $path) || die "can't open $path";
167            while ( read $sock, $buff, $block ) {
168                    print $fh $buff;
169            }
170            close($fh);
171    }
172    
173  sub merge_out {  sub merge_out {
174          my $new = shift;          my ( $from_node, $new ) = @_;
175    
176            pull_node_file $from_node => 'nr_md5';
177            pull_node_file $from_node => 'md5';
178    
179            my $remote_digest = Sack::Digest->new( port => $from_node );
180            my ( $local, $remote ) = ( 0, 0 );
181    
182            my $tick = 0;
183            print STDERR "[$port] merge $from_node";
184    
185            my $missing;
186    
187          foreach my $k1 ( keys %$new ) {          foreach my $k1 ( keys %$new ) {
188    
189                  foreach my $k2 ( keys %{ $new->{$k1} } ) {                  foreach my $k2 ( keys %{ $new->{$k1} } ) {
190    
191                          my $n   = delete $new->{$k1}->{$k2};                          my $n   = delete $new->{$k1}->{$k2};
                         my $ref = ref    $out->{$k1}->{$k2};  
192    
193                            if ( $k1 =~ m{#} ) {
194                                    die "ASSERT $k1 $k2" unless $k2 =~ m{^\d+$};
195    #warn "XXX $k1 $k2";
196                                    my $md5 = $remote_digest->{nr_md5}->[$k2];
197    
198                                    if ( ! $md5 ) {
199                                            $missing->{$from_node}++; # FIXME die?
200                                            next;
201                                    }
202    
203                                    if ( my $local_k2 = $digest->{md5_nr}->{$md5} ) {
204                                            $k2 = $local_k2;
205                                            $local++;
206                                    } else {
207                                            $k2 = $digest->to_int( $remote_digest->{md5}->{$md5} );
208                                            $remote++;
209                                    }
210                            }
211    
212                            my $ref = ref    $out->{$k1}->{$k2};
213    #warn "XXXX $k1 $k2 $ref";
214                          if ( ! defined $out->{$k1}->{$k2} ) {                          if ( ! defined $out->{$k1}->{$k2} ) {
215                                  $out->{$k1}->{$k2} = $n;                                  $out->{$k1}->{$k2} = $n;
216                          } elsif ( $k1 =~ m{\+} ) {                          } elsif ( $k1 =~ m{\+} ) {
# Line 164  sub merge_out { Line 227  sub merge_out {
227                          } else {                          } else {
228                                  die "can't merge $k2 [$ref] from ",dump($n), " into ", dump($out->{$k1}->{$k2});                                  die "can't merge $k2 [$ref] from ",dump($n), " into ", dump($out->{$k1}->{$k2});
229                          }                          }
230    
231                            if ( $tick++ % 1000 == 0 ) {
232                                    print STDERR ".";
233                            } elsif ( $tick % 10000 == 0 ) {
234                                    print STDERR $tick;
235                            }
236                  }                  }
237          }          }
238    
239            warn "$tick $local/$remote\n";
240    
241            warn "[$port] missing ", dump $missing if $missing;
242    
243          warn "## merge out ", dump $out if $debug;          warn "## merge out ", dump $out if $debug;
244  }  }
245    
# Line 211  sub run_code { Line 284  sub run_code {
284                  foreach my $node ( keys %$connected ) {                  foreach my $node ( keys %$connected ) {
285                          warn "[$port] get_node $node\n";                          warn "[$port] get_node $node\n";
286                          my $o = get_node $node;                          my $o = get_node $node;
287                            next unless $o;
288                          my $s = length $o;                          my $s = length $o;
289                          $o = thaw $o;                          $o = thaw $o;
290                          warn "[$port] merge $node $s bytes\n";                          warn "[$port] merge $node $s bytes\n";
291                          merge_out $o;                          merge_out $node => $o;
292                            report "[$port] merged $node";
293                  }                  }
294          }          }
295  }  }
# Line 298  if ( $port ) { Line 373  if ( $port ) {
373                  } elsif ( $header[0] eq 'exit' ) {                  } elsif ( $header[0] eq 'exit' ) {
374                          warn "[$port] exit";                          warn "[$port] exit";
375                          exit;                          exit;
376                    } elsif ( $header[0] eq 'file' ) {
377                            $digest->close;
378                            my $path = "/dev/shm/sack.$port.$header[1]";
379                            my $size = -s $path;
380                            warn "[$port] >>>> file $path $size bytes\n";
381                            print $client "$size\n";
382                            my $block = 4096;
383                            my $buff  = ' ' x $block;
384                            open(my $fh, '<', $path) || die "can't open $path";
385                            while ( read $fh, $buff, $block ) {
386                                    print $client $buff;
387                            }
388                  } else {                  } else {
389                          warn "[$port] UNKNOWN $header[0]";                          warn "[$port] UNKNOWN $header[0]";
390                  }                  }
# Line 343  Sacks Lorry v$VERSION - path: $path offs Line 430  Sacks Lorry v$VERSION - path: $path offs
430  __HELP__  __HELP__
431          } elsif ( $cmd =~ m{^(vi|\\e|o)}i ) {          } elsif ( $cmd =~ m{^(vi|\\e|o)}i ) {
432                  #system "vi out/*";                  #system "vi out/*";
433                    $digest->sync;
434                  system "bin/storableedit.pl", (glob('out/*.storable'))[0];                  system "bin/storableedit.pl", (glob('out/*.storable'))[0];
435          } elsif ( $cmd =~ m{^i(?:nfo)?\s?(.+)?$}i ) {          } elsif ( $cmd =~ m{^i(?:nfo)?\s?(.+)?$}i ) {
436                  info;                  info;
# Line 352  __HELP__ Line 440  __HELP__
440                  exit;                  exit;
441          } elsif ( $cmd =~ m{^(v|r)}i ) {          } elsif ( $cmd =~ m{^(v|r)}i ) {
442                  run_views;                  run_views;
443            } elsif ( $cmd =~ m{^n(ode)?\s*(\d+)}i ) {
444                    push @nodes, $2;
445                    info;
446          } elsif ( $cmd ) {          } elsif ( $cmd ) {
447                  warn "UNKNOWN ", dump $cmd;                  warn "UNKNOWN ", dump $cmd;
448          }          }
Legend:
Removed from v.46  
changed lines
  Added in v.64
  ViewVC Help
Powered by ViewVC 1.1.26