// File: crn_threading_pthreads.cpp
// See Copyright Notice and license at the end of include/crnlib.h
#include "crn_core.h"
#include "crn_threading_pthreads.h"
#include "crn_timer.h"

#if CRNLIB_USE_PTHREADS_API

#ifdef WIN32
#pragma comment(lib, "../ext/libpthread/lib/pthreadVC2.lib")
#include "crn_winhdr.h"
#endif

#ifdef __GNUC__
#include <sys/sysinfo.h>
#endif

#ifdef WIN32
#include <process.h>
#endif

namespace crnlib
{
   uint g_number_of_processors = 1;

   void crn_threading_init()
   {
#ifdef WIN32
      SYSTEM_INFO g_system_info;
      GetSystemInfo(&g_system_info);
      g_number_of_processors = math::maximum<uint>(1U, g_system_info.dwNumberOfProcessors);
#elif defined(__GNUC__)
      g_number_of_processors = math::maximum<int>(1, get_nprocs());
#else
      g_number_of_processors = 1;
#endif
   }

   crn_thread_id_t crn_get_current_thread_id()
   {
      // FIXME: Not portable
      return static_cast<crn_thread_id_t>(pthread_self());
   }

   void crn_sleep(unsigned int milliseconds)
   {
#ifdef WIN32
      struct timespec interval;
      interval.tv_sec = milliseconds / 1000;
      interval.tv_nsec = (milliseconds % 1000) * 1000000L;
      pthread_delay_np(&interval);
#else
      while (milliseconds)
      {
         int msecs_to_sleep = CRNLIB_MIN(milliseconds, 1000);
         usleep(msecs_to_sleep * 1000);
         milliseconds -= msecs_to_sleep;
      }
#endif
   }

   mutex::mutex(unsigned int spin_count)
   {
      spin_count;

      if (pthread_mutex_init(&m_mutex, NULL))
         crnlib_fail("mutex::mutex: pthread_mutex_init() failed", __FILE__, __LINE__);

#ifdef CRNLIB_BUILD_DEBUG
      m_lock_count = 0;
#endif
   }

   mutex::~mutex()
   {
#ifdef CRNLIB_BUILD_DEBUG
      if (m_lock_count)
         crnlib_assert("mutex::~mutex: mutex is still locked", __FILE__, __LINE__);
#endif
      if (pthread_mutex_destroy(&m_mutex))
         crnlib_assert("mutex::~mutex: pthread_mutex_destroy() failed", __FILE__, __LINE__);
   }

   void mutex::lock()
   {
      pthread_mutex_lock(&m_mutex);
#ifdef CRNLIB_BUILD_DEBUG
      m_lock_count++;
#endif
   }

   void mutex::unlock()
   {
#ifdef CRNLIB_BUILD_DEBUG
      if (!m_lock_count)
         crnlib_assert("mutex::unlock: mutex is not locked", __FILE__, __LINE__);
      m_lock_count--;
#endif
      pthread_mutex_unlock(&m_mutex);
   }

   void mutex::set_spin_count(unsigned int count)
   {
      count;
   }

   semaphore::semaphore(long initialCount, long maximumCount, const char* pName)
   {
      maximumCount, pName;
      CRNLIB_ASSERT(maximumCount >= initialCount);
      if (sem_init(&m_sem, 0, initialCount))
      {
         CRNLIB_FAIL("semaphore: sem_init() failed");
      }
   }

   semaphore::~semaphore()
   {
      sem_destroy(&m_sem);
   }

   void semaphore::release(long releaseCount)
   {
      CRNLIB_ASSERT(releaseCount >= 1);

      int status = 0;
#ifdef WIN32
      if (1 == releaseCount)
         status = sem_post(&m_sem);
      else
         status = sem_post_multiple(&m_sem, releaseCount);
#else
      while (releaseCount > 0)
      {
         status = sem_post(&m_sem);
         if (status)
            break;
         releaseCount--;
      }
#endif

      if (status)
      {
         CRNLIB_FAIL("semaphore: sem_post() or sem_post_multiple() failed");
      }
   }

   void semaphore::try_release(long releaseCount)
   {
      CRNLIB_ASSERT(releaseCount >= 1);

#ifdef WIN32
      if (1 == releaseCount)
         sem_post(&m_sem);
      else
         sem_post_multiple(&m_sem, releaseCount);
#else
      while (releaseCount > 0)
      {
         sem_post(&m_sem);
         releaseCount--;
      }
#endif
   }

   bool semaphore::wait(uint32 milliseconds)
   {
      int status;
      if (milliseconds == cUINT32_MAX)
      {
         status = sem_wait(&m_sem);
      }
      else
      {
         struct timespec interval;
         interval.tv_sec = milliseconds / 1000;
         interval.tv_nsec = (milliseconds % 1000) * 1000000L;
         status = sem_timedwait(&m_sem, &interval);
      }

      if (status)
      {
         if (errno != ETIMEDOUT)
         {
            CRNLIB_FAIL("semaphore: sem_wait() or sem_timedwait() failed");
         }
         return false;
      }

      return true;
   }

   spinlock::spinlock()
   {
      if (pthread_spin_init(&m_spinlock, 0))
      {
         CRNLIB_FAIL("spinlock: pthread_spin_init() failed");
      }
   }

   spinlock::~spinlock()
   {
      pthread_spin_destroy(&m_spinlock);
   }

