wait_queue.c

多线程库

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

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

#define OFFS(n)         ((n)%__THREAD_QUEUE_MAX)

wait_queue::wait_queue(int p_id)
{
  w_id = p_id;
  init(attributes::process_shared);
}

wait_queue::wait_queue(attributes::scope p_scope)
{
  w_id = shared_mem::share.create_proj( 0 );
  init(attributes::process_shared);
}

wait_queue::wait_queue()
{
  init(attributes::process_private);
}

wait_queue::~wait_queue()
{
  if ( _wq ) {
    while( !empty() )
      wake_up();
    if ( w_scope == attributes::process_private )
      delete _wq;
    else
      shared_mem::share.dealloc( _wq );
    _wq = 0;
  }
}

void wait_queue::init(attributes::scope p_scope)
{
  if ( w_id == -1 )
    w_scope     = attributes::process_private;
  else
    w_scope     = p_scope;
  if ( w_scope == attributes::process_shared ) {
    _wq         = (storage *)shared_mem::share.alloc( sizeof(storage),w_id );
    assert( _wq != (storage *)-1 );
    if ( _wq->w_magic != 0x212611 ) {
      _wq->top     = 0;
      _wq->bottom  = 0;
      _wq->w_magic = 0x212611;
    }
  } else {
    _wq         = new storage;
    _wq->top    = 0;
    _wq->bottom = 0;
  }
}

bool
wait_queue::empty()
{
  return ( _wq->bottom == _wq->top );
}

void
wait_queue::remove(int p_id)
{
  int pos;

  _wq->w_sync.acquire();
  if ( !empty() ) {
    for( pos = _wq->bottom;pos != OFFS(_wq->top-1);pos = OFFS(pos+1) )
      if ( _wq->pid[pos] == p_id ) {
	for( ;_wq->bottom != pos;pos = OFFS(pos-1) )
	  _wq->pid[pos] = _wq->pid[OFFS(pos-1)];
	_wq->bottom = OFFS(_wq->bottom+1);
	break;
      }
  }
  _wq->w_sync.release();
}

void
wait_queue::insert(int p_id)
{
  _wq->w_sync.acquire();
  if ( _wq->bottom == OFFS(_wq->top+1) )
    throw;
  _wq->pid[_wq->top] = p_id;
  _wq->top = OFFS(_wq->top+1);
  _wq->w_sync.release();
}

void
wait_queue::suspend_me()
{
  thread_list::iterator i = thread_list::__threads.locate(getpid());


  if ( i == thread_list::__threads.end() )
    return;
  insert(getpid());
  (*i)->suspend_with_cancelation();
  //__sched_yield();
  _wq->w_sync.acquire();
  //we need to synchronize with the wakeup below.  Maybe __sched.yield()?
  _wq->w_sync.release();
}

void
wait_queue::wake_up()
{
  _wq->w_sync.acquire();
  if ( !empty() ) {
    thread_list::__threads.restart(_wq->pid[_wq->bottom]);
    _wq->bottom = OFFS(_wq->bottom+1);
  }
  _wq->w_sync.release();
}