thread.cc

cygwin, 著名的在win32下模拟unix操作系统的东东

  if (cleanup_stack != NULL)
    {
      __pthread_cleanup_handler *handler = cleanup_stack;

      if (execute)
	(*handler->function) (handler->arg);
      cleanup_stack = handler->next;
    }

  mutex.UnLock ();
}

void
pthread::pop_all_cleanup_handlers ()
{
  while (cleanup_stack != NULL)
    pop_cleanup_handler (1);
}

void
pthread::cancel_self ()
{
  exit (PTHREAD_CANCELED);
}

DWORD
pthread::getThreadId ()
{
  return thread_id;
}

void
pthread::initCurrentThread ()
{
  cancel_event = ::CreateEvent (&sec_none_nih, TRUE, FALSE, NULL);
  if (!DuplicateHandle (GetCurrentProcess (), GetCurrentThread (),
                        GetCurrentProcess (), &win32_obj_id,
                        0, FALSE, DUPLICATE_SAME_ACCESS))
    win32_obj_id = NULL;
  setThreadIdtoCurrent ();
  setTlsSelfPointer (this);
}

/* static members */
bool
pthread_attr::isGoodObject (pthread_attr_t const *attr)
{
  if (verifyable_object_isvalid (attr, PTHREAD_ATTR_MAGIC) != VALID_OBJECT)
    return false;
  return true;
}

/* instance members */

pthread_attr::pthread_attr ():verifyable_object (PTHREAD_ATTR_MAGIC),
joinable (PTHREAD_CREATE_JOINABLE), contentionscope (PTHREAD_SCOPE_PROCESS),
inheritsched (PTHREAD_INHERIT_SCHED), stacksize (0)
{
  schedparam.sched_priority = 0;
}

pthread_attr::~pthread_attr ()
{
}

bool
pthread_condattr::isGoodObject (pthread_condattr_t const *attr)
{
  if (verifyable_object_isvalid (attr, PTHREAD_CONDATTR_MAGIC) != VALID_OBJECT)
    return false;
  return true;
}

pthread_condattr::pthread_condattr ():verifyable_object
  (PTHREAD_CONDATTR_MAGIC), shared (PTHREAD_PROCESS_PRIVATE)
{
}

pthread_condattr::~pthread_condattr ()
{
}

pthread_cond::pthread_cond (pthread_condattr *attr):verifyable_object (PTHREAD_COND_MAGIC)
{
  int temperr;
  this->shared = attr ? attr->shared : PTHREAD_PROCESS_PRIVATE;
  this->mutex = NULL;
  this->waiting = 0;

  this->win32_obj_id = ::CreateEvent (&sec_none_nih, false,	/* auto signal reset - which I think is pthreads like ? */
				      false,	/* start non signaled */
				      NULL /* no name */);
  /* TODO: make a shared mem mutex if out attributes request shared mem cond */
  cond_access = NULL;
  if ((temperr = pthread_mutex_init (&this->cond_access, NULL)))
    {
      system_printf ("couldn't init mutex, this %p errno %d", this, temperr);
      /* we need the mutex for correct behaviour */
      magic = 0;
    }

  if (!this->win32_obj_id)
    magic = 0;
  /* threadsafe addition is easy */
  next = (pthread_cond *) InterlockedExchangePointer (&MT_INTERFACE->conds, this);
}

pthread_cond::~pthread_cond ()
{
  if (win32_obj_id)
    CloseHandle (win32_obj_id);
  pthread_mutex_destroy (&cond_access);
  /* I'm not 100% sure the next bit is threadsafe. I think it is... */
  if (MT_INTERFACE->conds == this)
    InterlockedExchangePointer (&MT_INTERFACE->conds, this->next);
  else
    {
      pthread_cond *tempcond = MT_INTERFACE->conds;
      while (tempcond->next && tempcond->next != this)
	tempcond = tempcond->next;
      /* but there may be a race between the loop above and this statement */
      InterlockedExchangePointer (&tempcond->next, this->next);
    }
}

