lib/Sack/View.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 Sack::Color;

our $VERSION = '0.08';

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

        my $pid_path = "/tmp/sack.$port.pid";
        if ( -e $pid_path ) {
                my $pid = read_file $pid_path;
                kill 9, $pid; # warn "[$port] kill old $pid\n";
        }
        write_file $pid_path, $$;

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

        warn "[$port] accept $VERSION\n";

        my $client = $sock->accept();

        warn "[$port] connect from ", $client->peerhost, $/;

        while ( 1 ) {

                my $data = Storable::fd_retrieve( $client );
#               warn "[$port] <<<<\n";
                warn "[$port] data = ", dump( $data ) if $self->{debug};

                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";
                        close $sock;
                        exit;
                } elsif ( $data->{restart} ) {
                        warn "[$port] restart";
                        close $sock;
                        exec "$0 $port";
                } elsif ( $data->{info} ) {
                        $result = {
                                version => $VERSION,
                                size => $#{ $self->{data} } + 1,
                        };
                } 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;

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

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

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