trunk/bin/lorry.pl

#!/usr/bin/perl

use warnings;
use strict;

use Data::Dump qw(dump);
use File::Slurp;

use lib 'lib';
use Sack::From;
use Sack::Lorry;
use Sack::Color;
use Sack;

use Getopt::Long;
use Time::HiRes qw(time);


my $view   = 'views/00.demo.pl';
my $offset = 0;
my $limit  = 5000;
my $cloud_file;


GetOptions(
        'view=s'  => \$view,
        'limit=i' => \$limit,
        'cloud=s' => \$cloud_file,
) or die $!;


my $lorry = Sack::Lorry->new;

my @cloud = -e $cloud_file ? read_file $cloud_file : ( 'localhost' );
@cloud = map { chomp $_; $_; } @cloud;
my $cloud_size = scalar @cloud;

warn "cloud with $cloud_size nodes: ",join(' ', @cloud),$/;
my $from  = Sack::From->new(
        path => '/data/isi/full.txt',
        limit => $limit * $cloud_size,
);

duration 'load finished';

my $info;

my $port = 4000;

foreach my $host ( @cloud ) {

        if ( $lorry->start_node_port( $host, $port ) ) {

                warn "started [$port] on $host\n";

                my $data = $from->shard( $limit );
                $lorry->send_to( $port, { data => $data } ) || die "can't send to $port: $!";

                $info->{$port} = {
                        host => $host,
                        offset => $offset,
                        limit => $limit,
                };

                $port++;
                $offset += $limit;

        }
}

duration 'cloud up';

warn "info ",dump $info;

warn "load status ", dump( $lorry->get_from_all ),$/;

our $out;
our $repl = 1;

while ( $repl ) {
        print "sack> ";
        my $cmd = <STDIN>;
        last unless defined $cmd; # CTRL+D
        duration 'repl wait';
        chomp($cmd);
        if ( $cmd =~ m{^v} ) {
                $out = $lorry->view( $view );
                duration 'view';
        } elsif ( $cmd =~ m{^d} ) {
                warn dump $out;
                duration 'dump';
        } elsif ( $cmd =~ m{^x} ) {
                $repl = 0;
        } elsif ( $cmd =~ m{^r} ) {
                $lorry->restart_nodes;
        } elsif ( $cmd =~ m{^i} ) {
                $lorry->send_to_all({ info => 1 });
                my $info = $lorry->get_from_all;
                foreach my $port ( $lorry->connected ) {
                        warn "[$port] $lorry->{port_on_host}->{$port} ", dump( $info->{$port} ), "\n";
                }
        } elsif ( $cmd =~ m{^u} ) {
                my $updated;
                foreach my $host ( @cloud ) {
                        next if $updated->{$host}++;
                        warn "update $host\n";
                        system "find /srv/Sack/ | cpio --create | ssh -F etc/lib.ssh $host cpio --extract --make-directories --unconditional";
                }
        } elsif ( $cmd =~ m{^sh\s+(.+)} ) {
                $lorry->send_to_all({ sh => $1 });
                my $sh = $lorry->get_from_all;
                foreach my $port ( $lorry->connected ) {
                        warn "[$port]# $1\n$sh->{$port}->{sh}";
                }
        } else {
                warn "UNKNOWN $cmd\n" if $cmd;
        }
}

warn "exit all nodes\n";
$lorry->send_to_all( { exit => 1 } );

duration 'total usage';

