win32/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 <windows.h>
#include <process.h>
#include <ders/thread.hpp>
#include <algorithm>
#include <vector>
#include <ders/error.hpp>

namespace {

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

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

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

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

       void start();
};

struct win_mutex : public mutex {
       CRITICAL_SECTION cs;

       win_mutex();
       ~win_mutex();

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

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

class event_cache {
      vector<HANDLE> evs;

      event_cache(const event_cache&);
      event_cache& operator=(const event_cache&);

 public:
      event_cache();
      ~event_cache();

      HANDLE getEvent();
      void returnEvent(HANDLE ev);
};

struct win_cond_var : public cond_var {
       mutex& m;
       event_cache evc;
       vector<HANDLE> wts;

       win_cond_var(mutex& mtx);
       ~win_cond_var();

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

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

unsigned __stdcall win32_start(void* arg)
{
 static_cast<thr_core*>(arg)->start();
 return 0;
}

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

win_thread::~win_thread()
{
}

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

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

void win_thread::start()
{
 unsigned tid;
 HANDLE thn=(HANDLE)_beginthreadex(0, 0, win32_start, core.ptr, 0, &tid);
 if (thn==0) {
    core->clear_rng();

    throw newErrorException(mp, _FLINE_, "Can't begin thread", convert_errno(
      errno));
 }

 CloseHandle(thn);
}

win_mutex::win_mutex()
{
 InitializeCriticalSection(&cs);
}

win_mutex::~win_mutex()
{
 DeleteCriticalSection(&cs);
}

void win_mutex::lock()
{
 EnterCriticalSection(&cs);
}

void win_mutex::unlock()
{
 LeaveCriticalSection(&cs);
}

event_cache::event_cache()
{
 evs.reserve(16);
}

event_cache::~event_cache()
{
 for (int i=0, end=evs.size(); i<end; i++) CloseHandle(evs[i]);
}

HANDLE event_cache::getEvent()
{
 HANDLE ev;
 if (evs.size()>0) {
    ev=evs.back();
    evs.pop_back();
 }
 else {
      ev=CreateEvent(0, FALSE, FALSE, 0);
      hard_assert(ev!=0);
 }

 return ev;
}

void event_cache::returnEvent(HANDLE ev)
{
 assert(ev!=0);
 try { evs.push_back(ev); }
 catch (...) {
       CloseHandle(ev);
       throw;
 }
}

win_cond_var::win_cond_var(mutex& mtx) : m(mtx)
{
 wts.reserve(16);
}

win_cond_var::~win_cond_var()
{
 assert(wts.size()==0);
}

void win_cond_var::wait(long timeout)
{
 HANDLE ev=evc.getEvent();
 try { wts.push_back(ev); }
 catch (...) {
       CloseHandle(ev);
       throw;
 }

 m.unlock();

 DWORD rcd=WaitForSingleObject(ev, (timeout==infinite) ? timeout : INFINITE);
 hard_assert(rcd==WAIT_OBJECT_0 || rcd==WAIT_TIMEOUT);

 m.lock();

 if (rcd==WAIT_TIMEOUT) {
    vector<HANDLE>::iterator it=find(wts.begin(), wts.end(), ev);
    if (it==wts.end()) {
       BOOL rc=ResetEvent(ev);
       hard_assert(rc!=0);
    }
    else wts.erase(it);
 }

 evc.returnEvent(ev);
}

void win_cond_var::notify()
{
 if (wts.size()==0) return;

 HANDLE ev=wts.back();
 wts.pop_back();

 BOOL rc=SetEvent(ev);
 hard_assert(rc!=0);
}

void win_cond_var::notify_all()
{
 if (wts.size()==0) return;

 for (int i=0, end=wts.size(); i<end; i++) {
     BOOL rc=SetEvent(wts[i]);
     hard_assert(rc!=0);
 }

 wts.clear();
}

}  // unnamed

namespace ders {  // ::ders

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

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

 return ret;
}

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

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

}  // namespace ::ders

---