apprentice/src/compute_thread.rs

use std::collections::binary_heap::BinaryHeap;
use std::ops::DerefMut;
use std::sync::{Arc, Mutex};
use std::sync::mpsc::{Receiver, Sender, TryRecvError};
use std::time::Duration;

use ::task::Task;
use ::task_state::TaskState;
use ::thread::Thread;
use ::thread_state::ThreadState;



///
pub struct ComputeThread
{
   ///
   state: ThreadState,

   ///
   current_task: Option<Box<Task>>,

   ///
   task_queue: Arc<Mutex<BinaryHeap<Box<Task>>>>,

   ///
   shutdown_receiver: Receiver<bool>,

   ///
   state_sender: Sender<ThreadState>,

   ///
   continue_running: bool,
}



impl ComputeThread
{
   /// 
   pub fn new(queue: Arc<Mutex<BinaryHeap<Box<Task>>>>,
              shutdown_r: Receiver<bool>,
              state_s: Sender<ThreadState>)
      -> ComputeThread
   {
      ComputeThread
      {
         state: ThreadState::Starting,
         current_task: None,
         task_queue: queue,
         shutdown_receiver: shutdown_r,
         state_sender: state_s,
         continue_running: true
      }
   }

   ///
   fn change_state(&mut self, new_state: ThreadState)
   {
      self.state = new_state;
      match self.state_sender.send(self.state)
      {
         Ok(_) =>
         {
         }

         Err(error) =>
         {
            // We lost our connection to the
            // state channel.
            warn!("{}", error);
         }
      }
   }

   ///
   fn retrieve_task(&mut self)
   {
      // There is nothing to do if this thread already
      // has a task to work on.
      if self.current_task.is_some() == true
      {
         return;
      }

      // This thread does not have a current task.
      // Get another task to work on from the Queue.
      match self.task_queue.lock()
      {
         Ok(ref mut guard) =>
         {
            self.current_task = guard.deref_mut().pop();

            match self.current_task
            {
               Some(_) =>
               {
                  debug!("Received a task.");
               }

               None =>
               {
                  debug!("No new task to process.");
               }
            }
         }

         Err(error) =>
         {
            error!("{}", error);
         }
      }
   }

   /// Queues a task to be processed.
   fn queue_task(&mut self, task: Box<Task>)
   {
      // Just add the task to the queue.
      match self.task_queue.lock()
      {
         Ok(ref mut guard) =>
         {
            guard.deref_mut().push(task);
         }

         Err(error) =>
         {
            error!("{}", error);
         }
      }
   }

   /// Check for any new thread shutdown messages.
   fn process_shutdown_messages(&mut self)
   {
      let mut check_messages: bool;

      // Loop through all the messages in the
      // receivers queue.
      check_messages = true;
      while check_messages == true
      {
         match self.shutdown_receiver.try_recv()
         {
            Ok(val) =>
            {
               // Found a message.
               self.continue_running = !val;
            }

            Err(error) =>
            {
               // Determine the kind of error we received.
               match error
               {
                  TryRecvError::Empty =>
                  {
                     // No messages to handle.
                     check_messages = false;
                     debug!("There were no shutdown messages.");
                  }

                  TryRecvError::Disconnected =>
                  {
                     // We lost our connection to the
                     // shutdown channel.
                     // TODO: Handle this poisoning correctly.
                     warn!("{}", error);
                  }
               }
            }
         }
      }
   }
}

impl Thread for ComputeThread
{
   ///
   fn process(&mut self)
   {
      let mut check_messages: bool;
      let mut task_completed: bool;

      // Run this thread until the scheduler decides
      // to shut it down.
      self.change_state(ThreadState::Idle);
      while self.continue_running == true
      {
         // No task was recently completed.
         task_completed = false;

         // Make sure that this thread has a Task
         // to currently work on.
         match self.current_task
         {
            Some(ref mut task) =>
            {
               // Process the task this thread is
               // currently working on.
               task.process();

               match task.get_state()
               {
                  TaskState::Finished =>
                  {
                     task_completed = true;
                  }

                  _ =>
                  {
                  }
               }
            }

            None =>
            {
               // Try to get a task to work on.
               self.retrieve_task();
               match self.current_task
               {
                  Some(_) =>
                  {
                     // We have a task to work on, so switch to
                     // a Processing state.
                     self.change_state(ThreadState::Processing);
                  }

                  None =>
                  {
                     // If we don't have a task to process,
                     // then we may need to switch over to
                     // an idle state.
                     match self.state
                     {
                        ThreadState::Idle =>
                        {
                           // The thread is already sitting idle.
                        }

                        _ =>
                        {
                           // There is nothing for this thread
                           // to process, so mark the thread as idle.
                           self.change_state(ThreadState::Idle);
                        }
                     }
                  }
               }
            }
         }

         // Check to see if the task this thread
         // was processing was completed.
         if task_completed == true
         {
            println!("Task completed.");
            self.current_task = None;
            self.change_state(ThreadState::Idle);
         }

         // Sleep the thread so that other threads
         // get a chance to run 
         ::std::thread::sleep(Duration::new(0, 100));

         // Check to see if this thread should be shutdown.
         self.process_shutdown_messages();
      }

      // This thread is finished.
      println!("Shutting down thread.");
      self.change_state(ThreadState::Finished);
   }
}