/****************************************************************************
 * Copyright (C) 2009-2015 EPAM Systems
 * 
 * This file is part of Indigo toolkit.
 * 
 * This file may be distributed and/or modified under the terms of the
 * GNU General Public License version 3 as published by the Free Software
 * Foundation and appearing in the file LICENSE.GPL included in the
 * packaging of this file.
 * 
 * This file is provided AS IS with NO WARRANTY OF ANY KIND, INCLUDING THE
 * WARRANTY OF DESIGN, MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE.
 ***************************************************************************/

//
// Thread support based on command dispatcher:
//

#ifdef _WIN32
#include <windows.h>
#endif
#ifdef __APPLE__
#include <sys/sysctl.h>
#endif

#include "base_c/os_thread.h"
#include "base_cpp/os_thread_wrapper.h"
#include "base_cpp/exception.h"
#include "base_cpp/tlscont.h"
#include "base_cpp/auto_ptr.h"
#include "base_cpp/profiling.h"

using namespace indigo;

// Messages
enum {
   MSG_NEED_TASK,
   MSG_RECV_COMMAND,
   MSG_RECV_RESULT,
   MSG_RECV_INDEX,
   MSG_NO_TASK,
   MSG_HANDLE_RESULT,
   MSG_SYSPEND_RESULT,
   MSG_SYSPEND_SEMAPHORE,
   MSG_OK,
   MSG_CONNECT,
   MSG_HANDLE_EXCEPTION,
   MSG_TERMINATE
};

// Maximum number of results that are kept in queue if
// _handling_order is HANDLING_ORDER_SERIAL
static const int _MAX_RESULTS = 1000;

OsCommandDispatcher::OsCommandDispatcher (int handling_order, bool same_session_IDs)
{
   _storedResults.setSize(_MAX_RESULTS);
   _storedResults.zeroFill();
   _handling_order = handling_order;
   _session_id = TL_GET_SESSION_ID();
   _last_unique_command_id = 0;
   _same_session_IDs = same_session_IDs;
}

extern "C" THREAD_RET THREAD_MOD _threadFuncStatic (void *param)
{
   OsCommandDispatcher *dispatcher = (OsCommandDispatcher *)param;

   dispatcher->_threadFunc();
   THREAD_END;
}

void OsCommandDispatcher::run ()
{
   _run(3 * osGetProcessorsCount() / 2 + 1);
}

void OsCommandDispatcher::run (int nthreads)
{
   if (nthreads < 0)
      // Use automatic thread count selection
      run();
   else
      _run(nthreads);
}

void OsCommandDispatcher::_run (int nthreads)
{
   _last_command_index = 0;
   _expected_command_index = 0;
   _need_to_terminate = false;
   _exception_to_forward = NULL;

   _left_thread_count = nthreads;

   if (_left_thread_count == 0)
   {
      _startStandalone();
      return;
   }

   _parent_session_ID = TL_GET_SESSION_ID();

   // Create handling threads
   for (int i = 0; i < _left_thread_count; i++)
      osThreadCreate(_threadFuncStatic, this);

   _mainLoop();
}

void OsCommandDispatcher::_mainLoop ()
{
   profTimerStart(t, "dispatcher.main_loop");

   // Main loop
   int msg;
   while (_left_thread_count != 0)
   {
      void *parameter;
      _baseMessageSystem.RecvMsg(&msg, &parameter);

      if (msg == MSG_NEED_TASK)
         _onMsgNeedTask();
      if (msg == MSG_HANDLE_RESULT)
         _onMsgHandleResult();
      if (msg == MSG_HANDLE_EXCEPTION)
         _onMsgHandleException((Exception *)parameter);
   }

   if (_exception_to_forward != NULL)
   {
      Exception *cur = _exception_to_forward;
      _exception_to_forward = NULL;
      cur->throwSelf();
   }
}

void OsCommandDispatcher::markToTerminate ()
{
   _need_to_terminate = true;
}

void OsCommandDispatcher::terminate ()
{
   markToTerminate();
   _mainLoop();
}

OsCommand* OsCommandDispatcher::_getVacantCommand ()
{
   OsCommand *command;
   if (_availableCommands.size() == 0) {
      command = _allocateCommand();
      command->unique_id = _last_unique_command_id++;
   }
   else
      command = _availableCommands.pop();

   command->clear();

   return command;
}

OsCommandResult* OsCommandDispatcher::_getVacantResult ()
{
   OsCommandResult *result;

   if (_availableResults.size() == 0)
      result = _allocateResult();
   else
      result = _availableResults.pop();
   result->clear();

   return result;
}

