Author: spadkins
Date: Tue Aug 7 12:11:58 2007
New Revision: 9827
Modified:
p5ee/trunk/App-Context/lib/App/Context/POE/ClusterController.pm
p5ee/trunk/App-Context/lib/App/Context/POE/ClusterNode.pm
p5ee/trunk/App-Context/lib/App/Context/POE/Server.pm
Log:
first working version of POE cluster context
Modified: p5ee/trunk/App-Context/lib/App/Context/POE/ClusterController.pm
==============================================================================
--- p5ee/trunk/App-Context/lib/App/Context/POE/ClusterController.pm (original)
+++ p5ee/trunk/App-Context/lib/App/Context/POE/ClusterController.pm Tue Aug 7 12:11:58 2007
@@ -44,9 +44,10 @@
$self->{max_async_events} = 0; # start with 0 because there are no nodes up
push(@{$self->{poe_states}},
- "poe_remote_async_event_queued", "poe_set_node_status", "poe_run_event",
- "poe_register_node", "poe_set_node_up", "poe_set_node_down");
- push(@{$self->{poe_ikc_published_states}}, "poe_set_node_status");
+ "poe_receive_node_status",
+ "poe_run_event");
+ push(@{$self->{poe_ikc_published_states}},
+ "poe_receive_node_status");
$self->_init_poe($options);
@@ -83,21 +84,22 @@
my $event_token = $self->send_async_event_in_process($event, $callback_event);
}
elsif ($destination =~ /^([^:]+):([0-9]+)$/) {
+ my $controller = "$self->{host}:$self->{port}";
my $node_host = $1;
my $node_port = $2;
my $args = $event->{args};
my $remote_server_name = "poe_${node_host}_${node_port}";
my $remote_session_alias = $self->{poe_session_name}; # remote is same as local
- my $remote_session_state = "poe_send_async_event";
- my $local_callback_state = "poe_remote_async_event_queued";
+ my $remote_session_state = "poe_enqueue_async_event";
+ my $local_callback_state = "poe_enqueue_async_event_finished";
$self->{num_async_events}++;
$self->{node}{$destination}{num_async_events}++;
my $kernel = $self->{poe_kernel};
$kernel->post("IKC", "call", "poe://$remote_server_name/$remote_session_alias/$remote_session_state",
- [ $event, $callback_event ], "poe:$local_callback_state" );
+ [ $controller, $event, $callback_event ], "poe:$local_callback_state" );
}
else {
$self->SUPER::send_async_event_now($event, $callback_event);
@@ -109,8 +111,9 @@
&App::sub_entry if ($App::trace);
my ($self, $kernel, $session_name) = @_[OBJECT, KERNEL, ARG1];
$self->log({level=>2},"POE: ikc_register ($session_name)\n");
- if ($session_name =~ /^ikc_/) {
- # do something
+ if ($session_name =~ /^ikc_([^_]+)_(\d+)$/) {
+ my $node = "$1:$2";
+ $self->set_node_up($node);
}
my ($retval);
&App::sub_exit($retval) if ($App::trace);
@@ -119,8 +122,12 @@
sub ikc_unregister {
&App::sub_entry if ($App::trace);
- my ($self, $kernel, $remote_kernel_id) = @_[OBJECT, KERNEL, ARG0];
- $self->log({level=>2},"POE: ikc_unregister ($remote_kernel_id)\n");
+ my ($self, $kernel, $session_name) = @_[OBJECT, KERNEL, ARG1];
+ $self->log({level=>2},"POE: ikc_unregister ($session_name)\n");
+ if ($session_name =~ /^ikc_([^_]+)_(\d+)$/) {
+ my $node = "$1:$2";
+ $self->set_node_down($node);
+ }
&App::sub_exit() if ($App::trace);
}
@@ -132,14 +139,6 @@
return;
}
-sub poe_remote_async_event_queued {
- &App::sub_entry if ($App::trace);
- my ($self, $kernel, $runtime_event_token, $async_event) = @_[OBJECT, KERNEL, ARG0, ARG1];
- $self->log({level=>2},"POE: poe_remote_async_event_queued ($async_event->[0]{name}.$async_event->[0]{method} => $runtime_event_token\n");
- $self->{running_async_event}{$runtime_event_token} = $async_event;
- &App::sub_exit() if ($App::trace);
-}
-
# $runtime_event_tokens take the following forms:
# $runtime_event_token = $pid; -- App::Context::Server::send_async_event_now() and ::finish_pid()
# $runtime_event_token = "$host-$port-$serial"; -- i.e. a plain event token on the node
@@ -207,13 +206,14 @@
return($assigned);
}
-sub poe_set_node_status {
+sub poe_receive_node_status {
&App::sub_entry if ($App::trace);
my ($self, $kernel, $args) = @_[OBJECT, KERNEL, ARG0];
- $self->log("POE: poe_set_node_status args=(@$args)\n");
- my ($retval);
- &App::sub_exit($retval) if ($App::trace);
- return($retval);
+ my ($node, $sys_info) = @$args;
+ $self->log("POE: poe_receive_node_status ($node) - " .
+ "load=$sys_info->{load}, memfree=$sys_info->{memfree}/$sys_info->{memtotal} swapfree=$sys_info->{swapfree}/$sys_info->{swaptotal}\n");
+ $self->set_node_up($node, $sys_info);
+ &App::sub_exit() if ($App::trace);
}
sub poe_run_event {
@@ -277,49 +277,11 @@
return($state);
}
-sub poe_register_node {
- &App::sub_entry if ($App::trace);
- my ($self, $kernel, $remote_kernel_id) = @_[OBJECT, KERNEL, ARG0];
- $self->log({level=>2},"POE: poe_register_node ($remote_kernel_id)\n");
-
- my ($node);
- if ($remote_kernel_id =~ m!poe_([^_]+)_([0-9]+)!) {
- $node = "$1:$2";
- }
- else {
- $self->log("ERROR: poe_register_node: unparseable remote_kernel_id [$remote_kernel_id]\n");
- }
-
- if (!$self->{node}{$node}{up}) {
- my $remote_server_name = "poe_${node}";
- $remote_server_name =~ s/:/_/;
- $kernel->post("IKC", "monitor", "poe://$remote_server_name",
- {register => "poe_set_node_up",
- unregister => "poe_set_node_down",
- shutdown => "poe_set_node_down",
- data => $node});
- }
-
- &App::sub_exit() if ($App::trace);
-}
-
-sub poe_set_node_up {
+sub set_node_up {
&App::sub_entry if ($App::trace);
- my ($self, $kernel, $remote_kernel_id, $node) = @_[OBJECT, KERNEL, ARG0, ARG3];
- $self->log({level=>2},"POE: poe_set_node_up ($remote_kernel_id; node=$node)\n");
- my ($retval, $values);
+ my ($self, $node, $sys_info) = @_;
+ my ($retval);
if (!$self->{node}{$node}{up}) {
- if ($node =~ /^([^:]+:\d+):(.*)/) {
- $node = $1;
- $values = $2;
- if ($values) {
- foreach my $value (split(/,/, $values)) {
- if ($value =~ /^([^=]+)=(.*)/) {
- $self->{node}{$node}{$1} = $2;
- }
- }
- }
- }
$self->{node}{$node}{datetime} = time2str("%Y-%m-%d %H:%M:%S", time());
if ($self->{node}{$node}{up}) {
$retval = "ok";
@@ -330,14 +292,18 @@
$retval = "new";
}
}
+ if ($sys_info) {
+ foreach my $sys_var (keys %$sys_info) {
+ $self->{node}{$node}{$sys_var} = $sys_info->{$sys_var};
+ }
+ }
&App::sub_exit($retval) if ($App::trace);
return($retval);
}
-sub poe_set_node_down {
+sub set_node_down {
&App::sub_entry if ($App::trace);
- my ($self, $kernel, $remote_kernel_id, $node) = @_[OBJECT, KERNEL, ARG0, ARG3];
- $self->log({level=>2},"POE: poe_set_node_down ($remote_kernel_id; node=$node)\n");
+ my ($self, $node) = @_;
my $runtime_event_token_prefix = $node;
$runtime_event_token_prefix =~ s/:/-/;
$self->reset_running_async_events($runtime_event_token_prefix);
Modified: p5ee/trunk/App-Context/lib/App/Context/POE/ClusterNode.pm
==============================================================================
--- p5ee/trunk/App-Context/lib/App/Context/POE/ClusterNode.pm (original)
+++ p5ee/trunk/App-Context/lib/App/Context/POE/ClusterNode.pm Tue Aug 7 12:11:58 2007
@@ -61,7 +61,7 @@
&App::sub_entry if ($App::trace);
my ($self, $options) = @_;
- my $ikc_name = "ikc_$self->{host}_$self->{port}";
+ my $ikc_name = "poe_$self->{host}_$self->{port}";
### Set up a server
POE::Component::IKC::Responder->spawn();
POE::Component::IKC::Client->spawn(
@@ -74,7 +74,7 @@
my $session_name = $self->{poe_session_name};
POE::Component::Server::SimpleHTTP->new(
- 'ALIAS' => $self->{poe_kernel_httpd_name},
+ 'ALIAS' => $self->{poe_kernel_http_name},
'ADDRESS' => INADDR_ANY,
'PORT' => $self->{options}{http_port},
'HANDLERS' => [
@@ -140,9 +140,10 @@
sub ikc_unregister {
&App::sub_entry if ($App::trace);
- my ($self, $kernel, $remote_kernel_id, $node) = @_[OBJECT, KERNEL, ARG0, ARG3];
- $self->log({level=>2},"POE: ikc_unregister ($remote_kernel_id; node=$node)\n");
+ my ($self, $kernel, $remote_kernel_id, $session_name, $node) = @_[OBJECT, KERNEL, ARG0, ARG1, ARG3];
+ $self->log({level=>2},"POE: ikc_unregister ($remote_kernel_id; session_name=$session_name; node=$node)\n");
$self->{controller_up} = 0;
+ $kernel->yield("poe_shutdown");
&App::sub_exit() if ($App::trace);
}
@@ -161,7 +162,7 @@
my $node_heartbeat = $self->{options}{node_heartbeat} || 60;
$self->schedule_event(
method => "send_node_status",
- time => time(), # immediately ...
+ time => time()+5, # immediately ...
interval => $node_heartbeat, # and every X seconds hereafter
);
&App::sub_exit() if ($App::trace);
@@ -184,13 +185,21 @@
my $remote_server_name = "poe_${controller_host}_${controller_port}";
my $remote_session_alias = $self->{poe_session_name}; # remote is same as local
- my $remote_session_state = "poe_set_node_status";
+ my $remote_session_state = "poe_receive_node_status";
my $sys_info = $self->get_sys_info();
+ my $memfree = $sys_info->{memfree} + $sys_info->{buffers} + $sys_info->{cached};
+ my $s_info = {
+ load => $sys_info->{load},
+ memfree => $memfree,
+ memtotal => $sys_info->{memtotal},
+ swapfree => $sys_info->{swapfree},
+ swaptotal => $sys_info->{swaptotal},
+ };
if ($self->{controller_up}) {
my $kernel = $self->{poe_kernel};
$kernel->post("IKC", "post", "poe://$remote_server_name/$remote_session_alias/$remote_session_state",
- [ $sys_info ]);
+ [ "$node_host:$node_port", $s_info ]);
}
&App::sub_exit() if ($App::trace);
Modified: p5ee/trunk/App-Context/lib/App/Context/POE/Server.pm
==============================================================================
--- p5ee/trunk/App-Context/lib/App/Context/POE/Server.pm (original)
+++ p5ee/trunk/App-Context/lib/App/Context/POE/Server.pm Tue Aug 7 12:11:58 2007
@@ -44,7 +44,7 @@
$self->{port} = $options->{port};
$options->{http_port} ||= $options->{port}+1;
$self->{poe_kernel_name} = "poe_$self->{host}_$self->{port}";
- $self->{poe_kernel_httpd_name} = $self->{poe_kernel_name} . "_httpd";
+ $self->{poe_kernel_http_name} = $self->{poe_kernel_name} . "_httpd";
$self->{poe_session_name} = "poe_session";
$self->{poe_kernel} = $poe_kernel;
@@ -65,8 +65,13 @@
ikc_register ikc_unregister ikc_shutdown
poe_run_event poe_event_loop_extension poe_dispatch_pending_async_events
poe_server_state poe_http_server_state poe_http_test_run
+ poe_enqueue_async_event poe_enqueue_async_event_finished poe_remote_async_event_finished
+ )];
+ $self->{poe_ikc_published_states} = [qw(
+ poe_server_state
+ poe_enqueue_async_event
+ poe_remote_async_event_finished
)];
- $self->{poe_ikc_published_states} = ["poe_server_state"];
### Does nothing by default, used by ClusterController, maybe other subclasses?
$self->_init2a($options);
@@ -116,7 +121,7 @@
my $session_name = $self->{poe_session_name};
POE::Component::Server::SimpleHTTP->new(
- 'ALIAS' => $self->{poe_kernel_httpd_name},
+ 'ALIAS' => $self->{poe_kernel_http_name},
'ADDRESS' => INADDR_ANY,
'PORT' => $self->{options}{http_port},
'HANDLERS' => [
@@ -729,7 +734,7 @@
sub _stop {
&App::sub_entry if ($App::trace);
my ( $self, $kernel, $heap, $state, $args ) = @_[ OBJECT, KERNEL, HEAP, ARG0, ARG1 ];
- $self->log({level=>2},"POE: _start\n");
+ $self->log({level=>2},"POE: _stop\n");
#sleep(1); # take a second to let child processes to die (perhaps not necessary, perhaps necessary when using POE::Wheel::Run)
&App::sub_exit() if ($App::trace);
}
@@ -745,8 +750,8 @@
sub ikc_unregister {
&App::sub_entry if ($App::trace);
- my ($self, $kernel, $remote_kernel_id) = @_[OBJECT, KERNEL, ARG0];
- $self->log({level=>2},"POE: ikc_unregister ($remote_kernel_id)\n");
+ my ($self, $kernel, $session_name) = @_[OBJECT, KERNEL, ARG1];
+ $self->log({level=>2},"POE: ikc_unregister ($session_name)\n");
&App::sub_exit() if ($App::trace);
}
@@ -834,6 +839,9 @@
# get rid of external ref count
$kernel->refcount_decrement( $session, $self->{poe_session_name} );
+ $kernel->post( $self->{poe_kernel_http_name}, 'SHUTDOWN');
+ $kernel->post('IKC', 'shutdown');
+
# propagate the message to children
$kernel->post( $heap->{child_session}, 'poe_shutdown' );
&App::sub_exit() if ($App::trace);
@@ -855,18 +863,26 @@
$self->log({level=>2},"POE: poe_event_loop_extension\n");
my $event_loop_extensions = $self->{event_loop_extensions};
#$self->log({level=>2},"Event Loop extension ($event_loop_extensions: #=" . ($#$event_loop_extensions+1) . ").\n");
+ my $async_event_added = 0;
if ($event_loop_extensions && $#$event_loop_extensions > -1) {
my ($extension, $obj, $method, $args, $event_executed);
for (my $i = 0; $i <= $#$event_loop_extensions; $i++) {
$extension = $event_loop_extensions->[$i];
($obj, $method, $args) = @$extension;
$event_executed = $obj->$method(@$args); # execute extension
+ $async_event_added = 1 if ($event_executed);
#if ($event_executed) {
# $self->log({level=>2},"Event Loop extension: ${obj}->${method}(@$args) = $event_executed\n");
#}
}
}
- $kernel->delay_set("poe_event_loop_extension", 1);
+ if ($async_event_added) {
+ $kernel->yield("poe_dispatch_pending_async_events");
+ $kernel->yield("poe_event_loop_extension");
+ }
+ else {
+ $kernel->delay_set("poe_event_loop_extension", 1);
+ }
&App::sub_exit() if ($App::trace);
}
@@ -888,6 +904,77 @@
&App::sub_exit() if ($App::trace);
}
+# State on Node
+sub poe_enqueue_async_event {
+ &App::sub_entry if ($App::trace);
+ my ($self, $kernel, $args) = @_[OBJECT, KERNEL, ARG0];
+ my ($sender, $event, $callback_event) = @$args;
+ $self->log({level=>2},"POE: poe_enqueue_async_event ($event->{name}.$event->{method})\n");
+
+ my $runtime_event_token = $self->send_async_event($event, { method => "async_event_finished", args => [ $sender, $event, $callback_event ], });
+ $event->{event_token} = $runtime_event_token;
+
+ &App::sub_exit([$runtime_event_token, [$event, $callback_event]]) if ($App::trace);
+ return([$runtime_event_token, [$event, $callback_event]]);
+}
+
+# State on Controller
+sub poe_enqueue_async_event_finished {
+ &App::sub_entry if ($App::trace);
+ my ($self, $kernel, $return_values) = @_[OBJECT, KERNEL, ARG0];
+ my ($runtime_event_token, $async_event) = @$return_values;
+ $self->log({level=>2},"POE: poe_enqueue_async_event_finished ($async_event->[0]{name}.$async_event->[0]{method} => $runtime_event_token)\n");
+ $self->{running_async_event}{$runtime_event_token} = $async_event;
+ &App::sub_exit() if ($App::trace);
+}
+
+# Method on Node
+sub async_event_finished {
+ &App::sub_entry if ($App::trace);
+ my ($self, $sender, $event, $callback_event) = @_;
+
+ my $runtime_event_token = $event->{event_token};
+ my $remote_server_name = "poe_${sender}";
+ $remote_server_name =~ s/:/_/;
+ my $remote_session_alias = $self->{poe_session_name}; # remote is same as local
+ my $remote_session_state = "poe_remote_async_event_finished";
+
+ my $kernel = $self->{poe_kernel};
+ $kernel->post("IKC", "post", "poe://$remote_server_name/$remote_session_alias/$remote_session_state",
+ [ $runtime_event_token, $callback_event->{args} ]);
+
+ &App::sub_exit() if ($App::trace);
+}
+
+# State on Controller
+sub poe_remote_async_event_finished {
+ &App::sub_entry if ($App::trace);
+ my ($self, $kernel, $args) = @_[OBJECT, KERNEL, ARG0];
+ my ($runtime_event_token, $callback_args) = @$args;
+
+ my $async_event = $self->{running_async_event}{$runtime_event_token};
+
+ if ($async_event) {
+ my ($event, $callback_event) = @$async_event;
+ $self->log({level=>2},"POE: poe_remote_async_event_finished ($event->{name}.$event->{method} => $runtime_event_token)\n");
+ delete $self->{running_async_event}{$runtime_event_token};
+
+ my $destination = $event->{destination} || "local";
+ $self->{num_async_events}--;
+ $self->{node}{$destination}{num_async_events}--;
+
+ if ($callback_event) {
+ $callback_event->{args} = $callback_args;
+ $self->send_event($callback_event);
+ }
+ }
+ else {
+ $self->log({level=>2},"POE: poe_remote_async_event_finished ($runtime_event_token)\n");
+ }
+
+ &App::sub_exit() if ($App::trace);
+}
+
sub poe_server_state {
&App::sub_entry if ($App::trace);
my ( $self, $kernel, $heap ) = @_[ OBJECT, KERNEL, HEAP ];
@@ -909,7 +996,7 @@
$response->content($server_state);
# Signal that the request was handled okay.
- $kernel->post( $self->{poe_kernel_httpd_name}, 'DONE', $response );
+ $kernel->post( $self->{poe_kernel_http_name}, 'DONE', $response );
&App::sub_exit(RC_OK) if ($App::trace);
return RC_OK;
}
@@ -932,7 +1019,7 @@
$response->content("SessionObject(mvworkd).sleep(30)");
# Signal that the request was handled okay.
- $kernel->post( $self->{poe_kernel_httpd_name}, 'DONE', $response );
+ $kernel->post( $self->{poe_kernel_http_name}, 'DONE', $response );
&App::sub_exit(RC_OK) if ($App::trace);
return RC_OK;
}
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.