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	dpavlin	92	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	dpavlin	96	system "rsync -rav /srv/Sack/ $host:/srv/Sack/";
27
28	dpavlin	92	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	dpavlin	93	$self->{connected}->{$port} = $host;
57
58	dpavlin	92	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	dpavlin	96	/srv/Sack/bin/node.pl $port
74	dpavlin	92	\|;
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	dpavlin	93	warn "send_to [$port]\n";
86	dpavlin	92	Storable::store_fd( $data => $self->{sock}->{$port} );
87			}
88
89			sub get_from {
90			my ( $self, $port ) = @_;
91	dpavlin	93	warn "get_from [$port]\n";
92	dpavlin	92	Storable::fd_retrieve( $self->{sock}->{$port} );
93			}
94
95	dpavlin	93	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	dpavlin	92	sub DESTROY {
108			warn "pids ",dump( $pids );
109			foreach ( values %$pids ) {
110			warn "kill $_";
111			kill 1,$_ \|\| kill 9, $_;
112			}
113			}
114
115			1;