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 ) = @_;

        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";

                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|
                        perl -I/srv/Sack/lib -MSack::Node -e "Sack::Node->new($port)"
                |;

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

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

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

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

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	if ( my $pid = fork ) {
27	# parent
28	$pids->{ $host } = $pid;
29	$ports->{ $port } = $host;
30
31	my $sock;
32
33	print STDERR "waiting for $port";
34
35	while ( ! $sock ) {
36
37	$sock = IO::Socket::INET->new(
38	PeerAddr => '127.0.0.1',
39	PeerPort => $port,
40	Proto => 'tcp',
41	);
42
43	if ( ! $sock ) {
44	print STDERR ".";
45	sleep 1;
46	}
47
48	}
49
50	$self->{sock}->{$port} = $sock;
51
52	warn "\nconnected to $port\n";
53
54	return $port++;
55
56	} elsif ( ! defined $pid ) {
57	warn "can't fork $host $port";
58	return;
59	} else {
60	# child
61	my $cmd = $host !~ m{^(localhost\|127\.)}i ? qq\|
62	ssh
63	-S /tmp/sock.$port.ssh
64	-L $port:127.0.0.1:$port
65	$host
66	\| : '';
67
68	$cmd .= qq\|
69	perl -I/srv/Sack/lib -MSack::Node -e "Sack::Node->new($port)"
70	\|;
71
72	$cmd =~ s{\s+}{ }sg;
73
74	warn "exec: $cmd\n";
75	exec $cmd;
76	}
77	}
78
79	sub send_to {
80	my ( $self, $port, $data ) = @_;
81	warn ">>>> [$port] ", dump( keys %$data ), $/;
82	Storable::store_fd( $data => $self->{sock}->{$port} );
83	}
84
85	sub get_from {
86	my ( $self, $port ) = @_;
87	warn "<<<< [$port]\n";
88	Storable::fd_retrieve( $self->{sock}->{$port} );
89	}
90
91	sub DESTROY {
92	warn "pids ",dump( $pids );
93	foreach ( values %$pids ) {
94	warn "kill $_";
95	kill 1,$_ \|\| kill 9, $_;
96	}
97	}
98
99	1;