ThreadMgr.cpp

Go to the documentation of this file.

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

//----- Included files ---------------------------------------------------------

// 1. Precompiled header

// 2. My own header
#include <xmsinterp/thread/ThreadMgr.h>

// 3. Standard library headers

// 4. External library headers
#include <boost/thread.hpp>

// 5. Shared code headers
#include <xmscore/stl/vector.h>
#include <xmscore/misc/Observer.h>
#include <xmsinterp/thread/ThreadLoop.h>

// 6. Non-shared code headers

//----- Forward declarations ---------------------------------------------------

//----- External globals -------------------------------------------------------

//----- Namespace declaration --------------------------------------------------
namespace xms
{
//----- Constants / Enumerations -----------------------------------------------

//----- Classes / Structs ------------------------------------------------------
class ThreadMgrImpl : public ThreadMgr
{
public:
  struct myThread
  {
    myThread(BSHP<ThreadLoop> a_)
    : m_(a_)
    {
    }

    void operator()();   
    BSHP<ThreadLoop> m_; 
  };

  ThreadMgrImpl()
  : m_prog()
  , m_nThread(-1)
  , m_nIter(-1)
  {
  }

  void SetThreadLoopClass(BSHP<ThreadLoop> a_) override { m_threader = a_; }
  void ExplicitlySetNumThreads(int a_nThreads) override { m_nThread = a_nThreads; }
  void RunThreads(int a_nIter) override;
  void SetObserver(BSHP<Observer> a_prog) override { m_prog = a_prog; }
  std::vector<BSHP<ThreadLoop>> Threads() override;

  void SetupThreads();
  void MonitorThreads();
  void FinishThreads();

  BSHP<ThreadLoop> m_threader; 
  BSHP<Observer> m_prog;       
  int m_nThread;               
  int m_nIter;                 
  VecInt m_beg;                
  VecInt m_thrdNumIter;        
  VecInt m_counts;             
  std::vector<myThread> m_vMyThread;     
  std::vector<boost::thread> m_vThreads; 
};
//----- Internal functions -----------------------------------------------------

//----- Class / Function definitions -------------------------------------------

//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
void ThreadMgrImpl::myThread::operator()()
{
  m_->DoWork();
} // InterpIdwImpl::myThread::operator
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
BSHP<ThreadMgr> ThreadMgr::New()
{
  BSHP<ThreadMgr> ptr(new ThreadMgrImpl());
  return ptr;
} // ThreadMgr::New
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
ThreadMgr::ThreadMgr()
{
} // ThreadMgr::ThreadMgr
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
ThreadMgr::~ThreadMgr()
{
} // ThreadMgr::~ThreadMgr
  //------------------------------------------------------------------------------
  //------------------------------------------------------------------------------
std::vector<BSHP<ThreadLoop>> ThreadMgrImpl::Threads()
{
  std::vector<BSHP<ThreadLoop>> threads;
  for (size_t i = 0; i < m_vMyThread.size(); ++i)
  {
    threads.push_back(m_vMyThread[i].m_);
  }
  return threads;
} // ThreadMgrImpl::Threads
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
void ThreadMgrImpl::RunThreads(int a_nIter)
{
  m_nIter = a_nIter;
  if (m_nThread < 1)
    m_nThread = boost::thread::hardware_concurrency() - 1;
  if (m_nThread < 1)
    m_nThread = 1;
  SetupThreads();
  MonitorThreads();
  FinishThreads();
} // ThreadMgrImpl::RunThreads
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
void ThreadMgrImpl::SetupThreads()
{
  m_beg.assign(m_nThread, 0);
  m_counts.assign(m_nThread, 0);

  int cnt = (m_nIter / m_nThread) + 1;
  m_thrdNumIter.assign(m_nThread, cnt);
  for (int i = 1; i < m_nThread; ++i)
    m_beg[i] = m_beg[i - 1] + cnt;
  m_thrdNumIter.back() = m_nIter - m_beg.back();

  for (int i = 0; i < m_nThread; ++i)
  {
    BSHP<ThreadLoop> ptr = m_threader->CreateForNewThread();
    ptr->SetStartNumIterCnt(m_beg[i], m_thrdNumIter[i], &m_counts[i]);
    m_vMyThread.push_back(myThread(ptr));
  }
  // create the threads
  for (int i = 0; i < m_nThread; ++i)
  {
    m_vThreads.push_back(boost::thread(m_vMyThread[i]));
  }
} // ThreadMgrImpl::SetupThreads
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
void ThreadMgrImpl::MonitorThreads()
{
  double p(0.01);
  if (m_prog)
    m_prog->ProgressStatus(p);

  bool done(false);
  int sum, cnt, cntBase;
  cntBase = cnt = m_nIter / 100;
  while (!done)
  {
    sum = 0;
    for (size_t i = 0; i < m_counts.size(); ++i)
      sum += m_counts[i];
    while (sum >= cnt && p < 1)
    {
      p += .01;
      if (m_prog)
        m_prog->ProgressStatus(p);
      cnt += cntBase;
    }
    boost::this_thread::sleep(boost::posix_time::millisec(100));
    if (p >= 1)
      done = true;
  }
} // ThreadMgrImpl::MonitorThreads
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
void ThreadMgrImpl::FinishThreads()
{
  for (size_t i = 0; i < m_vThreads.size(); ++i)
    m_vThreads[i].join();
} // ThreadMgrImpl::FinishThreads

} // namespace xms