grain.h

/* MLPACK 0.2
 *
 * Copyright (c) 2008, 2009 Alexander Gray,
 *                          Garry Boyer,
 *                          Ryan Riegel,
 *                          Nikolaos Vasiloglou,
 *                          Dongryeol Lee,
 *                          Chip Mappus, 
 *                          Nishant Mehta,
 *                          Hua Ouyang,
 *                          Parikshit Ram,
 *                          Long Tran,
 *                          Wee Chin Wong
 *
 * Copyright (c) 2008, 2009 Georgia Institute of Technology
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License as
 * published by the Free Software Foundation; either version 2 of the
 * License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
 * 02110-1301, USA.
 */
#ifndef PAR_GRAIN_H
#define PAR_GRAIN_H

#include "fastlib/col/heap.h"

#include "fastlib/par/thread.h"
//#include "thread.h"

template<typename TGrain>
class GrainQueue {
 public:
  typedef TGrain Grain;

 private:
  MinHeap<double, Grain*> queue_;
  Mutex mutex_;
  
 public:
  GrainQueue() {}
  ~GrainQueue() {}
  
  void Init() {
    queue_.Init();
  }
  
  void Put(double difficulty, Grain *grain) {
    mutex_.Lock();
    queue_.Put(-difficulty, grain);
    mutex_.Unlock();
  }
  
  Grain *Pop() {
    mutex_.Lock();
    Grain *result = likely(queue_.size() != 0) ? queue_.Pop() : NULL;
    mutex_.Unlock();
    return result;
  }
  
  index_t size() const {
    return queue_.size();
  }
};

template<typename TGrain, typename TContext = int>
class ThreadedGrainRunner {
  FORBID_ACCIDENTAL_COPIES(ThreadedGrainRunner);
 public:
  typedef TGrain Grain;
  typedef TContext Context;

 private:
  struct ThreadTask : public Task {
    ThreadedGrainRunner *runner_;
    
    ThreadTask(ThreadedGrainRunner *runner_in) {
      runner_ = runner_in;
    }
    
    void Run() {
      while (runner_->RunOneGrain()) {}
      delete this;
    }
  };
  
 private:
  GrainQueue<Grain> *queue_;
  Context context_;
  
 public:
  ThreadedGrainRunner() {}
  ~ThreadedGrainRunner() {}
  
  void Init(GrainQueue<Grain> *queue_in, Context context_in) {
    queue_ = queue_in;
    context_ = context_in;
  }
    
  bool RunOneGrain() {
    Grain *grain = queue_->Pop();
    if (unlikely(!grain)) {
      return false;
    } else {
      grain->Run(context_);
      delete grain;
      return true;
    }
  }
  
  Thread *SpawnThread() {
    ThreadTask *task = new ThreadTask(this);
    Thread *thread = new Thread();
    thread->Init(task);
    thread->Start();
    return thread;
  }

  void RunThreads(int num_threads) {
    ArrayList<Thread*> threads;
    
    threads.Init(num_threads);
    
    for (int i = 0; i < num_threads; i++) {
      threads[i] = SpawnThread();
    }
    for (int i = 0; i < num_threads; i++) {
      threads[i]->WaitStop();
      delete threads[i];
    }
  }
};

#endif