Thread.h

// -*- indent-tabs-mode: nil -*-

#ifndef __ARC_THREAD_H__
#define __ARC_THREAD_H__

#include <map>

#include <glibmm/thread.h>

namespace Arc {

  class SimpleCondition;
  class SimpleCounter;

  // This module provides convenient helpers for Glibmm interface for thread
  // management. So far it takes care of automatic initialization
  // of threading environment and creation of simple detached threads.
  // Always use it instead of glibmm/thread.h and keep among first
  // includes. It safe to use it multiple times and to include it both
  // from source files and other include files.

  const size_t thread_stacksize = (16 * 1024 * 1024);


  bool CreateThreadFunction(void (*func)(void*), void *arg, SimpleCounter* count = NULL);

  class ThreadId {
  private:
    Glib::Mutex mutex;
    std::map<unsigned long int, unsigned long int> thread_ids;
    unsigned long int thread_no;
    ThreadId();
   public:
    static ThreadId& getInstance();
    void add();
    void remove();
    unsigned long int get();
  };
  class ThreadData;


  class ThreadDataItem {
  friend class ThreadData;
  private:
    ThreadDataItem(const ThreadDataItem& it);
  protected:
    virtual ~ThreadDataItem(void);
  public:

    ThreadDataItem(void);

    ThreadDataItem(std::string& key);
    ThreadDataItem(const std::string& key);

    void Attach(std::string& key);

    void Attach(const std::string& key);

    static ThreadDataItem* Get(const std::string& key);

    virtual void Dup(void);
  };


  class SimpleCondition {
  private:
    Glib::Cond cond_;
    Glib::Mutex lock_;
    unsigned int flag_;
    unsigned int waiting_;
  public:
    SimpleCondition(void)
      : flag_(0), waiting_(0) {}
    ~SimpleCondition(void) {
      /* race condition ? */
      broadcast();
    }
    void lock(void) {
      lock_.lock();
    }
    void unlock(void) {
      lock_.unlock();
    }
    void signal(void) {
      lock_.lock();
      flag_ = 1;
      cond_.signal();
      lock_.unlock();
    }

    void signal_nonblock(void) {
      flag_ = 1;
      cond_.signal();
    }

    void broadcast(void) {
      lock_.lock();
      flag_ = waiting_?waiting_:1;
      cond_.broadcast();
      lock_.unlock();
    }
    void wait(void) {
      lock_.lock();
      ++waiting_;
      while (!flag_) cond_.wait(lock_);
      --waiting_;
      --flag_;
      lock_.unlock();
    }

    void wait_nonblock(void) {
      ++waiting_;
      while (!flag_) cond_.wait(lock_);
      --waiting_;
      --flag_;
    }

    bool wait(int t) {
      lock_.lock();
      Glib::TimeVal etime;
      etime.assign_current_time();
      etime.add_milliseconds(t);
      bool res(true);
      ++waiting_;
      while (!flag_) {
        res = cond_.timed_wait(lock_, etime);
        if (!res) break;
      }
      --waiting_;
      if(res) --flag_;
      lock_.unlock();
      return res;
    }
    void reset(void) {
      lock_.lock();
      flag_ = 0;
      lock_.unlock();
    }

    void forceReset(void) {
      flag_ = 0;
      waiting_ = 0;
      lock_.unlock();
    }
  };


  class SimpleCounter {
  private:
    Glib::Cond cond_;
    Glib::Mutex lock_;
    int count_;
  public:
    SimpleCounter(void) : count_(0) {}
    virtual ~SimpleCounter(void);

    virtual int inc(void);

    virtual int dec(void);
    virtual int get(void) const;

    virtual int set(int v);
    virtual void wait(void) const;

    virtual bool wait(int t) const;

    virtual void forceReset(void) {
      lock_.unlock();
    }
  };


