lib/Sack/Lorry.pm

package Sack::Lorry;

use warnings;
use strict;

use IO::Socket::INET;
use Data::Dump qw(dump);
use Storable;

our $pids;
our $ports;

$SIG{CHLD} = 'IGNORE';

my $port = 4000;

sub new {
        my $class = shift;
        my $self  = bless {@_}, $class;
        return $self;
}

sub start_node {
        my ( $self, $host ) = @_;

        system "rsync -rav /srv/Sack/ $host:/srv/Sack/";

        if ( my $pid = fork ) {
                # parent
                $pids->{ $host } = $pid;
                $ports->{ $port } = $host;

                my $sock;

                print STDERR "waiting for $port";

                while ( ! $sock ) {

                        $sock = IO::Socket::INET->new(
                                PeerAddr => '127.0.0.1',
                                PeerPort => $port,
                                Proto    => 'tcp',
                        );

                        if ( ! $sock ) {
                                print STDERR ".";
                                sleep 1;
                        }

                }

                $self->{sock}->{$port} = $sock;

                warn "\nconnected to $port\n";

                $self->{connected}->{$port} = $host;

                return $port++;

        } elsif ( ! defined $pid ) {
                warn "can't fork $host $port";
                return;
        } else {
                # child
                my $cmd = $host !~ m{^(localhost|127\.)}i ? qq|
                        ssh
                                -S /tmp/sock.$port.ssh
                                -L $port:127.0.0.1:$port
                        $host
                | : '';

                $cmd .= qq|
                        /srv/Sack/bin/node.pl $port
                |;

                $cmd =~ s{\s+}{ }sg;

                warn "exec: $cmd\n";
                exec $cmd;
        }
}

sub send_to {
        my ( $self, $port, $data ) = @_;
        warn "send_to [$port]\n";
        Storable::store_fd( $data => $self->{sock}->{$port} );
}

sub get_from {
        my ( $self, $port )  = @_;
        warn "get_from [$port]\n";
        Storable::fd_retrieve( $self->{sock}->{$port} );
}

sub send_to_all {
        my ( $self, $data ) = @_;
        $self->send_to( $_, $data ) foreach keys %{ $self->{connected} };
}

sub get_from_all {
        my ( $self ) = @_;
        my $result;
        $result->{$_} = $self->get_from( $_ ) foreach keys %{ $self->{connected} };
        return $result;
}

sub DESTROY {
        warn "pids ",dump( $pids );
        foreach ( values %$pids ) {
                warn "kill $_";
                kill 1,$_ || kill 9, $_;
        }
}

1;
1	package Sack::Lorry;
2
3	use warnings;
4	use strict;
5
6	use IO::Socket::INET;
7	use Data::Dump qw(dump);
8	use Storable;
9
10	our $pids;
11	our $ports;
12
13	$SIG{CHLD} = 'IGNORE';
14
15	my $port = 4000;
16
17	sub new {
18	my $class = shift;
19	my $self = bless {@_}, $class;
20	return $self;
21	}
22
23	sub start_node {
24	my ( $self, $host ) = @_;
25
26	system "rsync -rav /srv/Sack/ $host:/srv/Sack/";
27
28	if ( my $pid = fork ) {
29	# parent
30	$pids->{ $host } = $pid;
31	$ports->{ $port } = $host;
32
33	my $sock;
34
35	print STDERR "waiting for $port";
36
37	while ( ! $sock ) {
38
39	$sock = IO::Socket::INET->new(
40	PeerAddr => '127.0.0.1',
41	PeerPort => $port,
42	Proto => 'tcp',
43	);
44
45	if ( ! $sock ) {
46	print STDERR ".";
47	sleep 1;
48	}
49
50	}
51
52	$self->{sock}->{$port} = $sock;
53
54	warn "\nconnected to $port\n";
55
56	$self->{connected}->{$port} = $host;
57
58	return $port++;
59
60	} elsif ( ! defined $pid ) {
61	warn "can't fork $host $port";
62	return;
63	} else {
64	# child
65	my $cmd = $host !~ m{^(localhost\|127\.)}i ? qq\|
66	ssh
67	-S /tmp/sock.$port.ssh
68	-L $port:127.0.0.1:$port
69	$host
70	\| : '';
71
72	$cmd .= qq\|
73	/srv/Sack/bin/node.pl $port
74	\|;
75
76	$cmd =~ s{\s+}{ }sg;
77
78	warn "exec: $cmd\n";
79	exec $cmd;
80	}
81	}
82
83	sub send_to {
84	my ( $self, $port, $data ) = @_;
85	warn "send_to [$port]\n";
86	Storable::store_fd( $data => $self->{sock}->{$port} );
87	}
88
89	sub get_from {
90	my ( $self, $port ) = @_;
91	warn "get_from [$port]\n";
92	Storable::fd_retrieve( $self->{sock}->{$port} );
93	}
94
95	sub send_to_all {
96	my ( $self, $data ) = @_;
97	$self->send_to( $_, $data ) foreach keys %{ $self->{connected} };
98	}
99
100	sub get_from_all {
101	my ( $self ) = @_;
102	my $result;
103	$result->{$_} = $self->get_from( $_ ) foreach keys %{ $self->{connected} };
104	return $result;
105	}
106
107	sub DESTROY {
108	warn "pids ",dump( $pids );
109	foreach ( values %$pids ) {
110	warn "kill $_";
111	kill 1,$_ \|\| kill 9, $_;
112	}
113	}
114
115	1;