   void spinlock::lock()
   {
      if (pthread_spin_lock(&m_spinlock))
      {
         CRNLIB_FAIL("spinlock: pthread_spin_lock() failed");
      }
   }

   void spinlock::unlock()
   {
      if (pthread_spin_unlock(&m_spinlock))
      {
         CRNLIB_FAIL("spinlock: pthread_spin_unlock() failed");
      }
   }

   task_pool::task_pool() :
      m_num_threads(0),
      m_tasks_available(0, 32767),
      m_all_tasks_completed(0, 1),
      m_total_submitted_tasks(0),
      m_total_completed_tasks(0),
      m_exit_flag(false)
   {
      utils::zero_object(m_threads);
   }

   task_pool::task_pool(uint num_threads) :
      m_num_threads(0),
      m_tasks_available(0, 32767),
      m_all_tasks_completed(0, 1),
      m_total_submitted_tasks(0),
      m_total_completed_tasks(0),
      m_exit_flag(false)
   {
      utils::zero_object(m_threads);

      bool status = init(num_threads);
      CRNLIB_VERIFY(status);
   }

   task_pool::~task_pool()
   {
      deinit();
   }

   bool task_pool::init(uint num_threads)
   {
      CRNLIB_ASSERT(num_threads <= cMaxThreads);
      num_threads = math::minimum<uint>(num_threads, cMaxThreads);

      deinit();

      bool succeeded = true;

      m_num_threads = 0;
      while (m_num_threads < num_threads)
      {
         int status = pthread_create(&m_threads[m_num_threads], NULL, thread_func, this);
         if (status)
         {
            succeeded = false;
            break;
         }

         m_num_threads++;
      }

      if (!succeeded)
      {
         deinit();
         return false;
      }

      return true;
   }

   void task_pool::deinit()
   {
      if (m_num_threads)
      {
         join();

         atomic_exchange32(&m_exit_flag, true);

         m_tasks_available.release(m_num_threads);

         for (uint i = 0; i < m_num_threads; i++)
            pthread_join(m_threads[i], NULL);

         m_num_threads = 0;

         atomic_exchange32(&m_exit_flag, false);
      }

      m_task_stack.clear();
      m_total_submitted_tasks = 0;
      m_total_completed_tasks = 0;
   }

   bool task_pool::queue_task(task_callback_func pFunc, uint64 data, void* pData_ptr)
   {
      CRNLIB_ASSERT(pFunc);

      task tsk;
      tsk.m_callback = pFunc;
      tsk.m_data = data;
      tsk.m_pData_ptr = pData_ptr;
      tsk.m_flags = 0;

      atomic_increment32(&m_total_submitted_tasks);
      if (!m_task_stack.try_push(tsk))
      {
         atomic_increment32(&m_total_completed_tasks);
         return false;
      }

      m_tasks_available.release(1);

      return true;
   }

   // It's the object's responsibility to delete pObj within the execute_task() method, if needed!
   bool task_pool::queue_task(executable_task* pObj, uint64 data, void* pData_ptr)
   {
      CRNLIB_ASSERT(pObj);

      task tsk;
      tsk.m_pObj = pObj;
      tsk.m_data = data;
      tsk.m_pData_ptr = pData_ptr;
      tsk.m_flags = cTaskFlagObject;

      atomic_increment32(&m_total_submitted_tasks);
      if (!m_task_stack.try_push(tsk))
      {
         atomic_increment32(&m_total_completed_tasks);
         return false;
      }

      m_tasks_available.release(1);

      return true;
   }

   void task_pool::process_task(task& tsk)
   {
      if (tsk.m_flags & cTaskFlagObject)
         tsk.m_pObj->execute_task(tsk.m_data, tsk.m_pData_ptr);
      else
         tsk.m_callback(tsk.m_data, tsk.m_pData_ptr);

      if (atomic_increment32(&m_total_completed_tasks) == m_total_submitted_tasks)
      {
         // Try to signal the semaphore (the max count is 1 so this may actually fail).
         m_all_tasks_completed.try_release();
      }
   }

   void task_pool::join()
   {
      // Try to steal any outstanding tasks. This could cause one or more worker threads to wake up and immediately go back to sleep, which is wasteful but should be harmless.
      task tsk;
      while (m_task_stack.pop(tsk))
         process_task(tsk);

      // At this point the task stack is empty.
      // Now wait for all concurrent tasks to complete. The m_all_tasks_completed semaphore has a max count of 1, so it's possible it could have saturated to 1 as the tasks
      // where issued and asynchronously completed, so this loop may iterate a few times.
      const int total_submitted_tasks = atomic_add32(&m_total_submitted_tasks, 0);
      while (m_total_completed_tasks != total_submitted_tasks)
      {
         // If the previous (m_total_completed_tasks != total_submitted_tasks) check failed the semaphore MUST be eventually signalled once the last task completes.
         // So I think this can actually be an INFINITE delay, but it shouldn't really matter if it's 1ms.
         m_all_tasks_completed.wait(1);
      }
   }

   void * task_pool::thread_func(void *pContext)
   {
      task_pool* pPool = static_cast<task_pool*>(pContext);
      task tsk;

      for ( ; ; )
      {
         if (!pPool->m_tasks_available.wait())
            break;

         if (pPool->m_exit_flag)
            break;

         if (pPool->m_task_stack.pop(tsk))
         {
            pPool->process_task(tsk);
         }
      }

      return NULL;
   }

} // namespace crnlib

#endif // CRNLIB_USE_PTHREADS_API