/[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 173 by dpavlin, Tue Nov 3 17:28:57 2009 UTC trunk/lib/Sack/Server.pm revision 192 by dpavlin, Sun Nov 8 14:12:38 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 9  use IO::Select; Line 11  use IO::Select;
11  use Data::Dump qw(dump);  use Data::Dump qw(dump);
12  use Storable qw();  use Storable qw();
13  use File::Slurp;  use File::Slurp;
14    use Cwd qw(abs_path);
15    
16  my @cloud = qw(localhost tab.lan llin.lan);  my @cloud;
17    my $cloud_path = $ENV{CLOUD} || die "start with: CLOUD=etc/cloud perl -I/srv/Sack/lib $0\n";
 my $cloud_path = $ENV{CLOUD} || '/srv/Sack/etc/lib';  
18  @cloud = read_file $cloud_path;  @cloud = read_file $cloud_path;
19  @cloud = map { chomp $_; $_ } @cloud;  @cloud = map { chomp $_; $_ } @cloud;
20    
# Line 20  warn "# cloud ",dump( @cloud ); Line 22  warn "# cloud ",dump( @cloud );
22    
23  my $listen_port = 4444;  my $listen_port = 4444;
24    
25  my $node_path = $0;  my $node_path = abs_path $0;
26  $node_path =~ s{server.pl}{client.pl};  $node_path =~ s{Server}{Client};
27    
28  my $lsn = IO::Socket::INET->new(Listen => 1, LocalPort => $listen_port, Reuse => 1) or die $!;  my $lsn = IO::Socket::INET->new(Listen => 1, LocalPort => $listen_port, Reuse => 1) or die $!;
29  my $sel = IO::Select->new($lsn);  my $sel = IO::Select->new($lsn);
# Line 32  sub info { Line 34  sub info {
34          push @{ $info->{$port} }, [ @_ ];          push @{ $info->{$port} }, [ @_ ];
35  }  }
36    
37  sub fork_node {  sub fork_ssh {
38          my ( $port, $host ) = @_;          my ( $port, $host ) = @_;
39    
40          if ( my $pid = fork ) {          if ( my $pid = fork ) {
# Line 54  sub fork_node { Line 56  sub fork_node {
56  my $node_port = 4000;  my $node_port = 4000;
57    
58  foreach my $host ( @cloud ) {  foreach my $host ( @cloud ) {
59          system "echo $node_path | cpio --create --dereference | ssh -T -F $cloud_path.ssh $host cpio --extract --make-directories --unconditional --verbose";          system "find /srv/Sack/ | cpio --create --dereference | ssh -T -F $cloud_path.ssh $host cpio --extract --make-directories --unconditional";
60          fork_node( $node_port++, $host );          fork_ssh( $node_port++, $host );
61  }  }
62    
63  my $session;  my $session;
64    
65    sub to_all {
66            my $data = shift;
67            foreach my $port ( keys %{ $session->{port} } ) {
68                    warn ">>>> [$port]\n";
69                    Storable::store_fd( $data, $session->{port}->{$port} );
70            }
71    }
72    
73    my @shard_paths;
74    
75  while (1) {  while (1) {
76          for my $sock ($sel->can_read(1)) {          for my $sock ($sel->can_read(1)) {
77                  if ($sock == $lsn) {                  if ($sock == $lsn) {
78                          my $new = $lsn->accept;                          my $new = $lsn->accept;
79                          $sel->add($new);                          $sel->add($new);
80                          $session->{$new} = $new->peerport;                          $session->{peerport}->{ $new->peerport } = $new;
81                          warn "[socket] connect\n";                          warn "[socket] connect\n";
82                          Storable::store_fd( { ping => 1 }, $new );                          Storable::store_fd( { ping => 1 }, $new );
83                          info 0 => 'ping', $new->peerport;                          info 0 => 'ping', $new->peerport;
# Line 78  while (1) { Line 90  while (1) {
90                                  $sock->close;                                  $sock->close;
91                          } else {                          } else {
92                                  warn "<<<< ", dump($data), $/;                                  warn "<<<< ", dump($data), $/;
93                                  if ( $data->{repl} ) {                                  if ( my $repl = $data->{repl} ) {
94                                          my $response = { repl => $$, info => $info };                                          my $response = { repl_pid => $$ };
95                                            if ( $repl =~ m/ping/ ) {
96                                                    to_all { ping => 1 };
97                                            } elsif ( $repl =~ m/info/ ) {
98                                                    $response->{info} = $info;
99                                            } elsif ( $repl =~ m{load\s*(\S+)?} ) {
100                                                    my $name = $1 || 'shard';
101                                                    @shard_paths = glob "/tmp/sack/$name/*";
102                                                    warn "loading shards ", dump( @shard_paths );
103                                                    to_all { load => $name };
104                                            } else {
105                                                    $response->{error}->{unknown} = $data;
106                                            }
107                                          Storable::store_fd( $response, $sock );                                          Storable::store_fd( $response, $sock );
108                                  } elsif ( $data->{ping} ) {                                  } elsif ( $data->{ping} ) {
109                                          $info->{_peer_port}->{$sock->peerport} = $data->{port};                                          my $port = $data->{port};
110                                          info $data->{port} => 'peer port', $sock->peerport;                                          info $port => 'ping', $port;
111                                            $session->{port}->{ $data->{port} } = $sock;
112                                    } elsif ( $data->{load} eq 'shard' ) {
113                                            if ( my $path = shift @shard_paths ) {
114                                                    warn "retrieve $path ", -s $path;
115                                                    my $shard = Storable::retrieve $path;
116                                                    warn ">>>> [", $data->{port}, "] sending shard $path\n";
117                                                    Storable::store_fd( { path => $path, shard => $shard }, $sock );
118                                            } else {
119                                                    warn "no more shards for [", $data->{port}, "]\n";
120                                            }
121                                    } else {
122                                            warn "UNKNOWN ",dump($data);
123                                  }                                  }
124                          }                          }
125                  }                  }
Legend:
Removed from v.173  
changed lines
  Added in v.192
  ViewVC Help
Powered by ViewVC 1.1.26