/*************************************************************************************************

 * Threading devices

 *                                                               Copyright (C) 2009-2012 FAL Labs

 * This file is part of Kyoto Cabinet.

 * 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

 * 3 of the License, or 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, see <http://www.gnu.org/licenses/>.

 *************************************************************************************************/

#ifndef _KCTHREAD_H                      // duplication check

#define _KCTHREAD_H

#include <kccommon.h>

#include <kcutil.h>

namespace kyotocabinet {                 // common namespace

/**

 * Threading device.

 */

class Thread {

 public:

  /**

   * Default constructor.

   */

  explicit Thread();

  /**

   * Destructor.

   */

  virtual ~Thread();

  /**

   * Perform the concrete process.

   */

  virtual void run() = 0;

  /**

   * Start the thread.

   */

  void start();

  /**

   * Wait for the thread to finish.

   */

  void join();

  /**

   * Put the thread in the detached state.

   */

  void detach();

  /**

   * Terminate the running thread.

   */

  static void exit();

  /**

   * Yield the processor from the current thread.

   */

  static void yield();

  /**

   * Chill the processor by suspending execution for a quick moment.

   */

  static void chill();

  /**

   * Suspend execution of the current thread.

   * @param sec the interval of the suspension in seconds.

   * @return true on success, or false on failure.

   */

  static bool sleep(double sec);

  /**

   * Get the hash value of the current thread.

   * @return the hash value of the current thread.

   */

  static int64_t hash();

 private:

  /** Dummy constructor to forbid the use. */

  Thread(const Thread&);

  /** Dummy Operator to forbid the use. */

  Thread& operator =(const Thread&);

  /** Opaque pointer. */

  void* opq_;

};

/**

 * Basic mutual exclusion device.

 */

class Mutex {

  friend class CondVar;

 public:

  /**

   * Type of the behavior for double locking.

   */

  enum Type {

    FAST,                ///< no operation

    ERRORCHECK,          ///< check error

    RECURSIVE            ///< allow recursive locking

  };

  /**

   * Default constructor.

   */

  explicit Mutex();

  /**

   * Constructor.

   * @param type the behavior for double locking.

   */

  explicit Mutex(Type type);

  /**

   * Destructor.

   */

  ~Mutex();

  /**

   * Get the lock.

   */

  void lock();

  /**

   * Try to get the lock.

   * @return true on success, or false on failure.

   */

  bool lock_try();

  /**

   * Try to get the lock.

   * @param sec the interval of the suspension in seconds.

   * @return true on success, or false on failure.

   */

  bool lock_try(double sec);

  /**

   * Release the lock.

   */

  void unlock();

 private:

  /** Dummy constructor to forbid the use. */

  Mutex(const Mutex&);

  /** Dummy Operator to forbid the use. */

  Mutex& operator =(const Mutex&);

  /** Opaque pointer. */

  void* opq_;

};

/**

 * Scoped mutex device.

 */

class ScopedMutex {

 public:

  /**

   * Constructor.

   * @param mutex a mutex to lock the block.

   */

  explicit ScopedMutex(Mutex* mutex) : mutex_(mutex) {

    _assert_(mutex);

    mutex_->lock();

  }

  /**

   * Destructor.

   */

  ~ScopedMutex() {

    _assert_(true);

    mutex_->unlock();

  }

 private:

  /** Dummy constructor to forbid the use. */

  ScopedMutex(const ScopedMutex&);

  /** Dummy Operator to forbid the use. */

  ScopedMutex& operator =(const ScopedMutex&);

  /** The inner device. */

  Mutex* mutex_;

};

/**

 * Slotted mutex device.

 */

class SlottedMutex {

 public:

  /**

   * Constructor.

   * @param slotnum the number of slots.

   */

  explicit SlottedMutex(size_t slotnum);

  /**

   * Destructor.

   */

  ~SlottedMutex();

  /**

   * Get the lock of a slot.

   * @param idx the index of a slot.

   */

  void lock(size_t idx);

  /**

   * Release the lock of a slot.

   * @param idx the index of a slot.

   */

  void unlock(size_t idx);

  /**

   * Get the locks of all slots.

   */

