thread.c

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

  //
  // detached
  //
  // return if the thread is detached, or joinable.
  bool Pthread::detached() const
  {
    return _attributes.a_state == attributes::attr_detached_e;
  }

  //
  // terminated
  //
  // wether the thread has terminated.
  bool Pthread::terminated() const
  {
    return (_state == p_terminated_e) || exited();
  }

  //
  // exited
  //
  // wether the thread has exited.
  bool Pthread::exited() const
  {
    return _state == p_exited_e;
  }

  //
  // canceled
  //
  // weather the thread is cancelled
  bool Pthread::canceled() const
  {
    return _state == p_canceled_e;
  }

  //
  // joinable
  //
  // the opposite of detached.
  bool Pthread::joinable() const
  {
    return _attributes.a_state == attributes::attr_joinable_e;
  }

  //
  // restart
  //
  // send a restart signal to the process associated with this thread.
  void Pthread::restart()
  {
    Pthread::debug("%d is signalling %d",getpid(),id());
    if( id() != getpid() )
      kill(id(),CPPTHREAD_SIG_RESTART );
  }

  //
  // stop
  //
  // Stop or suspend the process, until a restart or continue
  // signal is given.
  void Pthread::stop()
  {
    if( id() == getpid() )
      suspend();
    else
      kill(id(),s_stop);
  }

  //
  // continue
  //
  // Continue or restart the process, if it has been stopped
  // and suspended.
  void Pthread::resume()
  {
    if ( _state == p_suspended_e )
      kill(id(),s_continue);
  }

  //
  // suspend
  //
  // suspend the process, until a restart signal is received.
  //
  // the thread suspended, is the thread associated with the
  // calling process.
  void Pthread::suspend()
  {
    Pthread *th = ThreadList::__threads.self();
    sigset_t mask;
    Pthread::thread_state old_state;

    if ( th ) {
      old_state = th->state();
      ThreadList::__waiting_s.pushBack(th);
      sigprocmask(SIG_SETMASK, NULL, &mask);
      sigdelset(&mask, CPPTHREAD_SIG_RESTART );
      th->state( p_suspended_e );
      do {
	th->signal(0);
	sigsuspend(&mask);
      } while(th->signal() != CPPTHREAD_SIG_RESTART );
      th->state( old_state );
    }
  }

  //
  // suspend_with_cancelation
  //
  // suspend the thread, associated with the calling process
  // until a signal is received or it is canceled.
  void Pthread::suspend_with_cancelation()
  {
    Pthread *th = ThreadList::__threads.self();
    sigset_t mask;
    sigjmp_buf jmpbuf;
    Pthread::thread_state old_state;
    
    if ( th ) {
      old_state = th->state();
      ThreadList::__waiting_s.pushBack(th);
      sigprocmask(SIG_SETMASK, NULL, &mask);
      sigdelset(&mask, CPPTHREAD_SIG_RESTART );
      th->state( p_suspended_e );
      if (sigsetjmp(jmpbuf, 0) == 0) {
	th->set(Pthread::cancel_jmp_e, &jmpbuf);
	if (!(th->canceled() && th->cancelstate()==Pthread::cancel_enable_e)) {
	  do {
	    th->signal(0);
	    sigsuspend(&mask);
	  } while(th->signal() != CPPTHREAD_SIG_RESTART );
	}
	th->set(Pthread::cancel_jmp_e, 0);
      } else {
	sigaddset(&mask, CPPTHREAD_SIG_RESTART );
	sigprocmask(SIG_SETMASK, &mask, NULL);
      }
      Pthread::debug("running again");
      th->state( old_state );
    }
  }

  //
  // signal
  //
  // set the p_signal variable to the given value.  Can only be
  // done if called from the process that the thread is associated
  // with. If called from a different process, the given signal
  // is sent to the process.
  int Pthread::signal(int sig_p)
  {
    if (getpid() == _tid)
      _signal = sig_p;
    else {
      kill(_tid,sig_p);
      // give the signal a time to kick in
      sched_yield();
    }
    return _signal;
  }

  //
  // cancel
  //
  // send a cancel signal to the process assiated with thread. Setting
  // the cancelation flag.
  void Pthread::cancel()
  {
    Pthread::debug("Send %d a cancel",id());
    set(Pthread::set_canceled_e, true);
    kill(_tid, CPPTHREAD_SIG_CANCEL );
    // let the signal kick in.
    sched_yield();
  }

  //
  // running
  //
  // return a true value, if the process associated with the thread
  // can be found on the schedulers list.
  bool Pthread::running()
  {
    return (sched_getscheduler(_tid) != -1);
  }

  //
  // exit
  //
  // perform an exit of the process.  This means marking it as
  // terminated and signalling any processes that are joining on
  // it.
  void Pthread::exit(int retcode_p)
  {
    int joining = 0;

    if (getpid() == _tid && !exited()) {
      Pthread::debug("exit(%d)",_retcode);
      _spinlock.acquire();
      _retcode     = _retcode;
      _state       = p_exited_e;
      joining      = _joining;
      _spinlock.release();
      if ( joining ) {
	Pthread::debug("joining %d",joining);
	ThreadList::__threads.restart(joining);
      }
      __main_program_thread.exit_child( retcode_p );
    }
  }

  //
  // set - booleans
  //
  // set any of the boolean values, terminated exited canceled or
  // detached.
  void Pthread::set(booleans _set, bool _val)
  {
    switch(_set) {
    case Pthread::set_terminated_e:
      _state = p_terminated_e;
      break;
    case Pthread::set_detached_e:
      if ( _val )
	_attributes.a_state = attributes::attr_detached_e;
      else
	_attributes.a_state = attributes::attr_joinable_e;
      break;
    case Pthread::set_exited_e:
      _state = p_exited_e;
      break;
    case Pthread::set_canceled_e:
      _state = p_canceled_e;
      break;
    case Pthread::set_private_e:
      break;
    }
  }

  //
  // set - jumps
  //
  // set the signalling jumps, signal or cancel.  
  void Pthread::set(jumps typ_p, sigjmp_buf *jbuf_p)
  {
    switch(typ_p) {
    case cancel_jmp_e:
      _cancel_jmp = jbuf_p;
      break;
    case signal_jmp_e:
      _signal_jmp = jbuf_p;
      break;
    }
  }

  //
  // join
  //
  // suspend the calling process (thread), until the thread
  // owning the call has terminated.
  int Pthread::join()
  {
    Pthread *th = ThreadList::__threads.ptr(getpid());
    
    if ( _tid == getpid() )
      return EDEADLK;
    if ( th == 0 )
      return ESRCH;
    if ( detached() || joining() != 0 || !joinable() )
      return EINVAL;
    if (!terminated()) {
      joining(getpid());
      Pthread::debug("wait for %d",id());
      suspend_with_cancelation();
      Pthread::debug("joins with %d",id());
      if (canceled() && cancelstate() == Pthread::cancel_enable_e) {
	joining(0);
	exit(CPPTHREAD_CANCELED);
      }
    }
    return 0;
  }

  //
  // jump
  //
  // do a jump, to either signal or cancel locations.  If they
  // are specified.
  void Pthread::jump(jumps typ_p)
  {
    sigjmp_buf *jmp_buf = 0;
    
    switch(typ_p) {
    case cancel_jmp_e:
      jmp_buf = _cancel_jmp;
      break;
    case signal_jmp_e:
      jmp_buf = _signal_jmp;
      break;
    }
    if (jmp_buf != 0)
      siglongjmp(*jmp_buf, 1);
  }

  //
  // has jump
  //
  // return true, if a jump as been set for a signal.
  bool Pthread::isSet(jumps typ_p)
  {
    bool rv = false;

    switch(typ_p) {
    case cancel_jmp_e:
      rv = (_cancel_jmp != 0);
      break;
    case signal_jmp_e:
      rv = (_signal_jmp != 0);
      break;
    }
    return rv;
  }

  //
  // Reschedule the thread
  //
  void Pthread::schedule(attributes& p_attr)
  {
    _attributes = p_attr;
    sched_setscheduler(id(),_attributes.a_policy,&_attributes.a_param);
  }

  //
  // set the inheritance of this thread.
  //
  void Pthread::inherit(inheritance_t inherit_p)
  {
    _attributes.a_inherit = inherit_p;
  }

  //
  // Return the inheritance of this thread.
  //
  inheritance_t Pthread::inherit() const
  {
    return _attributes.a_inherit;
  }

  //
  // the scope of the thread.
  scope_t Pthread::scope() const
  {
    return _attributes.a_scope;
  }

  //
  // The scheduling policy of the thread
  policy_t Pthread::policy() const
  {
    return _attributes.a_policy;
  }

  //
  // main_thread
  //
  Pthread *Pthread::_main()
  {
    return &__main_program_thread;
  }

  //
  // Stack address
  //
  ThreadStack *Pthread::stack()
  {
    return _sp;
  }

  //
  // the running thread
  //
  Pthread *Pthread::self()
  {
    return ThreadList::__threads.self();
  }

  //
  // a particular thread, by process id.
  //
  Pthread *Pthread::ptr(int id_p)
  {
    return ThreadList::__threads.ptr(id_p);
  }

  //
  // set_project
  //
  // This is a way for the user to communicate with the
  // shared memory control.
  void Pthread::setProject(const std::string& name_p)
  {
    SharedMemory::share.changeProj( name_p );
  }

  //
  // set_permission
  //
  // Set the project permission attributes.
  void Pthread::setPermission(int perm_p)
  {
    SharedMemory::share.changePerm( perm_p );
  }
  
  //
  // shalloc
  //
  // allocate shared memory
  void *Pthread::shalloc(size_t size_p)
  {
    return SharedMemory::share.alloc(size_p, 0);
  }

  //
  // shdealloc
  //
  // deallocate shared memory
  void Pthread::shdealloc(void *ptr_p)
  {
    SharedMemory::share.deAlloc(ptr_p);
  }

  void Pthread::debug(const char *format_p,...)
  {
#ifdef THREADS_DEBUG
    static spinlock _bug_lock;

    _bug_lock.acquire();
    va_list ap;
    va_start(ap,format_p);
    std::cout << "<" << getpid() << "> ";
#if( GCC_VERSION < 3000 )
    std::cout.vform(format_p,ap);
#else
    char *data_string;
    vasprintf(&data_string,format_p,ap);
    std::cout << data_string;
#endif
    std::cout << std::endl;
    va_end(ap);
    _bug_lock.release();
#endif
  }

  Pthread::operator attributes()
  {
    return _attributes;
  }

  Pthread& Pthread::operator =(attributes& attr_p)
  {
    schedule(attr_p);
    return (*this);
  }

  Pthread::operator bool()
  {
    return !terminated();
  }
    
}; // namespace