trunk/bin/sack.pl

#!/usr/bin/perl

use warnings;
use strict;

our $VERSION = '0.06';

use Time::HiRes qw(time);
use Data::Dump qw(dump);
use File::Slurp;
use Getopt::Long;
use IO::Socket::INET;
use Storable qw/freeze thaw store/;


my $debug  = 0;
my $path   = '/data/isi/full.txt';
my $limit  = 5000;
my $offset = 0; 
my @views;
my $port   = 0; # interactive
my @nodes;


GetOptions(
        'path=s'   => \$path,
        'offset=i' => \$offset,
        'limit=i'  => \$limit,
        'view=s'   => \@views,
        'port|listen=i' => \$port,
        'node|connect=i'   => \@nodes,
        'debug!'   => \$debug,
) or die $!;

my $t = time;


sub send_nodes;

our $prefix;
sub BEGIN {
        $prefix = $0;
        if ( $prefix !~ m{^/} ) {
                chomp( my $pwd = `pwd` );
                $prefix = "$pwd/$prefix";
        }
        $prefix =~ s{^(.*)/srv/Sack/.+$}{$1};
        warn "# prefix $prefix";

        $SIG{INT} = sub {
                my $signame = shift;
                send_nodes 'exit';
                #clean if $clean;       # FIXME
                die "SIG$signame";
        };

        sub port2color {
                my $port = shift;
                return "\e[1m0\e[0m" if $port == 0;

                my $c = ( $port % 6 ) + 31;
                return "\e[${c}m$port\e[0m";
        }

        $SIG{__WARN__} = sub {
                return unless @_;
                my $msg = join('', @_);
                if ( $msg !~ m{[\n\r]$} ) {
                        my @loc = caller(1);
                        $msg .= " in $loc[1] +$loc[2]\n" if @loc;
                }
                $msg =~ s{\[(0|\d\d\d\d)\]}{ '[' . port2color($1) . ']' }eg;
                print STDERR $msg;
                return 1;
        };

}

our $digest_module = 'Sack::Digest::' . ( $ENV{SACK_DIGEST} || 'BerkeleyDB' );
use lib "$prefix/srv/Sack/lib/";
my $digest_path = $digest_module;
$digest_path =~ s{::}{/}g;
require "lib/$digest_path.pm";

our $digest = $digest_module->new( port => $port, clean => 1 );
sub digest { $digest->to_int($_[0]) }

warn "[$port] using $digest_module\n";

use lib "$prefix/srv/webpac2/lib/";
use WebPAC::Input::ISI;

$WebPAC::Input::ISI::subfields = undef; # disable parsing of subfields

my $input = WebPAC::Input::ISI->new(
        path   => "$prefix/$path",
        offset => $offset,
        limit  => $limit,
);

our $num_records = $input->size;
our @reports;

sub report {
        my $description = join(' ',@_);
        my $dt = time - $t;
        my $report = [ $description, $dt, $input->size / $dt ];
        warn sprintf "[$port] %s in %1.4fs %.2f/s\n", @$report;
        push @reports, $report;
        $t = time;
}

sub show_report {
        "\n" . join( "\n", map { sprintf "%8.4fs %10.2f/s  %s", $_->[1], $_->[2], $_->[0] } @reports ) . "\n";
}

report $input->size , 'records loaded';

mkdir 'out' unless -e 'out';

our $out;

our $cache;

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 @_;

        warn "# send_nodes ", dump(@_), " to ", dump @nodes;

        foreach my $node ( @nodes ) {

                my $sock = node_sock($node) || next;

                warn "[$port] >>>> [$node] $header\n";
                print $sock "$header\n$content" || warn "can't send $header to $node: $!";

                $connected->{$node} = $sock;
        }
}

sub get_node {
        my $node = shift;

        my $sock = $connected->{$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" if $debug || $size > 1024;
        my $data;
        read $sock, $data, $size;
        return $data;
}

sub send_sock {
        my ( $sock, $data ) = @_;
        my $size   = length $data;
        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);
}

my $merge_digest_mapping;

