///////////////////////////////////////////////////////////////////////
// ThreadPool.cpp - Enqueue work items to run on one of a collection //
//                  of threads, created at startup.                  //
//                                                                   //
// Jim Fawcett, CSE687 - Object Oriented Design, Spring 2016         //
///////////////////////////////////////////////////////////////////////

#include <thread>
#include <chrono>
#include <iostream>
#include <sstream>
#include "ThreadPool.h"

using namespace Async;
using Rslt = Logging::StaticLogger<0>;
using Demo = Logging::StaticLogger<1>;
using Dbug = Logging::StaticLogger<2>;

//----< define threadProc and start threads >------------------------

void ThreadPool::start(size_t numThrds)
{
  threadProc_ = [this]()  // all threads use this to acquire their callable objects.
  {
    while (true)
    {
      Demo::write(
        "\n--threadpool queue size = " +
        Utilities::Converter<size_t>::toString(Q_.size())
      );
      CallObj co = Q_.deQ();
      if (!co())
      {
        Dbug::write("\n  received exit request");
        Q_.enQ(co);
        break;
      }
    }
    Demo::write(
      "\n  thread " +
      Utilities::Converter<std::thread::id>::toString(std::this_thread::get_id()) +
      " terminating"
      );
  };
  for (size_t i = 0; i < numThrds; ++i)
  {
    ++numThrds_;
    std::thread t(threadProc_);
    Demo::write(
      "\n--started threadpool thread " +
      Utilities::Converter<std::thread::id>::toString(t.get_id())
    );
    threads_.push_back(std::move(t));
  }
}

void ThreadPool::workItem(CallObj& co)
{
  Q_.enQ(co);
}

void ThreadPool::wait()
{
  Demo::write(
    "\n  entering wait with queue size = " +
    Utilities::Converter<size_t>::toString(Q_.size())
    );
  for (auto& thrd : threads_)
    thrd.join();
  Q_.clear();
  Demo::write(
    "\n--leaving wait with queue size = " +
    Utilities::Converter<size_t>::toString(Q_.size())
    );
}

ThreadPool::~ThreadPool()
{
  for (auto& thrd : threads_)
  {
    if (thrd.joinable())
      thrd.join();
  }
}
//----< test stub >--------------------------------------------------

#ifdef TEST_THREADPOOL

#include <iostream>
using namespace std::chrono_literals;
using namespace Utilities;
using Utils = StringHelper;

int main()
{
  Rslt::attach(&std::cout);
  Rslt::start();
  Demo::attach(&std::cout);
  Demo::start();

  Utils::Title("Testing ThreadPool");

  ThreadPool trpl;
  trpl.start(6);

  ThreadPool::CallObj co = [&trpl]() ->bool { 
    Rslt::write(
      "\n  hello from thread " + 
      Utilities::Converter<std::thread::id>::toString(std::this_thread::get_id())
    ); 
    std::this_thread::sleep_for(20ms);
    return true;
  };

  for (size_t i = 0; i < 20; ++i)
  {
    trpl.workItem(co);
    std::this_thread::sleep_for(1ms);
  }
  ThreadPool::CallObj exit = []() ->bool { return false; };
  trpl.workItem(exit);
  trpl.wait();
}

#endif