posix.cc

编译工具

void
omni_thread::common_constructor(void* arg, priority_t pri, int det)
{
    _state = STATE_NEW;
    _priority = pri;

    next_id_mutex->lock();
    _id = next_id++;
    next_id_mutex->unlock();

    thread_arg = arg;
    detached = det;	// may be altered in start_undetached()

    _dummy       = 0;
    _values      = 0;
    _value_alloc = 0;
    // posix_thread is set up in initialisation routine or start().
}


//
// Destructor for omni_thread.
//

omni_thread::~omni_thread(void)
{
    DB(cerr << "destructor called for thread " << id() << endl);
    if (_values) {
        for (key_t i=0; i < _value_alloc; i++) {
	    if (_values[i]) {
	        delete _values[i];
	    }
        }
	delete [] _values;
    }
}


//
// Start the thread
//

void
omni_thread::start(void)
{
    omni_mutex_lock l(mutex);

    if (_state != STATE_NEW)
	throw omni_thread_invalid();

    pthread_attr_t attr;

#if (PthreadDraftVersion == 4)
    pthread_attr_create(&attr);
#else
    pthread_attr_init(&attr);
#endif

#if (PthreadDraftVersion == 8)
    pthread_attr_setdetachstate(&attr,PTHREAD_CREATE_UNDETACHED);
#endif

#ifdef PthreadSupportThreadPriority

#if (PthreadDraftVersion <= 6)

    THROW_ERRORS(pthread_attr_setprio(&attr, posix_priority(_priority)));

#else

    struct sched_param sparam;

    sparam.sched_priority = posix_priority(_priority);

    THROW_ERRORS(pthread_attr_setschedparam(&attr, &sparam));

#endif	/* PthreadDraftVersion */

#endif	/* PthreadSupportThreadPriority */

    if (stack_size) {
      THROW_ERRORS(pthread_attr_setstacksize(&attr, stack_size));
    }

#if (PthreadDraftVersion == 4)
    THROW_ERRORS(pthread_create(&posix_thread, attr, omni_thread_wrapper,
				(void*)this));
    pthread_attr_delete(&attr);
#else
    THROW_ERRORS(pthread_create(&posix_thread, &attr, omni_thread_wrapper,
				(void*)this));
    pthread_attr_destroy(&attr);
#endif

    _state = STATE_RUNNING;

    if (detached) {

#if (PthreadDraftVersion <= 6)
	THROW_ERRORS(pthread_detach(&posix_thread));
#else
	THROW_ERRORS(pthread_detach(posix_thread));
#endif
    }
}


//
// Start a thread which will run the member function run_undetached().
//

void
omni_thread::start_undetached(void)
{
    if ((fn_void != NULL) || (fn_ret != NULL))
	throw omni_thread_invalid();

    detached = 0;
    start();
}


//
// join - simply check error conditions & call pthread_join.
//

void
omni_thread::join(void** status)
{
    mutex.lock();

    if ((_state != STATE_RUNNING) && (_state != STATE_TERMINATED)) {
	mutex.unlock();
	throw omni_thread_invalid();
    }

    mutex.unlock();

    if (this == self())
	throw omni_thread_invalid();

    if (detached)
	throw omni_thread_invalid();

    DB(cerr << "omni_thread::join: doing pthread_join\n");

    THROW_ERRORS(pthread_join(posix_thread, status));

    DB(cerr << "omni_thread::join: pthread_join succeeded\n");

#if (PthreadDraftVersion == 4)
    // With draft 4 pthreads implementations (HPUX 10.x and
    // Digital Unix 3.2), have to detach the thread after 
    // join. If not, the storage for the thread will not be
    // be reclaimed.
    THROW_ERRORS(pthread_detach(&posix_thread));
#endif

    delete this;
}


//
// Change this thread's priority.
//