  void lock_all();

  /**

   * Release the locks of all slots.

   */

  void unlock_all();

 private:

  /** Opaque pointer. */

  void* opq_;

};

/**

 * Lightweight mutual exclusion device.

 */

class SpinLock {

 public:

  /**

   * Default constructor.

   */

  explicit SpinLock();

  /**

   * Destructor.

   */

  ~SpinLock();

  /**

   * Get the lock.

   */

  void lock();

  /**

   * Try to get the lock.

   * @return true on success, or false on failure.

   */

  bool lock_try();

  /**

   * Release the lock.

   */

  void unlock();

 private:

  /** Dummy constructor to forbid the use. */

  SpinLock(const SpinLock&);

  /** Dummy Operator to forbid the use. */

  SpinLock& operator =(const SpinLock&);

  /** Opaque pointer. */

  void* opq_;

};

/**

 * Scoped spin lock device.

 */

class ScopedSpinLock {

 public:

  /**

   * Constructor.

   * @param spinlock a spin lock to lock the block.

   */

  explicit ScopedSpinLock(SpinLock* spinlock) : spinlock_(spinlock) {

    _assert_(spinlock);

    spinlock_->lock();

  }

  /**

   * Destructor.

   */

  ~ScopedSpinLock() {

    _assert_(true);

    spinlock_->unlock();

  }

 private:

  /** Dummy constructor to forbid the use. */

  ScopedSpinLock(const ScopedSpinLock&);

  /** Dummy Operator to forbid the use. */

  ScopedSpinLock& operator =(const ScopedSpinLock&);

  /** The inner device. */

  SpinLock* spinlock_;

};

/**

 * Slotted spin lock devices.

 */

class SlottedSpinLock {

 public:

  /**

   * Constructor.

   * @param slotnum the number of slots.

   */

  explicit SlottedSpinLock(size_t slotnum);

  /**

   * Destructor.

   */

  ~SlottedSpinLock();

  /**

   * Get the lock of a slot.

   * @param idx the index of a slot.

   */

  void lock(size_t idx);

  /**

   * Release the lock of a slot.

   * @param idx the index of a slot.

   */

  void unlock(size_t idx);

  /**

   * Get the locks of all slots.

   */

  void lock_all();

  /**

   * Release the locks of all slots.

   */

  void unlock_all();

 private:

  /** Opaque pointer. */

  void* opq_;

};

/**

 * Reader-writer locking device.

 */

class RWLock {

 public:

  /**

   * Default constructor.

   */

  explicit RWLock();

  /**

   * Destructor.

   */

  ~RWLock();

  /**

   * Get the writer lock.

   */

  void lock_writer();

  /**

   * Try to get the writer lock.

   * @return true on success, or false on failure.

   */

  bool lock_writer_try();

  /**

   * Get a reader lock.

   */

  void lock_reader();

  /**

   * Try to get a reader lock.

   * @return true on success, or false on failure.

   */

  bool lock_reader_try();

  /**

   * Release the lock.

   */

  void unlock();

 private:

  /** Dummy constructor to forbid the use. */

  RWLock(const RWLock&);

  /** Dummy Operator to forbid the use. */

  RWLock& operator =(const RWLock&);

  /** Opaque pointer. */

  void* opq_;

};

/**

 * Scoped reader-writer locking device.

 */

class ScopedRWLock {

 public:

  /**

   * Constructor.

   * @param rwlock a rwlock to lock the block.

   * @param writer true for writer lock, or false for reader lock.

   */

  explicit ScopedRWLock(RWLock* rwlock, bool writer) : rwlock_(rwlock) {

    _assert_(rwlock);

    if (writer) {

      rwlock_->lock_writer();

    } else {

      rwlock_->lock_reader();

    }

  }

  /**

   * Destructor.

   */

  ~ScopedRWLock() {

    _assert_(true);

    rwlock_->unlock();

  }

 private:

  /** Dummy constructor to forbid the use. */

  ScopedRWLock(const ScopedRWLock&);

  /** Dummy Operator to forbid the use. */

  ScopedRWLock& operator =(const ScopedRWLock&);

  /** The inner device. */

