[svn:p5ee] r9922 - p5ee/trunk/App-Context/lib/App/Context/POE

Author: spadkins
Date: Fri Sep  7 12:42:17 2007
New Revision: 9922

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:
updates to get cancel to work more correctly, X cancel still broken

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	Fri Sep  7 12:42:17 2007
@@ -76,7 +76,7 @@
     &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};
+    #$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) {
@@ -146,26 +146,30 @@
 #    $runtime_event_token = "$host-$port-$serial"; -- i.e. a plain event token on the node
 sub _abort_running_async_event {
     &App::sub_entry if ($App::trace);
-    my ($self, $runtime_event_token) = @_;
-    my $async_event = $self->{running_async_event}{$runtime_event_token};
-    my ($event, $callback_event) = @$async_event;
-    if ($runtime_event_token =~ /^[0-9]+$/) {
-        kill(9, $runtime_event_token);
-    }
-    elsif ($runtime_event_token =~ /^([^-]+)-([0-9]+)-/) {
-        my $node_host = $1;
-        my $node_port = $2;
+    my ($self, $runtime_event_token, $event, $callback_event) = @_;
+    #$self->log({level=>2}, "CC: _abort_running_async_event : runtime_event_token=[$runtime_event_token] : event=[$event] : callback_event=[$callback_event]\n");
+    #my $async_event = $self->{running_async_event}{$runtime_event_token};
+    if ($runtime_event_token && $event && $callback_event) {
+        #my ($event, $callback_event) = @$async_event;
+        if ($runtime_event_token =~ /^[0-9]+$/) {
+            kill(9, $runtime_event_token);
+        }
+        elsif ($runtime_event_token =~ /^([^-]+)-([0-9]+)-/) {
+            my $node_host = $1;
+            my $node_port = $2;
 
-        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_cancel_async_event";
+            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_cancel_async_event";
 
-        my $kernel = $self->{poe_kernel};
-        $kernel->post("IKC", "post", "poe://$remote_server_name/$remote_session_alias/$remote_session_state",
-            [ $runtime_event_token ]);
-    }
-    else {
-        $self->log("ERROR: _abort_running_async_event() $event->{name}.$event->{method} : unparseable runtime event token [$runtime_event_token]\n");
+            my $kernel = $self->{poe_kernel};
+            #$self->log({level=>2},"CC: _abort_running_async_event : calling remote cancel for $runtime_event_token\n");
+            $kernel->post("IKC", "post", "poe://$remote_server_name/$remote_session_alias/$remote_session_state",
+                [ $runtime_event_token ]);
+        }
+        else {
+            $self->log("ERROR: _abort_running_async_event() $event->{name}.$event->{method} : unparseable runtime event token [$runtime_event_token]\n");
+        }
     }
     &App::sub_exit() if ($App::trace);
 }

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	Fri Sep  7 12:42:17 2007
@@ -49,8 +49,8 @@
     die "Node must have a controller host and port defined (\$context->{options}{controller_host} and {controller_port})"
         if (!$self->{controller_host} || !$self->{controller_port});
 
-    #push(@{$self->{poe_states}}, "foo", "bar");
-    #push(@{$self->{poe_ikc_published_states}}, "more", "states");
+    push(@{$self->{poe_states}}, "poe_cancel_async_event");
+    push(@{$self->{poe_ikc_published_states}}, "poe_cancel_async_event");
 
     $self->_init_poe($options);
 
@@ -222,7 +222,7 @@
     my ($self) = @_;
 
     my $datetime = time2str("%Y-%m-%d %H:%M:%S", time());
-    my $state = "Cluster Node: $self->{host}:$self->{port}  procs[$self->{num_procs}/$self->{max_procs}:max]  async_events[$self->{num_async_events}/$self->{max_async_events}:max]\n[$datetime]\n";
+    my $state = "Cluster Node: Node: $self->{host}:$self->{port}  procs[$self->{num_procs}/$self->{max_procs}:max]  async_events[$self->{num_async_events}/$self->{max_async_events}:max]\n[$datetime]\n";
     $state .= "\n";
     $state .= $self->_state();
 
@@ -242,5 +242,52 @@
     return($state);
 }
 