void
omni_thread::set_priority(priority_t pri)
{
    omni_mutex_lock l(mutex);

    if (_state != STATE_RUNNING)
	throw omni_thread_invalid();

    _priority = pri;

#ifdef PthreadSupportThreadPriority

#if (PthreadDraftVersion == 4)

    THROW_ERRORS(pthread_setprio(posix_thread, posix_priority(pri)));

#elif (PthreadDraftVersion == 6)

    pthread_attr_t attr;
    pthread_attr_init(&attr);

    THROW_ERRORS(pthread_attr_setprio(&attr, posix_priority(pri)));

    THROW_ERRORS(pthread_setschedattr(posix_thread, attr));

#else

    struct sched_param sparam;

    sparam.sched_priority = posix_priority(pri);

    THROW_ERRORS(pthread_setschedparam(posix_thread, SCHED_OTHER, &sparam));

#endif   /* PthreadDraftVersion */

#endif   /* PthreadSupportThreadPriority */
}


//
// create - construct a new thread object and start it running.  Returns thread
// object if successful, null pointer if not.
//

// detached version

omni_thread*
omni_thread::create(void (*fn)(void*), void* arg, priority_t pri)
{
    omni_thread* t = new omni_thread(fn, arg, pri);

    t->start();

    return t;
}

// undetached version

omni_thread*
omni_thread::create(void* (*fn)(void*), void* arg, priority_t pri)
{
    omni_thread* t = new omni_thread(fn, arg, pri);

    t->start();

    return t;
}


//
// exit() _must_ lock the mutex even in the case of a detached thread.  This is
// because a thread may run to completion before the thread that created it has
// had a chance to get out of start().  By locking the mutex we ensure that the
// creating thread must have reached the end of start() before we delete the
// thread object.  Of course, once the call to start() returns, the user can
// still incorrectly refer to the thread object, but that's their problem.
//

void
omni_thread::exit(void* return_value)
{
    omni_thread* me = self();

    if (me)
      {
	me->mutex.lock();

	me->_state = STATE_TERMINATED;

	me->mutex.unlock();

	DB(cerr << "omni_thread::exit: thread " << me->id() << " detached "
	   << me->detached << " return value " << return_value << endl);

	if (me->_values) {
	  for (key_t i=0; i < me->_value_alloc; i++) {
	    if (me->_values[i]) {
	      delete me->_values[i];
	    }
	  }
	  delete [] me->_values;
	  me->_values = 0;
	}

	if (me->detached)
	  delete me;
      }
    else
      {
	DB(cerr << "omni_thread::exit: called with a non-omnithread. Exit quietly." << endl);
      }

    pthread_exit(return_value);
}


omni_thread*
omni_thread::self(void)
{
    omni_thread* me;

#if (PthreadDraftVersion <= 6)

    THROW_ERRORS(pthread_getspecific(self_key, (void**)&me));

#else

    me = (omni_thread *)pthread_getspecific(self_key);

#endif

    if (!me) {
      // This thread is not created by omni_thread::start because it
      // doesn't has a class omni_thread instance attached to its key.
      DB(cerr << "omni_thread::self: called with a non-omnithread. NULL is returned." << endl);
    }

    return me;
}


void
omni_thread::yield(void)
{
#if (PthreadDraftVersion == 6)

    pthread_yield(NULL);

#elif (PthreadDraftVersion < 9)

    pthread_yield();

#else

    THROW_ERRORS(sched_yield());

#endif
}


