Changeset 10566

Timestamp:

02/07/10 03:24:00 (7 months 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?