Changeset 10566

Timestamp:

02/07/10 03:24:00 (2 years ago)

Author:

tsee

Message:

got experimental slave driving to work

Location:

trunk/Padre/lib/Padre

Files:

• SlaveDriver.pm (modified) (8 diffs)
• Task.pm (modified) (2 diffs)
• TaskManager.pm (modified) (7 diffs)

trunk/Padre/lib/Padre/SlaveDriver.pm

@@ -47 +47 @@
 use Thread::Queue 2.11;
 
+# This event is triggered by the worker thread main loop after
+# finishing a task.
+our $TASK_DONE_EVENT : shared;
+BEGIN { $TASK_DONE_EVENT = Wx::NewEventType; }
+
+# This event is triggered by the worker thread main loop before
+# running a task.
+our $TASK_START_EVENT : shared;
+BEGIN { $TASK_START_EVENT = Wx::NewEventType; }
+
 =head3 C<new>
 
@@ -55 +65 @@
 =cut
 
-{
+SCOPE: {
     my $SlaveDriver;
     sub new {
         my $class = shift;
         return $SlaveDriver if defined $SlaveDriver;
-        $SlaveDriver = bless {} => $class;
         @_ = ();
-        $SlaveDriver->{cmd_queue}  = Thread::Queue->new();
-        $SlaveDriver->{tid_queue}  = Thread::Queue->new();
+        $SlaveDriver = bless {
+            cmd_queue  => Thread::Queue->new(),
+            tid_queue  => Thread::Queue->new(),
+            task_queue => Thread::Queue->new(),
+        } => $class;
         $SlaveDriver->{master} = threads->create(
             \&_slave_driver_loop,
@@ -73 +85 @@
 
     END {
-        if (defined $SlaveDriver and defined $SlaveDriver->{master} and defined $SlaveDriver->{cmd_queue}) {
-            $SlaveDriver->{cmd_queue}->enqueue('STOP');
-            $SlaveDriver->{master}->join();
-        }
-    }
-
+        $SlaveDriver->cleanup(), undef $SlaveDriver if defined $SlaveDriver;
+    }
 }
 
@@ -95 +103 @@
     my $task_manager = shift;
     require Storable;
-    $self->{cmd_queue}->enqueue(Storable::freeze([$main, $task_manager->task_queue]));
+    $self->{cmd_queue}->enqueue(Storable::freeze([$task_manager->task_queue]));
     my $tid = $self->{tid_queue}->dequeue();
     return threads->object($tid);
+}
+
+
+=head3 task_queue
+
+Returns the task queue (C<Thread::Queue> object) for use by the
+L<Padre::TaskManager> for passing processing tasks to the worker
+threads.
+
+This queue is instantiated by the slave driver because it needs to be available
+early for passing to the master thread.
+
+=cut
+
+sub task_queue {
+    my $self = shift;
+    return $self->{task_queue};
+}
+
+=head3 cleanup
+
+Reaps the master thread. Will be called by the TaskManager on shutdown and
+on global destruction.
+
+=cut
+
+sub cleanup {
+    my $self = shift;
+    if (defined $self->{master} and defined $self->{cmd_queue}) {
+        $self->{cmd_queue}->enqueue('STOP');
+        require Time::HiRes;
+        Time::HiRes::usleep(5000); # 5 milli-sec
+        if ($self->{master}->is_joinable) {
+            $self->{master}->join;
+        }
+    }
+    # TaskManager does handle thread *killing*
+}
+
+sub DESTROY {
+    my $self = shift;
+    $self->cleanup();
 }
 
@@ -103 +153 @@
 # Worker thread main loop
 sub _worker_loop {
-    my ( $main, $queue ) = @_; @_ = (); # hack to avoid "Scalars leaked"
+    my ( $queue ) = @_; @_ = (); # hack to avoid "Scalars leaked"
     require Storable;
     require Padre::TaskManager;
 
     # Set the thread-specific main-window pointer
+    my $main = Wx::wxTheApp->GetTopWindow();
     $Padre::TaskManager::_main = $main;
 
@@ -123 +174 @@
 
         my $thread_start_event =
-            Wx::PlThreadEvent->new( -1, $Padre::TaskManager::TASK_START_EVENT, $task->{__thread_id} . ";" . ref($task) );
+            Wx::PlThreadEvent->new( -1, $TASK_START_EVENT, $task->{__thread_id} . ";" . ref($task) );
         Wx::PostEvent( $main, $thread_start_event );
 
@@ -133 +184 @@
         $task->serialize( \$frozen_task );
 
-        my $thread_done_event = Wx::PlThreadEvent->new( -1, $Padre::TaskManager::TASK_DONE_EVENT, $frozen_task );
+        my $thread_done_event = Wx::PlThreadEvent->new( -1, $TASK_DONE_EVENT, $frozen_task );
         Wx::PostEvent( $main, $thread_done_event );
 
@@ -150 +201 @@
         last if $args eq 'STOP';
         
-        my $worker_thread = threads->create(\&_worker_loop, @{Storable::thaw($args)});
+        my $task_queue = Padre::SlaveDriver->new->task_queue;
+        my $worker_thread = threads->create(\&_worker_loop, $task_queue);
         my $tid = $worker_thread->tid();
         $outqueue->enqueue($tid);

trunk/Padre/lib/Padre/Task.pm

@@ -133 +133 @@
 
 # set up the stdout/stderr printing events
-our $STDOUT_EVENT : shared = Wx::NewEventType();
-our $STDERR_EVENT : shared = Wx::NewEventType();
+our $STDOUT_EVENT : shared;
+BEGIN { $STDOUT_EVENT = Wx::NewEventType(); }
+our $STDERR_EVENT : shared;
+BEGIN { $STDERR_EVENT = Wx::NewEventType(); }
 
 =pod
@@ -486 +488 @@
 like this:
 
-  our $FUN_EVENT_TYPE =  : shared = Wx::NewEventType();
+  our $FUN_EVENT_TYPE : shared;
+  BEGIN { $FUN_EVENT_TYPE = Wx::NewEventType(); }
 
 Then you have to setup the event handler (for example in the

trunk/Padre/lib/Padre/TaskManager.pm

@@ -93 +93 @@
 };
 
-# This event is triggered by the worker thread main loop after
-# finishing a task.
-our $TASK_DONE_EVENT : shared = Wx::NewEventType;
-
-# This event is triggered by the worker thread main loop before
-# running a task.
-our $TASK_START_EVENT : shared = Wx::NewEventType;
-
 # This event is triggered by a worker thread DURING ->run to incrementally
 # communicate to the main thread over the life of a service.
-our $SERVICE_POLL_EVENT : shared = Wx::NewEventType;
+our $SERVICE_POLL_EVENT : shared;
+BEGIN { $SERVICE_POLL_EVENT = Wx::NewEventType; }
 
 # remember whether the event handlers were initialized...
@@ -129 +122 @@
         @_,
         workers       => [],
-        task_queue    => undef,
+                # grab a copy of the task_queue that's now handled by the slave driver
+        task_queue    => Padre::SlaveDriver->new()->task_queue,
         running_tasks => {},
     }, $class;
@@ -141 +135 @@
     _init_events($main);
 
-    $self->{task_queue} = Thread::Queue->new;
+        # To be removed: Old task queue instantiation => Padre::SlaveDriver
+    #$self->{task_queue} = Thread::Queue->new;
 
     # Set up a regular action for reaping dead workers
@@ -174 +169 @@
         Wx::Event::EVT_COMMAND(
             $main, -1,
-            $TASK_DONE_EVENT,
+            $Padre::SlaveDriver::TASK_DONE_EVENT,
             \&on_task_done_event,
         );
         Wx::Event::EVT_COMMAND(
             $main, -1,
-            $TASK_START_EVENT,
+            $Padre::SlaveDriver::TASK_START_EVENT,
             \&on_task_start_event,
         );
@@ -294 +289 @@
     return unless $self->use_threads;
 
-    @_ = (); # avoid "Scalars leaked"
+
+        # To be removed: Old worker thread cration. => Padre::SlaveDriver
+#   @_ = (); # avoid "Scalars leaked"
 #   my $worker = threads->create(
 #       { 'exit' => 'thread_only' }, \&worker_loop,
@@ -413 +410 @@
     my @workers = $self->workers;
     $self->task_queue->insert( 0, ("STOP") x scalar(@workers) );
-    while ( threads->list(threads::running) >= 1 ) {
+    while ( threads->list(threads::running) >= 2 ) {
         $_->join for threads->list(threads::joinable);
     }
@@ -420 +417 @@
         $thread->join;
     }
+
+        # cleanup master thread, too
+        Padre::SlaveDriver->new->cleanup();
 
     # didn't work the nice way?