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 Sack::Color;

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\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 = { view => $self->view( $data->{view} ) };
                } elsif ( $data->{data} ) {
                        $self->{data} = delete $data->{data};
                        $result = { data => 'loaded' };
                } elsif ( $data->{exit} ) {
                        warn "[$port] exit";
                        exit;
                } 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, scalar $self->{data} / $dt ];
        warn sprintf "[%d] %d affected in %1.4fs %.2f/s\n", @$report;

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

        return $out;
}

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