void
pthread_cond::BroadCast ()
{
  /* TODO: implement the same race fix as Signal has */
  if (pthread_mutex_lock (&cond_access))
    system_printf ("Failed to lock condition variable access mutex, this %p", this);
  int count = waiting;
  if (!pthread_mutex::isGoodObject (&mutex))
    {
      if (pthread_mutex_unlock (&cond_access))
	system_printf ("Failed to unlock condition variable access mutex, this %p", this);
      /* This isn't and API error - users are allowed to call this when no threads
	 are waiting
	 system_printf ("Broadcast called with invalid mutex");
      */
      return;
    }
  while (count--)
    PulseEvent (win32_obj_id);
  if (pthread_mutex_unlock (&cond_access))
    system_printf ("Failed to unlock condition variable access mutex, this %p", this);
}

void
pthread_cond::Signal ()
{
  if (pthread_mutex_lock (&cond_access))
    system_printf ("Failed to lock condition variable access mutex, this %p", this);
  if (!pthread_mutex::isGoodObject (&mutex))
    {
      if (pthread_mutex_unlock (&cond_access))
	system_printf ("Failed to unlock condition variable access mutex, this %p",
		       this);
      return;
    }
  int temp = waiting;
  if (!temp)
    /* nothing to signal */
    {
      if (pthread_mutex_unlock (&cond_access))
	system_printf ("Failed to unlock condition variable access mutex, this %p", this);
      return;
    }
  /* Prime the detection flag */
  ExitingWait = 1;
  /* Signal any waiting thread */
  PulseEvent (win32_obj_id);
  /* No one can start waiting until we release the condition access mutex */
  /* The released thread will decrement waiting when it gets a time slice...
     without waiting for the access mutex
   * InterLockedIncrement on 98 +, NT4 + returns the incremented value.
   * On 95, nt 3.51 < it returns a sign correct number - 0=0, + for greater than 0, -
   * for less than 0.
   * Because of this we cannot spin on the waiting count, but rather we need a
   * dedicated flag for a thread exiting the Wait function.
   * Also not that Interlocked* sync CPU caches with memory.
   */
  int spins = 10;
  /* When ExitingWait is nonzero after a decrement, the leaving thread has
   * done it's thing
   */
  while (InterlockedDecrement (&ExitingWait) == 0 && spins)
    {
      InterlockedIncrement (&ExitingWait);
      /* give up the cpu to force a context switch. */
      low_priority_sleep (0);
      if (spins == 5)
	/* we've had 5 timeslices, and the woken thread still hasn't done it's
	 * thing - maybe we raced it with the event? */
	PulseEvent (win32_obj_id);
      spins--;
    }
  if (waiting + 1 != temp)
    system_printf ("Released too many threads - %d now %d originally", waiting, temp);
  if (pthread_mutex_unlock (&cond_access))
    system_printf ("Failed to unlock condition variable access mutex, this %p", this);
}

int
pthread_cond::TimedWait (DWORD dwMilliseconds)
{
  DWORD rv;
  if (!wincap.has_signal_object_and_wait ())
    {
      // FIXME: race condition (potentially drop events
      // Possible solution (single process only) - place this in a critical section.
      ReleaseMutex (mutex->win32_obj_id);
      rv = WaitForSingleObject (win32_obj_id, dwMilliseconds);
    }
  else
    {
      LeaveCriticalSection (&mutex->criticalsection);
      rv = WaitForSingleObject (win32_obj_id, dwMilliseconds);
#if 0
    /* we need to use native win32 mutex's here, because the cygwin ones now use
     * critical sections, which are faster, but introduce a race _here_. Until then
     * The NT variant of the code is redundant.
     */

    rv = SignalObjectAndWait (mutex->win32_obj_id, win32_obj_id, dwMilliseconds,
			 false);
#endif
    }
  switch (rv)
    {
    case WAIT_FAILED:
      return 0;			/* POSIX doesn't allow errors after we modify the mutex state */
    case WAIT_ABANDONED:
    case WAIT_TIMEOUT:
      return ETIMEDOUT;
    case WAIT_OBJECT_0:
      return 0;			/* we have been signaled */
    default:
      return 0;
    }
}