  class TimedMutex {
  private:
    Glib::Cond cond_;
    Glib::Mutex lock_;
    bool locked_;
  public:
    TimedMutex(void):locked_(false) { };
    ~TimedMutex(void) { };

    bool lock(int t = -1) {
      lock_.lock();
      if(t < 0) { // infinite
        while(locked_) {
          cond_.wait(lock_);
        };
      } else if(t > 0) { // timed
        Glib::TimeVal etime;
        etime.assign_current_time();
        etime.add_milliseconds(t);
        while(locked_) {
          if(!cond_.timed_wait(lock_, etime)) break;
        };
      };
      bool res = !locked_;
      locked_=true;
      lock_.unlock();
      return res;
    };
    bool trylock(void) {
      return lock(0);
    };
    bool unlock(void) {
      lock_.lock();
      bool res = locked_;
      if(res) {
        locked_ = false;
        cond_.signal();
      };
      lock_.unlock();
      return true;
    };

    void forceReset(void) {
      locked_ = false;
      lock_.unlock();
    }
  };


  class SharedMutex {
  private:
    Glib::Cond cond_;
    Glib::Mutex lock_;
    unsigned int exclusive_;
    Glib::Thread* thread_;
    typedef std::map<Glib::Thread*,unsigned int> shared_list;
    shared_list shared_;
    void add_shared_lock(void);
    void remove_shared_lock(void);
    bool have_shared_lock(void);
    inline bool have_exclusive_lock(void) {
      if(!exclusive_) return false;
      if(thread_ == Glib::Thread::self()) return false;
      return true;
    };
  public:
    SharedMutex(void):exclusive_(0),thread_(NULL) { };
    ~SharedMutex(void) { };
    void lockShared(void);
    void unlockShared(void);
    bool isLockShared(void) {
      return (shared_.size() > 0); // Is it safe?
    };
    void lockExclusive(void);
    void unlockExclusive(void);
    bool isLockExclusive(void) {
      return (exclusive_ > 0);
    };

    void forceReset(void) {
      exclusive_ = 0;
      thread_ = NULL;
      shared_.clear();
      lock_.unlock();
    };
  };

  class ThreadedPointerBase {
  private:
    Glib::Mutex lock_;
    Glib::Cond cond_;
    unsigned int cnt_;
    void *ptr_;
    bool released_;
    ThreadedPointerBase(ThreadedPointerBase&);
    ~ThreadedPointerBase(void);
  public:
    ThreadedPointerBase(void *p);
    ThreadedPointerBase* add(void);
    void* rem(void);
    void* ptr(void) const { return ptr_; };
    void rel(void) { released_ = true; };
    unsigned int cnt(void) const { return cnt_; };
    void lock(void) { lock_.lock(); };
    void unlock(void) { lock_.unlock(); };
    void wait(void) { cond_.wait(lock_); };
    bool wait(Glib::TimeVal etime) {
      return cond_.timed_wait(lock_,etime);
    };
  };


  template<typename T>
  class ThreadedPointer {
  private:
    ThreadedPointerBase *object_;
  public:
    ThreadedPointer(T *p)
      : object_(new ThreadedPointerBase(p)) {}
    ThreadedPointer(const ThreadedPointer<T>& p)
      : object_(p.object_->add()) {}
    ThreadedPointer(void)
      : object_(new ThreadedPointerBase(NULL)) {}
    ~ThreadedPointer(void) {
      delete((T*)(object_->rem()));
    }
    ThreadedPointer<T>& operator=(T *p) {
      if (p != object_->ptr()) {
        delete((T*)(object_->rem()));
        object_ = new ThreadedPointerBase(p);
      }
      return *this;
    }
    ThreadedPointer<T>& operator=(const ThreadedPointer<T>& p) {
      if (p.object_->ptr() != object_->ptr()) {
        delete((T*)(object_->rem()));
        object_ = p.object_->add();
      }
      return *this;
    }
    T& operator*(void) const {
      return *(T*)(object_->ptr());
    }
    T* operator->(void) const {
      return (T*)(object_->ptr());
    }
    operator bool(void) const {
      return ((object_->ptr()) != NULL);
    }
    bool operator!(void) const {
      return ((object_->ptr()) == NULL);
    }
    bool operator==(const ThreadedPointer& p) const {
      return ((T*)(object_->ptr()) == (T*)(p.object_->ptr()));
    }
    bool operator!=(const ThreadedPointer& p) const {
      return ((T*)(object_->ptr()) != (T*)(p.object_->ptr()));
    }
    bool operator<(const ThreadedPointer& p) const {
      return ((T*)(object_->ptr()) < (T*)(p.object_->ptr()));
    }
    T* Ptr(void) const {
      return (T*)(object_->ptr());
    }

