thread.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_THREAD_H
#define PAR_THREAD_H

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

#include "fastlib/base/base.h"

#include <pthread.h>

class Thread {
  FORBID_ACCIDENTAL_COPIES(Thread);
  
 public:
  enum {
    LOW_PRIORITY = 20,
    NORMAL_PRIORITY = 0
  };
  
 private:
#ifdef DEBUG
  enum {UNINIT, READY, ATTACHED, DETACHED, DONE} status_;
#endif
  pthread_t thread_;
  Task *task_;
  
  static void *ThreadMain_(void *self) {
    Thread* thread = reinterpret_cast<Thread*>(self);
    
    thread->task_->Run();
    
    return NULL;
  }

  void Exit_() {
    pthread_exit(NULL);
  }
  
 public:
  Thread() {
#ifdef DEBUG
    DEBUG_ONLY(status_ = UNINIT);
#endif
  }
  ~Thread() {
#ifdef DEBUG
    DEBUG_ASSERT(status_ == DETACHED || status_ == READY || status_ == DONE || status_ == UNINIT);
    DEBUG_ONLY(status_ = UNINIT);
#endif
  }
  
  void Init(Task* task_in) {
#ifdef DEBUG
    DEBUG_ASSERT(status_ == UNINIT);
#endif
    task_ = task_in;
#ifdef DEBUG
    DEBUG_ONLY(status_ = READY);
#endif
  }
  
  void Start() {
#ifdef DEBUG
    DEBUG_ASSERT(status_ == READY);
#endif
    pthread_create(&thread_, NULL,
        ThreadMain_, reinterpret_cast<void*>(this));
#ifdef DEBUG
    DEBUG_ONLY(status_ = ATTACHED);
#endif
  }
  
  void Start(int prio) {
    pthread_attr_t tattr;
    sched_param param;

#ifdef DEBUG
    DEBUG_ASSERT(status_ == READY);
#endif
    pthread_attr_init(&tattr);
    pthread_attr_getschedparam(&tattr, &param);
    param.sched_priority = prio;
    pthread_attr_setschedparam(&tattr, &param);
    pthread_create(&thread_, &tattr,
        ThreadMain_, reinterpret_cast<void*>(this));
    pthread_attr_destroy(&tattr);
#ifdef DEBUG
    DEBUG_ONLY(status_ = ATTACHED);
#endif
  }
  
  void Detach() {
#ifdef DEBUG
    DEBUG_ASSERT(status_ == ATTACHED);
#endif
    pthread_detach(thread_);
#ifdef DEBUG
    DEBUG_ONLY(status_ = DETACHED);
#endif
  }
  
  void WaitStop() {
#ifdef DEBUG
    DEBUG_ASSERT(status_ == ATTACHED);
#endif
    pthread_join(thread_, NULL);
#ifdef DEBUG
    DEBUG_ONLY(status_ = DONE);
#endif
  }
  
  Task* task() const {
    return task_;
  }
};

class Mutex {
  FORBID_ACCIDENTAL_COPIES(Mutex);
  friend class WaitCondition;
 
 public:
  struct DummyRecursiveAttribute {};

 private:
  mutable pthread_mutex_t mutex_;

 public:
  static Mutex global;

 public:
  Mutex() {
#if defined(DEBUG) && defined(PTHREAD_ERRORCHECK_MUTEX_INITIALIZER_NP)
    mutex_ = (pthread_mutex_t)PTHREAD_ERRORCHECK_MUTEX_INITIALIZER_NP;
#else
    mutex_ = (pthread_mutex_t)PTHREAD_MUTEX_INITIALIZER;
#endif
  }
  Mutex(DummyRecursiveAttribute v) {
#ifdef PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP
    mutex_ = (pthread_mutex_t)PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP;
#else
    pthread_mutexattr_t attr;
    pthread_mutexattr_init(&attr);
    pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE);
    pthread_mutex_init(&mutex_, &attr);
    pthread_mutexattr_destroy(&attr);
#endif
  }

  ~Mutex() {
    pthread_mutex_destroy(&mutex_);
  }
  
  void Lock() const {
    int t = pthread_mutex_lock(&mutex_);
    (void)t; // avoid an "unused variable" warning
    DEBUG_ASSERT_MSG(t == 0, "Error locking mutex -- relocking a non-recursive mutex?");
  }
  
  bool TryLock() const {
    return likely(pthread_mutex_trylock(&mutex_) == 0);
  }
  
  void Unlock() const {
    pthread_mutex_unlock(&mutex_);
  }
};

class RecursiveMutex : public Mutex {
  FORBID_ACCIDENTAL_COPIES(RecursiveMutex);

 public:
  RecursiveMutex() : Mutex(DummyRecursiveAttribute()) {}
};

class WaitCondition {
  FORBID_ACCIDENTAL_COPIES(WaitCondition);
  
 private:
  pthread_cond_t cond_;
 
 public:
  WaitCondition() {
    pthread_cond_init(&cond_, NULL);
  }
  
  ~WaitCondition() {
    pthread_cond_destroy(&cond_);
  }
  
  void Signal() {
    pthread_cond_signal(&cond_);
  }
  
  void Broadcast() {
    pthread_cond_broadcast(&cond_);
  }
  
  void Wait(Mutex* mutex_to_unlock) {
    pthread_cond_wait(&cond_, &mutex_to_unlock->mutex_);
  }
  
  void WaitMillis(Mutex& mutex_to_unlock, unsigned millis) {
    struct timespec ts;
    
    ts.tv_sec = millis / 1000;
    ts.tv_nsec = (millis % 1000) * 1000000;
    
    pthread_cond_timedwait(&cond_, &mutex_to_unlock.mutex_, &ts);
  }
  
  void WaitSec(Mutex& mutex_to_unlock, unsigned sec) {
    struct timespec ts;
    
    ts.tv_sec = sec;
    ts.tv_nsec = 0;
    
    pthread_cond_timedwait(&cond_, &mutex_to_unlock.mutex_, &ts);
  }
};

class DoneCondition {
  Mutex mutex_;
  WaitCondition cond_;
  bool done_;

 public:
  DoneCondition() { done_ = false; }
  ~DoneCondition() {}

  void Wait() {
    mutex_.Lock();
    while (!done_) {
      cond_.Wait(&mutex_);
    }
    done_ = false;
    mutex_.Unlock();
  }

  void Done() {
    mutex_.Lock();
    DEBUG_ASSERT_MSG(done_ == false, "Doesn't do a counter -- should it?");
    done_ = true;
    cond_.Signal();
    mutex_.Unlock();
  }
};

class ValueCondition {
  Mutex mutex_;
  WaitCondition cond_;
  int value_;

 public:
  ValueCondition() { value_ = 0; }
  ~ValueCondition() {}

  void Wait(int v) {
    mutex_.Lock();
    while (value_ != v) {
      cond_.Wait(&mutex_);
    }
    mutex_.Unlock();
  }

  void WaitNot(int v) {
    mutex_.Lock();
    while (value_ == v) {
      cond_.Wait(&mutex_);
    }
    mutex_.Unlock();
  }

  void Set(int v) {
    mutex_.Lock();
    if (value_ != v) {
      value_ = v;
      cond_.Broadcast();
    }
    mutex_.Unlock();
  }
};

template<class TContained>
class Lockable : public TContained, public Mutex {
  FORBID_ACCIDENTAL_COPIES(Lockable);

  Lockable() {}
  ~Lockable() {}
};

#endif