void
pthread_cond::fixup_after_fork ()
{
  debug_printf ("cond %x in fixup_after_fork", this);
  if (shared != PTHREAD_PROCESS_PRIVATE)
    api_fatal ("doesn't understand PROCESS_SHARED condition variables");
  /* FIXME: duplicate code here and in the constructor. */
  this->win32_obj_id = ::CreateEvent (&sec_none_nih, false, false, NULL);
  if (!win32_obj_id)
    api_fatal ("failed to create new win32 mutex");
#if DETECT_BAD_APPS
  if (waiting)
    api_fatal ("Forked () while a condition variable has waiting threads.\nReport to [e-mail]");
#else
  waiting = 0;
  mutex = NULL;
#endif
}

/* pthread_key */
/* static members */
/* This stores pthread_key information across fork() boundaries */
List<pthread_key> pthread_key::keys;

void
pthread_key::saveAKey (pthread_key *key)
{
  key->saveKeyToBuffer ();
}

void
pthread_key::fixup_before_fork ()
{
  keys.forEach (saveAKey);
}

void
pthread_key::restoreAKey (pthread_key *key)
{
  key->recreateKeyFromBuffer ();
}

void
pthread_key::fixup_after_fork ()
{
  keys.forEach (restoreAKey);
}

void
pthread_key::destroyAKey (pthread_key *key)
{
  key->run_destructor ();
}

void
pthread_key::runAllDestructors ()
{
  keys.forEach (destroyAKey);
}

bool
pthread_key::isGoodObject (pthread_key_t const *key)
{
  if (verifyable_object_isvalid (key, PTHREAD_KEY_MAGIC) != VALID_OBJECT)
    return false;
  return true;
}

/* non-static members */

pthread_key::pthread_key (void (*aDestructor) (void *)):verifyable_object (PTHREAD_KEY_MAGIC), destructor (aDestructor)
{
  dwTlsIndex = TlsAlloc ();
  if (dwTlsIndex == TLS_OUT_OF_INDEXES)
    magic = 0;
  else
    keys.Insert (this);
}

pthread_key::~pthread_key ()
{
  /* We may need to make the list code lock the list during operations
   */
  if (magic != 0)
    {
      keys.Remove (this);
      TlsFree (dwTlsIndex);
    }
}

int
pthread_key::set (const void *value)
{
  /* the OS function doesn't perform error checking */
  TlsSetValue (dwTlsIndex, (void *) value);
  return 0;
}

void *
pthread_key::get () const
{
  int savedError = ::GetLastError ();
  void *result = TlsGetValue (dwTlsIndex);
  ::SetLastError (savedError);
  return result;
}

void
pthread_key::saveKeyToBuffer ()
{
  fork_buf = get ();
}

void
pthread_key::recreateKeyFromBuffer ()
{
  dwTlsIndex = TlsAlloc ();
  if (dwTlsIndex == TLS_OUT_OF_INDEXES)
    api_fatal ("pthread_key::recreateKeyFromBuffer () failed to reallocate Tls storage");
  set (fork_buf);
}

void
pthread_key::run_destructor ()
{
  if (destructor)
    {
      void *oldValue = get ();
      if (oldValue)
	{
    	  set (NULL);
    	  destructor (oldValue);
	}
    }
}

/* pshared mutexs:

   REMOVED FROM CURRENT. These can be reinstated with the daemon, when all the
   gymnastics can be a lot easier.

   the mutex_t (size 4) is not used as a verifyable object because we cannot
   guarantee the same address space for all processes.
   we use the following:
   high bit set (never a valid address).
   second byte is reserved for the priority.
   third byte is reserved
   fourth byte is the mutex id. (max 255 cygwin mutexs system wide).
   creating mutex's does get slower and slower, but as creation is a one time
   job, it should never become an issue

   And if you're looking at this and thinking, why not an array in cygwin for all mutexs,
   - you incur a penalty on _every_ mutex call and you have toserialise them all.
   ... Bad karma.

   option 2? put everything in userspace and update the ABI?
   - bad karma as well - the HANDLE, while identical across process's,
   Isn't duplicated, it's reopened. */

/* static members */
bool
pthread_mutex::isGoodObject (pthread_mutex_t const *mutex)
{
  if (verifyable_object_isvalid (mutex, PTHREAD_MUTEX_MAGIC) != VALID_OBJECT)
    return false;
  return true;
}

