--- trunk/bin/sack.pl	2009/09/24 19:21:55	50
+++ trunk/bin/sack.pl	2009/09/25 16:29:14	68
@@ -27,8 +27,8 @@
 	'offset=i' => \$offset,
 	'limit=i'  => \$limit,
 	'view=s'   => \@views,
-	'listen|port=i' => \$port,
-	'connect=s'   => \@nodes,
+	'port|listen=i' => \$port,
+	'node|connect=i'   => \@nodes,
 	'debug!'   => \$debug,
 ) or die $!;
 
@@ -44,7 +44,7 @@
 		chomp( my $pwd = `pwd` );
 		$prefix = "$pwd/$prefix";
 	}
-	$prefix =~ s{^(.*)/srv/Sack/[\./]+bin.+$}{$1};
+	$prefix =~ s{^(.*)/srv/Sack/.+$}{$1};
 	warn "# prefix $prefix";
 
 	$SIG{INT} = sub {
@@ -91,24 +91,32 @@
 
 our $connected;
 
+sub node_sock {
+	my $node = shift;
+	my $sock = IO::Socket::INET->new(
+			PeerAddr => '127.0.0.1',
+			PeerPort => $node,
+			Proto    => 'tcp',
+	);
+	
+	return $sock if $sock && $sock->connected;
+
+	warn "[$port] can't connect to $node - $!\n"; # FIXME die?
+	return;
+}
+
 sub send_nodes {
 	my $content = $#_ > 0 ? pop @_ : '';	# no content with just one argument!
 	my $header = defined $content ? length($content) : 0;
 	$header .= ' ' . join(' ', @_) if @_;
 
-	foreach my $node ( @nodes ) {
+	warn "# send_nodes ", dump @nodes;
 
-		my $sock = IO::Socket::INET->new(
-			PeerAddr => $node,
-			Proto    => 'tcp',
-		);
+	foreach my $node ( @nodes ) {
 
-		if ( ! $sock ) {
-			warn "can't connect to $node - $!"; # FIXME die?
-			next;
-		}
+		my $sock = node_sock($node) || next;
 
-		warn "[$port] >>>> $node $header\n";
+		warn "[$port] >>>> [$node] $header\n";
 		print $sock "$header\n$content" || warn "can't send $header to $node: $!";
 
 		$connected->{$node} = $sock;
@@ -119,13 +127,13 @@
 	my $node = shift;
 
 	my $sock = $connected->{$node};
-	if ( ! $sock ) {
-		warn "[$port] ERROR lost connection to $node";
+	if ( ! $sock || ! $sock->connected ) {
+		warn "[$port] no connection to $node";
 		delete $connected->{$node};
 		return;
 	}
 	chomp( my $size = <$sock> );
-	warn "[$port] <<<< $node $size bytes\n";
+	warn "[$port] <<<< [$node] $size bytes\n" if $debug || $size > 1024;
 	my $data;
 	read $sock, $data, $size;
 	return $data;
@@ -134,21 +142,50 @@
 sub send_sock {
 	my ( $sock, $data ) = @_;
 	my $size   = length $data;
-	warn "[$port] >>>> ", $sock->peerhost, " $size bytes\n";
+	warn "[$port] >>>> $size bytes\n" if $debug || $size > 1024;
 	print $sock "$size\n$data" || warn "can't send $size bytes to ", $sock->peerhost;
 }
 
+sub pull_node_file {
+	my ( $node, $file ) = @_;
+
+	my $path = "/dev/shm/sack.$node.$file";
+#	return if -e $path; # FIXME
+
+	my $sock = node_sock $node || die "not connected to $node";
+
+	print $sock "0 file $file\n";
+
+	my $size = <$sock>;
+	chomp($size);
+	warn "[$port] pull_node_file $node $file $size bytes\n";
+
+	my $block = 4096;
+	my $buff  = ' ' x $block;
+
+	open(my $fh, '>', $path) || die "can't open $path";
+	while ( read $sock, $buff, $block ) {
+		print $fh $buff;
+	}
+	close($fh);
+}
+
 sub merge_out {
 	my ( $from_node, $new ) = @_;
 
-	warn "### merge $from_node";
+	my $t_merge = time();
 
-	my $from_port = $from_node;
-	$from_port =~ s{.+:(\d+)$}{$1};
+	pull_node_file $from_node => 'nr_md5';
+	pull_node_file $from_node => 'md5';
 
-	my $remote_digest = Sack::Digest->new( port => $from_port );
+	my $remote_digest = Sack::Digest->new( port => $from_node );
 	my ( $local, $remote ) = ( 0, 0 );
 
+	my $tick = 0;
+	print STDERR "[$port] merge [$from_node]";
+
+	my $missing;
+
 	foreach my $k1 ( keys %$new ) {
 
 		foreach my $k2 ( keys %{ $new->{$k1} } ) {
@@ -158,13 +195,21 @@
 			if ( $k1 =~ m{#} ) {
 				die "ASSERT $k1 $k2" unless $k2 =~ m{^\d+$};
 #warn "XXX $k1 $k2";
-				my $md5 = $remote_digest->{nr_md5}->[$k2] || warn "[$port] no2md5 $n not found in $from_port\n";
+				my $md5 = $remote_digest->{nr_md5}->[$k2];
+
+				if ( ! $md5 ) {
+					$missing->{nr_md5}->{$from_node}++; # FIXME die?
+					next;
+				}
+
 				if ( my $local_k2 = $digest->{md5_nr}->{$md5} ) {
 					$k2 = $local_k2;
 					$local++;
-				} else {
+				} elsif ( my $full = $remote_digest->{md5}->{$md5} ) {
 					$k2 = $digest->to_int( $remote_digest->{md5}->{$md5} );
 					$remote++;
+				} else {
+					$missing->{md5}->{$from_node}++;
 				}
 			}
 
@@ -186,10 +231,20 @@
 			} else {
 				die "can't merge $k2 [$ref] from ",dump($n), " into ", dump($out->{$k1}->{$k2});
 			}
+
+			if ( $tick++ % 1000 == 0 ) {
+				print STDERR ".";
+			} elsif ( $tick % 10000 == 0 ) {
+				print STDERR $tick;
+			}
 		}
 	}
 
-	warn "[$port] merge local $local remote $remote from $from_port\n";
+	$t_merge = time - $t_merge;
+	printf STDERR "%d in %.4fs %.2f/s local %.1f%% %d/%d\n", $tick, $t_merge, $tick / $t_merge, $local * 100 / $tick, $local, $remote;
+
+	warn "[$port] missing ", dump $missing if $missing;
+
 	warn "## merge out ", dump $out if $debug;
 }
 
@@ -234,6 +289,7 @@
 		foreach my $node ( keys %$connected ) {
 			warn "[$port] get_node $node\n";
 			my $o = get_node $node;
+			next unless $o;
 			my $s = length $o;
 			$o = thaw $o;
 			warn "[$port] merge $node $s bytes\n";
@@ -321,6 +377,19 @@
 		} elsif ( $header[0] eq 'exit' ) {
 			warn "[$port] exit";
 			exit;
+		} elsif ( $header[0] eq 'file' ) {
+			$digest->close;
+			my $path = "/dev/shm/sack.$port.$header[1]";
+			my $size = -s $path;
+			warn "[$port] >>>> file $path $size bytes\n";
+			print $client "$size\n";
+			my $block = 4096;
+			my $buff  = ' ' x $block;
+			open(my $fh, '<', $path) || die "can't open $path";
+			while ( read $fh, $buff, $block ) {
+				print $client $buff;
+			}
+			$digest->open;
 		} else {
 			warn "[$port] UNKNOWN $header[0]";
 		}
@@ -366,6 +435,7 @@
 __HELP__
 	} elsif ( $cmd =~ m{^(vi|\\e|o)}i ) {
 		#system "vi out/*";
+		$digest->sync;
 		system "bin/storableedit.pl", (glob('out/*.storable'))[0];
 	} elsif ( $cmd =~ m{^i(?:nfo)?\s?(.+)?$}i ) {
 		info;
@@ -375,6 +445,9 @@
 		exit;
 	} elsif ( $cmd =~ m{^(v|r)}i ) {
 		run_views;
+	} elsif ( $cmd =~ m{^n(ode)?\s*(\d+)}i ) {
+		push @nodes, $2;
+		info;
 	} elsif ( $cmd ) {
 		warn "UNKNOWN ", dump $cmd;
 	}