void
omni_thread::sleep(unsigned long secs, unsigned long nanosecs)
{
    timespec rqts = { secs, nanosecs };

#ifndef NoNanoSleep

    timespec remain;
    while (nanosleep(&rqts, &remain)) {
      if (errno == EINTR) {
	rqts.tv_sec  = remain.tv_sec;
	rqts.tv_nsec = remain.tv_nsec;
	continue;
      }
      else
	throw omni_thread_fatal(errno);
    }
#else

#if defined(__osf1__) && defined(__alpha__) || defined(__hpux__) && (__OSVERSION__ == 10) || defined(__VMS) || defined(__SINIX__) || defined (__POSIX_NT__)

    if (pthread_delay_np(&rqts) != 0)
	throw omni_thread_fatal(errno);

#elif defined(__linux__) || defined(__aix__)

    if (secs > 2000) {
      while ((secs = ::sleep(secs))) ;
    } else {
	usleep(secs * 1000000 + (nanosecs / 1000));
    }

#elif defined(__darwin__) || defined(__macos__)

    // Single UNIX Specification says argument of usleep() must be
    // less than 1,000,000.
    secs += nanosecs / 1000000000;
    nanosecs %= 1000000000;
    while ((secs = ::sleep(secs))) ;
    usleep(nanosecs / 1000);

#else

    throw omni_thread_invalid();

#endif
#endif	/* NoNanoSleep */
}


void
omni_thread::get_time(unsigned long* abs_sec, unsigned long* abs_nsec,
		      unsigned long rel_sec, unsigned long rel_nsec)
{
    timespec abs;

#if defined(__osf1__) && defined(__alpha__) || defined(__hpux__) && (__OSVERSION__ == 10) || defined(__VMS) || defined(__SINIX__) || defined(__POSIX_NT__)

    timespec rel;
    rel.tv_sec = rel_sec;
    rel.tv_nsec = rel_nsec;
    THROW_ERRORS(pthread_get_expiration_np(&rel, &abs));

#else

#if defined(__linux__) || defined(__aix__) || defined(__SCO_VERSION__) || defined(__darwin__) || defined(__macos__)

    struct timeval tv;
    gettimeofday(&tv, NULL); 
    abs.tv_sec = tv.tv_sec;
    abs.tv_nsec = tv.tv_usec * 1000;

#else	/* __linux__ || __aix__ */

    clock_gettime(CLOCK_REALTIME, &abs);

#endif	/* __linux__ || __aix__ */

    abs.tv_nsec += rel_nsec;
    abs.tv_sec += rel_sec + abs.tv_nsec / 1000000000;
    abs.tv_nsec = abs.tv_nsec % 1000000000;

#endif	/* __osf1__ && __alpha__ */

    *abs_sec = abs.tv_sec;
    *abs_nsec = abs.tv_nsec;
}


int
omni_thread::posix_priority(priority_t pri)
{
#ifdef PthreadSupportThreadPriority
    switch (pri) {

    case PRIORITY_LOW:
	return lowest_priority;

    case PRIORITY_NORMAL:
	return normal_priority;

    case PRIORITY_HIGH:
	return highest_priority;

    }
#endif

    throw omni_thread_invalid();
#if defined(_MSC_VER ) || defined(__HP_aCC)
    return 0;
#endif
}

void
omni_thread::stacksize(unsigned long sz)
{
  stack_size = sz;
}

unsigned long
omni_thread::stacksize()
{
  return stack_size;
}

//
// Dummy thread
//

class omni_thread_dummy : public omni_thread {
public:
  inline omni_thread_dummy() : omni_thread()
  {
    _dummy = 1;
    _state = STATE_RUNNING;
    posix_thread = pthread_self();
    THROW_ERRORS(pthread_setspecific(self_key, (void*)this));
  }
  inline ~omni_thread_dummy()
  {
    THROW_ERRORS(pthread_setspecific(self_key, 0));
  }
};

omni_thread*
omni_thread::create_dummy()
{
  if (omni_thread::self())
    throw omni_thread_invalid();

  return new omni_thread_dummy;
}

void
omni_thread::release_dummy()
{
  omni_thread* self = omni_thread::self();
  if (!self || !self->_dummy)
    throw omni_thread_invalid();

  omni_thread_dummy* dummy = (omni_thread_dummy*)self;
  delete dummy;
}


#define INSIDE_THREAD_IMPL_CC
#include "threaddata.cc"
#undef INSIDE_THREAD_IMPL_CC