mt_thread_pool.cpp

Go to the documentation of this file.

/*
 * Copyright (C) Sergey P. Derevyago, 2008.
 *
 * Permission to copy, use, modify, sell and distribute this software is granted
 * provided this copyright notice appears in all copies.
 * This software is provided "as is" without express or implied warranty, and
 * with no claim as to its suitability for any purpose.
 *
 */

#include <ders/thread_pool.hpp>
#include <ders/thread.hpp>

namespace {

using namespace ders;
using namespace std;

struct exec_data;

struct mt_dque : data_queue {
       exec_data& ed;
       sh_mutex mtx;
       data_queue* dq;

       mt_dque(mem_pool& mp, exec_data& e, data_queue* d) : ed(e), mtx(
         new_mutex(mp)), dq(d) {}

       virtual void destroy(mem_pool& mp2) { destroy_this(this, mp2); }
       virtual bool is_intr();
       virtual void set_intr(bool val);
       virtual result read_obj(object_io& rdr, bool wait);
       virtual result write_obj(const object_io& wrtr);
};

struct exec_data : task_opers {
       task& tsk;
       vector<sh_data_queue> dqs;
       dq_vec dqv;
       sh_barrier bar;
       sh_mutex mtx;

       exec_data(mem_pool& mp, task& tk, int bc, const dq_vec& dv);

       virtual void destroy(mem_pool& mp2) { destroy_this(this, mp2); }
       virtual void invoke(void (*fun)(void*), void* arg);
};

typedef sh_ptr<exec_data> sh_exec_data;

struct mt_thrpl : thread_pool {
       mem_pool ownmp;
       bool dstr;
       sh_barrier bar;
       vector<sh_thread> tds;
       sh_exec_data ed;

       mt_thrpl(int cnt);
       ~mt_thrpl();

       virtual void destroy(mem_pool& mp2) { destroy_this(this, mp2); }
       virtual void exec(task& tk, const dq_vec& dv);

       void start();
       void do_exec();
};

sh_data_queue new_mt_dque(mem_pool& mp, exec_data& e, data_queue* d)
{
 mp_newbuf<mt_dque> buf(mp);
 return sh_data_queue(buf.pool(), buf.rls(::new(buf.get()) mt_dque(mp, e, d)));
}

bool mt_dque::is_intr()
{
 auto_mutex amt(*mtx);
 return dq->is_intr();
}

void mt_dque::set_intr(bool val)
{
 {
  auto_mutex amt(*mtx);
  dq->set_intr(val);
 }

 ed.bar->release_all();
}

data_queue::result mt_dque::read_obj(object_io& rdr, bool wait)
{
 for (;;) {
     result rc;
     {
      auto_mutex amt(*mtx);
      rc=dq->read_obj(rdr, wait);
     }

     if (!wait || rc!=empt) return rc;

     if (ed.bar->wait()) return empt;
 }
}

data_queue::result mt_dque::write_obj(const object_io& wrtr)
{
 result rc;
 {
  auto_mutex amt(*mtx);
  rc=dq->write_obj(wrtr);
 }
 if (rc==ok) ed.bar->release_all();

 return rc;
}

sh_exec_data null_exec_data(mem_pool& mp)
{
 return sh_exec_data(mp, 0);
}

sh_exec_data new_exec_data(mem_pool& mp, task& tk, int bc, const dq_vec& dv)
{
 mp_newbuf<exec_data> buf(mp);
 return sh_exec_data(buf.pool(), buf.rls(::new(buf.get()) exec_data(mp, tk, bc,
   dv)));
}

exec_data::exec_data(mem_pool& mp, task& tk, int bc, const dq_vec& dv) :
  tsk(tk), bar(new_barrier(mp, bc)), mtx(new_mutex(mp))
{
 int sz=dv.size();
 dqs.reserve(sz);
 dqv.reserve(sz);

 for (int i=0; i<sz; i++) {
     dqs.push_back(new_mt_dque(mp, *this, dv[i]));
     dqv.push_back(dqs[i].get());
 }
}

void exec_data::invoke(void (*fun)(void*), void* arg)
{
 auto_mutex amt(*mtx);
 fun(arg);
}

void start_fun(void* arg)
{
 static_cast<mt_thrpl*>(arg)->start();
}

mt_thrpl::mt_thrpl(int cnt) :
  dstr(false), bar(new_barrier(ownmp, cnt+1)), ed(null_exec_data(ownmp))
{
 assert(cnt>0);
 for (int i=0; i<cnt; i++) tds.push_back(new_thread(ownmp, start_fun, this));
}

mt_thrpl::~mt_thrpl()
{
 assert(!ed.get());

 assert(!dstr);
 dstr=true;
 bar->wait();

 for (int i=0, end=tds.size(); i<end; i++) tds[i]->join();
}

void mt_thrpl::exec(task& tk, const dq_vec& dv)
{
 assert(!dstr);

 assert(!ed.get());
 ed=new_exec_data(ownmp, tk, bar->get_count(), dv);
 bar->wait();

 do_exec();

 bar->wait();
 ed=null_exec_data(ownmp);
}

void mt_thrpl::start()
{
 for (;;) {
     bar->wait();
     if (dstr) break;

     assert(ed.get());
     do_exec();

     bar->wait();
 }
}

void mt_thrpl::do_exec()
{
 mem_pool mp;

 void* arg;
 {
  auto_mutex amt(*ed->mtx);
  arg=ed->tsk.proc_arg();
 }

 try { ed->tsk.proc(mp, ed->dqv, arg, *ed); }
 catch (...) { hard_assert(false); }
}

}  // unnamed

namespace ders {  // ::ders

sh_thread_pool new_thread_pool(mem_pool& mp, int cnt)
{
 mp_newbuf<mt_thrpl> buf(mp);
 return sh_thread_pool(buf.pool(), buf.rls(::new(buf.get()) mt_thrpl(cnt)));
}

}  // namespace ::ders

---