  RWLock* rwlock_;

};

/**

 * Slotted reader-writer lock devices.

 */

class SlottedRWLock {

 public:

  /**

   * Constructor.

   * @param slotnum the number of slots.

   */

  explicit SlottedRWLock(size_t slotnum);

  /**

   * Destructor.

   */

  ~SlottedRWLock();

  /**

   * Get the writer lock of a slot.

   * @param idx the index of a slot.

   */

  void lock_writer(size_t idx);

  /**

   * Get the reader lock of a slot.

   * @param idx the index of a slot.

   */

  void lock_reader(size_t idx);

  /**

   * Release the lock of a slot.

   * @param idx the index of a slot.

   */

  void unlock(size_t idx);

  /**

   * Get the writer locks of all slots.

   */

  void lock_writer_all();

  /**

   * Get the reader locks of all slots.

   */

  void lock_reader_all();

  /**

   * Release the locks of all slots.

   */

  void unlock_all();

 private:

  /** Opaque pointer. */

  void* opq_;

};

/**

 * Lightweight reader-writer locking device.

 */

class SpinRWLock {

 public:

  /**

   * Default constructor.

   */

  explicit SpinRWLock();

  /**

   * Destructor.

   */

  ~SpinRWLock();

  /**

   * Get the writer lock.

   */

  void lock_writer();

  /**

   * Try to get the writer lock.

   * @return true on success, or false on failure.

   */

  bool lock_writer_try();

  /**

   * Get a reader lock.

   */

  void lock_reader();

  /**

   * Try to get a reader lock.

   * @return true on success, or false on failure.

   */

  bool lock_reader_try();

  /**

   * Release the lock.

   */

  void unlock();

  /**

   * Promote a reader lock to the writer lock.

   * @return true on success, or false on failure.

   */

  bool promote();

  /**

   * Demote the writer lock to a reader lock.

   */

  void demote();

 private:

  /** Dummy constructor to forbid the use. */

  SpinRWLock(const SpinRWLock&);

  /** Dummy Operator to forbid the use. */

  SpinRWLock& operator =(const SpinRWLock&);

  /** Opaque pointer. */

  void* opq_;

};

/**

 * Scoped reader-writer locking device.

 */

class ScopedSpinRWLock {

 public:

  /**

   * Constructor.

   * @param srwlock a spin rwlock to lock the block.

   * @param writer true for writer lock, or false for reader lock.

   */

  explicit ScopedSpinRWLock(SpinRWLock* srwlock, bool writer) : srwlock_(srwlock) {

    _assert_(srwlock);

    if (writer) {

      srwlock_->lock_writer();

    } else {

      srwlock_->lock_reader();

    }

  }

  /**

   * Destructor.

   */

  ~ScopedSpinRWLock() {

    _assert_(true);

    srwlock_->unlock();

  }

 private:

  /** Dummy constructor to forbid the use. */

  ScopedSpinRWLock(const ScopedSpinRWLock&);

  /** Dummy Operator to forbid the use. */

  ScopedSpinRWLock& operator =(const ScopedSpinRWLock&);

  /** The inner device. */

  SpinRWLock* srwlock_;

};

/**

 * Slotted lightweight reader-writer lock devices.

 */

class SlottedSpinRWLock {

 public:

  /**

   * Constructor.

   * @param slotnum the number of slots.

   */

  explicit SlottedSpinRWLock(size_t slotnum);

  /**

   * Destructor.

   */

  ~SlottedSpinRWLock();

  /**

   * Get the writer lock of a slot.

   * @param idx the index of a slot.

   */

  void lock_writer(size_t idx);

  /**

   * Get the reader lock of a slot.

   * @param idx the index of a slot.

   */

  void lock_reader(size_t idx);

  /**

   * Release the lock of a slot.

   * @param idx the index of a slot.

   */

  void unlock(size_t idx);

  /**

   * Get the writer locks of all slots.

   */

  void lock_writer_all();

  /**

   * Get the reader locks of all slots.

   */

  void lock_reader_all();

  /**

   * Release the locks of all slots.

   */

  void unlock_all();

 private:

  /** Opaque pointer. */

  void* opq_;

};

/**

 * Condition variable.

 */

class CondVar {