+sub poe_cancel_async_event {
+    &App::sub_entry if ($App::trace);
+    my ( $self, $kernel, $heap, $arg0 ) = @_[ OBJECT, KERNEL, HEAP, ARG0 ];
+    my ($runtime_event_token) = @$arg0;
+    $self->log({level=>2},"CN : poe_cancel_async_event : runtime_event_token=[$runtime_event_token]\n");
+    my $async_event = $self->{running_async_event}{$runtime_event_token};
+
+    ### Find if running
+    for my $pid (keys %{$self->{running_async_event}}) {
+        #$self->log({level=>2},"CN : poe_cancel_async_event : running_async_event : pid=[$pid]\n");
+        #my $ae = $self->{running_async_event}{$pid};
+        #my $e = join(", ", map {sprintf("$_ = [$ae->[0]{$_}]")} keys %{$ae->[0]});
+        #my $ce = join(", ", map {sprintf("$_ = [$ae->[1]{$_}]")} keys %{$ae->[1]});
+        #$self->log({level=>2},"CN : poe_cancel_async_event : e=[$e]\n");
+        #$self->log({level=>2},"CN : poe_cancel_async_event : ce=[$ce]\n");
+
+        my $event_token = $self->{running_async_event}{$pid}[0]{event_token};
+        if ($runtime_event_token eq $event_token) {
+            $self->log({level=>2},"CN : poe_cancel_async_event : running_async_event : found event_token=[$event_token] pid=[$pid]\n");
+
+            ### Kill it
+            if ($pid =~ /^[0-9]+$/) {
+                kill(9, $pid);
+            }
+
+            ### Remove from pending
+            delete $self->{running_async_event}{$pid};
+
+            last;
+        }
+    }
+
+    ### Find if pending
+    for (my $i = 0; $i < @{$self->{pending_async_events}}; $i++) {
+        my $event_token = $self->{pending_async_events}[$i][0]{event_token};
+        $self->log({level=>2},"CN : poe_cancel_async_event : pending_async_events : event_token=[$event_token]\n");
+        if ($runtime_event_token eq $event_token) {
+            splice(@{$self->{pending_async_events}}, $i, 1);
+        }
+        #my $ae = $self->{pending_async_events}{$foo};
+        #my $e = join(", ", map {sprintf("$_ = [$ae->[0]{$_}]")} keys %{$ae->[0]});
+        #my $ce = join(", ", map {sprintf("$_ = [$ae->[1]{$_}]")} keys %{$ae->[1]});
+        #$self->log({level=>2},"CN : poe_cancel_async_event : e=[$e]\n");
+    }
+     &App::sub_exit() if ($App::trace);
+}
+
 1;
 

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	Fri Sep  7 12:42:17 2007
@@ -23,6 +23,8 @@
 use POE::Component::IKC::Server;
 use HTTP::Status qw/RC_OK/;
 use Socket qw(INADDR_ANY);
+use Data::Dumper;
+$Data::Dumper::Terse = 1;
 
 sub _init {
     &App::sub_entry if ($App::trace);
@@ -416,7 +418,7 @@
     &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");
+    #$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);
@@ -549,7 +551,8 @@
     my ($runtime_event_token_found, $async_event);
     foreach my $runtime_event_token (keys %$running_async_event) {
         $async_event = $running_async_event->{$runtime_event_token};
-        if ($async_event->[0]{event_token} eq $event_token) {
+        ### async_event = [ event, callback_event ]
+        if ($async_event->[1]{event_token} eq $event_token) {
             $runtime_event_token_found = $runtime_event_token;
             last;
         }
@@ -591,6 +594,9 @@
     my $pending_async_events = $self->{pending_async_events};
     my ($async_event);
     my $aborted = 0;
+
+    #$self->log({level=>2}, "S: abort_async_event : event_token=[$event_token] : pending_async_events: ", Dumper($pending_async_events));
+
     # first look for it in the pending list
     for (my $i = 0; $i <= $#$pending_async_events; $i++) {
         $async_event = $pending_async_events->[$i];
@@ -614,10 +620,12 @@
 sub abort_running_async_event {
     &App::sub_entry if ($App::trace);
     my ($self, $runtime_event_token) = @_;
+    #$self->log({level=>2}, "S: abort_running_async_event : runtime_event_token=[$runtime_event_token]\n");
     my $running_async_event  = $self->{running_async_event};
     my $pending_async_events = $self->{pending_async_events};
     my $async_event = $running_async_event->{$runtime_event_token};
     if ($async_event) {
+        #$self->log({level=>2}, "S: abort_running_async_event : async_event=[$async_event]\n");
         $self->{num_async_events}--;
         delete $self->{running_async_event}{$runtime_event_token};
         unshift(@$pending_async_events, $async_event);