1
// Package : omnithread
2
// omnithread/nt.cc Created : 6/95 tjr
4
// Copyright (C) 2006 Free Software Foundation, Inc.
5
// Copyright (C) 1995-1999 AT&T Laboratories Cambridge
7
// This file is part of the omnithread library
9
// The omnithread library is free software; you can redistribute it and/or
10
// modify it under the terms of the GNU Library General Public
11
// License as published by the Free Software Foundation; either
12
// version 2 of the License, or (at your option) any later version.
14
// This library is distributed in the hope that it will be useful,
15
// but WITHOUT ANY WARRANTY; without even the implied warranty of
16
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
17
// Library General Public License for more details.
19
// You should have received a copy of the GNU Library General Public
20
// License along with this library; if not, write to the Free
21
// Software Foundation, Inc., 51 Franklin Street, Boston, MA
26
// Implementation of OMNI thread abstraction for NT threads
36
#include <omnithread.h>
40
//#include <iostream.h> or #include <iostream> if DB is on.
42
static void get_time_now(unsigned long* abs_sec, unsigned long* abs_nsec);
44
///////////////////////////////////////////////////////////////////////////
48
///////////////////////////////////////////////////////////////////////////
51
omni_mutex::omni_mutex(void)
53
InitializeCriticalSection(&crit);
56
omni_mutex::~omni_mutex(void)
58
DeleteCriticalSection(&crit);
63
///////////////////////////////////////////////////////////////////////////
67
///////////////////////////////////////////////////////////////////////////
71
// Condition variables are tricky to implement using NT synchronisation
72
// primitives, since none of them have the atomic "release mutex and wait to be
73
// signalled" which is central to the idea of a condition variable. To get
74
// around this the solution is to record which threads are waiting and
75
// explicitly wake up those threads.
77
// Here we implement a condition variable using a list of waiting threads
78
// (protected by a critical section), and a per-thread semaphore (which
79
// actually only needs to be a binary semaphore).
81
// To wait on the cv, a thread puts itself on the list of waiting threads for
82
// that cv, then releases the mutex and waits on its own personal semaphore. A
83
// signalling thread simply takes a thread from the head of the list and kicks
84
// that thread's semaphore. Broadcast is simply implemented by kicking the
85
// semaphore of each waiting thread.
87
// The only other tricky part comes when a thread gets a timeout from a timed
88
// wait on its semaphore. Between returning with a timeout from the wait and
89
// entering the critical section, a signalling thread could get in, kick the
90
// waiting thread's semaphore and remove it from the list. If this happens,
91
// the waiting thread's semaphore is now out of step so it needs resetting, and
92
// the thread should indicate that it was signalled rather than that it timed
95
// It is possible that the thread calling wait or timedwait is not a
96
// omni_thread. In this case we have to provide a temporary data structure,
97
// i.e. for the duration of the call, for the thread to link itself on the
98
// list of waiting threads. _internal_omni_thread_dummy provides such
99
// a data structure and _internal_omni_thread_helper is a helper class to
100
// deal with this special case for wait() and timedwait(). Once created,
101
// the _internal_omni_thread_dummy is cached for use by the next wait() or
102
// timedwait() call from a non-omni_thread. This is probably worth doing
103
// because creating a Semaphore is quite heavy weight.
105
class _internal_omni_thread_helper;
107
class _internal_omni_thread_dummy : public omni_thread {
109
inline _internal_omni_thread_dummy() : next(0) { }
110
inline ~_internal_omni_thread_dummy() { }
111
friend class _internal_omni_thread_helper;
113
_internal_omni_thread_dummy* next;
116
class _internal_omni_thread_helper {
118
inline _internal_omni_thread_helper() {
120
t = omni_thread::self();
122
omni_mutex_lock sync(cachelock);
128
d = new _internal_omni_thread_dummy;
133
inline ~_internal_omni_thread_helper() {
135
omni_mutex_lock sync(cachelock);
140
inline operator omni_thread* () { return t; }
141
inline omni_thread* operator->() { return t; }
143
static _internal_omni_thread_dummy* cache;
144
static omni_mutex cachelock;
147
_internal_omni_thread_dummy* d;
151
_internal_omni_thread_dummy* _internal_omni_thread_helper::cache = 0;
152
omni_mutex _internal_omni_thread_helper::cachelock;
155
omni_condition::omni_condition(omni_mutex* m) : mutex(m)
157
InitializeCriticalSection(&crit);
158
waiting_head = waiting_tail = NULL;
162
omni_condition::~omni_condition(void)
164
DeleteCriticalSection(&crit);
165
DB( if (waiting_head != NULL) {
166
cerr << "omni_condition::~omni_condition: list of waiting threads "
173
omni_condition::wait(void)
175
_internal_omni_thread_helper me;
177
EnterCriticalSection(&crit);
179
me->cond_next = NULL;
180
me->cond_prev = waiting_tail;
181
if (waiting_head == NULL)
184
waiting_tail->cond_next = me;
186
me->cond_waiting = TRUE;
188
LeaveCriticalSection(&crit);
192
DWORD result = WaitForSingleObject(me->cond_semaphore, INFINITE);
196
if (result != WAIT_OBJECT_0)
197
throw omni_thread_fatal(GetLastError());
202
omni_condition::timedwait(unsigned long abs_sec, unsigned long abs_nsec)
204
_internal_omni_thread_helper me;
206
EnterCriticalSection(&crit);
208
me->cond_next = NULL;
209
me->cond_prev = waiting_tail;
210
if (waiting_head == NULL)
213
waiting_tail->cond_next = me;
215
me->cond_waiting = TRUE;
217
LeaveCriticalSection(&crit);
221
unsigned long now_sec, now_nsec;
223
get_time_now(&now_sec, &now_nsec);
226
if ((abs_sec <= now_sec) && ((abs_sec < now_sec) || (abs_nsec < now_nsec)))
229
timeout = (abs_sec-now_sec) * 1000;
231
if( abs_nsec < now_nsec ) timeout -= (now_nsec-abs_nsec) / 1000000;
232
else timeout += (abs_nsec-now_nsec) / 1000000;
235
DWORD result = WaitForSingleObject(me->cond_semaphore, timeout);
237
if (result == WAIT_TIMEOUT) {
238
EnterCriticalSection(&crit);
240
if (me->cond_waiting) {
241
if (me->cond_prev != NULL)
242
me->cond_prev->cond_next = me->cond_next;
244
waiting_head = me->cond_next;
245
if (me->cond_next != NULL)
246
me->cond_next->cond_prev = me->cond_prev;
248
waiting_tail = me->cond_prev;
249
me->cond_waiting = FALSE;
251
LeaveCriticalSection(&crit);
258
// We timed out but another thread still signalled us. Wait for
259
// the semaphore (it _must_ have been signalled) to decrement it
260
// again. Return that we were signalled, not that we timed out.
263
LeaveCriticalSection(&crit);
265
result = WaitForSingleObject(me->cond_semaphore, INFINITE);
268
if (result != WAIT_OBJECT_0)
269
throw omni_thread_fatal(GetLastError());
277
omni_condition::signal(void)
279
EnterCriticalSection(&crit);
281
if (waiting_head != NULL) {
282
omni_thread* t = waiting_head;
283
waiting_head = t->cond_next;
284
if (waiting_head == NULL)
287
waiting_head->cond_prev = NULL;
288
t->cond_waiting = FALSE;
290
if (!ReleaseSemaphore(t->cond_semaphore, 1, NULL)) {
291
int rc = GetLastError();
292
LeaveCriticalSection(&crit);
293
throw omni_thread_fatal(rc);
297
LeaveCriticalSection(&crit);
302
omni_condition::broadcast(void)
304
EnterCriticalSection(&crit);
306
while (waiting_head != NULL) {
307
omni_thread* t = waiting_head;
308
waiting_head = t->cond_next;
309
if (waiting_head == NULL)
312
waiting_head->cond_prev = NULL;
313
t->cond_waiting = FALSE;
315
if (!ReleaseSemaphore(t->cond_semaphore, 1, NULL)) {
316
int rc = GetLastError();
317
LeaveCriticalSection(&crit);
318
throw omni_thread_fatal(rc);
322
LeaveCriticalSection(&crit);
327
///////////////////////////////////////////////////////////////////////////
329
// Counting semaphore
331
///////////////////////////////////////////////////////////////////////////
334
#define SEMAPHORE_MAX 0x7fffffff
337
omni_semaphore::omni_semaphore(unsigned int initial, unsigned int max_count)
339
if (max_count > SEMAPHORE_MAX)
340
max_count= SEMAPHORE_MAX;
342
nt_sem = CreateSemaphore(NULL, initial, max_count, NULL);
344
if (nt_sem == NULL) {
345
DB( cerr << "omni_semaphore::omni_semaphore: CreateSemaphore error "
346
<< GetLastError() << endl );
347
throw omni_thread_fatal(GetLastError());
352
omni_semaphore::~omni_semaphore(void)
354
if (!CloseHandle(nt_sem)) {
355
DB( cerr << "omni_semaphore::~omni_semaphore: CloseHandle error "
356
<< GetLastError() << endl );
357
throw omni_thread_fatal(GetLastError());
363
omni_semaphore::wait(void)
365
if (WaitForSingleObject(nt_sem, INFINITE) != WAIT_OBJECT_0)
366
throw omni_thread_fatal(GetLastError());
371
omni_semaphore::trywait(void)
373
switch (WaitForSingleObject(nt_sem, 0)) {
381
throw omni_thread_fatal(GetLastError());
382
return 0; /* keep msvc++ happy */
387
omni_semaphore::post(void)
389
if (!ReleaseSemaphore(nt_sem, 1, NULL)
390
&& GetLastError() != ERROR_TOO_MANY_POSTS ) // MinGW fix--see ticket:95 in trac
391
throw omni_thread_fatal(GetLastError());
396
///////////////////////////////////////////////////////////////////////////
400
///////////////////////////////////////////////////////////////////////////
407
omni_mutex* omni_thread::next_id_mutex;
408
int omni_thread::next_id = 0;
409
static DWORD self_tls_index;
411
static unsigned int stack_size = 0;
414
// Initialisation function (gets called before any user code).
417
static int& count() {
418
static int the_count = 0;
422
omni_thread::init_t::init_t(void)
424
if (count()++ != 0) // only do it once however many objects get created.
427
DB(cerr << "omni_thread::init: NT implementation initialising\n");
429
self_tls_index = TlsAlloc();
431
if (self_tls_index == 0xffffffff)
432
throw omni_thread_fatal(GetLastError());
434
next_id_mutex = new omni_mutex;
437
// Create object for this (i.e. initial) thread.
440
omni_thread* t = new omni_thread;
442
t->_state = STATE_RUNNING;
444
if (!DuplicateHandle(GetCurrentProcess(), GetCurrentThread(),
445
GetCurrentProcess(), &t->handle,
446
0, FALSE, DUPLICATE_SAME_ACCESS))
447
throw omni_thread_fatal(GetLastError());
449
t->nt_id = GetCurrentThreadId();
451
DB(cerr << "initial thread " << t->id() << " NT thread id " << t->nt_id
454
if (!TlsSetValue(self_tls_index, (LPVOID)t))
455
throw omni_thread_fatal(GetLastError());
457
if (!SetThreadPriority(t->handle, nt_priority(PRIORITY_NORMAL)))
458
throw omni_thread_fatal(GetLastError());
461
omni_thread::init_t::~init_t(void)
463
if (--count() != 0) return;
465
omni_thread* self = omni_thread::self();
468
TlsSetValue(self_tls_index, (LPVOID)0);
471
delete next_id_mutex;
473
TlsFree(self_tls_index);
477
// Wrapper for thread creation.
481
#ifndef __BCPLUSPLUS__
486
omni_thread_wrapper(void* ptr)
488
omni_thread* me = (omni_thread*)ptr;
490
DB(cerr << "omni_thread_wrapper: thread " << me->id()
493
if (!TlsSetValue(self_tls_index, (LPVOID)me))
494
throw omni_thread_fatal(GetLastError());
497
// Now invoke the thread function with the given argument.
500
if (me->fn_void != NULL) {
501
(*me->fn_void)(me->thread_arg);
505
if (me->fn_ret != NULL) {
506
void* return_value = (*me->fn_ret)(me->thread_arg);
507
omni_thread::exit(return_value);
511
me->run(me->thread_arg);
514
void* return_value = me->run_undetached(me->thread_arg);
515
omni_thread::exit(return_value);
518
// should never get here.
519
#ifndef __BCPLUSPLUS__
526
// Constructors for omni_thread - set up the thread object but don't
530
// construct a detached thread running a given function.
532
omni_thread::omni_thread(void (*fn)(void*), void* arg, priority_t pri)
534
common_constructor(arg, pri, 1);
539
// construct an undetached thread running a given function.
541
omni_thread::omni_thread(void* (*fn)(void*), void* arg, priority_t pri)
543
common_constructor(arg, pri, 0);
548
// construct a thread which will run either run() or run_undetached().
550
omni_thread::omni_thread(void* arg, priority_t pri)
552
common_constructor(arg, pri, 1);
557
// common part of all constructors.
560
omni_thread::common_constructor(void* arg, priority_t pri, int det)
565
next_id_mutex->lock();
567
next_id_mutex->unlock();
570
detached = det; // may be altered in start_undetached()
572
cond_semaphore = CreateSemaphore(NULL, 0, SEMAPHORE_MAX, NULL);
574
if (cond_semaphore == NULL)
575
throw omni_thread_fatal(GetLastError());
577
cond_next = cond_prev = NULL;
578
cond_waiting = FALSE;
589
// Destructor for omni_thread.
592
omni_thread::~omni_thread(void)
594
DB(cerr << "destructor called for thread " << id() << endl);
596
for (key_t i=0; i < _value_alloc; i++) {
603
if (handle && !CloseHandle(handle))
604
throw omni_thread_fatal(GetLastError());
605
if (cond_semaphore && !CloseHandle(cond_semaphore))
606
throw omni_thread_fatal(GetLastError());
615
omni_thread::start(void)
617
omni_mutex_lock l(mutex);
619
if (_state != STATE_NEW)
620
throw omni_thread_invalid();
622
#ifndef __BCPLUSPLUS__
623
// MSVC++ or compatiable
625
handle = (HANDLE)_beginthreadex(
634
throw omni_thread_fatal(GetLastError());
637
handle = (HANDLE)_beginthreadNT(omni_thread_wrapper,
643
if (handle == INVALID_HANDLE_VALUE)
644
throw omni_thread_fatal(errno);
647
if (!SetThreadPriority(handle, nt_priority(_priority)))
648
throw omni_thread_fatal(GetLastError());
650
if (ResumeThread(handle) == 0xffffffff)
651
throw omni_thread_fatal(GetLastError());
653
_state = STATE_RUNNING;
658
// Start a thread which will run the member function run_undetached().
662
omni_thread::start_undetached(void)
664
if ((fn_void != NULL) || (fn_ret != NULL))
665
throw omni_thread_invalid();
673
// join - simply check error conditions & call WaitForSingleObject.
677
omni_thread::join(void** status)
681
if ((_state != STATE_RUNNING) && (_state != STATE_TERMINATED)) {
683
throw omni_thread_invalid();
689
throw omni_thread_invalid();
692
throw omni_thread_invalid();
694
DB(cerr << "omni_thread::join: doing WaitForSingleObject\n");
696
if (WaitForSingleObject(handle, INFINITE) != WAIT_OBJECT_0)
697
throw omni_thread_fatal(GetLastError());
699
DB(cerr << "omni_thread::join: WaitForSingleObject succeeded\n");
702
*status = return_val;
709
// Change this thread's priority.
713
omni_thread::set_priority(priority_t pri)
715
omni_mutex_lock l(mutex);
717
if (_state != STATE_RUNNING)
718
throw omni_thread_invalid();
722
if (!SetThreadPriority(handle, nt_priority(pri)))
723
throw omni_thread_fatal(GetLastError());
728
// create - construct a new thread object and start it running. Returns thread
729
// object if successful, null pointer if not.
735
omni_thread::create(void (*fn)(void*), void* arg, priority_t pri)
737
omni_thread* t = new omni_thread(fn, arg, pri);
742
// undetached version
745
omni_thread::create(void* (*fn)(void*), void* arg, priority_t pri)
747
omni_thread* t = new omni_thread(fn, arg, pri);
754
// exit() _must_ lock the mutex even in the case of a detached thread. This is
755
// because a thread may run to completion before the thread that created it has
756
// had a chance to get out of start(). By locking the mutex we ensure that the
757
// creating thread must have reached the end of start() before we delete the
758
// thread object. Of course, once the call to start() returns, the user can
759
// still incorrectly refer to the thread object, but that's their problem.
763
omni_thread::exit(void* return_value)
765
omni_thread* me = self();
771
me->_state = STATE_TERMINATED;
775
DB(cerr << "omni_thread::exit: thread " << me->id() << " detached "
776
<< me->detached << " return value " << return_value << endl);
781
me->return_val = return_value;
786
DB(cerr << "omni_thread::exit: called with a non-omnithread. Exit quietly." << endl);
788
#ifndef __BCPLUSPLUS__
789
// MSVC++ or compatiable
790
// _endthreadex() does not automatically closes the thread handle.
791
// The omni_thread dtor closes the thread handle.
795
// _endthread() does not automatically closes the thread handle.
796
// _endthreadex() is only available if __MFC_COMPAT__ is defined and
797
// all it does is to call _endthread().
804
omni_thread::self(void)
808
me = TlsGetValue(self_tls_index);
811
DB(cerr << "omni_thread::self: called with a non-ominthread. NULL is returned." << endl);
813
return (omni_thread*)me;
818
omni_thread::yield(void)
824
#define MAX_SLEEP_SECONDS (DWORD)4294966 // (2**32-2)/1000
827
omni_thread::sleep(unsigned long secs, unsigned long nanosecs)
829
if (secs <= MAX_SLEEP_SECONDS) {
830
Sleep(secs * 1000 + nanosecs / 1000000);
834
DWORD no_of_max_sleeps = secs / MAX_SLEEP_SECONDS;
836
for (DWORD i = 0; i < no_of_max_sleeps; i++)
837
Sleep(MAX_SLEEP_SECONDS * 1000);
839
Sleep((secs % MAX_SLEEP_SECONDS) * 1000 + nanosecs / 1000000);
844
omni_thread::get_time(unsigned long* abs_sec, unsigned long* abs_nsec,
845
unsigned long rel_sec, unsigned long rel_nsec)
847
get_time_now(abs_sec, abs_nsec);
848
*abs_nsec += rel_nsec;
849
*abs_sec += rel_sec + *abs_nsec / 1000000000;
850
*abs_nsec = *abs_nsec % 1000000000;
855
omni_thread::nt_priority(priority_t pri)
860
return THREAD_PRIORITY_LOWEST;
862
case PRIORITY_NORMAL:
863
return THREAD_PRIORITY_NORMAL;
866
return THREAD_PRIORITY_HIGHEST;
869
throw omni_thread_invalid();
870
return 0; /* keep msvc++ happy */
875
get_time_now(unsigned long* abs_sec, unsigned long* abs_nsec)
877
static int days_in_preceding_months[12]
878
= { 0, 31, 59, 90, 120, 151, 181, 212, 243, 273, 304, 334 };
879
static int days_in_preceding_months_leap[12]
880
= { 0, 31, 60, 91, 121, 152, 182, 213, 244, 274, 305, 335 };
885
*abs_nsec = st.wMilliseconds * 1000000;
887
// this formula should work until 1st March 2100
889
DWORD days = ((st.wYear - 1970) * 365 + (st.wYear - 1969) / 4
891
? days_in_preceding_months[st.wMonth - 1]
892
: days_in_preceding_months_leap[st.wMonth - 1])
895
*abs_sec = st.wSecond + 60 * (st.wMinute + 60 * (st.wHour + 24 * days));
899
omni_thread::stacksize(unsigned long sz)
905
omni_thread::stacksize()
914
class omni_thread_dummy : public omni_thread {
916
inline omni_thread_dummy() : omni_thread()
919
_state = STATE_RUNNING;
921
if (!DuplicateHandle(GetCurrentProcess(), GetCurrentThread(),
922
GetCurrentProcess(), &handle,
923
0, FALSE, DUPLICATE_SAME_ACCESS))
924
throw omni_thread_fatal(GetLastError());
926
nt_id = GetCurrentThreadId();
928
if (!TlsSetValue(self_tls_index, (LPVOID)this))
929
throw omni_thread_fatal(GetLastError());
931
inline ~omni_thread_dummy()
933
if (!TlsSetValue(self_tls_index, (LPVOID)0))
934
throw omni_thread_fatal(GetLastError());
939
omni_thread::create_dummy()
941
if (omni_thread::self())
942
throw omni_thread_invalid();
944
return new omni_thread_dummy;
948
omni_thread::release_dummy()
950
omni_thread* self = omni_thread::self();
951
if (!self || !self->_dummy)
952
throw omni_thread_invalid();
954
omni_thread_dummy* dummy = (omni_thread_dummy*)self;
959
#if defined(__DMC__) && defined(_WINDLL)
960
BOOL WINAPI DllMain(HINSTANCE hinstDLL, DWORD fdwReason, LPVOID lpvReserved)
967
#define INSIDE_THREAD_IMPL_CC
968
#include "threaddata.cc"
969
#undef INSIDE_THREAD_IMPL_CC
added
removed
gnuradio-core/src/lib/omnithread/nt.cc
