semaphore.c

多线程库

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

#include <iostream.h>

#include <thread.h>
#include <thread_lists.h>
#include "wait_queue.h"
#include "thread_semaphore.h"
#include "shared.h"

char *semaphore::s_project = 0;

semaphore::semaphore(attributes::scope p_scope)
{
  if ( p_scope == attributes::process_private )
    _s = new storage;
  else {
    s_id = shared_mem::share.create_proj( s_project );
    _s = (storage *)shared_mem::share.alloc( sizeof(storage),s_id );
  }
  s_waiting = new wait_queue(s_id);
  s_scope = p_scope;
  if ( _s->s_magic != 0x212610 ) {
    _s->s_count = 0;
    _s->s_magic = 0x212610;
  }
}

semaphore::semaphore(attributes::scope p_scope, int p_count)
{
  if ( p_scope == attributes::process_private )
    _s = new storage;
  else {
    s_id = shared_mem::share.create_proj("sem");
    key_t key = shared_mem::share.make_key( s_id );
    _s = (storage *)shared_mem::share.alloc( key,sizeof(storage) );
  }
  s_waiting = new wait_queue(s_id);
  s_scope = p_scope;
  if ( _s->s_magic != 0x212610 ) {
    _s->s_count = p_count;
    _s->s_magic = 0x212610;
  }
}

semaphore::semaphore()
{
  s_waiting   = new wait_queue;
  _s          = new storage;
  _s->s_count = 0;
  s_scope = attributes::process_private;
}

semaphore::semaphore(int p_count)
{
  s_waiting   = new wait_queue;
  _s          = new storage;
  _s->s_count = p_count;
  s_scope = attributes::process_private;
}

semaphore::~semaphore()
{
  if ( _s ) {
    delete s_waiting;
    if ( s_scope == attributes::process_private )
      delete _s;
    else
      shared_mem::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;
  s_waiting->wake_up();
  _s->s_sync.release();
  return rval;
}

//
// wait
//
// Wait until a semaphore becomes non-zero, then atomically
// decrement it and return with the current count of the
// semaphore.
int
semaphore::wait()
{
  thread_list::iterator i = thread_list::__threads.locate(getpid());
  int val = -1;

  if (i != thread_list::__threads.end()) {
    _s->s_sync.acquire();
    if ( _s->s_count == 0 ) {
      s_waiting->insert((*i)->id());
      _s->s_sync.release();
      (*i)->suspend_with_cancelation();
      _s->s_sync.acquire();
      if ((*i)->canceled() && (*i)->cancelstate() == pthread::cancel_enable) {
	s_waiting->remove((*i)->id());
	_s->s_sync.release();
	(*i)->exit(PTHREAD_CANCELED);
      }
      if ( _s->s_count > 0 )
	_s->s_count--;
    } else
      _s->s_count--;
    val = _s->s_count;
    _s->s_sync.release();
  }
  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::project_part(const char *p_part)
{
  if ( s_project == 0 )
    s_project = new char[80];
  s_project[0] = 0;
  if ( p_part )
    strcpy( s_project,p_part );
}