1	dpavlin	92	#!/usr/bin/perl
2
3			use warnings;
4			use strict;
5
6			use Data::Dump qw(dump);
7	dpavlin	93	use File::Slurp;
8	dpavlin	92
9			use lib 'lib';
10			use Sack::From;
11			use Sack::Lorry;
12	dpavlin	102	use Sack::Color;
13	dpavlin	135	use Sack;
14
15	dpavlin	107	use Getopt::Long;
16			use Time::HiRes qw(time);
17	dpavlin	92
18	dpavlin	107
19	dpavlin	93	my $view = 'views/00.demo.pl';
20	dpavlin	92	my $offset = 0;
21	dpavlin	100	my $limit = 5000;
22	dpavlin	108	my $cloud_file;
23	dpavlin	92
24
25	dpavlin	107	GetOptions(
26			'view=s' => \$view,
27			'limit=i' => \$limit,
28			'cloud=s' => \$cloud_file,
29			) or die $!;
30
31
32	dpavlin	92	my $lorry = Sack::Lorry->new;
33
34	dpavlin	108	my @cloud = -e $cloud_file ? read_file $cloud_file : ( 'localhost' );
35	dpavlin	127	@cloud = map { chomp $_; $_; } @cloud;
36	dpavlin	135	my $cloud_size = scalar @cloud;
37	dpavlin	107
38	dpavlin	135	warn "cloud with $cloud_size nodes: ",join(' ', @cloud),$/;
39	dpavlin	95	my $from = Sack::From->new(
40			path => '/data/isi/full.txt',
41	dpavlin	135	limit => $limit * $cloud_size,
42	dpavlin	95	);
43
44	dpavlin	107	duration 'load finished';
45
46	dpavlin	92	my $info;
47
48	dpavlin	127	my $port = 4000;
49
50	dpavlin	92	foreach my $host ( @cloud ) {
51	dpavlin	101
52	dpavlin	127	if ( $lorry->start_node_port( $host, $port ) ) {
53	dpavlin	92
54			warn "started [$port] on $host\n";
55
56	dpavlin	95	my $data = $from->shard( $limit );
57			$lorry->send_to( $port, { data => $data } ) \|\| die "can't send to $port: $!";
58	dpavlin	92
59			$info->{$port} = {
60			host => $host,
61			offset => $offset,
62			limit => $limit,
63			};
64
65	dpavlin	127	$port++;
66	dpavlin	92	$offset += $limit;
67
68			}
69			}
70
71	dpavlin	107	duration 'cloud up';
72
73	dpavlin	92	warn "info ",dump $info;
74
75	dpavlin	119	warn "load status ", dump( $lorry->get_from_all ),$/;
76	dpavlin	92
77	dpavlin	110	our $out;
78	dpavlin	113	our $repl = 1;
79	dpavlin	110
80	dpavlin	113	while ( $repl ) {
81	dpavlin	100	print "sack> ";
82			my $cmd = <STDIN>;
83			last unless defined $cmd; # CTRL+D
84	dpavlin	110	duration 'repl wait';
85	dpavlin	100	chomp($cmd);
86	dpavlin	110	if ( $cmd =~ m{^v} ) {
87			$out = $lorry->view( $view );
88			duration 'view';
89			} elsif ( $cmd =~ m{^d} ) {
90			warn dump $out;
91			duration 'dump';
92	dpavlin	113	} elsif ( $cmd =~ m{^x} ) {
93			$repl = 0;
94	dpavlin	119	} elsif ( $cmd =~ m{^r} ) {
95			$lorry->restart_nodes;
96			} elsif ( $cmd =~ m{^i} ) {
97			$lorry->send_to_all({ info => 1 });
98			my $info = $lorry->get_from_all;
99	dpavlin	127	foreach my $port ( $lorry->connected ) {
100	dpavlin	119	warn "[$port] $lorry->{port_on_host}->{$port} ", dump( $info->{$port} ), "\n";
101			}
102	dpavlin	127	} elsif ( $cmd =~ m{^u} ) {
103			my $updated;
104			foreach my $host ( @cloud ) {
105			next if $updated->{$host}++;
106			warn "update $host\n";
107			system "find /srv/Sack/ \| cpio --create \| ssh -F etc/lib.ssh $host cpio --extract --make-directories --unconditional";
108			}
109	dpavlin	119	} elsif ( $cmd =~ m{^sh\s+(.+)} ) {
110			$lorry->send_to_all({ sh => $1 });
111			my $sh = $lorry->get_from_all;
112	dpavlin	127	foreach my $port ( $lorry->connected ) {
113			warn "[$port]# $1\n$sh->{$port}->{sh}";
114	dpavlin	119	}
115	dpavlin	110	} else {
116	dpavlin	119	warn "UNKNOWN $cmd\n" if $cmd;
117	dpavlin	110	}
118	dpavlin	100	}
119	dpavlin	93
120	dpavlin	92	warn "exit all nodes\n";
121	dpavlin	93	$lorry->send_to_all( { exit => 1 } );
122
123	dpavlin	107	duration 'total usage';
124