trunk/bin/sack.pl

#!/usr/bin/perl

use warnings;
use strict;

our $VERSION = '0.01';

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/;
use Digest::MD5 qw/md5/;


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


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

my $t = time;


sub send_nodes;

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

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


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;

sub report {
        my $description = shift;
        my $dt = time - $t;
        printf "%s in %1.4fs %.2f/s\n", $description, $dt, $input->size / $dt;
        $t = time;
}


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

mkdir 'out' unless -e 'out';

our $out;

our $cache;

our $connected;

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 ) {

                my $sock = IO::Socket::INET->new(
                        PeerAddr => $node,
                        Proto    => 'tcp',
                );

                if ( ! $sock ) {
                        warn "can't connect to $node - $!"; # FIXME die?
                        next;
                }

                print ">>>> $listen $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 ) {
                warn "ERROR: lost connection to $node";
                delete $connected->{$node};
                return;
        }
        chomp( my $size = <$sock> );
        warn "<<<< $listen $node $size bytes\n";
        my $data;
        read $sock, $data, $size;
        return $data;
}

sub send_sock {
        my ( $sock, $data ) = @_;
        my $size   = length $data;
        warn ">>>> $listen ", $sock->peerhost, " $size bytes\n";
        print $sock "$size\n$data" || warn "can't send $size bytes to ", $sock->peerhost;
}

sub merge_out {
        my $new = shift;

        foreach my $k1 ( keys %$new ) {

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

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

                        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});
                        }
                }
        }

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

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;

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

                eval "$code";
                if ( $@ ) {
                        warn "ABORT [$pos] $@\n";
                        last;
                } else {
                        $affected++;
                }
        };

        report "$affected affected records $view";

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

        if ( $connected ) {
                foreach my $node ( keys %$connected ) {
                        warn "# $listen get_node $node\n";
                        my $o = get_node $node;
                        my $s = length $o;
                        $o = thaw $o;
                        warn "# $listen merge $s bytes\n";
                        merge_out $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 $dump = dump $out;
                        my $len  = length $dump;

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

                        print "OUT $view $offset/$limit $len bytes $path"
                                , ( $len < 10000 ?  " \$out = $dump" : ' SAVED ONLY' )
                                , "\n"
                                ;

                        unlink "$path.last" if -e "$path.last";
                        rename $path, "$path.last";
                        write_file $path, $dump;
                        report "SAVE $path";
                }

        }

}

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

        while (1) {

                warn "NODE $listen ready - path: $path offset: $offset limit: $limit #recs: $num_records\n";

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

                warn "<<<< $listen connect from ", $client->peerhost, $/;

                my @header = split(/\s/, <$client>);
                warn "<<<< $listen 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 = "$listen\t$offset\t$limit\t$num_records\t$path";
                        $info .= "\t" . eval $header[1] if $header[1];
                        warn "info $info\n";
                        send_sock $client => $info;
                } elsif ( $header[0] eq 'exit' ) {
                        warn "exit $listen";
                        exit;
                } else {
                        warn "WARN $listen unknown";
                }

        }
}

run_views;

while ( 1 ) {

        print "sack> ";
        chomp( my $cmd = <STDIN> );

        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 [\$VERSION]        instrospect
        Quit EXit       shutdown

__HELP__
        } elsif ( $cmd =~ m{^(vi|\\e|o)}i ) {
                system "vi out/*";
        } elsif ( $cmd =~ m{^i(?:nfo)?\s?(.+)?$}i ) {
                print "# nodes: ", join(' ',@nodes), $/;

                send_nodes 'info' => $2;

                my @info = (
                        "node\toffset\tlimit\t#recs\tpath",
                        "----\t------\t-----\t-----\t----",
                        "0\t$offset\t$limit\t$num_records\t$path",
                );

                push @info, get_node $_ foreach @nodes;

                print "$_\n" foreach @info;

        } elsif ( $cmd =~ m{^(q|e|x)}i ) {
                warn "# exit";
                send_nodes 'exit';
                exit;
        } elsif ( $cmd =~ m{^(v|r)}i ) {
                run_views;
        } elsif ( $cmd ) {
                warn "UNKNOWN ", dump $cmd;
        }

}

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