#include "tthread.h"
#include "tthreadP.h"

#include "boost/thread/thread.hpp"
#include "boost/thread/condition.hpp"
#include "boost/thread/tss.hpp"
#include "boost/thread/xtime.hpp"
#include <queue>

#ifdef WIN32
#include <windows.h>
#define WM_THREAD_NOTIFICATION (WM_USER + 10)
#else

#endif

DEFINE_CLASS_CODE(TThread::Runnable, 21)

//==============================================================================

namespace {
#ifdef WIN32
HWND MainHandle;
#else
Display *TheDisplay;
Window TheMainWindow;
#endif

//------------------------------------------------------------------------------

class Thread {
public:
  Thread();
  Thread(const TThread::RunnableP &runnable);

  ~Thread();

  void join();
  void cancel();

#if defined(WIN32) && (_MSC_VER == 1200)
#pragma warning(disable : 4290)
#endif

  static void milestone() throw(TThread::Interrupt);

#if defined(WIN32) && (_MSC_VER == 1200)
#pragma warning(default : 4290)
#endif

  class Imp;

private:
  friend class ThreadGroup;
  Imp *m_imp;
};

//------------------------------------------------------------------------------

class ThreadGroup {
public:
  ThreadGroup();
  ~ThreadGroup();

  void add(Thread *thread);
  void joinAll();

private:
  class Imp;
  Imp *m_imp;
};

//------------------------------------------------------------------------------

template <class T>
class QueueT {
public:
  QueueT(int slotCount)
      : m_items()
      , m_slotCount(slotCount)
      , m_notEmpty()
      , m_notFull()
      , m_mutex() {}

  ~QueueT() {}

  void put(const T &item) {
    TThread::ScopedLock sl(m_mutex);
    while (m_items.size() == m_slotCount) m_notFull.wait(sl);
    m_items.push(item);
    m_notEmpty.notify_one();
  }

  T get() {
    TThread::ScopedLock sl(m_mutex);

    while (m_items.size() == 0) m_notEmpty.wait(sl);

    m_notFull.notify_one();
    T item = m_items.front();
    m_items.pop();
    return item;
  }

  int size() {
    TThread::ScopedLock sl(m_mutex);
    int size = m_items.size();
    return size;
  }

private:
  std::queue<T> m_items;
  TThread::Condition m_notEmpty;
  TThread::Condition m_notFull;
  TThread::Mutex m_mutex;
  int m_slotCount;
};

}  // anonymous namespace

//------------------------------------------------------------------------------

void TThread::setMainThreadId(TThread::ThreadInfo *info) {
  assert(info);
#ifdef WIN32
  MainHandle = info->mainHandle;
#else
  TheDisplay    = info->dpy;
  TheMainWindow = info->win;
#endif
}

//------------------------------------------------------------------------------

#ifdef WIN32
ULONG TThread::getMainShellHandle() { return ULONG(MainHandle); }
#endif

//==============================================================================

class BoostRunnable {
public:
  BoostRunnable(const TThread::RunnableP &runnable, Thread::Imp *threadImp)
      : m_runnable(runnable), m_threadImp(threadImp) {}

  void operator()();

  TThread::RunnableP m_runnable;
  Thread::Imp *m_threadImp;
};

//==============================================================================

class Thread::Imp {
public:
  Imp() : m_boostThread(0), m_isCanceled(false), m_stateMutex() {}

  Imp(const TThread::RunnableP &runnable)
      : m_isCanceled(false), m_stateMutex() {
    m_boostThread = new boost::thread(BoostRunnable(runnable, this));
  }

  ~Imp() {
    if (m_boostThread) delete m_boostThread;
  }

  boost::thread *m_boostThread;
  boost::mutex m_stateMutex;
  bool m_isCanceled;
  long m_kkkk;

  enum State { Running, Canceled };

  static boost::mutex m_mutex;
  static std::map<long, State> m_state;

  class Key {
  public:
    Key(long id) : m_id(id) {}
    long m_id;
  };

  static boost::thread_specific_ptr<Key> m_key;
};

boost::mutex Thread::Imp::m_mutex;
std::map<long, Thread::Imp::State> Thread::Imp::m_state;

