Changeset 5824

Timestamp:

07/04/09 07:51:16 (3 years ago)

Author:

submersible_toaster

Message:

merge branches/Padre-Taskmanager , introducing Padre::Service for persistent Padre::Tasks

Location:

trunk/Padre

Files:

• lib/Padre/Service.pm (modified) (9 diffs)
• lib/Padre/Task.pm (modified) (2 diffs)
• lib/Padre/TaskManager.pm (modified) (15 diffs)
• lib/Padre/Wx/Main.pm (modified) (3 diffs)
• t/85-task-manager.t (modified) (2 diffs)
• t/86-service.t (copied) (copied from branches/Padre-Taskmanager/t/86-service.t)

trunk/Padre/lib/Padre/Service.pm

@@ -2 +2 @@
 use strict;
 use warnings;
-
+use Carp qw( croak );
+
+use threads;
+use threads::shared;
+
+use Padre::Wx     ();
+use Padre::Task   ();
+use Thread::Queue ();
 our @ISA = 'Padre::Task';
 
@@ -11 +18 @@
 =head1 NAME
 
-Padre::Service - API for non trivial Padre::Task
+Padre::Service - persistent Padre::Task API
 
 =head2 SYNOPSIS
@@ -17 +24 @@
   # Create your service, default implementation warns to output
   #  sleeps 1 second and loops over.
