Author: spadkins
Date: Tue Aug 28 12:01:45 2007
New Revision: 9897
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:
tweaks on POE stuff
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 28 12:01:45 2007
@@ -63,12 +63,12 @@
sub dispatch_events_begin {
my ($self) = @_;
- $self->log({level=>2},"Starting Cluster Controller on $self->{host}:$self->{port}\n");
+ $self->log({level=>2},"Starting Cluster Controller on $self->{host}:$self->{port}\n") if $self->{options}{poe_trace};
}
sub dispatch_events_end {
my ($self) = @_;
- $self->log({level=>2},"Stopping Cluster Controller\n");
+ $self->log({level=>2},"Stopping Cluster Controller\n") if $self->{options}{poe_trace};
# nothing special yet
}
@@ -76,6 +76,8 @@
&App::sub_entry if ($App::trace);
my ($self, $event, $callback_event) = @_;
+ $self->log({level=>2}, "CC: send_async_event_now() $event->{name}.$event->{method} : $event->{destination}\n") if $self->{options}{poe_trace};
+
my $destination = $event->{destination};
if (! defined $destination) {
$self->log("ERROR: send_async_event_now() $event->{name}.$event->{method} : destination not assigned\n");
@@ -110,7 +112,7 @@
sub ikc_register {
&App::sub_entry if ($App::trace);
my ($self, $kernel, $session_name) = @_[OBJECT, KERNEL, ARG1];
- $self->log({level=>2},"POE: ikc_register ($session_name)\n");
+ $self->log({level=>2},"POE: ikc_register ($session_name)\n") if $self->{options}{poe_ikc_debug};
if ($session_name =~ /^ikc_([^_]+)_(\d+)$/) {
my $node = "$1:$2";
$self->set_node_up($node);
@@ -123,8 +125,8 @@
sub ikc_unregister {
&App::sub_entry if ($App::trace);
my ($self, $kernel, $session_name) = @_[OBJECT, KERNEL, ARG1];
- $self->log({level=>2},"POE: ikc_unregister ($session_name)\n");
- if ($session_name =~ /^ikc_([^_]+)_(\d+)$/) {
+ $self->log({level=>2},"POE: ikc_unregister ($session_name)\n") if $self->{options}{poe_ikc_debug};
+ if ($session_name =~ /^poe_([^_]+)_(\d+)$/) {
my $node = "$1:$2";
$self->set_node_down($node);
}
@@ -134,7 +136,7 @@
sub ikc_shutdown {
&App::sub_entry if ($App::trace);
my ($self, $kernel, $arg0, $arg1, $arg2, $arg3) = @_[OBJECT, KERNEL, ARG0, ARG1, ARG2, ARG3];
- $self->log({level=>2},"POE: ikc_shutdown args=($arg0, $arg1, $arg2, $arg3)\n");
+ $self->log({level=>2},"POE: ikc_shutdown args=($arg0, $arg1, $arg2, $arg3)\n") if $self->{options}{poe_ikc_debug};
&App::sub_exit() if ($App::trace);
return;
}
@@ -211,7 +213,9 @@
my ($self, $kernel, $args) = @_[OBJECT, KERNEL, ARG0];
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");
+ "load=$sys_info->{load}, " .
+ "memfree=$sys_info->{memfree}/$sys_info->{memtotal} " .
+ "swapfree=$sys_info->{swapfree}/$sys_info->{swaptotal}\n") if $self->{options}{poe_trace};
$self->set_node_up($node, $sys_info);
&App::sub_exit() if ($App::trace);
}
@@ -229,7 +233,7 @@
else {
$event_str = "$event->{method}($args_str)";
}
- $self->log({level=>2},"Run Event: $event_str\n");
+ $self->log({level=>2},"Run Event: $event_str\n") if $self->{options}{poe_trace};
$self->send_event($event);
&App::sub_exit() if ($App::trace);
}
@@ -381,7 +385,7 @@
$cmd = $cmd_fmt;
$cmd =~ s/{host}/$host/g;
$cmd =~ s/{port}/$port/g;
- $self->log("Starting Node [$node]: [$cmd]\n");
+ $self->log("Starting Node [$node]: [$cmd]\n") if $self->{options}{poe_trace};
system("$cmd < /dev/null &");
}
else {
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 28 12:01:45 2007
@@ -70,7 +70,7 @@
name => $ikc_name,
timeout => 60,
);
- $self->log({level=>2},"Listening for Inter-Kernel Communications on $self->{host}:$self->{port}\n");
+ $self->log({level=>2},"Listening for Inter-Kernel Communications on $self->{host}:$self->{port}\n") if $self->{options}{poe_ikc_debug};
my $session_name = $self->{poe_session_name};
POE::Component::Server::SimpleHTTP->new(
@@ -82,7 +82,7 @@
{ 'DIR' => '.*', 'SESSION' => $session_name, 'EVENT' => 'poe_http_server_state', },
],
);
- $self->log({level=>2},"Listening for HTTP Requests on $self->{host}:$self->{options}{http_port}\n");
+ $self->log({level=>2},"Listening for HTTP Requests on $self->{host}:$self->{options}{http_port}\n") if $self->{options}{poe_http_debug};
&App::sub_exit() if ($App::trace);
}
@@ -97,7 +97,7 @@
sub _start {
&App::sub_entry if ($App::trace);
my ( $self, $kernel, $heap ) = @_[ OBJECT, KERNEL, HEAP ];
- $self->log({level=>2},"POE: _start\n");
+ $self->log({level=>2},"POE: _start\n") if $self->{options}{poe_trace};
my $name = $self->{poe_session_name};
$kernel->alias_set($name);
@@ -131,7 +131,7 @@
sub ikc_register {
&App::sub_entry if ($App::trace);
my ($self, $kernel, $remote_kernel_id, $node) = @_[OBJECT, KERNEL, ARG0, ARG3];
- $self->log({level=>2},"POE: ikc_register ($remote_kernel_id; node=$node)\n");
+ $self->log({level=>2},"POE: ikc_register ($remote_kernel_id; node=$node)\n") if $self->{options}{poe_ikc_debug};
$self->{controller_up} = 1;
my ($retval);
&App::sub_exit($retval) if ($App::trace);
@@ -141,7 +141,7 @@
sub ikc_unregister {
&App::sub_entry if ($App::trace);
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->log({level=>2},"POE: ikc_unregister ($remote_kernel_id; session_name=$session_name; node=$node)\n") if $self->{options}{poe_ikc_debug};
$self->{controller_up} = 0;
$kernel->yield("poe_shutdown");
&App::sub_exit() if ($App::trace);
@@ -150,7 +150,7 @@
sub ikc_shutdown {
&App::sub_entry if ($App::trace);
my ( $self, $kernel, $session, $heap ) = @_[ OBJECT, KERNEL, SESSION, HEAP ];
- $self->log({level=>2},"POE: ikc_shutdown\n");
+ $self->log({level=>2},"POE: ikc_shutdown\n") if $self->{options}{poe_ikc_debug};
&App::sub_exit() if ($App::trace);
return;
}
@@ -158,20 +158,26 @@
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");
+ $self->log({level=>2},"Starting Cluster Node on $self->{host}:$self->{port}\n") if $self->{options}{poe_trace};
my $node_heartbeat = $self->{options}{node_heartbeat} || 60;
$self->schedule_event(
method => "send_node_status",
time => time()+5, # immediately ...
interval => $node_heartbeat, # and every X seconds hereafter
);
+ my $node_alarm_interval = $self->{options}{node_alarm_interval} || 5;
+ $self->schedule_event(
+ method => "alarm_noop",
+ #time => time()+5, # immediately ...
+ interval => $node_alarm_interval, # 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");
+ $self->log({level=>2},"Stopping Cluster Node\n") if $self->{options}{poe_trace};
&App::sub_exit() if ($App::trace);
}
@@ -205,6 +211,12 @@
&App::sub_exit() if ($App::trace);
}
+sub alarm_noop {
+ &App::sub_entry if ($App::trace);
+ &App::sub_exit() if ($App::trace);
+ return();
+}
+
sub state {
&App::sub_entry if ($App::trace);
my ($self) = @_;
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 28 12:01:45 2007
@@ -40,13 +40,13 @@
$self->{hostname} = $host;
$host =~ s/\..*//; # get rid of fully qualified domain name
$self->{host} = $host;
- $options->{port} ||= 8080;
- $self->{port} = $options->{port};
- $options->{http_port} ||= $options->{port}+1;
- $self->{poe_kernel_name} = "poe_$self->{host}_$self->{port}";
- $self->{poe_kernel_http_name} = $self->{poe_kernel_name} . "_httpd";
- $self->{poe_session_name} = "poe_session";
- $self->{poe_kernel} = $poe_kernel;
+ $options->{port} ||= 8080;
+ $self->{port} = $options->{port};
+ $options->{http_port} ||= $options->{port}+1;
+ $self->{poe_kernel_name} = "poe_$self->{host}_$self->{port}";
+ $self->{poe_kernel_http_name} = $self->{poe_kernel_name} . "_httpd";
+ $self->{poe_session_name} = "poe_session";
+ $self->{poe_kernel} = $poe_kernel;
$self->{num_procs} = 0;
$self->{max_procs} = $self->{options}{"app.context.max_procs"} || 10;
@@ -116,7 +116,7 @@
port => $self->{port},
name => $self->{poe_kernel_name},
);
- $self->log({level=>2},"Listening for Inter-Kernel Communications on $self->{host}:$self->{port}\n");
+ $self->log({level=>2},"Listening for Inter-Kernel Communications on $self->{host}:$self->{port}\n") if $self->{options}{poe_ikc_debug};
POE::Component::IKC::Responder->spawn();
my $session_name = $self->{poe_session_name};
@@ -129,7 +129,7 @@
{ 'DIR' => '.*', 'SESSION' => $session_name, 'EVENT' => 'poe_http_server_state', },
],
);
- $self->log({level=>2},"Listening for HTTP Requests on $self->{host}:$self->{options}{http_port}\n");
+ $self->log({level=>2},"Listening for HTTP Requests on $self->{host}:$self->{options}{http_port}\n") if $self->{options}{poe_http_debug};
&App::sub_exit() if ($App::trace);
}
@@ -415,6 +415,9 @@
sub send_async_event_now {
&App::sub_entry if ($App::trace);
my ($self, $event, $callback_event) = @_;
+
+ $self->log({level=>2}, "Server: send_async_event_now() $event->{name}.$event->{method} : $event->{destination}\n");
+
if ($event->{destination} eq "in_process") {
my $event_token = $self->send_async_event_in_process($event, $callback_event);
}
@@ -677,14 +680,14 @@
my (@args);
@args = @$args if (ref($args) eq "ARRAY");
@args = ($args) if (!ref($args));
- $self->log({level=>2},"POE: _default - WARNING: Entered an unhandled state ($state) with args (@args)\n");
+ $self->log({level=>2},"POE: _default - WARNING: Entered an unhandled state ($state) with args (@args)\n") if $self->{options}{poe_trace};
&App::sub_exit() if ($App::trace);
}
sub _start {
&App::sub_entry if ($App::trace);
my ( $self, $kernel, $heap ) = @_[ OBJECT, KERNEL, HEAP ];
- $self->log({level=>2},"POE: _start\n");
+ $self->log({level=>2},"POE: _start\n") if $self->{options}{poe_trace};
my $name = $self->{poe_session_name};
$kernel->alias_set($name);
@@ -714,7 +717,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: _stop\n");
+ $self->log({level=>2},"POE: _stop\n") if $self->{options}{poe_trace};
#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);
}
@@ -722,7 +725,7 @@
sub ikc_register {
&App::sub_entry if ($App::trace);
my ($self, $kernel, $session_name) = @_[OBJECT, KERNEL, ARG1];
- $self->log({level=>2},"POE: ikc_register ($session_name)\n");
+ $self->log({level=>2},"POE: ikc_register ($session_name)\n") if $self->{options}{poe_ikc_debug};
my ($retval);
&App::sub_exit($retval) if ($App::trace);
return($retval);
@@ -731,14 +734,14 @@
sub ikc_unregister {
&App::sub_entry if ($App::trace);
my ($self, $kernel, $session_name) = @_[OBJECT, KERNEL, ARG1];
- $self->log({level=>2},"POE: ikc_unregister ($session_name)\n");
+ $self->log({level=>2},"POE: ikc_unregister ($session_name)\n") if $self->{options}{poe_ikc_debug};
&App::sub_exit() if ($App::trace);
}
sub ikc_shutdown {
&App::sub_entry if ($App::trace);
my ($self, $kernel, $arg0, $arg1, $arg2, $arg3) = @_[OBJECT, KERNEL, ARG0, ARG1, ARG2, ARG3];
- $self->log({level=>2},"POE: ikc_shutdown args=($arg0, $arg1, $arg2, $arg3)\n");
+ $self->log({level=>2},"POE: ikc_shutdown args=($arg0, $arg1, $arg2, $arg3)\n") if $self->{options}{poe_ikc_debug};
&App::sub_exit() if ($App::trace);
return;
}
@@ -746,7 +749,7 @@
sub poe_sigterm {
&App::sub_entry if ($App::trace);
my ( $self, $kernel, $heap, $signame ) = @_[ OBJECT, KERNEL, HEAP, ARG0 ];
- $self->log({level=>2},"POE: poe_sigterm (Caught signal $signame. quitting.)\n");
+ $self->log({level=>2},"POE: poe_sigterm (Caught signal $signame. quitting.)\n") if $self->{options}{poe_trace};
# How do I shut down the POE kernel now and exit?
# I think I need to shut down the last session and the kernel will exit.
@@ -760,7 +763,7 @@
sub poe_sigignore {
&App::sub_entry if ($App::trace);
my ( $self, $kernel, $heap, $signame ) = @_[ OBJECT, KERNEL, HEAP, ARG0 ];
- $self->log({level=>2},"POE: poe_sigignore (Caught signal $signame. quitting.)\n");
+ $self->log({level=>2},"POE: poe_sigignore (Caught signal $signame. quitting.)\n") if $self->{options}{poe_trace};
&App::sub_exit() if ($App::trace);
}
@@ -770,7 +773,7 @@
#print STDERR "NOTICE: STATE (poe_sigchld) invoked with ($pid, $status) args\n";
my $exitval = $status >> 8;
my $sig = $status & 255;
- $self->log({level=>2},"POE: poe_sigchld (Child $pid finished [exitval=$exitval,sig=$sig])\n");
+ $self->log({level=>2},"POE: poe_sigchld (Child $pid finished [exitval=$exitval,sig=$sig])\n") if $self->{options}{poe_trace};
$self->finish_pid($pid, $exitval, $sig);
$kernel->yield("poe_dispatch_pending_async_events");
&App::sub_exit() if ($App::trace);
@@ -779,7 +782,17 @@
sub poe_alarm {
&App::sub_entry if ($App::trace);
my ( $self, $kernel, $heap ) = @_[ OBJECT, KERNEL, HEAP ];
- $self->log({level=>2},"POE: poe_alarm (Dispatching pending events and queueing scheduled events)\n");
+ $self->log({level=>2},"POE: poe_alarm (Dispatching pending events and queueing scheduled events)\n") if $self->{options}{poe_trace};
+ my $main_service = $self->{main_service};
+ if ($self->{options}{poe_trace}) {
+ for my $async_event (@{$self->{pending_async_events}}) {
+ if ($main_service && $main_service->can("format_async_event")) {
+ my ($event, $callback_event) = @$async_event;
+ my $str = $main_service->format_async_event($event, $callback_event);
+ $self->log({level=>2},"POE: poe_alarm : pending_async_events : $str\n");
+ }
+ }
+ }
$kernel->yield("poe_dispatch_pending_async_events");
my $time = time();
my (@events);
@@ -805,7 +818,7 @@
sub poe_shutdown {
&App::sub_entry if ($App::trace);
my ( $self, $kernel, $session, $heap ) = @_[ OBJECT, KERNEL, SESSION, HEAP ];
- $self->log({level=>2},"POE: poe_shutdown\n");
+ $self->log({level=>2},"POE: poe_shutdown\n") if $self->{options}{poe_trace};
# delete all wheels.
delete $heap->{wheel};
@@ -831,7 +844,7 @@
sub poe_dispatch_pending_async_events {
&App::sub_entry if ($App::trace);
my ( $self, $kernel, $heap ) = @_[ OBJECT, KERNEL, HEAP ];
- $self->log({level=>2},"POE: poe_dispatch_pending_async_events\n");
+ $self->log({level=>2},"POE: poe_dispatch_pending_async_events\n") if $self->{options}{poe_trace};
my $events_occurred = $self->dispatch_pending_async_events();
$kernel->yield("poe_dispatch_pending_async_events") if ($events_occurred > 0);
&App::sub_exit() if ($App::trace);
@@ -840,9 +853,9 @@
sub poe_event_loop_extension {
&App::sub_entry if ($App::trace);
my ( $self, $kernel, $heap ) = @_[ OBJECT, KERNEL, HEAP ];
- $self->log({level=>2},"POE: poe_event_loop_extension\n");
+ $self->log({level=>2},"POE: poe_event_loop_extension\n") if $self->{options}{poe_trace};
my $event_loop_extensions = $self->{event_loop_extensions};
- #$self->log({level=>2},"Event Loop extension ($event_loop_extensions: #=" . ($#$event_loop_extensions+1) . ").\n");
+ #$self->log({level=>2},"Event Loop extension ($event_loop_extensions: #=" . ($#$event_loop_extensions+1) . ").\n") if $self->{options}{poe_trace};
my $async_event_added = 0;
if ($event_loop_extensions && $#$event_loop_extensions > -1) {
my ($extension, $obj, $method, $args, $event_executed);
@@ -852,7 +865,7 @@
$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");
+ # $self->log({level=>2},"Event Loop extension: ${obj}->${method}(@$args) = $event_executed\n") if $self->{options}{poe_trace};
#}
}
}
@@ -879,7 +892,7 @@
else {
$event_str = "$event->{method}($args_str)";
}
- $self->log({level=>2},"POE: poe_run_event ($event_str)\n");
+ $self->log({level=>2},"POE: poe_run_event ($event_str)\n") if $self->{options}{poe_trace};
$self->send_event($event);
&App::sub_exit() if ($App::trace);
}
@@ -889,7 +902,7 @@
&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");
+ $self->log({level=>2},"POE: poe_enqueue_async_event ($event->{name}.$event->{method})\n") if $self->{options}{poe_trace};
my $runtime_event_token = $self->send_async_event($event, { method => "async_event_finished", args => [ $sender, $event, $callback_event ], });
$event->{event_token} = $runtime_event_token;
@@ -903,7 +916,7 @@
&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->log({level=>2},"POE: poe_enqueue_async_event_finished ($async_event->[0]{name}.$async_event->[0]{method} => $runtime_event_token)\n") if $self->{options}{poe_trace};
$self->{running_async_event}{$runtime_event_token} = $async_event;
&App::sub_exit() if ($App::trace);
}
@@ -936,7 +949,7 @@
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");
+ $self->log({level=>2},"POE: poe_remote_async_event_finished ($event->{name}.$event->{method} => $runtime_event_token)\n") if $self->{options}{poe_trace};
delete $self->{running_async_event}{$runtime_event_token};
my $destination = $event->{destination} || "local";
@@ -949,7 +962,7 @@
}
}
else {
- $self->log({level=>2},"POE: poe_remote_async_event_finished ($runtime_event_token)\n");
+ $self->log({level=>2},"POE: poe_remote_async_event_finished ($runtime_event_token)\n") if $self->{options}{poe_trace};
}
&App::sub_exit() if ($App::trace);
@@ -958,7 +971,7 @@
sub poe_server_state {
&App::sub_entry if ($App::trace);
my ( $self, $kernel, $heap ) = @_[ OBJECT, KERNEL, HEAP ];
- $self->log({level=>2},"POE: poe_server_state\n");
+ #$self->log({level=>2},"POE: poe_server_state\n") if $self->{options}{poe_trace};
my $server_state = $self->state();
&App::sub_exit($server_state) if ($App::trace);
return $server_state;
@@ -967,7 +980,7 @@
sub poe_http_server_state {
&App::sub_entry if ($App::trace);
my ( $self, $kernel, $heap, $request, $response ) = @_[ OBJECT, KERNEL, HEAP, ARG0, ARG1 ];
- $self->log({level=>2},"POE: poe_http_server_state\n");
+ $self->log({level=>2},"POE: poe_http_server_state\n") if $self->{options}{poe_http_debug};
my $server_state = $kernel->call( $self->{poe_session_name}, 'poe_server_state' );
# Build the response.
@@ -984,7 +997,7 @@
sub poe_http_test_run {
&App::sub_entry if ($App::trace);
my ( $self, $kernel, $heap, $request, $response ) = @_[ OBJECT, KERNEL, HEAP, ARG0, ARG1 ];
- $self->log({level=>2},"POE: poe_http_test_run\n");
+ $self->log({level=>2},"POE: poe_http_test_run\n") if $self->{options}{poe_http_debug};
my $event = {
service_type => "SessionObject",
name => "mvworkd",
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.