boost::thread_specific_ptr<Thread::Imp::Key> Thread::Imp::m_key;

void BoostRunnable::operator()() {
  Thread::Imp::m_key.reset(new Thread::Imp::Key(reinterpret_cast<long>(this)));
  m_threadImp->m_kkkk = reinterpret_cast<long>(this);

  {
    boost::mutex::scoped_lock sl(Thread::Imp::m_mutex);
    Thread::Imp::m_state[m_threadImp->m_kkkk] = Thread::Imp::Running;
  }

  assert(m_runnable);
  m_runnable->run();

  {
    boost::mutex::scoped_lock sl(Thread::Imp::m_mutex);
    Thread::Imp::m_state.erase(reinterpret_cast<long>(this));
  }
}

//==============================================================================

Thread::Thread() : m_imp(new Thread::Imp) {}

//------------------------------------------------------------------------------

Thread::Thread(const TThread::RunnableP &runnable) : m_imp(new Imp(runnable)) {}

//------------------------------------------------------------------------------

Thread::~Thread() { delete m_imp; }

//------------------------------------------------------------------------------

void Thread::join() {
  assert(m_imp->m_boostThread);
  m_imp->m_boostThread->join();
}

//------------------------------------------------------------------------------

void Thread::cancel() {
  boost::mutex::scoped_lock sl(Thread::Imp::m_mutex);

  std::map<long, Thread::Imp::State>::iterator it =
      Thread::Imp::m_state.find(m_imp->m_kkkk);
  if (it != Thread::Imp::m_state.end())
    Thread::Imp::m_state[m_imp->m_kkkk] = Thread::Imp::Canceled;
}

//------------------------------------------------------------------------------
// static member function
#if defined(WIN32) && (_MSC_VER == 1200)
#pragma warning(disable : 4290)
#endif

void Thread::milestone() throw(TThread::Interrupt) {
  boost::mutex::scoped_lock sl(Thread::Imp::m_mutex);
  Thread::Imp::Key key = *Thread::Imp::m_key.get();
  Thread::Imp::m_state.find(key.m_id);

  if (Thread::Imp::m_state[key.m_id]) throw TThread::Interrupt();
}

#if defined(WIN32) && (_MSC_VER == 1200)
#pragma warning(default : 4290)
#endif

//==============================================================================

class ThreadGroup::Imp {
public:
  Imp() : m_boostThreadGroup() {}
  ~Imp() {}

  boost::thread_group m_boostThreadGroup;
};

ThreadGroup::ThreadGroup() : m_imp(new Imp) {}

ThreadGroup::~ThreadGroup() { delete m_imp; }

void ThreadGroup::add(Thread *thread) {
  m_imp->m_boostThreadGroup.add_thread(thread->m_imp->m_boostThread);
}

void ThreadGroup::joinAll() { m_imp->m_boostThreadGroup.join_all(); }

//==============================================================================

#if defined(WIN32) && (_MSC_VER == 1200)
#pragma warning(disable : 4290)
#endif

void TThread::milestone() throw(TThread::Interrupt) { Thread::milestone(); }

#if defined(WIN32) && (_MSC_VER == 1200)
#pragma warning(default : 4290)
#endif

//==============================================================================

class TThread::Mutex::Imp {
public:
  boost::mutex m_mutex;

  Imp() : m_mutex() {}
  ~Imp() {}
};

TThread::Mutex::Mutex() : m_imp(new Imp) {}

TThread::Mutex::~Mutex() { delete m_imp; }

//==============================================================================

class TThread::ScopedLock::Imp {
public:
  boost::mutex::scoped_lock *m_sl;

  Imp(boost::mutex &mutex) : m_sl(new boost::mutex::scoped_lock(mutex)) {}

  ~Imp() {
    m_sl->unlock();
    delete m_sl;
  }
};

TThread::ScopedLock::ScopedLock(Mutex &mutex)
    : m_imp(new Imp(mutex.m_imp->m_mutex)) {}

TThread::ScopedLock::~ScopedLock() { delete m_imp; }

//==============================================================================

class TThread::Condition::Imp {
public:
  boost::condition m_condition;

  Imp() : m_condition() {}
  ~Imp() {}
};

TThread::Condition::Condition() : m_imp(new Imp()) {}

