cvs commit: p5ee/App-Context/lib/App/Context Cluster.pm

[email protected] (Stephen Adkins)
Newsgroups perl.cvs.p5ee
Message-ID <[email protected]> 
cvsuser     04/09/28 08:20:40

  Added:       App-Context/lib/App/Context Cluster.pm
  Log:
  new
  
  Revision  Changes    Path
  1.1                  p5ee/App-Context/lib/App/Context/Cluster.pm
  
  Index: Cluster.pm
  ===================================================================
  
  #############################################################################
  ## $Id: Cluster.pm,v 1.1 2004/09/28 15:20:40 spadkins Exp $
  #############################################################################
  
  package App::Context::Cluster;
  
  use App;
  use App::Context;
  
  @ISA = ( "App::Context" );
  
  use Sys::Hostname;
  use Socket;
  use IO::Socket;
  use IO::Socket::INET;
  use POSIX ":sys_wait_h";
  
  use strict;
  
  =head1 NAME
  
  App::Context::Cluster - a runtime environment with a Cluster Controller and many Cluster Nodes
  
  =head1 SYNOPSIS
  
     # ... official way to get a Context object ...
     use App;
     $context = App->context();
     $config = $context->config();   # get the configuration
     $config->dispatch_events();     # dispatch events
  
     # ... alternative way (used internally) ...
     use App::Context::Cluster;
     $context = App::Context::Cluster->new();
  
  =cut
  
  sub _init {
      &App::sub_entry if ($App::trace);
      my ($self, $options) = @_;
      $options = {} if (!defined $options);
  
      my $host = hostname;
      $self->{hostname} = $host;
      $host =~ s/\..*//;   # get rid of fully qualified domain name
      $self->{host} = $host;
  
      $self->{controller_host} = $options->{controller_host};
      $self->{controller_port} = $options->{controller_port};
      $self->{is_controller}   = $options->{controller};
      $self->{verbose}         = $options->{verbose};
  
      if ($self->{is_controller}) {
          $self->{port} = $options->{controller_port};
      }
      else {
          $self->{port} = $options->{node_port};
      }
  
      my $listen_socket = IO::Socket::INET->new(
          Proto     => "tcp",
          LocalPort => $self->{port},
          Listen    => SOMAXCONN,
      ) || die "Unable to listen on $host:$self->{port} - $!";
  
      $self->{listen_socket} = $listen_socket;
  
      if ($self->{is_controller}) {
          $self->ctrl_init();
      }
      else {
          $self->node_init();
      }
      &App::sub_exit() if ($App::trace);
  }
  
  sub close_listen_socket {
      &App::sub_entry if ($App::trace);
      my ($self) = @_;
      if ($self->{listen_socket}) {
          $self->{listen_socket}->close();
          delete $self->{listen_socket};
      }
      &App::sub_exit() if ($App::trace);
  }
  
  sub DESTROY {
      &App::sub_entry if ($App::trace);
      my ($self) = @_;
      $self->close_listen_socket();
      &App::sub_exit() if ($App::trace);
  }
  
  sub dispatch_events {
      &App::sub_entry if ($App::trace);
      my ($self) = @_;
  
      my ($role, $port, $startup, $shutdown, $process_msg);
      if ($self->{is_controller}) {
          $role        = "Controller";
          $startup     = "ctrl_startup";
          $shutdown    = "ctrl_shutdown";
          $process_msg = "ctrl_process_msg";
      }
      else {
          $role        = "Node";
          $startup     = "node_startup";
          $shutdown    = "node_shutdown";
          $process_msg = "node_process_msg";
      }
      my $verbose = $self->{verbose};
  
      print "[$$] Starting Cluster $role on $self->{host}:$self->{port}\n" if ($verbose);
  
      $self->$startup();
  
      my $quit = 0;
      $SIG{USR1} = sub { print "[$$] Caught Signal: @_\n" if ($verbose); };
      $SIG{USR2} = sub { print "[$$] Caught Signal: @_\n" if ($verbose); };
      $SIG{HUP}  = sub { print "[$$] Caught Signal: @_\n" if ($verbose); };
      $SIG{INT}  = sub { print "[$$] Caught Signal: @_\n" if ($verbose); $quit = 1; };
      $SIG{QUIT} = sub { print "[$$] Caught Signal: @_\n" if ($verbose); $quit = 1; };
      $SIG{TERM} = sub { print "[$$] Caught Signal: @_\n" if ($verbose); $quit = 1; };
  
      my $listen_socket = $self->{listen_socket};
      my ($connection_socket, $msg);
      my ($event, @events);
      my ($time, $time_of_next_event, $sleep_interval);
      my ($pid, $exitval, $sig);
      while (!$quit) {
          eval {
              print "[$$] Checking for scheduled events.\n" if ($verbose >= 8);
              $time = time();
              $time_of_next_event = $self->get_current_events(\@events, $time);
              if ($#events > -1) {
                  foreach $event (@events) {
                      $self->send_event($event);
                  }
                  $time = time();
              }
              while (($pid = waitpid(-1,WNOHANG)) > 0) {
                  $exitval = $? >> 8;
                  $sig     = $? & 255;
                  print "[$$] Child $pid finished [exitval=$exitval,sig=$sig]\n";
                  $self->cleanup_pid($pid, $exitval, $sig);
              }
              $time = time();
              if ($time_of_next_event > 0) {
                  $sleep_interval = $time_of_next_event - $time;
                  $sleep_interval = 0 if ($sleep_interval < 0);
              }
              else {
                  $sleep_interval = 15*60;
              }
              # TODO: if (sleep_interval == 0), use select() to see if anyone is waiting, else ...
              print "[$$] Sleeping on socket: timeout($sleep_interval)\n" if ($verbose >= 3);
              $listen_socket->timeout($sleep_interval);
              $connection_socket = $listen_socket->accept();
              if ($connection_socket) {
                  $connection_socket->autoflush(1);
                  $msg = <$connection_socket>;
                  $msg =~ s/[\015\012]+$//;
                  if ($msg =~ /^EXIT/i || $msg =~ /^QUIT/i) {
                      $quit = 1;
                  }
                  else {
                      $self->$process_msg($connection_socket, $msg);
                  }
                  $connection_socket->close();
              }
          };
          if ($@) {
              print STDERR "[$$] $@" if ($verbose);
          }
      }
  
      $self->close_listen_socket();
      $self->$shutdown();
      print "[$$] Stopping Cluster $role.\n" if ($verbose);
      $self->shutdown();
      &App::sub_exit() if ($App::trace);
  }
  
  # TODO: Implement this as a fork() or a context-level message to a node to fork().
  #       i.e. messages such as "EVENT:" and "EVENT-OK:"
  #       Save the callback_event according to an event_tag.
  #       Then implement cleanup_pid to send the callback_event.
  sub send_async_event {
      &App::sub_entry if ($App::trace);
      my ($self, $event, $callback_event) = @_;
      $self->send_event($event);
      if ($callback_event) {
          my $event_tag = "local-$$";
          if (! $callback_event->{args}) {
              $callback_event->{args} = [ $event_tag ];
          }
          $self->send_event($callback_event);
      }
      &App::sub_exit() if ($App::trace);
  }
  
  sub cleanup_pid {
      &App::sub_entry if ($App::trace);
      my ($self, $pid, $exitval, $sig) = @_;
      # nothing to do
      &App::sub_exit() if ($App::trace);
  }
  
  sub send_async_message {
      &App::sub_entry if ($App::trace);
      my ($self, $host, $port, $message) = @_;
      my $pid = fork();
      if (!$pid) {   # running in child
          $self->send_message($host, $port, $message);
          exit(0);
      }
      &App::sub_exit() if ($App::trace);
  }
  
  sub send_message {
      &App::sub_entry if ($App::trace);
      my ($self, $host, $port, $message) = @_;
      my $verbose = $self->{verbose};
      print STDERR "[$$] send_message($host, $port, $message)\n" if ($verbose >= 2);
  
      if (!$port && $host =~ /^([^:]+):([0-9]+)$/) {
          $host = $1;
          $port = $2;
      }
  
      my $sock = IO::Socket::INET->new(
          PeerAddr => $host,
          PeerPort => $port,
          Proto    => "tcp",
          Type     => SOCK_STREAM,
      );
  
      my ($response);
      if ($sock) {
          eval {
              $sock->autoflush(1);
              print $sock $message, "\n";
              $response = <$sock>;
              $response =~ s/[\r\n]+$//;
              close($sock);
          };
          if ($@) {
              $response = "SEND ERROR: $@";
          }
      }
      else {
          $response = "CONNECT ERROR: $!";
      }
  
      print "[$$] send_message($host, $port, ...) => [$response]\n" if ($verbose >= 2);
      &App::sub_exit($response) if ($App::trace);
      return($response);
  }
  
  ######################################################################
  # Controller
  ######################################################################
  
  sub ctrl_init {
      &App::sub_entry if ($App::trace);
      my ($self) = @_;
      &App::sub_exit() if ($App::trace);
  }
  
  sub ctrl_startup {
      &App::sub_entry if ($App::trace);
      my ($self) = @_;
      &App::sub_exit() if ($App::trace);
  }
  
  sub ctrl_shutdown {
      &App::sub_entry if ($App::trace);
      my ($self) = @_;
      &App::sub_exit() if ($App::trace);
  }
  
  sub ctrl_process_msg {
      &App::sub_entry if ($App::trace);
      my ($self, $connection_socket, $msg) = @_;
      my $verbose = $self->{verbose};
      print STDERR "[$$] process_msg($msg)\n" if ($verbose >= 2);
      my $processing_complete = $self->ctrl_process_app_msg($connection_socket, $msg);
      if (!$processing_complete) {
          if ($msg =~ /^NODE-UP:(.*)/) {
              my $resp = $self->ctrl_set_node_up($1) || "ok";
              $connection_socket->print("$resp\n");
          }
          elsif ($msg =~ /^NODE-DOWN:(.*)/) {
              $self->ctrl_set_node_down($1);
              $connection_socket->print("ok\n");
          }
          elsif ($msg =~ s/^GET//) {
              $connection_socket->print("HTTP/1.0 200 OK\n");
              $connection_socket->print("Content-type: text/plain\n");
              $connection_socket->print("\n");
              $connection_socket->print("$self->{host}:$self->{port} acting as cluster controller\n");
              $connection_socket->print($self->ctrl_state());
          }
          else {
              $connection_socket->print("unknown [$msg]\n");
          }
      }
      &App::sub_exit() if ($App::trace);
  }
  
  sub ctrl_process_app_msg {
      &App::sub_entry if ($App::trace);
      my ($self, $connection_socket, $msg) = @_;
      my $processing_complete = 0;
      &App::sub_exit($processing_complete) if ($App::trace);
      return($processing_complete);
  }
  
  sub ctrl_state {
      &App::sub_entry if ($App::trace);
      my ($self) = @_;
      my %state = (
          "node" => $self->{node},
          "nodes" => $self->{nodes},
          "scheduled_events" => $self->{scheduled_events},
      );
      my $d = Data::Dumper->new([ \%state ], [ "state" ]);
      $d->Indent(1);
      my $state = $d->Dump();
      &App::sub_exit($state) if ($App::trace);
      return($state);
  }
  
  sub ctrl_set_node_down {
      &App::sub_entry if ($App::trace);
      my ($self, $node) = @_;
      $self->{node}{$node}{up} = 0;
      $self->ctrl_set_nodes();
      &App::sub_exit() if ($App::trace);
  }
  
  sub ctrl_set_node_up {
      &App::sub_entry if ($App::trace);
      my ($self, $node) = @_;
      my ($retval);
      if ($self->{node}{$node}{up}) {
          $retval = "ok";
      }
      else {
          $self->{node}{$node}{up} = 1;
          $self->ctrl_set_nodes();
          $retval = "new";
      }
      &App::sub_exit($retval) if ($App::trace);
      return($retval);
  }
  
  sub ctrl_set_nodes {
      &App::sub_entry if ($App::trace);
      my ($self) = @_;
      my (@nodes);
      foreach my $node (sort keys %{$self->{node}}) {
          if ($self->{node}{$node}{up}) {
              push(@nodes, $node);
          }
      }
      $self->{nodes} = \@nodes;
      &App::sub_exit() if ($App::trace);
  }
  
  ######################################################################
  # Node
  ######################################################################
  
  sub node_init {
      &App::sub_entry if ($App::trace);
      my ($self) = @_;
      &App::sub_exit() if ($App::trace);
  }
  
  sub node_startup {
      &App::sub_entry if ($App::trace);
      my ($self) = @_;
      my $controller_host = $self->{controller_host};
      my $controller_port = $self->{controller_port};
      my $node_host       = $self->{host};
      my $node_port       = $self->{port};
      my $node_heartbeat  = $self->{context}{options}{node_heartbeat} || 60;
      $self->schedule_event(
          method => "send_async_message",
          args => [ $controller_host, $controller_port, "NODE-UP:$node_host:$node_port" ],
          time => time(),  # immediately ...
          interval => $node_heartbeat,  # and every X seconds hereafter
      );
      &App::sub_exit() if ($App::trace);
  }
  
  sub node_shutdown {
      &App::sub_entry if ($App::trace);
      my ($self) = @_;
      my $controller_host = $self->{controller_host};
      my $controller_port = $self->{controller_port};
      my $node_host       = $self->{host};
      my $node_port       = $self->{port};
      $self->send_async_message($controller_host, $controller_port, "NODE-DOWN:$node_host:$node_port");
      &App::sub_exit() if ($App::trace);
  }
  
  sub node_process_msg {
      &App::sub_entry if ($App::trace);
      my ($self, $connection_socket, $msg) = @_;
      my $verbose = $self->{verbose};
      print STDERR "[$$] process_msg($msg)\n" if ($verbose >= 2);
      my $processing_complete = $self->node_process_app_msg($connection_socket, $msg);
      if (!$processing_complete) {
          if ($msg =~ s/^GET//) {
              $connection_socket->print("HTTP/1.0 200 OK\n");
              $connection_socket->print("Content-type: text/plain\n");
              $connection_socket->print("\n");
              $connection_socket->print("$self->{host}:$self->{port} acting as cluster node\n");
              #$connection_socket->print($self->ctrl_state());
          }
          else {
              $connection_socket->print("unknown [$msg]\n");
          }
      }
      &App::sub_exit() if ($App::trace);
  }
  
  sub node_process_app_msg {
      &App::sub_entry if ($App::trace);
      my ($self, $connection_socket, $msg) = @_;
      my $processing_complete = 0;
      &App::sub_exit($processing_complete) if ($App::trace);
      return($processing_complete);
  }
  
  1;
lmpx.com only provides a reader for public news (NNTP) servers. It is not affiliated with the servers or forums shown here and is not responsible for the content of articles, which is written by their respective authors.