-  my $service = Padre::Service->new( 
-    main_thread_only => \&show_my_dialog,
+  my $service = Padre::Service->new();
+  Wx::Event::EVT_COMMAND(
+    $main , -1 , $service->event ,
+    \&receive_data_from_service
   );
   $service->schedule;
+  $service->
+  
   
   # Later
@@ -36 +47 @@
 
 C<service_loop> should not block forever. If there is no work for the service to do
-then return immediately, allowing the C<<Task->run>> loop to
+then return immediately, allowing the C<<Task->run>> loop to continue.
 
   package Padre::Service::HTTPD
@@ -52 +63 @@
   
   sub terminate { # Stop everything, brutally }
-  
-  sub service_results { # Returned as the task return to Padre, }
   
 =head1 METHODS
@@ -66 +75 @@
 
 =cut
+{
+my $running = 0;
+sub running { $running };
+
+sub stop { $running = 0 };
+sub start{ $running = 1 }; #??
 
 sub run {
-    my ( $self, $queue ) = @_;
-
-    my $tid   = threads->tid;
-    my $event = $self->{service_event};
-    $self->post_event( $event, "$tid;ALIVE" );
-    my $running = 1;
-
+    croak "Already running!" if $running;
+    
+    my ($self) = @_;
+    my $queue = $self->queue;
+    Padre::Util::debug( "Running queue $queue" );
+    my $tid = threads->tid;
+    my $event  = $self->event;
+    
+    
+    # Now we're in the worker thread, start our service
+    # and begin the select orbit around the manager's queue
+    #  , the service_loop and throwing ->event back at the main thread
+    $self->start;
+    $running = 1;
+    $self->post_event(  $event , "ALIVE" );
     while ($running) {
+        # Let the service provider have first chance.
+        #   and if nothing is waiting in the queue - tight loop.
         $self->service_loop;
         next unless $queue->pending;
 
-        my $incoming = $queue->peek(0);
-
-        # Peek at the queue - for something addressed to us
-        # YUK - how about a dedicated queue per service
-        # peek and poke went out in the dark ages didn't they
-        if ( $incoming =~ m{$tid;} ) {
-            my $instruction = $queue->dequeue;
-            my ($command) = $instruction =~ m{^$tid;(HANGUP|TERMINATE)};
+        my $command = $queue->dequeue;
+        Padre::Util::debug( "Service dequeued input" );
+        
+        # Respond to HANGUP TERMINATE and PING - 
+        if ( ref($command) ) {
+            $self->service_loop($command);
+        }
+        
+        # Or possibly a signal from the main thread
+        else {
+            Padre::Util::debug( "Caught command signal '$command'" );
             if ( $command eq 'HANGUP' ) {
-                $self->hangup;
-                $running = 0;
+                $self->hangup( \$running );
             } elsif ( $command eq 'TERMINATE' ) {
-                $self->terminate;
-                $running = 0;
+                $self->terminate( \$running );
             } elsif ( $command eq 'PING' ) {
-                $self->post_event( $event, "$tid;ALIVE" );
+                $self->post_event( $event, "ALIVE" );
             } else {
-                $self->task_warn( "$self : Unrecognised command event $command" );
+                Padre::Util::debug("Service does not recognise '$command' signal");
             }
-
         }
-
     }
+    
+    # Loop broken - cleanup
+    #$self->shutdown;
     return;
 }
+
+}
+
+=head2 start
+
+consider start the background_thread analog of C<prepare> and will be called
+in the service thread immediatly prior to the service loop starting.
+
+
+=cut
+
 
 =head2 hangup
@@ -113 +151 @@
 
 sub hangup {
-    my ($self) = @_;
-
+    my ($self,$running) = @_;
+    $$running = 0;
 }
 
@@ -126 +164 @@
 
 sub terminate {
-    my ($self) = @_;
-
+    my ($self,$running) = @_;
+    $$running = 0;
 }
 
@@ -139 +177 @@
 
 {
-    my $i = 0;
+    
 
     sub service_loop {
-        my ($self) = @_;
+        my ($self,$incoming) = @_;
+        $self->{iterator} = 0
+            unless exists $self->{iterator};
         my $tid = threads->tid;
-        $self->task_print("Service ($tid) Looped [$i]\n");
-        $i++;
+        $self->task_print('ok - entered service loop')
+            || print "ok - entered service loop\n";
+        
+        $self->task_print("# Service ($tid) Looped $self->{iterator}\n");
+        if (defined $incoming) {
+            $self->task_print("ok - got incoming service data '$incoming'");
+        }
+        # Tell the main thread some progress.
+        $self->post_event( $self->event , "$self->{iterator}" );
+        
+        $self->{iterator}++;
+        $self->tell('HANGUP') if $self->{iterator} > 10;
         sleep 1;
     }
 }
+
+=head2 event
+
+Accessor for this service's instance event, in the running service
+data may be posted to this event and the Wx subscribers will be notified
+ 
+=cut
+
+{
+our %ServiceEvents : shared = ();
+  sub event {
+    my $self = shift;
+    if ( exists $ServiceEvents{$self->{__service_refid}} ) {
+        return $ServiceEvents{ $self->{__service_refid} } ;
+    }
+    else {
+        croak "Cannot lookup shared event for $self";
+    }
+  }
+ 
+
+my %Queues : shared;
+ sub prepare {
+    my $self = shift;
+    my $queue : shared;
+    $queue = new Thread::Queue;
+    $Queues{"$self"} = $queue;
+    $self->{_refid} = "$self";
+    $self->SUPER::prepare(@_);
+  }
+ 
+=head2 queue
+
+accessor for the shared queue the service thread is polling for input.
+Calling C<enqueue> on reference sends data to the service thread. Storable
+serialization rules apply. See also L<"event"> for receiving data from 
+the service thread
+ 
+=cut
+ 
+ sub queue { 
+    my $self = shift;
+    if ( exists $self->{_refid} 
+        && exists $Queues{$self->{_refid}} ) {
+        return $Queues{$self->{_refid}} ;
+    }
+    elsif  ( exists $Queues{"$self"} ) {
+        return $Queues{"$self"};
+    }
+    else { croak "No such service queue "; }
+ 
+ }
+ 
+
+  sub serialize {
+    my $self = shift;
+  # croak "Serialized!!";
+    my $service_refid = "$self";
+    $self->{__service_refid} = $service_refid;
+    
+    # Wait until the last moment before we declare 
+    # the event
+    my $service_event : shared = Wx::NewEventType;
+    $ServiceEvents{$service_refid} = $service_event;
+    
+#   my $wx_attach;
+#   if ( exists $self->{_main_thread_only}
+#        && 
+#        _INSTANCE( $self->{_main_thread_only}, 'Wx::Object' )
+#       )
+#   {
+#       $wx_attach = $self->{_main_thread_only};
+#   }
+#   else {  $wx_attach = Padre->ide->wx->main };
+
+
+#   if (!exists $self->{__events_init}
+#       and !defined $self->{__events_init} ) 
+#   {
+#       $self->{__events_init} =
+#           Wx::Event::EVT_COMMAND(
+#           $wx_attach, -1,
+#           $service_event,
+#           sub{ $self->receive(@_) } ,
+#       );
+#   }
+
+
+    # FILO
+    my $payload = $self->SUPER::serialize(@_);
+    
+    return $payload;
+  }
+  
+  sub deserialize_hook {
+    my $self = shift;
+    # FILO
+    # Shutdown the queue and event ?;
+  }
+
+}
+
+  sub shutdown {
+    my $self = shift;
+    Padre::Util::debug( "shutdown - $self");
+    my $queue =$self->queue;
+    $queue->enqueue( 'HANGUP' );
+  }
+  
+  
+  sub cleanup {
+    my $self = shift;
+    Padre::Util::debug( "cleanup - $self" );
+  }
+
+=head2 tell
+
+Accepts a reference as it's argument, this is serialized and sent to
+the service thread
+
+=cut  
+  
+  ## MAIN
+  sub tell {
+    my ($self,$ref) = @_;
+    my $queue = $self->queue;
+    $queue->enqueue($ref);
+  }
+
 
 =head1 COPYRIGHT

trunk/Padre/lib/Padre/Task.pm

@@ -326 +326 @@
 
         my $obj = bless $padretask => $userclass;
+        # Xtra evil , let a subclass ducktype a hook here
+        $obj->deserialize_hook if $obj->can('deserialize_hook');
+        
         return $obj;
     }
@@ -506 +509 @@
 =cut
 
+
+use Carp qw( cluck );
+
 sub post_event {
-    my @stuff = @_;
+    my ($self,$eventid,$data) = @_;
     @_ = ();
+    cluck 'eventid is not defined' unless defined $eventid;
+    cluck "eventid[$eventid] , no data to post" 
+        unless ( defined $data and length($data) );
+        
     Wx::PostEvent(
         $Padre::TaskManager::_main,
-        Wx::PlThreadEvent->new( -1, $stuff[1], $stuff[2] ),
+        Wx::PlThreadEvent->new( -1, $eventid, $data ),
     );
     return ();

trunk/Padre/lib/Padre/TaskManager.pm

@@ -85 +85 @@
 use Class::XSAccessor getters => {
     task_queue     => 'task_queue',
-    service_queue  => 'service_queue',
     reap_interval  => 'reap_interval',
     use_threads    => 'use_threads',
@@ -112 +111 @@
 our $SINGLETON;
 
-# This is set in the worker threads only!
+# This is set in the worker threads only! 
 our $_main;
 
@@ -121 +120 @@
 
     my $self = $SINGLETON = bless {
-        min_no_workers => 2,
-        max_no_workers => 6,
+        min_no_workers => 2, # there were config settings for 
+        max_no_workers => 6, #  these long ago?
         use_threads    => 1,       # can be explicitly disabled
         reap_interval  => 15000,
@@ -140 +139 @@
 
     $self->{task_queue}    = Thread::Queue->new;
-    $self->{service_queue} = Thread::Queue->new;
+
 
     # Set up a regular action for reaping dead workers
@@ -208 +207 @@
         or die "Invalid task scheduled!";    # TODO: grace
 
+    if ( _INSTANCE( $task , 'Padre::Service' ) ) {
+        $self->{running_services}{$task} = $task;
+    }
     # Cleanup old threads and refill the pool
     $self->reap();
@@ -217 +219 @@
     }
 
+
     my $string;
     $task->serialize( \$string );
+    
     if ( $self->use_threads ) {
         require Time::HiRes;
@@ -289 +293 @@
 
     @_ = ();    # avoid "Scalars leaked"
-    my $worker = threads->create( { 'exit' => 'thread_only' }, \&worker_loop, $main, $self->task_queue );
+    my $worker = threads->create( 
+        { 'exit' => 'thread_only' }, \&worker_loop,
+         $main, $self->task_queue 
+    );
     push @{ $self->{workers} }, $worker;
 }
@@ -351 +358 @@
             unless $queue->pending()
                 and not ref( $queue->peek(0) );
-
-        # We don't actually need to wait for the soon-to-be-joinable threads
-        # since reap should be called regularly.
-        #while (threads->list(threads::running) >= $target_n_threads) {
-        #  $_->join for threads->list(threads::joinable);
-        #}
     }
 
@@ -389 +390 @@
 =head2 cleanup
 
-Stops all worker threads. Called on editor shutdown.
+Shutdown all services with a HANGUP, then stop all worker threads.
+Called on editor shutdown.
 
 =cut
@@ -396 +398 @@
     my $self = shift;
     return if not $self->use_threads;
-
+    
+    # Send all services a HANGUP , they will (hopefully)
+    # catch this and break the run loop, returning below as
+    # regular tasks. :|
+    Padre::Util::debug( 'Tell services to hangup' );
+    $self->shutdown_services;
+    
     # the nice way:
+    Padre::Util::debug( 'Tell all tasks to stop' );
     my @workers = $self->workers;
     $self->task_queue->insert( 0, ("STOP") x scalar(@workers) );
@@ -403 +412 @@
         $_->join for threads->list(threads::joinable);
     }
-    $_->join for threads->list(threads::joinable);
-
+    foreach my $thread ( threads->list(threads::joinable) ) {
+        Padre::Util::debug( 'Joining thread ' . $thread->tid );
+        $thread->join;
+    }
+    
     # didn't work the nice way?
     while ( threads->list(threads::running) >= 1 ) {
+        Padre::Util::debug( 'Killing thread ' . $_->tid );
         $_->detach(), $_->kill() for threads->list(threads::running);
     }
@@ -452 +465 @@
 =pod
 
-=head2 shutdown
-
-Gracefully shutdown the service queue by instructing them to hangup themselves
+=head2 shutdown_services
+
+Gracefully shutdown the services by instructing them to hangup themselves
 and return via the usual Task mechanism.
 
 =cut
 
-sub shutdown {
-    my $self = shift;
-
-    while ( my ( $type, $tasks ) = each %{ $self->{running_tasks} } ) {
-        next unless Params::Util::_CLASSISA( $type, 'Padre::Service' );
-        foreach my $threadid ( keys %$tasks ) {
-            Padre::Util::debug("Hangup $type in $threadid !");
-            $SINGLETON->service_queue->enqueue(
-                "$threadid;HANGUP",
-            );
-        }
-
-    }
-
+## ERM FIXME where are is the {running_services} populated then eh?
+sub shutdown_services {
+    my $self = shift;
+    Padre::Util::debug( 'Shutdown services' );
+    
+    while ( my  ($sid,$service) = each %{ $self->{running_services} } ) {
+        Padre::Util::debug( "Hangup service $sid!" );
+        $service->shutdown;
+    }
 }
 
@@ -505 +513 @@
     my ( $main, $event ) = @_; @_ = ();    # hack to avoid "Scalars leaked"
     my $frozen = $event->GetData;
+    
+    # FIXME - can we know the _real_ class so the an extender 
+    #  may hook de/serialize
     my $task   = Padre::Task->deserialize( \$frozen );
 
@@ -636 +647 @@
 
         # RUN
-        if ( $task->isa('Padre::Service') ) {
-            $task->run( $SINGLETON->service_queue );
-        } else {
-            $task->run;
-        }
+        $task->run;
 
         # FREEZE THE PROCESS AND PASS IT BACK
@@ -667 +674 @@
 Wx MainLoop isn't reached for processing finish events.
 
+Polling services 'aliveness' in a useful way , something a Wx::Taskmanager
+might like to display. Ability to selectivly kill tasks/services
+
 =head1 SEE ALSO
 

trunk/Padre/lib/Padre/Wx/Main.pm

@@ -2132 +2132 @@
     Padre::Util::debug("Files saved (or not), hiding window");
 
+
     # Immediately hide the window so that the user
     # perceives the application as closing faster.
@@ -2138 +2139 @@
     $self->Show(0);
 
-    Padre::Util::debug('Try to shutdown services');
-    $self->ide->task_manager->shutdown;
-
-    # Stop all Task Manager's worker threads
-    $self->ide->task_manager->cleanup;
-
-    Padre::Util::debug("Finished TaskManager's cleanup");
+
+
+
 
     # Save the window geometry
@@ -2180 +2177 @@
     $ide->save_config;
     $event->Skip;
+
+    Padre::Util::debug("Tell TaskManager to cleanup");
+    # Stop all Task Manager's worker threads
+    $self->ide->task_manager->cleanup;
 
     Padre::Util::debug("Closing Padre");

trunk/Padre/t/85-task-manager.t

@@ -17 +17 @@
 }
 
-plan( tests => 17 );
+plan( tests => 18 );
 
 # need to load these before padre!
@@ -30 +30 @@
 use_ok('Padre::TaskManager');
 use_ok('Padre::Task');
+use_ok('Padre::Service');
 require t::lib::Padre::Task::Test;