lib/Sack/Node.pm

package Sack::Node;

use warnings;
use strict;

use IO::Socket::INET;
use File::Slurp;
use Carp qw(confess);
use Data::Dump qw(dump);
use Storable;
use Time::HiRes qw(time);

use lib 'lib';
use base qw(Sack::Pid);
use Sack::Color;

our $VERSION = '0.09';

sub new {
        my $class = shift;
        my $port = shift;
        my $self  = bless { port => $port }, $class;

        $self->port_pid( $port );

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

        my $client;

        while ( 1 ) {

                if ( ! $client ) {
                        warn "[$port] accept $VERSION\n";
                        $client = $sock->accept();
                        warn "[$port] connect from ", $client->peerhost, $/;
                }

                my $data = eval { Storable::fd_retrieve( $client ) };
                if ( $@ ) {
                        warn "[$port] ERROR $@\n";
                        close $client;
                        next;
                }

                if ( defined $data->{data} ) {
                        warn "# [$port] <<<< data\n";
                } else {
                        warn "# [$port] <<<< ", dump( $data ), $/;
                }

                my $result;

                if ( $data->{view} ) {
                        $result = $self->view( $data->{view} );
                } elsif ( $data->{data} ) {
                        $self->{data} = delete $data->{data};
                        $result = { data => 'loaded' };
                } elsif ( $data->{exit} ) {
                        warn "[$port] exit\n";
                        close $sock;
                        exit;
                } elsif ( $data->{restart} ) {
                        warn "[$port] restart";
                        close $sock;
                        exec "$0 $port";
                } elsif ( $data->{info} ) {
                        $result = {
                                version => $VERSION,
                                size => $#{ $self->{data} } + 1,
                                reports => $self->{reports},
                        };
                } elsif ( my $sh = delete $data->{sh} ) {
                        $result = { sh => scalar `$sh` };
                } else {
                        warn "[$port] UNKNOWN ", dump( $data ), $/;
                        $result = { 'error' => $data };
                }

                warn "# [$port] >>>>\n";
                Storable::store_fd( $result => $client );
        }

}


our $rec;
our $out;

sub view {
        my ( $self, $code ) = @_;

        undef $out;

        my $affected = 0;
        my $start_t = time;

        my $coderef = eval "sub { $code }";
        if ( $@ ) {
                warn "ABORT code: $@";
                return;
        }


        foreach my $pos ( 0 .. $#{ $self->{data} } ) {
                $rec = $self->{data}->[$pos];
                if ( ! $rec ) {
                        print STDERR "END @ $pos";
                        last;
                }

                eval { $coderef->() };
                if ( $@ ) {
                        warn "ABORT $pos $@\n";
                        last;
                } else {
                        $affected++;
                }

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

        my $dt = time - $start_t;
        my $report = [ $self->{port}, $affected, $dt, $affected / $dt ];
        warn sprintf "[%d] %d affected in %1.4fs %.2f/s\n", @$report;

        push @{ $self->{reports} }, "$affected in ${dt}s";

#       warn "# out ", dump( $out );

        return {
                out => $out,
                report => $report,
        };
}

1;
1	package Sack::Node;
2
3	use warnings;
4	use strict;
5
6	use IO::Socket::INET;
7	use File::Slurp;
8	use Carp qw(confess);
9	use Data::Dump qw(dump);
10	use Storable;
11	use Time::HiRes qw(time);
12
13	use lib 'lib';
14	use base qw(Sack::Pid);
15	use Sack::Color;
16
17	our $VERSION = '0.09';
18
19	sub new {
20	my $class = shift;
21	my $port = shift;
22	my $self = bless { port => $port }, $class;
23
24	$self->port_pid( $port );
25
26	my $sock = IO::Socket::INET->new(
27	Listen => SOMAXCONN,
28	LocalAddr => '127.0.0.1',
29	LocalPort => $port,
30	Proto => 'tcp',
31	Reuse => 1,
32	) or die "[$port] die $!";
33
34	my $client;
35
36	while ( 1 ) {
37
38	if ( ! $client ) {
39	warn "[$port] accept $VERSION\n";
40	$client = $sock->accept();
41	warn "[$port] connect from ", $client->peerhost, $/;
42	}
43
44	my $data = eval { Storable::fd_retrieve( $client ) };
45	if ( $@ ) {
46	warn "[$port] ERROR $@\n";
47	close $client;
48	next;
49	}
50
51	if ( defined $data->{data} ) {
52	warn "# [$port] <<<< data\n";
53	} else {
54	warn "# [$port] <<<< ", dump( $data ), $/;
55	}
56
57	my $result;
58
59	if ( $data->{view} ) {
60	$result = $self->view( $data->{view} );
61	} elsif ( $data->{data} ) {
62	$self->{data} = delete $data->{data};
63	$result = { data => 'loaded' };
64	} elsif ( $data->{exit} ) {
65	warn "[$port] exit\n";
66	close $sock;
67	exit;
68	} elsif ( $data->{restart} ) {
69	warn "[$port] restart";
70	close $sock;
71	exec "$0 $port";
72	} elsif ( $data->{info} ) {
73	$result = {
74	version => $VERSION,
75	size => $#{ $self->{data} } + 1,
76	reports => $self->{reports},
77	};
78	} elsif ( my $sh = delete $data->{sh} ) {
79	$result = { sh => scalar `$sh` };
80	} else {
81	warn "[$port] UNKNOWN ", dump( $data ), $/;
82	$result = { 'error' => $data };
83	}
84
85	warn "# [$port] >>>>\n";
86	Storable::store_fd( $result => $client );
87	}
88
89	}
90
91
92	our $rec;
93	our $out;
94
95	sub view {
96	my ( $self, $code ) = @_;
97
98	undef $out;
99
100	my $affected = 0;
101	my $start_t = time;
102
103	my $coderef = eval "sub { $code }";
104	if ( $@ ) {
105	warn "ABORT code: $@";
106	return;
107	}
108
109
110	foreach my $pos ( 0 .. $#{ $self->{data} } ) {
111	$rec = $self->{data}->[$pos];
112	if ( ! $rec ) {
113	print STDERR "END @ $pos";
114	last;
115	}
116
117	eval { $coderef->() };
118	if ( $@ ) {
119	warn "ABORT $pos $@\n";
120	last;
121	} else {
122	$affected++;
123	}
124
125	$pos % 10000 == 0 ? print STDERR $pos :
126	$pos % 1000 == 0 ? print STDERR "." : 0 ;
127	};
128
129	my $dt = time - $start_t;
130	my $report = [ $self->{port}, $affected, $dt, $affected / $dt ];
131	warn sprintf "[%d] %d affected in %1.4fs %.2f/s\n", @$report;
132
133	push @{ $self->{reports} }, "$affected in ${dt}s";
134
135	# warn "# out ", dump( $out );
136
137	return {
138	out => $out,
139	report => $report,
140	};
141	}
142
143	1;