sub merge_out {
        my ( $from_node, $new ) = @_;

        my $t_merge = time();

        pull_node_file $from_node => 'nr_md5';
        pull_node_file $from_node => 'md5';

        my $remote_digest = $digest_module->new( port => $from_node );
        my ( $local, $remote ) = ( 0, 0 );

        my $tick = 0;
        warn "[$port] merge [$from_node]";

        my $missing;

        foreach my $k1 ( keys %$new ) {

                foreach my $k2 ( keys %{ $new->{$k1} } ) {

                        my $n   = delete $new->{$k1}->{$k2};

                        if ( $k1 =~ m{#} ) {
                                die "ASSERT $k1 $k2" unless $k2 =~ m{^\d+$};
#warn "XXX $k1 $k2";

                                if ( defined $merge_digest_mapping->{$from_node}->[ $k2 ] ) {
                                        $k2 = $merge_digest_mapping->{$from_node}->[ $k2 ];
                                } else {

                                        my $md5 = $remote_digest->{nr_md5}->[$k2];

                                        if ( ! $md5 ) {
                                                $missing->{nr_md5}->{$from_node}++; # FIXME die?
                                                next;
                                        }

                                        my $local_k2;

                                        if ( $local_k2 = $digest->{md5_nr}->{$md5} ) {
                                                $local++;
                                        } elsif ( my $full = $remote_digest->{md5}->{$md5} ) {
                                                $local_k2 = $digest->to_int( $remote_digest->{md5}->{$md5} );
                                                $remote++;
                                        } else {
                                                $missing->{md5}->{$from_node}++;
                                        }

                                        $k2 = $merge_digest_mapping->{$from_node}->[ $k2 ] = $local_k2;

                                }
                        }

                        my $ref = ref    $out->{$k1}->{$k2};
#warn "XXXX $k1 $k2 $ref";
                        if ( ! defined $out->{$k1}->{$k2} ) {
                                $out->{$k1}->{$k2} = $n;
                        } elsif ( $k1 =~ m{\+} ) {
#                               warn "## agregate $k1 $k2";
                                $out->{$k1}->{$k2} += $n;
                        } elsif ( $ref  eq 'ARRAY' ) {
                                if ( ref $n eq 'ARRAY' ) {
                                        push @{ $out->{$k1}->{$k2} }, $_ foreach @$n;
                                } else {
                                        push @{ $out->{$k1}->{$k2} }, $n;
                                }
                        } elsif ( $ref eq '' ) {
                                $out->{$k1}->{$k2} = [ $out->{$k1}->{$k2}, $n ];
                        } 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;
                        }
                }
        }

        $t_merge = time - $t_merge;
        my $digests = $local + $remote;
        warn sprintf "\n[$port] merge %d in %.4fs %.2f/s digests local: %.1f%% %d/%d\n", $tick, $t_merge, $digests / $t_merge, $local * 100 / ( $digests || 1 ), $local, $remote;
        push @reports, [ "$tick merged [$from_node]", $t_merge, $digests / $t_merge ];

        warn "[$port] missing ", dump $missing if $missing;

        warn "## merge out ", dump $out if $debug;
}

our $rec;

sub run_code {
        my ( $view, $code ) = @_;

        warn "\n#### CODE $view START ####\n$code\n#### CODE $view END ####\n" if $debug;

        send_nodes view => $view => $code;

        undef $out;

        my $affected = 0;
        $t = time;

        my $coderef = eval "sub { $code }";
        if ( $@ ) {
                warn "ABORT code: $@";
                return;
        }

        foreach my $pos ( $offset + 1 .. $offset + $input->size ) {
                $rec = $cache->{$pos} ||= $input->fetch_rec( $pos );
                if ( ! $rec ) {
                        print STDERR "END @ $pos";
                        last;
                }

                eval { $coderef->() };
                if ( $@ ) {
                        warn "ABORT $pos $@\n";
                        last;
                } else {
                        $affected++;
                }

                $pos % 10000 == 0 ? print STDERR $pos :
                $pos % 1000  == 0 ? print STDERR "."  : 0 ;
        };

        report "$affected affected $view";

        warn "WARN no \$out defined!" unless defined $out;

        $digest->sync;

        if ( $connected ) {
                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] got $s bytes from [$node]\n";
                        merge_out $node => $o;
                }
        }
}