void OsCommandDispatcher::_onMsgNeedTask()
{
   if (_need_to_terminate)
   {
      _privateMessageSystem.SendMsg(MSG_NO_TASK, NULL);
      _left_thread_count--;
      return;
   }

   OsCommandResult *result = _getVacantResult();
   OsCommand *command = _getVacantCommand();

   if (!_setupCommand(*command))
   {
      _availableResults.add(result);
      _availableCommands.add(command);

      _privateMessageSystem.SendMsg(MSG_NO_TASK, NULL);
      _left_thread_count--;
      return;
   }
   _privateMessageSystem.SendMsg(MSG_RECV_INDEX, &_last_command_index);
   _privateMessageSystem.SendMsg(MSG_RECV_COMMAND, command);
   _privateMessageSystem.SendMsg(MSG_RECV_RESULT, result);

   _last_command_index++;
}

void OsCommandDispatcher::_handleResultWithCheck (OsCommandResult *result)
{
   Exception *exception = 0;
   try 
   {
      if (!_need_to_terminate)
         _handleResult(*result);
   }
   catch (Exception &e)
   {
      exception = e.clone();
   }
   catch (...)
   {
      exception = Exception("Unknown exception").clone();
   }
   if (exception != NULL)
      _handleException(exception);
}

void OsCommandDispatcher::_onMsgHandleResult ()
{
   OsCommandResult *result;
   OsCommand *command;
   int index, msg;
   void *param;

   _privateMessageSystem.RecvMsg(&msg, &param);
   if (msg != MSG_RECV_INDEX)
      throw Exception("cmdDispatcher::_OnMsgHandleResult: internal error");
   index = *(int *)param;

   if (_handling_order == HANDLING_ORDER_SERIAL)
      if (!_storedResults.isInBound(index))
      {
         profIncCounter("dispatcher.syspend_count", 1);

         _privateMessageSystem.SendMsg(MSG_SYSPEND_RESULT);
         _privateMessageSystem.RecvMsg(&msg, &param);
         if (msg != MSG_SYSPEND_SEMAPHORE)
            throw Exception("cmdDispatcher::_OnMsgHandleResult: internal error #2");

         OsSemaphore *sem = (OsSemaphore *)param;
         _syspendedThreads.push(sem);

         return;
      }

   _privateMessageSystem.SendMsg(MSG_OK);
   _recvCommandAndResult(result, command);
   _availableCommands.add(command);

   if (_handling_order == HANDLING_ORDER_ANY)
   {
      // Handle result in parallel mode
      _handleResultWithCheck(result);
      _availableResults.add(result);
   }
   else
   {
      // Handle results in correct order
      _storedResults[index] = result;
      while (_storedResults[_expected_command_index] != NULL)
      {
         OsCommandResult *current = _storedResults[_expected_command_index];
         _storedResults[_expected_command_index] = NULL;

         _handleResultWithCheck(current);
         _availableResults.add(current);
         _expected_command_index++;
      }
      _storedResults.setOffset(_expected_command_index);

      _wakeSuspended();
   }
}

void OsCommandDispatcher::_wakeSuspended ()
{
   // Wake up all syspended threads
   for (int i = 0; i < _syspendedThreads.size(); i++)
      _syspendedThreads[i]->Post();
   _syspendedThreads.clear();
}


void OsCommandDispatcher::_onMsgHandleException (Exception *exception)
{
   OsCommandResult *result;
   OsCommand *command;

   _recvCommandAndResult(result, command);
   _availableResults.add(result);
   _availableCommands.add(command);

   _handleException(exception);
}

void OsCommandDispatcher::_handleException (Exception *exception)
{
   if (!_need_to_terminate)
   {
      _need_to_terminate = true;
      // Store exception to correct memory deallocation in the next time...
      TL_DECL(AutoPtr<Exception>, exception_ptr);
      TL_GET(AutoPtr<Exception>, exception_ptr);

      exception_ptr.reset(exception);
      _exception_to_forward = exception;

      _wakeSuspended();
   }
   else
   {
      // This is second exception. Skip it
      delete exception;
   }
}

