trunk/bin/sack.pl

#!/usr/bin/perl

use warnings;
use strict;

our $VERSION = '0.02';

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,
        'listen|port=i' => \$port,
        'connect=s'   => \@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/[\./]+bin.+$}{$1};
        warn "# prefix $prefix";

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


# digest experiment
use BerkeleyDB;

our $seq = 0;
our $btree;
my $db_file = "/dev/shm/sack.digest.$port.$offset-$limit";
sub digest {
        my $nr;
        if ( ! $btree ) {
                if ( -e $db_file ) {
                        warn "[$port] CLEAN removed $db_file ", -s $db_file, " bytes\n";
                        unlink $db_file;
                }

                our $btree = BerkeleyDB::Btree->new(
                        -Filename  => $db_file,
                        -Cachesize => 700_000_000,
                        -Flags     => DB_CREATE,
                ) || die "$db_file: $!";

                warn "[$port] BDB created $db_file\n";
        }
        $btree->db_get( $_[0] => $nr    ) == 0 && return $nr;
        $btree->db_put( $_[0] => ++$seq ) == 0 || die "$_[0] [$seq] $!";
        $btree->db_put(  $seq => $_[0]  ) == 0 || die "[$seq] $_[0] $!";
        return $seq;
}


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 = join(' ',@_);
        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;
                }

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

sub send_sock {
        my ( $sock, $data ) = @_;
        my $size   = length $data;
        warn "[$port] >>>> ", $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   = delete $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++;
                }

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

        report "\n[$port] RECS $affected $view";

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

        if ( $connected ) {
                foreach my $node ( keys %$connected ) {
                        warn "[$port] get_node $node\n";
                        my $o = get_node $node;
                        my $s = length $o;
                        $o = thaw $o;
                        warn "[$port] merge $node $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 $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 "[$port] SAVE $offset/$limit $view $path", -s $path, " bytes";
                }

        }

}

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

        while (1) {

                warn "[$port] READY 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 = "$port\t$offset\t$limit\t$num_records\t$path";
                        $info .= "\t" . eval $header[1] if $header[1];
                        warn "[$port] info $info\n";
                        send_sock $client => $info;
                } elsif ( $header[0] eq 'exit' ) {
                        warn "[$port] exit";
                        exit;
                } else {
                        warn "[$port] UNKNOWN $header[0]";
                }

        }
}

sub info {
        send_nodes 'info' => $2;

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

        push @info, get_node $_ foreach @nodes;

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

        return @info;
}

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