thread.C

/*
 * This file is part of the "Archon" framework.
 * (http://files3d.sourceforge.net)
 *
 * Copyright © 2002 by Kristian Spangsege and Brian Kristiansen.
 *
 * Permission to use, copy, modify, and distribute this software and
 * its documentation under the terms of the GNU General Public License is
 * hereby granted. No representations are made about the suitability of
 * this software for any purpose. It is provided "as is" without express
 * or implied warranty. See the GNU General Public License
 * (http://www.gnu.org/copyleft/gpl.html) for more details.
 *
 * The characters in this file are ISO8859-1 encoded.
 *
 * The documentation in this file is in "Doxygen" style
 * (http://www.doxygen.org).
 */

#include <errno.h>
#include <iostream>

#include <archon/util/thread.H>

namespace Archon
{
  namespace Utilities
  {
    using namespace std;

    namespace
    {
      pthread_key_t selfKey;

      pthread_once_t selfKeyOnce = PTHREAD_ONCE_INIT;

      pthread_mutex_t activeThreadsMutex = PTHREAD_MUTEX_INITIALIZER;

      pthread_cond_t lastActiveThread = PTHREAD_COND_INITIALIZER;

      unsigned long activeThreads = 0;

      void lockMutex(pthread_mutex_t *m)
      {
        int e = pthread_mutex_lock(m);
        if(e == 0) return;
        if(e == EDEADLK)
          ARCHON_THROW1(StateException,
                        "Attempt to lock a locked mutex");
        if(e == EINVAL)
          ARCHON_THROW1(InternalException,
                        "Attempt to lock uninitialized mutex");
        ARCHON_THROW1(InternalException,
                      "Attempt to lock mutex failed");
      }

      void unlockMutex(pthread_mutex_t *m)
      {
        int e = pthread_mutex_unlock(m);
        if(e == 0) return;
        if(e == EPERM)
          ARCHON_THROW1(StateException,
                        "Attempt to unlock mutex that is locked by another thread");
        if(e == EINVAL)
          ARCHON_THROW1(InternalException,
                        "Attempt to unlock uninitialized mutex");
        ARCHON_THROW1(InternalException,
                      "Attempt to unlock mutex failed");
      }
    }


    void Thread::activate(Thread *t)
    {
      lockMutex(&activeThreadsMutex);
      ++activeThreads;
      unlockMutex(&activeThreadsMutex);
      t->refInc();
    }

    void Thread::deactivate(Thread *t)
    {
      t->refDec();
      lockMutex(&activeThreadsMutex);
      bool l = --activeThreads == 1;
      unlockMutex(&activeThreadsMutex);
      if(l)
      {
        const int e = pthread_cond_broadcast(&lastActiveThread);
        if(e != 0)
          ARCHON_THROW1(InternalException,
                        "Attempt to broadcast on condition failed");
      }
    }

    void Thread::selfKeyDestroy(void *v) throw()
    {
      Thread *t = reinterpret_cast<Thread *>(v);
      {
        Mutex::Lock l(*t);
        t->terminated = true;
      }
      t->termination.notifyAll();
      deactivate(t);
    }

    void Thread::selfKeyAlloc() throw()
    {
      int error = pthread_key_create(&selfKey, &Thread::selfKeyDestroy);
      if(error)
      {
        if(error == EAGAIN)
          ARCHON_THROW1(ResourceException,
                        "Too many TSD keys");
        ARCHON_THROW1(InternalException,
                      "Could not create new TSD key");
      }
    }

    void Thread::registerSelf(Thread *t)
    {
      int error = pthread_once(&selfKeyOnce, &selfKeyAlloc);
      if(error)
      {
        deactivate(t);
        ARCHON_THROW1(InternalException, "pthread_once failed");
      }
      error = pthread_setspecific(selfKey, reinterpret_cast<void *>(t));
      if(error)
      {
        deactivate(t);
        
        if(error == EINVAL)
          ARCHON_THROW1(InternalException,
                        "Invalid TSD key in pthread_setspecific");
        ARCHON_THROW1(InternalException,
                      "pthread_setspecific failed");
      }
    }