 public:

  /**

   * Default constructor.

   */

  explicit CondVar();

  /**

   * Destructor.

   */

  ~CondVar();

  /**

   * Wait for the signal.

   * @param mutex a locked mutex.

   */

  void wait(Mutex* mutex);

  /**

   * Wait for the signal.

   * @param mutex a locked mutex.

   * @param sec the interval of the suspension in seconds.

   * @return true on catched signal, or false on timeout.

   */

  bool wait(Mutex* mutex, double sec);

  /**

   * Send the wake-up signal to another waiting thread.

   * @note The mutex used for the wait method should be locked by the caller.

   */

  void signal();

  /**

   * Send the wake-up signals to all waiting threads.

   * @note The mutex used for the wait method should be locked by the caller.

   */

  void broadcast();

 private:

  /** Dummy constructor to forbid the use. */

  CondVar(const CondVar&);

  /** Dummy Operator to forbid the use. */

  CondVar& operator =(const CondVar&);

  /** Opaque pointer. */

  void* opq_;

};

/**

 * Assosiative condition variable.

 */

class CondMap {

 private:

  struct Count;

  struct Slot;

  /** An alias of set of counters. */

  typedef std::map<std::string, Count> CountMap;

  /** The number of slots. */

  static const size_t SLOTNUM = 64;

 public:

  /**

   * Default constructor.

   */

  explicit CondMap() : slots_() {

    _assert_(true);

  }

  /**

   * Destructor.

   */

  ~CondMap() {

    _assert_(true);

  }

  /**

   * Wait for a signal.

   * @param kbuf the pointer to the key region.

   * @param ksiz the size of the key region.

   * @param sec the interval of the suspension in seconds.  If it is negative, no timeout is

   * specified.

   * @return true on catched signal, or false on timeout.

   */

  bool wait(const char* kbuf, size_t ksiz, double sec = -1) {

    _assert_(kbuf && ksiz <= MEMMAXSIZ);

    std::string key(kbuf, ksiz);

    return wait(key, sec);

  }

  /**

   * Wait for a signal by a key.

   * @param key the key.

   * @param sec the interval of the suspension in seconds.  If it is negative, no timeout is

   * specified.

   * @return true on catched signal, or false on timeout.

   */

  bool wait(const std::string& key, double sec = -1) {

    _assert_(true);

    double invtime = sec < 0 ? 1.0 : sec;

    double curtime = time();

    double endtime = curtime + (sec < 0 ? UINT32MAX : sec);

    Slot* slot = get_slot(key);

    while (curtime < endtime) {

      ScopedMutex lock(&slot->mutex);

      CountMap::iterator cit = slot->counter.find(key);

      if (cit == slot->counter.end()) {

        Count cnt = { 1, false };

        slot->counter[key] = cnt;

      } else {

        cit->second.num++;

      }

      slot->cond.wait(&slot->mutex, invtime);

      cit = slot->counter.find(key);

      cit->second.num--;

      if (cit->second.wake > 0) {

        cit->second.wake--;

        if (cit->second.num < 1) slot->counter.erase(cit);

        return true;

      }

      if (cit->second.num < 1) slot->counter.erase(cit);

      curtime = time();

    }

    return false;

  }

  /**

   * Send a wake-up signal to another thread waiting by a key.

   * @param kbuf the pointer to the key region.

   * @param ksiz the size of the key region.

   * @return the number of threads waiting for the signal.

   */

  size_t signal(const char* kbuf, size_t ksiz) {

    _assert_(kbuf && ksiz <= MEMMAXSIZ);

    std::string key(kbuf, ksiz);

    return signal(key);

  }

  /**

   * Send a wake-up signal to another thread waiting by a key.

   * @param key the key.

   * @return the number of threads waiting for the signal.

   */

  size_t signal(const std::string& key) {

    _assert_(true);

    Slot* slot = get_slot(key);

    ScopedMutex lock(&slot->mutex);

    CountMap::iterator cit = slot->counter.find(key);

    if (cit == slot->counter.end() || cit->second.num < 1) return 0;

    if (cit->second.wake < cit->second.num) cit->second.wake++;

    slot->cond.broadcast();

    return cit->second.num;

  }

