Author: spadkins
Date: Sat Apr 8 08:41:03 2006
New Revision: 5875
Added:
p5ee/trunk/App-Context/lib/App/Context/ClusterNode.pm
Modified:
p5ee/trunk/App-Context/lib/App/Context/Cluster.pm
Log:
moving clustering context toward maturity
Modified: p5ee/trunk/App-Context/lib/App/Context/Cluster.pm
==============================================================================
--- p5ee/trunk/App-Context/lib/App/Context/Cluster.pm (original)
+++ p5ee/trunk/App-Context/lib/App/Context/Cluster.pm Sat Apr 8 08:41:03 2006
@@ -11,17 +11,13 @@
@ISA = ( "App::Context::Server" );
-use Sys::Hostname;
-use Socket;
-use IO::Socket;
-use IO::Socket::INET;
-use POSIX ":sys_wait_h";
+use Date::Format;
use strict;
=head1 NAME
-App::Context::Cluster - a runtime environment with a Cluster Controller and many Cluster Nodes
+App::Context::Cluster - a runtime environment of a Cluster Controller served by many Cluster Nodes
=head1 SYNOPSIS
@@ -40,115 +36,133 @@
sub _init2a {
&App::sub_entry if ($App::trace);
my ($self, $options) = @_;
- $self->{controller_host} = $options->{controller_host};
- $self->{controller_port} = $options->{controller_port};
- $self->{is_controller} = $options->{controller};
- if ($self->{is_controller}) {
- $self->{port} = $options->{controller_port};
- }
- else {
- $self->{port} = $options->{node_port};
- }
+ die "Controller must have a port defined (\$context->{options}{port})" if (!$self->{port});
+ $self->{num_node_events} = 0;
+ $self->{max_node_events} = $self->{options}{"App.Context.Cluster.max_node_events"} || 10;
&App::sub_exit() if ($App::trace);
}
sub _init2b {
&App::sub_entry if ($App::trace);
my ($self, $options) = @_;
- if ($self->{is_controller}) {
- $self->ctrl_init();
- }
- else {
- $self->node_init();
- }
+ # nothing yet
&App::sub_exit() if ($App::trace);
}
sub dispatch_events_begin {
my ($self) = @_;
-
- my $verbose = $self->{verbose};
- my $role = $self->{is_controller} ? "Controller" : "Node";
-
- print "[$$] Starting Cluster $role on $self->{host}:$self->{port}\n" if ($verbose);
-
- if ($self->{is_controller}) {
- $self->ctrl_startup();
- }
- else {
- $self->node_startup();
- }
+ $self->log({level=>2},"Starting Cluster Controller on $self->{host}:$self->{port}\n");
+ # nothing special yet
}
sub dispatch_events_end {
my ($self) = @_;
+ $self->log({level=>2},"Stopping Cluster Controller\n");
+ # nothing special yet
+}
- my $verbose = $self->{verbose};
- my $role = $self->{is_controller} ? "Controller" : "Node";
+# CONTROLLER ONLY
+sub send_async_event_now {
+ &App::sub_entry if ($App::trace);
+ my ($self, $event, $callback_event) = @_;
- if ($self->{is_controller}) {
- $self->ctrl_shutdown();
+ my $destination = $event->{destination};
+ if (! defined $destination) {
+ $self->log("ERROR: send_async_event_now(): node not assigned\n");
}
- else {
- $self->node_shutdown();
- }
-
- print "[$$] Stopping Cluster $role\n" if ($verbose);
-}
-
-sub process_msg {
- my ($self, $connection_socket, $msg) = @_;
- if ($self->{is_controller}) {
- $self->ctrl_process_msg($connection_socket, $msg);
+ elsif ($destination =~ /^([^:]+):([0-9]+)$/) {
+ my $node_host = $1;
+ my $node_port = $2;
+ my $args = "";
+ if ($event->{args}) {
+ $args = $self->{rpc_serializer}->serialize($event->{args});
+ }
+ my $response = $self->send_message($node_host, $node_port, "ASYNC-EVENT:$event->{service_type}:$event->{name}:$event->{method}:$args");
+ $self->{num_node_events}++;
+ if ($callback_event) {
+ if ($response =~ /^ASYNC-EVENT-TOKEN:(.+)/) {
+ my $remote_event_token = $1;
+ $self->{pending_callback_event}{$remote_event_token} = $callback_event;
+ }
+ }
}
else {
- $self->node_process_msg($connection_socket, $msg);
+ $self->SUPER::send_async_event_now($event, $callback_event);
}
-}
-
-######################################################################
-# Controller
-######################################################################
-
-sub ctrl_init {
- &App::sub_entry if ($App::trace);
- my ($self) = @_;
&App::sub_exit() if ($App::trace);
}
-sub ctrl_startup {
+sub assign_event_destination {
&App::sub_entry if ($App::trace);
- my ($self) = @_;
- &App::sub_exit() if ($App::trace);
+ my ($self, $event) = @_;
+ my $assigned = 0;
+ if ($self->{num_node_events} < $self->{max_node_events}) {
+ $event->{destination} = $self->{host};
+ $assigned = $self->assign_event_destination_by_round_robin($event);
+ }
+ &App::sub_exit($assigned) if ($App::trace);
+ return($assigned);
}
-sub ctrl_shutdown {
+sub assign_event_destination_by_round_robin {
&App::sub_entry if ($App::trace);
- my ($self) = @_;
- &App::sub_exit() if ($App::trace);
+ my ($self, $event) = @_;
+
+ my $assigned = 0;
+ my $nodes = $self->{nodes};
+ if ($#$nodes > -1) {
+ my $node_idx = $self->{last_node_idx};
+ $node_idx = (defined $node_idx) ? $node_idx + 1 : 0;
+ $node_idx = 0 if ($node_idx > $#$nodes);
+ $event->{destination} = $nodes->[$node_idx];
+ $self->{last_node_idx} = $node_idx;
+ $assigned = 1;
+ }
+
+ &App::sub_exit($assigned) if ($App::trace);
+ return($assigned);
}
-sub ctrl_process_msg {
+sub 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);
+ $self->log({level=>3},"process_msg: [$msg]\n");
+ my $processing_complete = $self->process_custom_msg($connection_socket, $msg);
if (!$processing_complete) {
if ($msg =~ /^NODE-UP:(.*)/) {
- my $resp = $self->ctrl_set_node_up($1) || "ok";
+ my $resp = $self->set_node_up($1) || "ok";
$connection_socket->print("$resp\n");
}
elsif ($msg =~ /^NODE-DOWN:(.*)/) {
- $self->ctrl_set_node_down($1);
+ $self->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());
+ $connection_socket->print($self->state());
+ }
+ elsif ($msg =~ /^ASYNC-EVENT-RESULTS:([^:]+):(.*)$/) {
+ my $remote_event_token = $1;
+ my $results = $2;
+ if ($results ne "") {
+ $results = $self->{rpc_serializer}->deserialize($results);
+ if ($results && ref($results) eq "ARRAY" && $#$results == 0) {
+ $results = $results->[0];
+ }
+ }
+ $connection_socket->print("OK\n");
+ $self->{num_node_events}--;
+ my $callback_event = $self->{pending_callback_event}{$remote_event_token};
+ if ($callback_event) {
+ delete $self->{pending_callback_event}{$remote_event_token};
+ $callback_event->{args} = [] if (! $callback_event->{args});
+ push(@{$callback_event->{args}},
+ {event_token => $callback_event->{event_token}, returnval => $results, errnum => 0, errmsg => ""});
+ $self->send_event($callback_event);
+ }
}
else {
$connection_socket->print("unknown [$msg]\n");
@@ -157,7 +171,8 @@
&App::sub_exit() if ($App::trace);
}
-sub ctrl_process_app_msg {
+# Can be overridden to provide customized processing.
+sub process_custom_msg {
&App::sub_entry if ($App::trace);
my ($self, $connection_socket, $msg) = @_;
my $processing_complete = 0;
@@ -165,30 +180,40 @@
return($processing_complete);
}
-sub ctrl_state {
+sub state {
+ &App::sub_entry if ($App::trace);
+ my ($self) = @_;
+
+ my $datetime = time2str("%Y-%m-%d %H:%M:%S", time());
+ my $state = "Cluster Controller: $self->{host}:$self->{port}\n[$datetime]\n";
+ $state .= "\n";
+ $state .= $self->_state();
+
+ &App::sub_exit($state) if ($App::trace);
+ return($state);
+}
+
+sub _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();
+
+ my $state = "";
+
+ $state .= $self->SUPER::_state();
+
&App::sub_exit($state) if ($App::trace);
return($state);
}
-sub ctrl_set_node_down {
+sub set_node_down {
&App::sub_entry if ($App::trace);
my ($self, $node) = @_;
$self->{node}{$node}{up} = 0;
- $self->ctrl_set_nodes();
+ $self->set_nodes();
&App::sub_exit() if ($App::trace);
}
-sub ctrl_set_node_up {
+sub set_node_up {
&App::sub_entry if ($App::trace);
my ($self, $node) = @_;
my ($retval);
@@ -197,14 +222,14 @@
}
else {
$self->{node}{$node}{up} = 1;
- $self->ctrl_set_nodes();
+ $self->set_nodes();
$retval = "new";
}
&App::sub_exit($retval) if ($App::trace);
return($retval);
}
-sub ctrl_set_nodes {
+sub set_nodes {
&App::sub_entry if ($App::trace);
my ($self) = @_;
my (@nodes);
@@ -217,72 +242,5 @@
&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;
Added: p5ee/trunk/App-Context/lib/App/Context/ClusterNode.pm
==============================================================================
--- (empty file)
+++ p5ee/trunk/App-Context/lib/App/Context/ClusterNode.pm Sat Apr 8 08:41:03 2006
@@ -0,0 +1,161 @@
+
+#############################################################################
+## $Id: ClusterNode.pm 3666 2006-03-11 20:34:10Z spadkins $
+#############################################################################
+
+package App::Context::ClusterNode;
+$VERSION = (q$Revision: 3666 $ =~ /(\d[\d\.]*)/)[0]; # VERSION numbers generated by svn
+
+use App;
+use App::Context::Server;
+
+@ISA = ( "App::Context::Server" );
+
+use Date::Format;
+
+use strict;
+
+=head1 NAME
+
+App::Context::ClusterNode - a runtime environment for a Cluster Node that serves a Cluster Controller
+
+=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::ClusterNode;
+ $context = App::Context::ClusterNode->new();
+
+=cut
+
+sub _init2a {
+ &App::sub_entry if ($App::trace);
+ my ($self, $options) = @_;
+ $self->{controller_host} = $options->{controller_host};
+ $self->{controller_port} = $options->{controller_port};
+ die "Node must have a controller host and port defined (\$context->{options}{controller_host} and {controller_port})"
+ if (!$self->{controller_host} || !$self->{controller_port});
+ &App::sub_exit() if ($App::trace);
+}
+
+sub _init2b {
+ &App::sub_entry if ($App::trace);
+ my ($self, $options) = @_;
+ # nothing yet
+ &App::sub_exit() if ($App::trace);
+}
+
+sub dispatch_events_begin {
+ &App::sub_entry if ($App::trace);
+ my ($self) = @_;
+ $self->log({level=>2},"Starting Cluster Node on $self->{host}:$self->{port}\n");
+ 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->{options}{node_heartbeat} || 300;
+ $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 dispatch_events_end {
+ &App::sub_entry if ($App::trace);
+ my ($self) = @_;
+ $self->log({level=>2},"Stopping Cluster Node\n");
+ 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 process_msg {
+ &App::sub_entry if ($App::trace);
+ my ($self, $connection_socket, $msg) = @_;
+ my $verbose = $self->{verbose};
+ $self->log({level=>3},"process_msg: [$msg]\n");
+ my $processing_complete = $self->process_custom_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->state());
+ }
+ elsif ($msg =~ /^ASYNC-EVENT:([^:]+):([^:]+):([^:]+):(.*)$/) {
+ my %event = (
+ service_type => $1,
+ name => $2,
+ method => $3,
+ );
+ my $args = $4;
+ $event{args} = $self->{rpc_serializer}->deserialize($args) if ($args ne "");
+ my $event_token = $self->send_async_event({method => "process_async_event", args => [\%event],});
+ $event{event_token} = $event_token;
+ $connection_socket->print("ASYNC-EVENT-TOKEN:$event_token\n");
+ }
+ else {
+ $connection_socket->print("unknown [$msg]\n");
+ }
+ }
+ &App::sub_exit() if ($App::trace);
+}
+
+# Can be overridden to provide customized processing.
+sub process_custom_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 state {
+ &App::sub_entry if ($App::trace);
+ my ($self) = @_;
+
+ my $datetime = time2str("%Y-%m-%d %H:%M:%S", time());
+ my $state = "Cluster Node: $self->{host}:$self->{port}\n[$datetime]\n";
+ $state .= "\n";
+ $state .= $self->_state();
+
+ &App::sub_exit($state) if ($App::trace);
+ return($state);
+}
+
+sub _state {
+ &App::sub_entry if ($App::trace);
+ my ($self) = @_;
+
+ my $state = "";
+
+ $state .= $self->SUPER::_state();
+
+ &App::sub_exit($state) if ($App::trace);
+ return($state);
+}
+
+sub process_async_event {
+ &App::sub_entry if ($App::trace);
+ my ($self, $event) = @_;
+ my $results = $self->send_event($event);
+ my $results_txt = $self->{rpc_serializer}->serialize($results);
+ my $msg = "ASYNC-EVENT-RESULTS:$event->{event_token}:$results_txt";
+ $self->send_message($self->{controller_host}, $self->{controller_port}, $msg);
+ &App::sub_exit(undef) if ($App::trace);
+ return(undef);
+}
+
+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.