TThread::Condition::~Condition() { delete m_imp; }

void TThread::Condition::wait(ScopedLock &lock) {
  m_imp->m_condition.wait(*(lock.m_imp->m_sl));
}

bool TThread::Condition::wait(ScopedLock &lock, long timeout) {
  boost::xtime xt;
  boost::xtime_get(&xt, boost::TIME_UTC);
  xt.nsec += timeout * 1000;
  xt.sec += timeout / 1000;
  return m_imp->m_condition.timed_wait(*(lock.m_imp->m_sl), xt);
}

void TThread::Condition::notifyOne() { m_imp->m_condition.notify_one(); }

void TThread::Condition::notifyAll() { m_imp->m_condition.notify_all(); }

//==============================================================================

TThread::Msg::Msg() {}

//------------------------------------------------------------------------------

void TThread::Msg::send() {
  Msg *msg = clone();

#ifdef WIN32
  /*
Non viene utilizzato PostThreadMessage perche' se l'applicazione
si trova in un modal loop (esempio MessageBox) oppure si sta
facendo move o resize di una finestra i messaggi non giungono al
message loop.
http://support.microsoft.com/default.aspx?scid=KB;EN-US;q183116&
*/

  /*
BOOL rc = PostThreadMessage(
getMainThreadId(),      // thread identifier
WM_THREAD_NOTIFICATION, // message
WPARAM(msg),            // first message parameter
0);                     // second message parameter
*/
  PostMessage(HWND(getMainShellHandle()), WM_THREAD_NOTIFICATION, WPARAM(msg),
              0);
#else
  XClientMessageEvent clientMsg;
  clientMsg.type         = ClientMessage;
  clientMsg.window       = TheMainWindow;
  clientMsg.format       = 32;
  clientMsg.message_type = Msg::MsgId();
  clientMsg.data.l[0]    = (long)msg;
  // Status status =
  XSendEvent(TheDisplay, TheMainWindow, 0, NoEventMask, (XEvent *)&clientMsg);
  XFlush(TheDisplay);
#endif
}

//------------------------------------------------------------------------------

// statica
UINT TThread::Msg::MsgId() {
#ifdef WIN32
  return WM_THREAD_NOTIFICATION;
#else
  static Atom atom = 0;
  if (!atom) {
    atom = XInternAtom(TheDisplay, "ThreadMessage", false);
    assert(atom);
  }
  return atom;
#endif
}

//==============================================================================

class TThread::Executor::Imp : public TSmartObject {
public:
  typedef TSmartPointerT<TThread::Executor::Imp> ImpP;

  //---------------------------------------------------

  class Worker : public Runnable {
  public:
    Worker(ImpP owner) : Runnable(), m_owner(owner) {}
    ~Worker() {}

    void run();
    void doCleanup();

    ImpP m_owner;
  };

  //---------------------------------------------------

  bool m_suspend;
  bool m_threadHasToDie;

  UINT m_threadCount;
  Mutex m_mutex;

  Condition m_cond;
  Condition m_taskQueueEmpty;

  //------
  Condition m_taskQueueNotEmpty;
  //------

  std::queue<TThread::RunnableP> m_tasks;
  std::map<long, Thread *> m_workerThreads;

  Imp(int threadsCount, bool suspend)
      : TSmartObject()
      , m_suspend(suspend)
      , m_threadHasToDie(false)
      , m_threadCount(threadsCount)
      , m_tasks()
      , m_workerThreads()
      , m_mutex()
      , m_cond(){};

  ~Imp() {}
};

//------------------------------------------------------------------------------

TThread::Executor::Executor(int threadsCount, bool suspend)
    : m_imp(new Imp(threadsCount, suspend)) {
  m_imp->addRef();
}

//------------------------------------------------------------------------------

TThread::Executor::~Executor() {
  {
    TThread::ScopedLock sl(m_imp->m_mutex);

    if (m_imp->m_suspend) {
      m_imp->m_threadHasToDie = true;
      m_imp->m_taskQueueNotEmpty.notifyAll();
    }
  }

  m_imp->release();
}

//------------------------------------------------------------------------------

