/[Sack]/trunk/lib/Sack/Server.pm
This is repository of my old source code which isn't updated any more. Go to git.rot13.org for current projects!
ViewVC logotype

Diff of /trunk/lib/Sack/Server.pm

Parent Directory Parent Directory | Revision Log Revision Log | View Patch Patch

trunk/experiments/protocol-v3/server.pl revision 164 by dpavlin, Sun Nov 1 21:00:30 2009 UTC trunk/lib/Sack/Server.pm revision 203 by dpavlin, Sun Nov 8 22:56:46 2009 UTC
# Line 1  Line 1 
1  #!/usr/bin/perl  #!/usr/bin/perl
2    
3    package Sack::Server;
4    
5  use warnings;  use warnings;
6  use strict;  use strict;
7    
# Line 8  use IO::Select; Line 10  use IO::Select;
10    
11  use Data::Dump qw(dump);  use Data::Dump qw(dump);
12  use Storable qw();  use Storable qw();
13    use File::Slurp;
14    use Cwd qw(abs_path);
15    
16    use lib '/srv/Sack/lib';
17    use Sack::Merge;
18    
19    my @cloud;
20    my $cloud_path = $ENV{CLOUD} || die "start with: CLOUD=etc/cloud perl -I/srv/Sack/lib $0\n";
21    @cloud = read_file $cloud_path;
22    @cloud = map { chomp $_; $_ } @cloud;
23    
24    warn "# cloud ",dump( @cloud );
25    
26  my $listen_port = 4444;  my $listen_port = 4444;
27    
28  my $node_path = '/tmp/client.pl';  my $node_path = abs_path $0;
29    $node_path =~ s{Server}{Client};
30    
31  my $lsn = IO::Socket::INET->new(Listen => 1, LocalPort => $listen_port, Reuse => 1) or exit;  my $lsn = IO::Socket::INET->new(Listen => 1, LocalPort => $listen_port, Reuse => 1) or die $!;
32  my $sel = IO::Select->new($lsn);  my $sel = IO::Select->new($lsn);
33    
34  sub fork_node {  my $info;
35    sub info {
36            my $port = shift;
37            push @{ $info->{$port} }, [ @_ ];
38    }
39    
40    sub fork_ssh {
41          my ( $port, $host ) = @_;          my ( $port, $host ) = @_;
42    
43          if ( my $pid = fork ) {          if ( my $pid = fork ) {
44                  # parent                  # parent
45                    info $port => 'forked', $pid;
46                  return $port;                  return $port;
47    
48          } elsif ( ! defined $pid ) {          } elsif ( ! defined $pid ) {
# Line 29  sub fork_node { Line 50  sub fork_node {
50                  return;                  return;
51          } else {          } else {
52                  # child                  # child
53                  my $cmd = qq|ssh -R $port:127.0.0.1:$listen_port $host $node_path $port|;                  my $cmd = qq|ssh -F $cloud_path.ssh -R $port:127.0.0.1:$listen_port $host $node_path $port|;
54                  warn "# exec: $cmd\n";                  warn "# exec: $cmd\n";
55                  exec $cmd;                  exec $cmd;
56          }          }
# Line 37  sub fork_node { Line 58  sub fork_node {
58    
59  my $node_port = 4000;  my $node_port = 4000;
60    
61  foreach my $host ( qw/localhost tab.lan llin.lan/ ) {  foreach my $host ( @cloud ) {
62          system "scp client.pl $host:$node_path";          system "find /srv/Sack/ | cpio --create --dereference | ssh -T -F $cloud_path.ssh $host cpio --extract --make-directories --unconditional";
63          fork_node( $node_port++, $host );          fork_ssh( $node_port++, $host );
64  }  }
65    
66  my $session;  my $session;
67    
68    sub to_all {
69            my $data = shift;
70            foreach my $port ( keys %{ $session->{port} } ) {
71                    warn ">>>> [$port]\n";
72                    Storable::store_fd( $data, $session->{port}->{$port} );
73            }
74    }
75    
76    my @shard_paths;
77    
78  while (1) {  while (1) {
79          for my $sock ($sel->can_read(1)) {          for my $sock ($sel->can_read(1)) {
80                  if ($sock == $lsn) {                  if ($sock == $lsn) {
81                          my $new = $lsn->accept;                          my $new = $lsn->accept;
82                          $sel->add($new);                          $sel->add($new);
83                          $session->{$new} = $new->peerhost;                          $session->{peerport}->{ $new->peerport } = $new;
84                          warn "[socket] connect\n";                          warn "[socket] connect\n";
85                          Storable::store_fd( { ping => 1 }, $new );                          Storable::store_fd( { ping => 1 }, $new );
86                            info 0 => 'ping', $new->peerport;
87                  } else {                  } else {
88                          if ( $sock->connected ) {                          my $data = eval { Storable::fd_retrieve( $sock ) };
89                                  my $data = Storable::fd_retrieve( $sock );                          if ( $@ ) {
                                 warn "<<<< ", dump($data), $/;  
                                 if ( $data->{repl} ) {  
                                         Storable::store_fd( { repl => $data }, $sock );  
                                 }  
                         } else {  
90                                  delete $session->{$sock};                                  delete $session->{$sock};
91                                  warn "[socket] disconnect\n";                                  warn "[socket] disconnect: $@\n";
92                                  $sel->remove($sock);                                  $sel->remove($sock);
93                                  $sock->close;                                  $sock->close;
94                            } else {
95                                    warn "<<<< ", dump($data), $/;
96    
97                                    if ( my $path = $data->{shard} ) {
98                                            push @{ $info->{shard}->{ $data->{port} } }, $path;
99                                    }
100    
101                                    if ( my $repl = $data->{repl} ) {
102                                            my $response = { repl_pid => $$ };
103                                            if ( $repl =~ m/ping/ ) {
104                                                    to_all { ping => 1 };
105                                            } elsif ( $repl =~ m/info/ ) {
106                                                    $response->{info} = $info;
107                                            } elsif ( $repl =~ m{load\s*(\S+)?} ) {
108                                                    my $name = $1 || 'shard';
109                                                    @shard_paths = glob "/tmp/sack/$name/*";
110                                                    warn "loading shard $name from ", dump( @shard_paths );
111                                                    to_all { load => $name };
112                                            } elsif ( $repl =~ m{view\s*(\S+)?} ) {
113                                                    my $path = $1 || '/srv/Sack/views/00.demo.pl';
114                                                    my $code = read_file $path;
115                                                    Sack::Merge->clean;
116                                                    to_all { view => $code, path => $path };
117                                                    $response->{view}->{$path}->{running};
118                                            } elsif ( $repl =~ m{debug\s*(.+)?} ) {
119                                                    to_all { debug => $1 };
120                                            } elsif ( $repl =~ m{out} ) {
121                                                    my $out = Sack::Merge->out;    
122                                                    warn "out ",dump( $out );
123                                                    $response->{out} = $out;
124                                            } elsif ( $repl =~ m{clean} ) {
125                                                    delete $info->{shard};
126                                                    to_all { clean => 1 };
127                                            } else {
128                                                    $response->{error}->{unknown} = $data;
129                                            }
130                                            Storable::store_fd( $response, $sock );
131                                    } elsif ( $data->{ping} ) {
132                                            my $port = $data->{port};
133                                            info $port => 'ping', $port;
134                                            $session->{port}->{ $data->{port} } = $sock;
135                                    } elsif ( $data->{load} eq 'shard' ) {
136                                            if ( my $path = shift @shard_paths ) {
137                                                    warn "retrieve $path ", -s $path;
138                                                    my $shard = Storable::retrieve $path;
139                                                    warn ">>>> [", $data->{port}, "] sending shard $path\n";
140                                                    Storable::store_fd( { path => $path, shard => $shard }, $sock );
141                                            } else {
142                                                    warn "no more shards for [", $data->{port}, "]\n";
143                                            }
144                                    } elsif ( defined $data->{out} ) {
145                                            Sack::Merge->add( $data->{out} );
146                                    } else {
147                                            warn "UNKNOWN ",dump($data);
148                                    }
149                          }                          }
150                  }                  }
151          }          }
Legend:
Removed from v.164  
changed lines
  Added in v.203
  ViewVC Help
Powered by ViewVC 1.1.26