void OsCommandDispatcher::_threadFunc (void)
{
   qword initial_SID = TL_GET_SESSION_ID();

   if (_same_session_IDs)
      TL_SET_SESSION_ID(_parent_session_ID);

   _prepareThread();

   OsSemaphore syspendSem(0, 1);
   while (true)
   {
      int msg;
      _baseMessageSystem.SendMsg(MSG_NEED_TASK, NULL);

      void *param;
      _privateMessageSystem.RecvMsg(&msg, &param);
      if (msg == MSG_NO_TASK)
         break;
      if (msg != MSG_RECV_INDEX)
         throw Exception("cmdDispatcher::_ThreadFunc: internal error");

      int index;
      index = *(int *)param;

      OsCommandResult *result;
      OsCommand *command;
      _recvCommandAndResult(result, command);

      Exception *exception = NULL;
      try 
      {
         result->clear();
         command->execute(*result);
      }
      catch (Exception &e)
      {
         exception = e.clone();
      }
      catch (...)
      {
         exception = Exception("Unknown exception").clone();
      }

      if (exception != NULL)
      {
         // Forward exception into main thread
         _baseMessageSystem.SendMsg(MSG_HANDLE_EXCEPTION, exception);
         // Send current command and result to add 
         // them to the vacant list
         _privateMessageSystem.SendMsg(MSG_RECV_COMMAND, command);
         _privateMessageSystem.SendMsg(MSG_RECV_RESULT, result);
         continue;
      }

      while (true)
      {
         _baseMessageSystem.SendMsg(MSG_HANDLE_RESULT, NULL);

         _privateMessageSystem.SendMsg(MSG_RECV_INDEX, &index);
         _privateMessageSystem.RecvMsg(&msg, &param);

         if (msg == MSG_SYSPEND_RESULT)
         {
            // Enter syspend mode
            _privateMessageSystem.SendMsg(MSG_SYSPEND_SEMAPHORE, &syspendSem);
            syspendSem.Wait();
            continue;
         }
         else if (msg == MSG_OK)
         {
            _privateMessageSystem.SendMsg(MSG_RECV_COMMAND, command);
            _privateMessageSystem.SendMsg(MSG_RECV_RESULT, result);
            break;
         }
         else
            // This should terminate application
            throw Exception("cmdDispatcher::_ThreadFunc: internal error #2");
      }
   }

   _cleanupThread();

   TL_RELEASE_SESSION_ID(initial_SID);
}

void OsCommandDispatcher::_recvCommandAndResult (OsCommandResult *&result, OsCommand *&command)
{
   for (int i = 0; i < 2; i++)
   {
      void *param;
      int msg;

      _privateMessageSystem.RecvMsg(&msg, &param);
      if (msg == MSG_RECV_RESULT)
         result = (OsCommandResult *)param;
      else if (msg == MSG_RECV_COMMAND)
         command = (OsCommand *)param;
      else
         throw Exception("cmdDispatcher::_RecvCommandAndResult");
   }
}

OsCommandResult* OsCommandDispatcher::_allocateResult ()
{
   // Create empty results
   return new OsCommandResult;
}

void OsCommandDispatcher::_startStandalone ()
{
   OsCommandResult *result = _getVacantResult();
   OsCommand *command = _getVacantCommand();

   while (_setupCommand(*command))
   {
      command->execute(*result);
      _handleResult(*result);

      command->clear();
      result->clear();
   }

   _availableResults.add(result);
   _availableCommands.add(command);
}


int osGetProcessorsCount (void)
{
   static ThreadSafeStaticObj<OsLock> _processors_lock;
   OsLocker locker(_processors_lock.ref());

   static int processors_count = 0;
   if (processors_count == 0)
   {
#ifdef _WIN32
      SYSTEM_INFO info;
      GetSystemInfo(&info);
      processors_count = info.dwNumberOfProcessors;
#elif __APPLE__ // MacOS X
      int mib[2];
      size_t len = sizeof(processors_count);

      /* set the mib for hw.ncpu */
      mib[0] = CTL_HW;
      mib[1] = HW_AVAILCPU;  // alternatively, try HW_NCPU;

      /* get the number of CPUs from the system */
      sysctl(mib, 2, &processors_count, &len, NULL, 0);

      if (processors_count < 1)
      {
         mib[1] = HW_NCPU;
         sysctl(mib, 2, &processors_count, &len, NULL, 0);

         if (processors_count < 1)
            processors_count = 1;
      }

#else
      char line[200];
      FILE *cpuinfo_file = fopen("/proc/cpuinfo", "rt");
      if (cpuinfo_file)
      {
         while (fgets(line, sizeof(line), cpuinfo_file))
            if (!strncmp("processor", line, 9))
               processors_count++;
         fclose(cpuinfo_file);
      }
      if (processors_count == 0)
         processors_count = 1;
#endif
   }

   return processors_count;
}