semaphore.c

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

//
// Semaphores
//
// This is a basic semaphore implementation, not very rich.
//


#include <thread.h>

#include <iostream>

#include "thread_lists.h"
#include "wait_queue.h"
#include "semaphore.h"
#include "shared.h"

#define MAGIC_SEMAPHORE_NO   0x212610

namespace cpp_threads {

  std::string Semaphore::_project = "";

  Semaphore::Semaphore(scope_t scope_p,int count_p)
  {
    init(scope_p,count_p);
  }

  Semaphore::Semaphore(int count_p)
  {
    Pthread *p = ThreadList::__threads.self();
    scope_t scope = attributes::process_private_e;

    if( p && p->inherit() )
      scope = p->scope();
    init( scope,count_p );
  }

  void Semaphore::init(attributes::scope scope_p, int count_p)
  {
    _id = -1;
    if ( scope_p == attributes::process_private_e )
      _s  = new storage;
    else {
      _id = SharedMemory::share.createProj( _project );
      _s  = (storage *)SharedMemory::share.alloc( sizeof(storage),_id );
    }
    _waiting = new WaitQueue(_id);
    _scope = scope_p;
    if ( _s->s_magic != MAGIC_SEMAPHORE_NO ) {
      Pthread::debug(" - Empty semaphore space.");
      _s->s_count = count_p;
      _s->s_magic = MAGIC_SEMAPHORE_NO;
    }
  }

  Semaphore::~Semaphore()
  {
    if ( _s ) {
      _s->s_sync.acquire();
      delete _waiting;
      if ( _scope == attributes::process_private_e )
	delete _s;
      else
	SharedMemory::share.deAlloc( _s );
      _s = 0;
    }
  }


  //
  // post()
  //
  // Increment the count associated with the semaphore, and restart
  // any processes that are blocking on it.
  int
  Semaphore::post()
  {
    int rval;

    _s->s_sync.acquire();
    rval = ++_s->s_count;
    Pthread::debug(" semaphore post %d",rval);
    _waiting->wakeUp();
    _s->s_sync.release();
    return rval;
  }

  //
  // wait
  //
  // Wait until a semaphore becomes non-zero, then automically
  // decrement it and return with the current count of the
  // semaphore.
  int
  Semaphore::wait(int pri_p)
  {
    Pthread *th = ThreadList::__threads.self();
    int val = -1;

    if ( th ) {
      _s->s_sync.acquire();
      while ( _s->s_count == 0 ) {
	_waiting->insert(th->id(),pri_p);
	_s->s_sync.release();
	th->suspend_with_cancelation();
	_s->s_sync.acquire();
	if (th->canceled() && th->cancelstate() == Pthread::cancel_enable_e) {
	  _waiting->remove(th->id());
	  _s->s_sync.release();
	  th->exit(CPPTHREAD_CANCELED);
	}
      }
      _s->s_count--;
      val = _s->s_count;
      _s->s_sync.release();
      Pthread::debug("semaphore count %d",_s->s_count);
    } else
      Pthread::debug("thread %d is not in our list!",getpid());
    return val;
  }

  //
  // trywait
  //
  // This will decrement the semaphore by one, without blocking it.
  // it will return with -1, if the semaphore could not be decremented,
  // and the current semaphore count otherwise.
  int
  Semaphore::tryWait()
  {
    int retcode = 0;

    _s->s_sync.acquire();
    retcode = _s->s_count;
    if ( retcode-- > 0 )
      _s->s_count--;
    _s->s_sync.release();
    return retcode;
  }

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

}; // Namespace