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 base qw(Sack::Pid);
use Sack::Color;
use Sack;

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 $Sack::VERSION\n";
                        $client = $sock->accept();
                        warn "[$port] connect from ", $client->peerhost, $/;
                }

                my $data = Storable::fd_retrieve( $client );

                if ( defined $data->{data} ) {
                        warn "# [$port] <<<< data\n" if $self->{debug};
                } else {
                        warn "# [$port] <<<< ", 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\n";
                        close $sock;
                        exit;
                } elsif ( $data->{restart} ) {
                        warn "[$port] restart";
                        close $sock;
                        exec "$0 $port";
                } elsif ( $data->{info} ) {
                        $result = {
                                version => $Sack::VERSION,
                                size => $#{ $self->{data} } + 1,
                                reports => $self->{reports},
                        };
                } elsif ( my $sh = delete $data->{sh} ) {
                        $result = { sh => scalar `$sh` };
                } elsif ( defined $data->{debug} ) {
                        $result = { debug => $self->{debug} = $data->{debug} };
                } else {
                        warn "[$port] UNKNOWN ", dump( $data ), $/;
                        $result = { 'error' => 'unknown', data => $data };
                }

                $result = { 'error' => 'result not reference', result => $result, data => $data } unless ref($result);

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

}

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

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

        my $out;

        my $coderef = eval "sub { my \$rec = \$_[0]; $code }";
        if ( $@ ) {
                warn "ABORT code: $@";
                return;
        }


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

                eval { $coderef->( $self->{data}->[$pos] ) };

                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 "[$self->{port}] out ", dump( $out ),$/ if $self->{debug};

        return {
                out => $out,
        };
}

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