  /**

   * Send wake-up signals to all threads waiting by a key.

   * @param kbuf the pointer to the key region.

   * @param ksiz the size of the key region.

   * @return the number of threads waiting for the signal.

   */

  size_t broadcast(const char* kbuf, size_t ksiz) {

    _assert_(kbuf && ksiz <= MEMMAXSIZ);

    std::string key(kbuf, ksiz);

    return broadcast(key);

  }

  /**

   * Send wake-up signals to all threads waiting by a key.

   * @param key the key.

   * @return the number of threads waiting for the signal.

   */

  size_t broadcast(const std::string& key) {

    _assert_(true);

    Slot* slot = get_slot(key);

    ScopedMutex lock(&slot->mutex);

    CountMap::iterator cit = slot->counter.find(key);

    if (cit == slot->counter.end() || cit->second.num < 1) return 0;

    cit->second.wake = cit->second.num;

    slot->cond.broadcast();

    return cit->second.num;

  }

  /**

   * Send wake-up signals to all threads waiting by each key.

   * @return the number of threads waiting for the signal.

   */

  size_t broadcast_all() {

    _assert_(true);

    size_t sum = 0;

    for (size_t i = 0; i < SLOTNUM; i++) {

      Slot* slot = slots_ + i;

      ScopedMutex lock(&slot->mutex);

      CountMap::iterator cit = slot->counter.begin();

      CountMap::iterator citend = slot->counter.end();

      while (cit != citend) {

        if (cit->second.num > 0) {

          cit->second.wake = cit->second.num;

          sum += cit->second.num;

        }

        slot->cond.broadcast();

        ++cit;

      }

    }

    return sum;

  }

  /**

   * Get the total number of threads waiting for signals.

   * @return the total number of threads waiting for signals.

   */

  size_t count() {

    _assert_(true);

    size_t sum = 0;

    for (size_t i = 0; i < SLOTNUM; i++) {

      Slot* slot = slots_ + i;

      ScopedMutex lock(&slot->mutex);

      CountMap::iterator cit = slot->counter.begin();

      CountMap::iterator citend = slot->counter.end();

      while (cit != citend) {

        sum += cit->second.num;

        ++cit;

      }

    }

    return sum;

  }

 private:

  /**

   * Counter for waiting threads.

   */

  struct Count {

    size_t num;                          ///< waiting threads

    size_t wake;                         ///< waking threads

  };

  /**

   * Slot of a key space.

   */

  struct Slot {

    CondVar cond;                        ///< condition variable

    Mutex mutex;                         ///< mutex

    CountMap counter;                    ///< counter

  };

  /**

   * Get the slot corresponding a key.

   * @param key the key.

   * @return the slot corresponding the key.

   */

  Slot* get_slot(const std::string& key) {

    return slots_ + hashmurmur(key.data(), key.size()) % SLOTNUM;

  }

  /** The slot array. */

  Slot slots_[SLOTNUM];

};

/**

 * Key of thread specific data.

 */

class TSDKey {

 public:

  /**

   * Default constructor.

   */

  explicit TSDKey();

  /**

   * Constructor.

   * @param dstr the destructor for the value.

   */

  explicit TSDKey(void (*dstr)(void*));

  /**

   * Destructor.

   */

  ~TSDKey();

  /**

   * Set the value.

   * @param ptr an arbitrary pointer.

   */

  void set(void* ptr);

  /**

   * Get the value.

   * @return the value.

   */

  void* get() const ;

 private:

  /** Opaque pointer. */

  void* opq_;

};

/**

 * Smart pointer to thread specific data.

 */

template <class TYPE>

class TSD {

 public:

  /**

   * Default constructor.

   */

  explicit TSD() : key_(delete_value) {

    _assert_(true);

  }

  /**

   * Destructor.

   */

  ~TSD() {

    _assert_(true);

    TYPE* obj = (TYPE*)key_.get();

    if (obj) {

      delete obj;

      key_.set(NULL);

    }

  }

  /**

   * Dereference operator.

   * @return the reference to the inner object.

   */

  TYPE& operator *() {

    _assert_(true);