    T* Release(void) {
      T* tmp = (T*)(object_->ptr());
      object_->rel();
      return tmp;
    }
    unsigned int Holders(void) {
      return object_->cnt();
    }
    /* Returns current number of instances. */
    unsigned int WaitOutRange(unsigned int minThr, unsigned int maxThr) {
      unsigned int r = 0;
      object_->lock();
      for(;;) {
        r = object_->cnt();
        if(r <= minThr) break;
        if(r >= maxThr) break;
        object_->wait();
      };
      object_->unlock();
      return r;
    }

    unsigned int WaitOutRange(unsigned int minThr, unsigned int maxThr, int timeout) {
      if(timeout < 0) return WaitOutRange(minThr, maxThr);
      unsigned int r = 0;
      object_->lock();
      Glib::TimeVal etime;
      etime.assign_current_time();
      etime.add_milliseconds(timeout);
      for(;;) {
        r = object_->cnt();
        if(r <= minThr) break;
        if(r >= maxThr) break;
        if(!object_->wait(etime)) break;
      };
      object_->unlock();
      return r;
    }
    /* Returns current number of instances. */
    unsigned int WaitInRange(unsigned int minThr, unsigned int maxThr) {
      unsigned int r = 0;
      object_->lock();
      for(;;) {
        r = object_->cnt();
        if((r >= minThr) && (r <= maxThr)) break;
        object_->wait();
      };
      object_->unlock();
      return r;
    }

    unsigned int WaitInRange(unsigned int minThr, unsigned int maxThr, int timeout) {
      if(timeout < 0) return WaitInRange(minThr, maxThr);
      unsigned int r = 0;
      object_->lock();
      Glib::TimeVal etime;
      etime.assign_current_time();
      etime.add_milliseconds(timeout);
      for(;;) {
        r = object_->cnt();
        if((r >= minThr) && (r <= maxThr)) break;
        if(!object_->wait(etime)) break;
      };
      object_->unlock();
      return r;
    }

  };


  class ThreadRegistry {
  private:
    int counter_;
    bool cancel_;
    Glib::Cond cond_;
    Glib::Mutex lock_;
  public:
    ThreadRegistry(void);
    ~ThreadRegistry(void);
    void RegisterThread(void);
    void UnregisterThread(void);

    bool WaitOrCancel(int timeout);

    bool WaitForExit(int timeout = -1);
    void RequestCancel(void);

    void forceReset(void) {
      counter_ = 0;
      cancel_ = false;
      lock_.unlock();
    }
  };

  void GlibThreadInitialize(void);

  class ThreadInitializer {
  public:
    ThreadInitializer(void) {
      GlibThreadInitialize();
    }

    void forceReset(void);

    void waitExit(void);
  };

  // This is done intentionally to make sure glibmm is
  // properly initialized before every module starts
  // using threads functionality. To make it work this
  // header must be included before defining any 
  // variable/class instance using static threads-related
  // elements. The simplest way to do that is to use
  // this header instead of glibmm/thread.h
  static ThreadInitializer _local_thread_initializer;

} // namespace Arc

#endif /* __ARC_THREAD_H__ */