bool
pthread_mutex::isGoodInitializer (pthread_mutex_t const *mutex)
{
  if (verifyable_object_isvalid (mutex, PTHREAD_MUTEX_MAGIC, PTHREAD_MUTEX_INITIALIZER) != VALID_STATIC_OBJECT)
    return false;
  return true;
}

bool
pthread_mutex::isGoodInitializerOrObject (pthread_mutex_t const *mutex)
{
  if (verifyable_object_isvalid (mutex, PTHREAD_MUTEX_MAGIC, PTHREAD_MUTEX_INITIALIZER) == INVALID_OBJECT)
    return false;
  return true;
}

bool
pthread_mutex::isGoodInitializerOrBadObject (pthread_mutex_t const *mutex)
{
    verifyable_object_state objectState = verifyable_object_isvalid (mutex, PTHREAD_MUTEX_MAGIC, PTHREAD_MUTEX_INITIALIZER);
    if (objectState == VALID_OBJECT)
	return false;
    return true;
}

/* This is used for mutex creation protection within a single process only */
pthread_mutex::nativeMutex pthread_mutex::mutexInitializationLock NO_COPY;

/* We can only be called once.
   TODO: (no rush) use a non copied memory section to
   hold an initialization flag.  */
void
pthread_mutex::initMutex ()
{
  if (!mutexInitializationLock.init ())
    api_fatal ("Could not create win32 Mutex for pthread mutex static initializer support.");
}

pthread_mutex::pthread_mutex (pthread_mutexattr *attr):verifyable_object (PTHREAD_MUTEX_MAGIC)
{
  /* attr checked in the C call */
  if (attr && attr->pshared == PTHREAD_PROCESS_SHARED)
    {
      // fail
      magic = 0;
      return;
    }
  if (wincap.has_try_enter_critical_section ())
    InitializeCriticalSection (&criticalsection);
  else
    {
      this->win32_obj_id = ::CreateMutex (&sec_none_nih, false, NULL);
      if (!win32_obj_id)
	magic = 0;
    }
  condwaits = 0;
  pshared = PTHREAD_PROCESS_PRIVATE;
  /* threadsafe addition is easy */
  next = (pthread_mutex *) InterlockedExchangePointer (&MT_INTERFACE->mutexs, this);
}

pthread_mutex::~pthread_mutex ()
{
  if (wincap.has_try_enter_critical_section ())
    DeleteCriticalSection (&criticalsection);
  else
    {
      if (win32_obj_id)
	CloseHandle (win32_obj_id);
      win32_obj_id = NULL;
    }
  /* I'm not 100% sure the next bit is threadsafe. I think it is... */
  if (MT_INTERFACE->mutexs == this)
    /* TODO: printf an error if the return value != this */
    InterlockedExchangePointer (&MT_INTERFACE->mutexs, next);
  else
    {
      pthread_mutex *tempmutex = MT_INTERFACE->mutexs;
      while (tempmutex->next && tempmutex->next != this)
	tempmutex = tempmutex->next;
      /* but there may be a race between the loop above and this statement */
      /* TODO: printf an error if the return value != this */
      InterlockedExchangePointer (&tempmutex->next, this->next);
    }
}

int
pthread_mutex::Lock ()
{
  if (wincap.has_try_enter_critical_section ())
    {
      EnterCriticalSection (&criticalsection);
      return 0;
    }
  /* FIXME: Return 0 on success */
  return WaitForSingleObject (win32_obj_id, INFINITE);
}

/* returns non-zero on failure */
int
pthread_mutex::TryLock ()
{
  if (wincap.has_try_enter_critical_section ())
    return (!TryEnterCriticalSection (&criticalsection));
  return (WaitForSingleObject (win32_obj_id, 0) == WAIT_TIMEOUT);
}

int
pthread_mutex::UnLock ()
{
  if (wincap.has_try_enter_critical_section ())
    {
      LeaveCriticalSection (&criticalsection);
      return 0;
    }
  return (!ReleaseMutex (win32_obj_id));
}