sub run_views {
        @views = sort glob 'views/*.pl' unless @views;
        warn "# views ", dump @views;

        foreach my $view ( @views ) {

                next if system("perl -c $view") != 0;

                my $code = read_file $view;

                run_code $view => $code;

                if ( defined $out ) {

                        my $path = $view;
                        $path =~ s{views?/}{out/} || die "no view in $view";
                        $path =~ s{\.pl}{.storable};

                        unlink "$path.last" if -e "$path.last";
                        rename $path, "$path.last";

                        store $out => $path;
                        report "save $path", -s $path, "bytes";

                        if ( -s $path < 4096 ) {
                                print '$out = ', dump $digest->undigest_out($out);
                        }
                }

        }

}


sub info_tabs {
        my $port_col = port2color($port);
        "$port_col\t$offset\t$limit\t$num_records\t$path\t"
        . join("\t", map {
                my $b = $_;
                $b =~ s{^.+\.$port\.([^/]+)$}{$1};
                "$b " . -s $_
        } glob "/dev/shm/sack.$port.*" );
}


if ( $port ) {

        my $pid_path = "/tmp/sack.$port.pid";
        if ( -e $pid_path ) {
                my $pid = read_file $pid_path;
                kill 9, $pid && warn "[$port] kill old $pid\n";
        }
        write_file $pid_path, $$;


        my $sock = IO::Socket::INET->new(
                Listen    => SOMAXCONN,
                LocalAddr => '127.0.0.1',
                LocalPort => $port,
                Proto     => 'tcp',
                Reuse     => 1,
        ) or die "[$port] die $!";

        while (1) {

                warn "[$port] accept path: $path offset: $offset limit: $limit #recs: $num_records\n";

                my $client = $sock->accept();

                warn "[$port] <<<< connect from ", $client->peerhost, $/;

                my @header = split(/\s/, <$client>);
                warn "[$port] <<<< header ",dump(@header),$/;

                my $size = shift @header;

                my $content;
                read $client, $content, $size;

                if ( $header[0] eq 'view' ) {
                        run_code $header[1] => $content;
                        send_sock $client => freeze $out;
                } elsif ( $header[0] eq 'info' ) {
                        my $info = info_tabs;
                        warn "[$port] info $info\n";
                        $info .= "\n" . show_report if $content =~ m{r}i;
                        send_sock $client => $info;
                } 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]";
                }

                $client->close;

        }
}

sub info {
        my $detail = shift || '';

        send_nodes 'info' => $detail;

        my @info = (
                "port\toffset\tlimit\t#recs\tpath",
                "----\t------\t-----\t-----\t----",
                info_tabs,
        );

        push @info, show_report if $detail =~ m{r}i;

        push @info, get_node $_ foreach @nodes;

        print "[$port] INFO", $detail ? " $detail" : '', " \n"
                , join("\n", @info)
                , "\n" ;

        return @info;
}

info;
while ( keys %$connected != @nodes ) {
        warn "[$port] wait for [", join('] [', grep { ! defined $connected->{$_} } @nodes ), "]\n";
        sleep 1;
        info;
}
run_views;

while ( 1 ) {

        print "sack> ";

        my $cmd = <STDIN>;
        exit unless defined $cmd; # CTRL+D
        chomp $cmd;

        if ( $cmd =~ m{^(h|\?)} ) {
                print << "__HELP__"
Sacks Lorry v$VERSION - path: $path offset: $offset limit: $limit

        View Run        run views
        VI \\e Output   show output of last run
        Info [Report]   info with optional report
        Quit EXit       shutdown

__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?(\S+)?$}i ) {
                info $1;
        } elsif ( $cmd =~ m{^(q|e|x)}i ) {
                warn "# exit";
                send_nodes 'exit';
                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;
        }

}