    Thread::Thread():
      started(false), terminated(false), termination(*this),
      terminateRequest(false), waitCondition(0)
    {
    }

    Thread::~Thread()
    {
    }

    void *Thread::entry(Thread *t) throw()
    {
      int error = pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, 0);
      if(error) ARCHON_THROW1(ResourceException,
                              "pthread_setcancelstate failed");
      registerSelf(t);
      try
      {
        t->main();
      }
      catch(ThreadTerminatedException &) {}
      return 0;
    }

    void Thread::start(Ref<Thread> t) throw(AlreadyStartedException)
    {
      if(!t) ARCHON_THROW1(ArgumentException, "Got null thread");
      {
        Mutex::Lock l(*t.get());
        if(t->started) ARCHON_THROW(AlreadyStartedException);
        t->started = true;
      }

      pthread_attr_t attr;
      int error = pthread_attr_init(&attr);
      if(error) ARCHON_THROW1(ResourceException, "pthread_attr_init failed");
      error = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
      if(error) ARCHON_THROW1(ResourceException,
                              "pthread_attr_setdetachstate failed");

      activate(t.get());
      error = pthread_create
        (&t->pthread, 0, reinterpret_cast<void *(*)(void *)>(t->entry), t.get());
      pthread_attr_destroy(&attr);

      if(!error) return;
      deactivate(t.get());
      if(error == EAGAIN)
        ARCHON_THROW1(ResourceException,
                      "Not enough system resources to create a new thread");
      ARCHON_THROW1(InternalException,
                    "Could not create a new thread");
    }


    struct Thread::SelfThread: Thread
    {
      SelfThread()
      {
        pthread = pthread_self();
        started = true;
      }

      void main() {} // Dummy
    };

    Ref<Thread> Thread::self()
    {
      int error = pthread_once(&selfKeyOnce, &selfKeyAlloc);
      if(error) ARCHON_THROW1(InternalException, "pthread_once failed");
      Thread *t = reinterpret_cast<Thread *>(pthread_getspecific(selfKey));
      if(t) return t;
      t = new SelfThread;
      activate(t);
      registerSelf(t);
      return t;
    }

    void Thread::terminate()
    {
      terminateRequest = true;
      Mutex::Lock l1(waitConditionMutex);
      if(waitCondition)
      {
        Mutex::Lock l2(*waitCondition->mutex);
        waitCondition->notifyAll();
      }
      l1.release();
    }

    void Thread::selfResurrect()
    {
      self()->terminateRequest = false;
    }

    void Thread::wait() throw(UnexpectedException, NotStartedException)
    {
      Mutex::Lock l(*this);
      if(!started) ARCHON_THROW(NotStartedException);
      while(!terminated) termination.wait();
    }

    void Thread::sleep(const Time &period) throw(UnexpectedException)
    {
      Time timeout(Time::now());
      timeout += period;
      Mutex mutex;
      Condition cond(mutex);
      Mutex::Lock lock(mutex);
      while(!cond.timedWait(timeout));
    }

    void Thread::sleepUntil(const Time &timeout) throw(UnexpectedException)
    {
      Mutex mutex;
      Condition cond(mutex);
      Mutex::Lock lock(mutex);
      while(!cond.timedWait(timeout));
    }

    void Thread::acceptTermination() throw(UnexpectedException)
    {
      if(self()->terminateRequest) ARCHON_THROW(ThreadTerminatedException);
    }

    void Thread::mainExitWait()
    {
      Thread::self(); // Must make sure that the calling thread counts
                      // as active.
      lockMutex(&activeThreadsMutex);
      while(activeThreads > 1)
        if(pthread_cond_wait(&lastActiveThread, &activeThreadsMutex))
          ARCHON_THROW1(InternalException, "pthread_cond_wait failed");
      unlockMutex(&activeThreadsMutex);
    }


    void Semaphore::down() throw(UnexpectedException)
    {
      Mutex::Lock l(mutex);
      while(value == 0) nonZero.wait();
      --value;
    }

    void Semaphore::up()
    {
      Mutex::Lock l(mutex);
      ++value;
      l.release();
      nonZero.notifyAll();
    }
  }
}

---