void
pthread_mutex::fixup_after_fork ()
{
  debug_printf ("mutex %x in fixup_after_fork", this);
  if (pshared != PTHREAD_PROCESS_PRIVATE)
    api_fatal ("pthread_mutex::fixup_after_fork () doesn'tunderstand PROCESS_SHARED mutex's");
  /* FIXME: duplicate code here and in the constructor. */
  if (wincap.has_try_enter_critical_section ())
    InitializeCriticalSection (&criticalsection);
  else
    {
      win32_obj_id = ::CreateMutex (&sec_none_nih, false, NULL);
      if (!win32_obj_id)
	api_fatal ("pthread_mutex::fixup_after_fork () failed to create new win32 mutex");
    }
#if DETECT_BAD_APPS
  if (condwaits)
    api_fatal ("Forked () while a mutex has condition variables waiting on it.\nReport to [e-mail]");
#else
  condwaits = 0;
#endif
}

bool
pthread_mutex::nativeMutex::init ()
{
  theHandle = CreateMutex (&sec_none_nih, FALSE, NULL);
  if (!theHandle)
    {
      debug_printf ("CreateMutex failed. %E");
      return false;
    }
  return true;
}

bool
pthread_mutex::nativeMutex::lock ()
{
  DWORD waitResult = WaitForSingleObject (theHandle, INFINITE);
  if (waitResult != WAIT_OBJECT_0)
    {
      system_printf ("Received unexpected wait result %d on handle %p, %E", waitResult, theHandle);
      return false;
    }
  return true;
}

void
pthread_mutex::nativeMutex::unlock ()
{
  if (!ReleaseMutex (theHandle))
    system_printf ("Received a unexpected result releasing mutex. %E");
}

bool
pthread_mutexattr::isGoodObject (pthread_mutexattr_t const * attr)
{
  if (verifyable_object_isvalid (attr, PTHREAD_MUTEXATTR_MAGIC) != VALID_OBJECT)
    return false;
  return true;
}

pthread_mutexattr::pthread_mutexattr ():verifyable_object (PTHREAD_MUTEXATTR_MAGIC),
pshared (PTHREAD_PROCESS_PRIVATE), mutextype (PTHREAD_MUTEX_DEFAULT)
{
}

pthread_mutexattr::~pthread_mutexattr ()
{
}

semaphore::semaphore (int pshared, unsigned int value):verifyable_object (SEM_MAGIC)
{
  this->win32_obj_id = ::CreateSemaphore (&sec_none_nih, value, LONG_MAX,
					  NULL);
  if (!this->win32_obj_id)
    magic = 0;
  this->shared = pshared;
  currentvalue = value;
  /* threadsafe addition is easy */
  next = (semaphore *) InterlockedExchangePointer (&MT_INTERFACE->semaphores, this);
}

semaphore::~semaphore ()
{
  if (win32_obj_id)
    CloseHandle (win32_obj_id);
  /* I'm not 100% sure the next bit is threadsafe. I think it is... */
  if (MT_INTERFACE->semaphores == this)
    InterlockedExchangePointer (&MT_INTERFACE->semaphores, this->next);
  else
    {
      semaphore *tempsem = MT_INTERFACE->semaphores;
      while (tempsem->next && tempsem->next != this)
	tempsem = tempsem->next;
      /* but there may be a race between the loop above and this statement */
      InterlockedExchangePointer (&tempsem->next, this->next);
    }
}

void
semaphore::Post ()
{
  /* we can't use the currentvalue, because the wait functions don't let us access it */
  ReleaseSemaphore (win32_obj_id, 1, NULL);
  currentvalue++;
}

int
semaphore::TryWait ()
{
  /* FIXME: signals should be able to interrupt semaphores...
   *We probably need WaitForMultipleObjects here.
   */
  if (WaitForSingleObject (win32_obj_id, 0) == WAIT_TIMEOUT)
    {
      set_errno (EAGAIN);
      return -1;
    }
  currentvalue--;
  return 0;
}

void
semaphore::Wait ()
{
  WaitForSingleObject (win32_obj_id, INFINITE);
  currentvalue--;
}

void
semaphore::fixup_after_fork ()
{
  debug_printf ("sem %x in fixup_after_fork", this);
  if (shared != PTHREAD_PROCESS_PRIVATE)