posix/mt_threadimpl.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.hpp>
#include <errno.h>
#include <pthread.h>
#include <time.h>
#include <ders/error.hpp>

namespace {

using namespace std;
using namespace ders;
using namespace ders::detail;

struct pos_thread : public thread {
       mem_pool& mp;
       tc_ptr core;

       pos_thread(mem_pool& m, void (*strt)(void*), void* arg);
       ~pos_thread();

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

       void start();
};

struct pos_mutex : public mutex {
       pthread_mutex_t mtx;

       pos_mutex();
       ~pos_mutex();

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

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

struct pos_cond_var : public cond_var {
       pos_mutex& pm;
       pthread_cond_t cv;

       pos_cond_var(mutex& mtx);
       ~pos_cond_var();

       virtual void wait(long timeout);
       virtual void notify();
       virtual void notify_all();

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

extern "C" void* pos_start(void* arg)
{
 static_cast<thr_core*>(arg)->start();
 return 0;
}

pos_thread::pos_thread(mem_pool& m, void (*strt)(void*), void* arg) :
  mp(m), core(true, new thr_core(strt, arg))
{
}

pos_thread::~pos_thread()
{
}

bool pos_thread::is_running()
{
 return core->is_running();
}

void pos_thread::join()
{
 core->join();
}

void pos_thread::start()
{
 pthread_attr_t attr;
 int rc=pthread_attr_init(&attr);
 hard_assert(rc==0);

 rc=pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
 hard_assert(rc==0);

 pthread_t thr;
 rc=pthread_create(&thr, &attr, pos_start, core.ptr);

 int rc2=pthread_attr_destroy(&attr);
 hard_assert(rc2==0);

 if (rc!=0) {
    core->clear_rng();

    throw newErrorException(mp, _FLINE_, "Can't create thread", convert_errno(
      rc));
 }
}

pos_mutex::pos_mutex()
{
 int rc=pthread_mutex_init(&mtx, NULL);
 hard_assert(rc==0);
}

pos_mutex::~pos_mutex()
{
 int rc=pthread_mutex_destroy(&mtx);
 hard_assert(rc==0);
}

void pos_mutex::lock()
{
 int rc=pthread_mutex_lock(&mtx);
 hard_assert(rc==0);
}

void pos_mutex::unlock()
{
 int rc=pthread_mutex_unlock(&mtx);
 hard_assert(rc==0);
}

pos_cond_var::pos_cond_var(mutex& mtx) : pm(dynamic_cast<pos_mutex&>(mtx))
{
 int rc=pthread_cond_init(&cv, NULL);
 hard_assert(rc==0);
}

pos_cond_var::~pos_cond_var()
{
 int rc=pthread_cond_destroy(&cv);
 hard_assert(rc==0);
}

void pos_cond_var::wait(long timeout)
{
 if (timeout==infinite) {
    int rc=pthread_cond_wait(&cv, &pm.mtx);
    hard_assert(rc==0);
 }
 else {
      timespec ts;
      clock_gettime(CLOCK_REALTIME, &ts);
      ts.tv_sec+=timeout/1000;
      ts.tv_nsec+=timeout%1000*1000;

      int rc=pthread_cond_timedwait(&cv, &pm.mtx, &ts);
      hard_assert(rc==0 || rc==ETIMEDOUT);
 }
}

void pos_cond_var::notify()
{
 int rc=pthread_cond_signal(&cv);
 hard_assert(rc==0);
}

void pos_cond_var::notify_all()
{
 int rc=pthread_cond_broadcast(&cv);
 hard_assert(rc==0);
}

}  // unnamed

namespace ders {  // ::ders

sh_thread new_thread(mem_pool& mp, void (*start)(void*), void* arg)
{
 mp_newbuf<pos_thread> buf(mp);
 sh_thread ret(buf.pool(), buf.rls(::new(buf.get()) pos_thread(buf.pool(),
   start, arg)));

 static_cast<pos_thread*>(ret.get())->start();

 return ret;
}

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

sh_cond_var new_cond_var(mem_pool& mp, mutex& mtx)
{
 mp_newbuf<pos_cond_var> buf(mp);
 return sh_cond_var(buf.pool(), buf.rls(::new(buf.get()) pos_cond_var(mtx)));
}

}  // namespace ::ders

---