void TThread::Executor::Imp::Worker::run() {
  try {
    while (true) {
      // check if thread has been canceled
      Thread::milestone();

      // get the next task
      RunnableP task = 0;

      {
        ScopedLock sl(m_owner->m_mutex);
        if (m_owner->m_tasks.empty()) {
          // la lista di task e' stata esaurita
          if (m_owner->m_suspend) {
            // il thread deve sospendersi
            m_owner->m_taskQueueNotEmpty.wait(sl);

            // a questo punto il thread e' stato risvegliato
            if (m_owner->m_threadHasToDie) {
              doCleanup();
              return;
            }
          } else {
            // il thread sta per morire -> bisogna eliminarlo dalla lista dei
            // worker thread
            doCleanup();
            return;
          }
        }

        if (!m_owner->m_tasks.empty()) {
          task = m_owner->m_tasks.front();
          m_owner->m_tasks.pop();
        }
      }

      if (task) task->run();

      // check if thread has been canceled
      Thread::milestone();
    }
  } catch (TThread::Interrupt &) {
    // m_owner->m_cond.notifyOne();
  } catch (...) {
    // eccezione non prevista --> bisogna eliminare il thread
    // dalla lista dei worker thread

    ScopedLock sl(m_owner->m_mutex);
    doCleanup();
  }
}

//------------------------------------------------------------------------------

void TThread::Executor::Imp::Worker::doCleanup() {
  std::map<long, Thread *>::iterator it =
      m_owner->m_workerThreads.find(reinterpret_cast<long>(this));

  if (it != m_owner->m_workerThreads.end()) {
    Thread *thread = it->second;
    delete thread;
    m_owner->m_workerThreads.erase(it);
  }

  if (m_owner->m_workerThreads.size() == 0)
    m_owner->m_taskQueueEmpty.notifyAll();
}

//------------------------------------------------------------------------------

void TThread::Executor::addTask(const RunnableP &task) {
  TThread::ScopedLock sl(m_imp->m_mutex);
  m_imp->m_tasks.push(task);
  if (m_imp->m_workerThreads.size() < m_imp->m_threadCount) {
    TThread::Executor::Imp::Worker *worker =
        new TThread::Executor::Imp::Worker(m_imp);

    m_imp->m_workerThreads[reinterpret_cast<long>(worker)] = new Thread(worker);
  } else {
    if (m_imp->m_suspend)
      // risveglia uno dei thread in attesa
      m_imp->m_taskQueueNotEmpty.notifyOne();
  }
}

//------------------------------------------------------------------------------

void TThread::Executor::clear() {
  ScopedLock sl(m_imp->m_mutex);

  while (!m_imp->m_tasks.empty()) m_imp->m_tasks.pop();
}

//------------------------------------------------------------------------------

void TThread::Executor::cancel() {
  {
    ScopedLock sl(m_imp->m_mutex);

    while (!m_imp->m_tasks.empty()) m_imp->m_tasks.pop();
  }

  while (true) {
    Thread *thread = 0;
    {
      ScopedLock sl(m_imp->m_mutex);
      if (m_imp->m_workerThreads.empty())
        break;
      else {
        std::map<long, Thread *>::iterator it = m_imp->m_workerThreads.begin();
        thread = it->second;
        m_imp->m_workerThreads.erase(it);

        if (thread) thread->cancel();
      }
    }
  }
}

//------------------------------------------------------------------------------

void TThread::Executor::wait() {
  TThread::ScopedLock sl(m_imp->m_mutex);

  while (m_imp->m_workerThreads.size()) m_imp->m_taskQueueEmpty.wait(sl);
}

//------------------------------------------------------------------------------

bool TThread::Executor::wait(long timeout) {
  TThread::ScopedLock sl(m_imp->m_mutex);

  bool expired = false;
  while (m_imp->m_workerThreads.size())
    expired = m_imp->m_taskQueueEmpty.wait(sl, timeout);

  return expired;
}

//------------------------------------------------------------------------------

int TThread::Executor::getThreadCount() {
  TThread::ScopedLock sl(m_imp->m_mutex);
  return m_imp->m_workerThreads.size();
}

//------------------------------------------------------------------------------

int TThread::Executor::getTaskCount() {
  TThread::ScopedLock sl(m_imp->m_mutex);
  return m_imp->m_tasks.size();
}