cond.c

用c++包装好的线程库,直接拿来使用,提高效率.

#include "thread.h"
#include "thread_lists.h"
#include "signal_num.h"

#include "wait_queue.h"
#include "shared.h"

#include <cerrno>
#include <csignal>
#include <ctime>

extern "C" {

#       include <sys/time.h>

};

namespace cpp_threads {

  std::string Cond::_project = "";

  //
  // Conditional variables.  This variable consists of a single
  // spinlock structure, and a couple of function to control
  // activity around it.  And a list of threads that are waiting
  // on that condition.
  Cond::Cond(scope_t scope_p)
  {
    _id = -1;
    if ( _scope == attributes::process_shared_e )
      _id = SharedMemory::share.createProj( _project );
    _waiting = new WaitQueue( _id );
    _scope = scope_p;
  }

  Cond::Cond()
  {
    Pthread *p = ThreadList::__threads.self();

    _id = -1;
    _scope = attributes::process_private_e;
    if( p && p->inherit() )
      _scope = p->scope();
    if( _scope == attributes::process_shared_e )
      _id = SharedMemory::share.createProj( _project );
    _waiting = new WaitQueue( _id );
  }
  
  Cond::~Cond()
  {
    if ( _waiting->empty() == false )
      broadcast();
    delete _waiting;
  }

  scope_t Cond::scope()
  {
    return _scope;
  }

  //
  // Wait for a signal to arrive through the conditional variable.
  // mutex is considered lock on entry, and is unlock during the
  // time we wait for the signal to arrive.  The mutex is relocked
  // before returning.
  int Cond::wait(Mutex& mutx_p, int pri_p)
  {
    Pthread *th = ThreadList::__threads.self();
    
    if ( th ) {
      sched_yield();
      mutx_p.unLock();
      _waiting->suspendMe(pri_p);
      mutx_p.lock();
      if (th->canceled() && th->cancelstate() == Pthread::cancel_enable_e)
	th->exit(CPPTHREAD_CANCELED);
    }
    return 0;
  }

  //
  // timedwait relative.
  //
  // wait for a specific time, for a signal to arrive, or for a
  // cancelation to occur.  As with wait, mutex is considered
  // lock on entry, and is unlocked during the wait for the
  // signal.  It is relocked, after the duration transpires.
  int Cond::timedWaitRel(Mutex& mutx_p,const struct timespec *spec_p,int pri_p)
  {
    Pthread *th = ThreadList::__threads.self();
    sigset_t unblock, initial_mask;
    int retsleep = -1;
    sigjmp_buf jmpbuf;

    if ( th == 0 )
      return ESRCH;
    _waiting->insert(th->id(),pri_p);
    sched_yield();
    mutx_p.unLock();
    if (sigsetjmp(jmpbuf, 0) == 0) {
      th->set(Pthread::signal_jmp_e, &jmpbuf);
      th->signal(0);
      if (!(th->canceled() && th->cancelstate() == Pthread::cancel_enable_e)) {
	sigemptyset(&unblock);
	sigaddset(&unblock, CPPTHREAD_SIG_RESTART );
	sigprocmask(SIG_UNBLOCK, &unblock, &initial_mask);
	retsleep = nanosleep(spec_p, NULL);
	sigprocmask(SIG_SETMASK, &initial_mask, NULL);
      }
    }
    th->set(Pthread::signal_jmp_e, 0);
    _waiting->remove(th->id());
    mutx_p.lock();
    if( th->canceled() && th->cancelstate() == Pthread::cancel_enable_e )
      th->exit(CPPTHREAD_CANCELED);
    if( th->signal() != CPPTHREAD_SIG_RESTART )
      return retsleep ? EINTR : ETIMEDOUT;
    return 0;
  }

  //
  // timedwait
  //
  // Wait for a specific time period, for a signal to occur.  This
  // routine simply calculates the relative time, from the given
  // time and current time.  Then uses timedwait relative to
  // perform the actual wait.
  int Cond::timedWait(Mutex& mutx_p,const struct timespec *spec_p,int pri_p)
  {
    struct timeval now;
    struct timespec reltime;
    
    gettimeofday(&now, NULL);
    reltime.tv_sec = spec_p->tv_sec - now.tv_sec;
    reltime.tv_nsec = spec_p->tv_nsec - now.tv_usec * 1000;
    if (reltime.tv_nsec < 0) {
      reltime.tv_nsec += 1000000000;
      reltime.tv_sec -= 1;
    }
    if (reltime.tv_sec < 0)
      return ETIMEDOUT;
    return timedWaitRel( mutx_p,&reltime,pri_p );
  }

  //
  // timedwait
  //
  // Wait for some period of time, micro seconds, for a signal to occur
  // this will use timedwait relative, by filling out the seconds
  // part in a timespec that it will pass to it.
  int Cond::timedWait(Mutex& mutx_p,long int usec_p,int pri_p)
  {
    struct timespec reltime;

    reltime.tv_sec  = usec_p/1000000;
    reltime.tv_nsec = (usec_p % 1000000)*1000;
    return timedWaitRel( mutx_p,&reltime,pri_p );
  }

  //
  // signal
  //
  // send a signal to the next process waiting on the list.
  int Cond::signal()
  {
    _waiting->wakeUp();
    return 0;
  }

  //
  // broadcast
  //
  // send a signal to every process on the list.
  int Cond::broadcast()
  {
    while( !_waiting->empty() )
      _waiting->wakeUp();
    return 0;
  }

  //
  // project_part
  //
  // change the project part
  void
  Cond::projectPart(const std::string& part_p)
  {
    _project = part_p;
  }

}; // Namespace