mt_thread.cpp

Go to the documentation of this file.

/*
 * Copyright (C) Sergey P. Derevyago, 2008-2009.
 *
 * 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.hpp>

namespace {

using namespace ders;

struct long_mutex : public mutex {
       bool locked;
       sh_mutex mtx;
       sh_cond_var cv;

       long_mutex(mem_pool& mp);
       ~long_mutex();

       virtual void lock();
       virtual void unlock();

       virtual void destroy(mem_pool& mp2) { destroy_this(this, mp2); }
};

long_mutex::long_mutex(mem_pool& mp) :
  locked(false), mtx(new_mutex(mp)), cv(new_cond_var(mp, *mtx))
{
}

long_mutex::~long_mutex()
{
 assert(!locked);
}

void long_mutex::lock()
{
 auto_mutex amt(*mtx);

 while (locked)
       cv->wait();

 locked=true;
}

void long_mutex::unlock()
{
 auto_mutex amt(*mtx);

 assert(locked);
 locked=false;
 cv->notify();
}

}  // unnamed

namespace ders {  // ::ders

sh_mutex new_long_mutex(mem_pool& mp)
{
 mp_newbuf<long_mutex> buf(mp);
 return sh_mutex(buf.pool(), buf.rls(::new(buf.get()) long_mutex(mp)));
}

barrier::barrier(mem_pool& mp, int cnt) :
  mtx(new_mutex(mp)), cv1(new_cond_var(mp, *mtx)), cv2(new_cond_var(mp, *mtx)),
  maxCnt(cnt), wtCnt(0), relCnt(0), ret(false), blnr(0)
{
 assert(maxCnt>=0);
}

barrier::~barrier()
{
 assert(wtCnt==0);
}

int barrier::inc_count()
{
 auto_mutex amt(*mtx);
 return ++maxCnt;
}

int barrier::dec_count()
{
 auto_mutex amt(*mtx);
 assert(wtCnt<maxCnt);

 relCnt=--maxCnt;
 cv1->notify_all();

 return maxCnt;
}

int barrier::get_count()
{
 auto_mutex amt(*mtx);
 assert(maxCnt>=0);
 return maxCnt;
}

bool barrier::wait()
{
 auto_mutex amt(*mtx);
 assert(maxCnt>0);

 while (wtCnt>0 && relCnt>0)
       cv2->wait();

 assert(wtCnt<maxCnt);
 if (++wtCnt==maxCnt) {
    barrier_listener::action act=barrier_listener::release_all_true;
    if (blnr) {
       try { act=blnr->on_release(*this); }
       catch (...) { hard_assert(false); }
    }

    switch (act) {
           case barrier_listener::release_one_false: {
                if (relCnt<maxCnt) relCnt++;

                break;
           }
           case barrier_listener::release_all_false: {
                relCnt=maxCnt;
                cv1->notify_all();

                break;
           }
           case barrier_listener::release_all_true: {
                relCnt=maxCnt;
                ret=true;
                cv1->notify_all();

                break;
           }
    }
 }

 for (;;) {
     if (relCnt>0) {
        relCnt--;
        break;
     }

     cv1->wait();
 }

 bool sav=ret;
 assert(wtCnt>0);
 if (--wtCnt==0) ret=false;

 if (wtCnt==0 || relCnt==0) cv2->notify_all();

 return sav;
}

void barrier::release_one()
{
 auto_mutex amt(*mtx);

 if (relCnt<maxCnt) relCnt++;
 cv1->notify();
}

void barrier::release_all()
{
 auto_mutex amt(*mtx);

 relCnt=maxCnt;
 cv1->notify_all();
}

barrier_listener* barrier::set_listener(barrier_listener* bl)
{
 barrier_listener* ret=blnr;
 blnr=bl;

 return ret;
}

sh_barrier new_barrier(mem_pool& mp, int cnt)
{
 mp_newbuf<barrier> buf(mp);
 return sh_barrier(buf.pool(), buf.rls(::new(buf.get()) barrier(mp, cnt)));
}

namespace detail {  // ::ders::detail

thr_core::thr_core(void (*st)(void*), void* ar) :
  strt(st), arg(ar), ownmp(0, 1), mtx(new_mutex(ownmp)), cv(new_cond_var(ownmp,
  *mtx)), refs(0), rng(true)
{
}

void thr_core::inc_refs()
{
 auto_mutex amt(*mtx);
 refs++;
}

void thr_core::dec_refs(bool parent)
{
 bool del=false;

 if (parent) {
    auto_mutex amt(*mtx);
    del=(--refs==0 && !rng);
 }
 else {
      auto_mutex amt(*mtx);

      del=(--refs==0);

      clear_rng();
      cv->notify_all();
 }

 if (del) delete this;
}

void thr_core::clear_rng()
{
 rng=false;
}

void thr_core::start()
{
 tc_ptr child(false, this);
 strt(arg);
}

bool thr_core::is_running()
{
 auto_mutex amt(*mtx);
 return rng;
}

void thr_core::join()
{
 auto_mutex amt(*mtx);
 while (rng) cv->wait();
}

}  // namespace ::ders::detail

}  // namespace ::ders

---