1	dpavlin	1	#!/usr/bin/perl
2
3			use warnings;
4			use strict;
5
6	dpavlin	82	our $VERSION = '0.06';
7	dpavlin	21
8	dpavlin	1	use Time::HiRes qw(time);
9			use Data::Dump qw(dump);
10			use File::Slurp;
11	dpavlin	4	use Getopt::Long;
12	dpavlin	11	use IO::Socket::INET;
13	dpavlin	30	use Storable qw/freeze thaw store/;
14	dpavlin	1
15	dpavlin	4
16	dpavlin	19	my $debug = 0;
17	dpavlin	4	my $path = '/data/isi/full.txt';
18	dpavlin	11	my $limit = 5000;
19	dpavlin	4	my $offset = 0;
20	dpavlin	6	my @views;
21	dpavlin	30	my $port = 0; # interactive
22	dpavlin	11	my @nodes;
23	dpavlin	4
24
25			GetOptions(
26			'path=s' => \$path,
27			'offset=i' => \$offset,
28			'limit=i' => \$limit,
29	dpavlin	6	'view=s' => \@views,
30	dpavlin	64	'port\|listen=i' => \$port,
31			'node\|connect=i' => \@nodes,
32	dpavlin	19	'debug!' => \$debug,
33	dpavlin	5	) or die $!;
34	dpavlin	4
35	dpavlin	1	my $t = time;
36
37	dpavlin	12
38	dpavlin	21	sub send_nodes;
39
40	dpavlin	12	our $prefix;
41	dpavlin	21	sub BEGIN {
42	dpavlin	12	$prefix = $0;
43	dpavlin	30	if ( $prefix !~ m{^/} ) {
44	dpavlin	12	chomp( my $pwd = `pwd` );
45			$prefix = "$pwd/$prefix";
46			}
47	dpavlin	51	$prefix =~ s{^(.*)/srv/Sack/.+$}{$1};
48	dpavlin	12	warn "# prefix $prefix";
49	dpavlin	21
50			$SIG{INT} = sub {
51			my $signame = shift;
52			send_nodes 'exit';
53	dpavlin	33	#clean if $clean; # FIXME
54	dpavlin	21	die "SIG$signame";
55			};
56	dpavlin	81
57			sub port2color {
58			my $port = shift;
59	dpavlin	82	return "\e[1m0\e[0m" if $port == 0;
60
61			my $c = ( $port % 6 ) + 31;
62			return "\e[${c}m$port\e[0m";
63	dpavlin	81	}
64
65			$SIG{__WARN__} = sub {
66	dpavlin	82	return unless @_;
67	dpavlin	81	my $msg = join('', @_);
68	dpavlin	82	if ( $msg !~ m{[\n\r]$} ) {
69			my @loc = caller(1);
70			$msg .= " in $loc[1] +$loc[2]\n" if @loc;
71			}
72			$msg =~ s{\[(0\|\d\d\d\d)\]}{ '[' . port2color($1) . ']' }eg;
73			print STDERR $msg;
74	dpavlin	81	return 1;
75			};
76
77	dpavlin	12	}
78
79	dpavlin	87	our $digest_module = 'Sack::Digest::' . ( $ENV{SACK_DIGEST} \|\| 'BerkeleyDB' );
80	dpavlin	33	use lib "$prefix/srv/Sack/lib/";
81	dpavlin	87	my $digest_path = $digest_module;
82			$digest_path =~ s{::}{/}g;
83			require "lib/$digest_path.pm";
84
85			our $digest = $digest_module->new( port => $port, clean => 1 );
86	dpavlin	45	sub digest { $digest->to_int($_[0]) }
87	dpavlin	12
88	dpavlin	87	warn "[$port] using $digest_module\n";
89
90	dpavlin	12	use lib "$prefix/srv/webpac2/lib/";
91	dpavlin	1	use WebPAC::Input::ISI;
92	dpavlin	24
93			$WebPAC::Input::ISI::subfields = undef; # disable parsing of subfields
94
95	dpavlin	1	my $input = WebPAC::Input::ISI->new(
96	dpavlin	12	path => "$prefix/$path",
97	dpavlin	4	offset => $offset,
98			limit => $limit,
99	dpavlin	1	);
100
101	dpavlin	25	our $num_records = $input->size;
102	dpavlin	74	our @reports;
103	dpavlin	1
104			sub report {
105	dpavlin	30	my $description = join(' ',@_);
106	dpavlin	1	my $dt = time - $t;
107	dpavlin	74	my $report = [ $description, $dt, $input->size / $dt ];
108	dpavlin	82	warn sprintf "[$port] %s in %1.4fs %.2f/s\n", @$report;
109	dpavlin	74	push @reports, $report;
110	dpavlin	10	$t = time;
111	dpavlin	1	}
112
113	dpavlin	74	sub show_report {
114			"\n" . join( "\n", map { sprintf "%8.4fs %10.2f/s %s", $_->[1], $_->[2], $_->[0] } @reports ) . "\n";
115			}
116	dpavlin	1
117	dpavlin	74	report $input->size , 'records loaded';
118	dpavlin	1
119			mkdir 'out' unless -e 'out';
120
121	dpavlin	5	our $out;
122
123	dpavlin	8	our $cache;
124
125	dpavlin	13	our $connected;
126
127	dpavlin	64	sub node_sock {
128			my $node = shift;
129			my $sock = IO::Socket::INET->new(
130			PeerAddr => '127.0.0.1',
131			PeerPort => $node,
132			Proto => 'tcp',
133			);
134
135			return $sock if $sock && $sock->connected;
136
137			warn "[$port] can't connect to $node - $!\n"; # FIXME die?
138			return;
139			}
140
141	dpavlin	11	sub send_nodes {
142	dpavlin	17	my $content = $#_ > 0 ? pop @_ : ''; # no content with just one argument!
143	dpavlin	21	my $header = defined $content ? length($content) : 0;
144	dpavlin	11	$header .= ' ' . join(' ', @_) if @_;
145
146	dpavlin	74	warn "# send_nodes ", dump(@_), " to ", dump @nodes;
147	dpavlin	60
148	dpavlin	11	foreach my $node ( @nodes ) {
149
150	dpavlin	64	my $sock = node_sock($node) \|\| next;
151	dpavlin	11
152	dpavlin	67	warn "[$port] >>>> [$node] $header\n";
153	dpavlin	11	print $sock "$header\n$content" \|\| warn "can't send $header to $node: $!";
154
155	dpavlin	13	$connected->{$node} = $sock;
156	dpavlin	11	}
157			}
158
159	dpavlin	16	sub get_node {
160			my $node = shift;
161
162			my $sock = $connected->{$node};
163	dpavlin	64	if ( ! $sock \|\| ! $sock->connected ) {
164			warn "[$port] no connection to $node";
165	dpavlin	16	delete $connected->{$node};
166			return;
167			}
168			chomp( my $size = <$sock> );
169	dpavlin	67	warn "[$port] <<<< [$node] $size bytes\n" if $debug \|\| $size > 1024;
170	dpavlin	16	my $data;
171			read $sock, $data, $size;
172			return $data;
173			}
174
175			sub send_sock {
176			my ( $sock, $data ) = @_;
177			my $size = length $data;
178	dpavlin	58	warn "[$port] >>>> $size bytes\n" if $debug \|\| $size > 1024;
179	dpavlin	16	print $sock "$size\n$data" \|\| warn "can't send $size bytes to ", $sock->peerhost;
180			}
181
182	dpavlin	64	sub pull_node_file {
183			my ( $node, $file ) = @_;
184
185			my $path = "/dev/shm/sack.$node.$file";
186	dpavlin	72	return if -e $path; # FIXME
187	dpavlin	64
188			my $sock = node_sock $node \|\| die "not connected to $node";
189
190			print $sock "0 file $file\n";
191
192			my $size = <$sock>;
193			chomp($size);
194	dpavlin	82	warn "[$port] pull_node_file [$node] $file $size bytes\n";
195	dpavlin	64
196			my $block = 4096;
197			my $buff = ' ' x $block;
198
199			open(my $fh, '>', $path) \|\| die "can't open $path";
200			while ( read $sock, $buff, $block ) {
201			print $fh $buff;
202			}
203			close($fh);
204			}
205
206	dpavlin	80	my $merge_digest_mapping;
207
208	dpavlin	13	sub merge_out {
209	dpavlin	50	my ( $from_node, $new ) = @_;
210	dpavlin	13
211	dpavlin	67	my $t_merge = time();
212
213	dpavlin	64	pull_node_file $from_node => 'nr_md5';
214			pull_node_file $from_node => 'md5';
215
216	dpavlin	91	my $remote_digest = $digest_module->new( port => $from_node );
217	dpavlin	59	my ( $local, $remote ) = ( 0, 0 );
218	dpavlin	50
219	dpavlin	59	my $tick = 0;
220	dpavlin	82	warn "[$port] merge [$from_node]";
221	dpavlin	50
222	dpavlin	64	my $missing;
223
224	dpavlin	13	foreach my $k1 ( keys %$new ) {
225
226			foreach my $k2 ( keys %{ $new->{$k1} } ) {
227
228	dpavlin	30	my $n = delete $new->{$k1}->{$k2};
229	dpavlin	50
230			if ( $k1 =~ m{#} ) {
231			die "ASSERT $k1 $k2" unless $k2 =~ m{^\d+$};
232			#warn "XXX $k1 $k2";
233	dpavlin	64
234	dpavlin	80	if ( defined $merge_digest_mapping->{$from_node}->[ $k2 ] ) {
235			$k2 = $merge_digest_mapping->{$from_node}->[ $k2 ];
236			} else {
237	dpavlin	64
238	dpavlin	80	my $md5 = $remote_digest->{nr_md5}->[$k2];
239
240			if ( ! $md5 ) {
241			$missing->{nr_md5}->{$from_node}++; # FIXME die?
242			next;
243			}
244
245			my $local_k2;
246
247			if ( $local_k2 = $digest->{md5_nr}->{$md5} ) {
248			$local++;
249			} elsif ( my $full = $remote_digest->{md5}->{$md5} ) {
250			$local_k2 = $digest->to_int( $remote_digest->{md5}->{$md5} );
251			$remote++;
252			} else {
253			$missing->{md5}->{$from_node}++;
254			}
255
256			$k2 = $merge_digest_mapping->{$from_node}->[ $k2 ] = $local_k2;
257
258	dpavlin	50	}
259			}
260
261	dpavlin	30	my $ref = ref $out->{$k1}->{$k2};
262	dpavlin	50	#warn "XXXX $k1 $k2 $ref";
263	dpavlin	13	if ( ! defined $out->{$k1}->{$k2} ) {
264			$out->{$k1}->{$k2} = $n;
265			} elsif ( $k1 =~ m{\+} ) {
266	dpavlin	19	# warn "## agregate $k1 $k2";
267	dpavlin	13	$out->{$k1}->{$k2} += $n;
268	dpavlin	21	} elsif ( $ref eq 'ARRAY' ) {
269			if ( ref $n eq 'ARRAY' ) {
270			push @{ $out->{$k1}->{$k2} }, $_ foreach @$n;
271			} else {
272			push @{ $out->{$k1}->{$k2} }, $n;
273			}
274	dpavlin	13	} elsif ( $ref eq '' ) {
275			$out->{$k1}->{$k2} = [ $out->{$k1}->{$k2}, $n ];
276			} else {
277			die "can't merge $k2 [$ref] from ",dump($n), " into ", dump($out->{$k1}->{$k2});
278			}
279	dpavlin	59
280			if ( $tick++ % 1000 == 0 ) {
281			print STDERR ".";
282			} elsif ( $tick % 10000 == 0 ) {
283			print STDERR $tick;
284			}
285	dpavlin	13	}
286			}
287
288	dpavlin	67	$t_merge = time - $t_merge;
289	dpavlin	79	my $digests = $local + $remote;
290	dpavlin	82	warn sprintf "\n[$port] merge %d in %.4fs %.2f/s digests local: %.1f%% %d/%d\n", $tick, $t_merge, $digests / $t_merge, $local * 100 / ( $digests \|\| 1 ), $local, $remote;
291			push @reports, [ "$tick merged [$from_node]", $t_merge, $digests / $t_merge ];
292	dpavlin	59
293	dpavlin	64	warn "[$port] missing ", dump $missing if $missing;
294
295	dpavlin	19	warn "## merge out ", dump $out if $debug;
296	dpavlin	13	}
297
298	dpavlin	84	our $rec;
299
300	dpavlin	11	sub run_code {
301			my ( $view, $code ) = @_;
302
303	dpavlin	21	warn "\n#### CODE $view START ####\n$code\n#### CODE $view END ####\n" if $debug;
304	dpavlin	11
305			send_nodes view => $view => $code;
306
307			undef $out;
308
309			my $affected = 0;
310			$t = time;
311
312	dpavlin	84	my $coderef = eval "sub { $code }";
313			if ( $@ ) {
314			warn "ABORT code: $@";
315			return;
316			}
317
318	dpavlin	11	foreach my $pos ( $offset + 1 .. $offset + $input->size ) {
319	dpavlin	84	$rec = $cache->{$pos} \|\|= $input->fetch_rec( $pos );
320	dpavlin	11	if ( ! $rec ) {
321	dpavlin	41	print STDERR "END @ $pos";
322	dpavlin	11	last;
323			}
324
325	dpavlin	84	eval { $coderef->() };
326	dpavlin	11	if ( $@ ) {
327	dpavlin	41	warn "ABORT $pos $@\n";
328	dpavlin	20	last;
329	dpavlin	11	} else {
330			$affected++;
331			}
332	dpavlin	30
333	dpavlin	40	$pos % 10000 == 0 ? print STDERR $pos :
334			$pos % 1000 == 0 ? print STDERR "." : 0 ;
335	dpavlin	11	};
336
337	dpavlin	74	report "$affected affected $view";
338	dpavlin	11
339			warn "WARN no \$out defined!" unless defined $out;
340	dpavlin	13
341	dpavlin	46	$digest->sync;
342
343	dpavlin	13	if ( $connected ) {
344	dpavlin	24	foreach my $node ( keys %$connected ) {
345	dpavlin	74	warn "[$port] get_node [$node]\n";
346	dpavlin	24	my $o = get_node $node;
347	dpavlin	64	next unless $o;
348	dpavlin	24	my $s = length $o;
349			$o = thaw $o;
350	dpavlin	74	warn "[$port] got $s bytes from [$node]\n";
351	dpavlin	50	merge_out $node => $o;
352	dpavlin	24	}
353	dpavlin	13	}
354	dpavlin	11	}
355
356	dpavlin	1	sub run_views {
357	dpavlin	6	@views = sort glob 'views/*.pl' unless @views;
358	dpavlin	1	warn "# views ", dump @views;
359
360			foreach my $view ( @views ) {
361
362			next if system("perl -c $view") != 0;
363
364			my $code = read_file $view;
365
366	dpavlin	11	run_code $view => $code;
367	dpavlin	1
368	dpavlin	11	if ( defined $out ) {
369	dpavlin	1
370	dpavlin	11	my $path = $view;
371			$path =~ s{views?/}{out/} \|\| die "no view in $view";
372	dpavlin	30	$path =~ s{\.pl}{.storable};
373	dpavlin	1
374	dpavlin	9	unlink "$path.last" if -e "$path.last";
375			rename $path, "$path.last";
376	dpavlin	30
377			store $out => $path;
378	dpavlin	74	report "save $path", -s $path, "bytes";
379	dpavlin	35
380			if ( -s $path < 4096 ) {
381	dpavlin	45	print '$out = ', dump $digest->undigest_out($out);
382	dpavlin	35	}
383	dpavlin	11	}
384
385			}
386
387			}
388
389	dpavlin	35
390			sub info_tabs {
391	dpavlin	82	my $port_col = port2color($port);
392			"$port_col\t$offset\t$limit\t$num_records\t$path\t"
393	dpavlin	36	. join("\t", map {
394	dpavlin	35	my $b = $_;
395	dpavlin	46	$b =~ s{^.+\.$port\.([^/]+)$}{$1};
396	dpavlin	35	"$b " . -s $_
397	dpavlin	46	} glob "/dev/shm/sack.$port.*" );
398	dpavlin	35	}
399
400
401	dpavlin	30	if ( $port ) {
402	dpavlin	75
403			my $pid_path = "/tmp/sack.$port.pid";
404			if ( -e $pid_path ) {
405			my $pid = read_file $pid_path;
406	dpavlin	79	kill 9, $pid && warn "[$port] kill old $pid\n";
407	dpavlin	75	}
408			write_file $pid_path, $$;
409
410
411	dpavlin	11	my $sock = IO::Socket::INET->new(
412			Listen => SOMAXCONN,
413	dpavlin	21	LocalAddr => '127.0.0.1',
414	dpavlin	30	LocalPort => $port,
415	dpavlin	11	Proto => 'tcp',
416			Reuse => 1,
417	dpavlin	82	) or die "[$port] die $!";
418	dpavlin	11
419			while (1) {
420
421	dpavlin	74	warn "[$port] accept path: $path offset: $offset limit: $limit #recs: $num_records\n";
422	dpavlin	11
423			my $client = $sock->accept();
424
425	dpavlin	30	warn "[$port] <<<< connect from ", $client->peerhost, $/;
426	dpavlin	11
427			my @header = split(/\s/, <$client>);
428	dpavlin	30	warn "[$port] <<<< header ",dump(@header),$/;
429	dpavlin	11
430			my $size = shift @header;
431
432			my $content;
433			read $client, $content, $size;
434
435			if ( $header[0] eq 'view' ) {
436			run_code $header[1] => $content;
437	dpavlin	16	send_sock $client => freeze $out;
438			} elsif ( $header[0] eq 'info' ) {
439	dpavlin	35	my $info = info_tabs;
440	dpavlin	30	warn "[$port] info $info\n";
441	dpavlin	74	$info .= "\n" . show_report if $content =~ m{r}i;
442	dpavlin	16	send_sock $client => $info;
443			} elsif ( $header[0] eq 'exit' ) {
444	dpavlin	30	warn "[$port] exit";
445	dpavlin	16	exit;
446	dpavlin	64	} elsif ( $header[0] eq 'file' ) {
447			$digest->close;
448			my $path = "/dev/shm/sack.$port.$header[1]";
449			my $size = -s $path;
450			warn "[$port] >>>> file $path $size bytes\n";
451			print $client "$size\n";
452			my $block = 4096;
453			my $buff = ' ' x $block;
454			open(my $fh, '<', $path) \|\| die "can't open $path";
455			while ( read $fh, $buff, $block ) {
456			print $client $buff;
457			}
458	dpavlin	67	$digest->open;
459	dpavlin	5	} else {
460	dpavlin	30	warn "[$port] UNKNOWN $header[0]";
461	dpavlin	1	}
462
463	dpavlin	75	$client->close;
464
465	dpavlin	1	}
466			}
467
468	dpavlin	30	sub info {
469	dpavlin	74	my $detail = shift \|\| '';
470	dpavlin	30
471	dpavlin	74	send_nodes 'info' => $detail;
472
473	dpavlin	30	my @info = (
474	dpavlin	36	"port\toffset\tlimit\t#recs\tpath",
475			"----\t------\t-----\t-----\t----",
476	dpavlin	35	info_tabs,
477	dpavlin	30	);
478
479	dpavlin	74	push @info, show_report if $detail =~ m{r}i;
480
481	dpavlin	30	push @info, get_node $_ foreach @nodes;
482
483	dpavlin	74	print "[$port] INFO", $detail ? " $detail" : '', " \n"
484	dpavlin	30	, join("\n", @info)
485	dpavlin	74	, "\n" ;
486	dpavlin	30
487			return @info;
488			}
489
490			info;
491	dpavlin	76	while ( keys %$connected != @nodes ) {
492	dpavlin	82	warn "[$port] wait for [", join('] [', grep { ! defined $connected->{$_} } @nodes ), "]\n";
493	dpavlin	76	sleep 1;
494	dpavlin	82	info;
495	dpavlin	76	}
496	dpavlin	3	run_views;
497
498	dpavlin	1	while ( 1 ) {
499
500			print "sack> ";
501
502	dpavlin	89	my $cmd = <STDIN>;
503			exit unless defined $cmd; # CTRL+D
504			chomp $cmd;
505
506	dpavlin	21	if ( $cmd =~ m{^(h\|\?)} ) {
507			print << "__HELP__"
508			Sacks Lorry v$VERSION - path: $path offset: $offset limit: $limit
509
510			View Run run views
511			VI \\e Output show output of last run
512	dpavlin	82	Info [Report] info with optional report
513	dpavlin	21	Quit EXit shutdown
514
515			__HELP__
516			} elsif ( $cmd =~ m{^(vi\|\\e\|o)}i ) {
517	dpavlin	33	#system "vi out/*";
518	dpavlin	53	$digest->sync;
519	dpavlin	33	system "bin/storableedit.pl", (glob('out/*.storable'))[0];
520	dpavlin	74	} elsif ( $cmd =~ m{^i(?:nfo)?\s?(\S+)?$}i ) {
521			info $1;
522	dpavlin	21	} elsif ( $cmd =~ m{^(q\|e\|x)}i ) {
523	dpavlin	16	warn "# exit";
524			send_nodes 'exit';
525			exit;
526	dpavlin	21	} elsif ( $cmd =~ m{^(v\|r)}i ) {
527	dpavlin	3	run_views;
528	dpavlin	55	} elsif ( $cmd =~ m{^n(ode)?\s*(\d+)}i ) {
529	dpavlin	60	push @nodes, $2;
530	dpavlin	55	info;
531	dpavlin	21	} elsif ( $cmd ) {
532			warn "UNKNOWN ", dump $cmd;
533	dpavlin	3	}
534
535	dpavlin	1	}
536