xdasd_thread.c

Go to the documentation of this file.

/*----------------------------------------------------------------------------
 * Copyright (c) 2006, Novell, Inc.
 * All rights reserved.
 * 
 * Redistribution and use in source and binary forms, with or without 
 * modification, are permitted provided that the following conditions are 
 * met:
 * 
 *     * Redistributions of source code must retain the above copyright 
 *       notice, this list of conditions and the following disclaimer.
 *     * Redistributions in binary form must reproduce the above copyright 
 *       notice, this list of conditions and the following disclaimer in the 
 *       documentation and/or other materials provided with the distribution.
 *     * Neither the name of the Novell nor the names of its contributors 
 *       may be used to endorse or promote products derived from this 
 *       software without specific prior written permission.
 * 
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 
 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
 * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR 
 * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR 
 * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, 
 * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, 
 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR 
 * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF 
 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING 
 * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS 
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *--------------------------------------------------------------------------*/

#include "xdasd_thread.h"

#include <xdas_base.h>

#include <stddef.h>

thread_t thread_create(void *(*thread_proc)(void *), void * arg)
{
#ifdef _WIN32
   return (thread_t)CreateThread(0, 0, 
         (LPTHREAD_START_ROUTINE)thread_proc, arg, 0, 0);
#else
   pthread_t th;
   return (thread_t)(pthread_create(&th, 0, thread_proc, arg)? 0: th);
#endif
}

void * thread_wait(thread_t th)
{
   void * result = 0;
#ifdef _WIN32
   DWORD dwres;
   WaitForSingleObject((HANDLE)th, INFINITE);
   if (GetExitCodeThread((HANDLE)th, &dwres))
      result = (void *)(intptr_t)dwres;
   CloseHandle((HANDLE)th);
#else
   pthread_join((pthread_t)th, &result);
#endif
   return result;
}

#ifdef _WIN32

/* This is the largest value we can pass as the maxCount parameter 
   to CreateSemaphore on any Windows platform. */
#define MAX_SEM_COUNT   ((LONG)(0x7FFFFFFFL))

static int cv_unblock(cv_t * cvp, int unblock_all)
{
   int nSignalsToIssue;

   /* s_assert(cvp); */

   EnterCriticalSection(&cvp->csUnblockLock);

   if (cvp->waitersToUnblock != 0)
   {
      if (cvp->waitersBlocked == 0)
      {
         LeaveCriticalSection(&cvp->csUnblockLock);
         return 0;
      }
      if (unblock_all)
      {
         cvp->waitersToUnblock += (nSignalsToIssue = cvp->waitersBlocked);
         cvp->waitersBlocked = 0;
      }
      else
      {
         nSignalsToIssue = 1;
         cvp->waitersToUnblock++;
         cvp->waitersBlocked--;
      }
   }
   else if (cvp->waitersBlocked > cvp->waitersGone)
   {
      /* Use the non-cancellable version of sem_wait() */
      if (!ReleaseSemaphore(cvp->semBlockLock, 1, 0))
      {
         int err = (int)GetLastError();
         LeaveCriticalSection(&cvp->csUnblockLock);
         return err;
      }
      if (cvp->waitersGone != 0)
      {
         cvp->waitersBlocked -= cvp->waitersGone;
         cvp->waitersGone = 0;
      }
      if (unblock_all)
      {
         nSignalsToIssue = cvp->waitersToUnblock = cvp->waitersBlocked;
         cvp->waitersBlocked = 0;
      }
      else
      {
         nSignalsToIssue = cvp->waitersToUnblock = 1;
         cvp->waitersBlocked--;
      }
   }
   else
   {
      LeaveCriticalSection(&cvp->csUnblockLock);
      return 0;
   }

   LeaveCriticalSection(&cvp->csUnblockLock);

   return ReleaseSemaphore(cvp->semBlockQueue, 
         nSignalsToIssue, 0)? 0: (int)GetLastError();
}

int cv_create(cv_t * cvp)
{
   /* s_assert(cvp); */

   memset(cvp, 0, sizeof(*cvp));
   if ((cvp->semBlockLock = CreateSemaphore(0, 1, MAX_SEM_COUNT, 0)) == 0
         || (cvp->semBlockQueue = CreateSemaphore(0, 0, MAX_SEM_COUNT, 0)) == 0)
   {
      if (cvp->semBlockLock)
         CloseHandle(cvp->semBlockLock);
      return 0;
   }
   InitializeCriticalSection(&cvp->csUnblockLock);
   return 0;
}

void cv_destroy(cv_t * cvp)
{
   /* s_assert(cvpp && *cvpp); */

   /* Close the gate; this will synchronize this thread with
      all already signaled waiters to let them retract their
      waiter status - SEE NOTE 1 ABOVE!!! */

   if (WaitForSingleObject(cvp->semBlockLock, INFINITE) != WAIT_OBJECT_0)
      return;

   /* !TRY! lock csUnblockLock; try will detect busy condition
      and will not cause a deadlock with respect to concurrent
      signal/broadcast. */

   if (TryEnterCriticalSection(&cvp->csUnblockLock) == FALSE)
   {
      ReleaseSemaphore(cvp->semBlockLock, 1, 0);
      return;
   }

   /* Check whether cv is still busy (still has waiters) */
   if (cvp->waitersBlocked > cvp->waitersGone)
   {
      ReleaseSemaphore(cvp->semBlockLock, 1, 0);
      LeaveCriticalSection(&cvp->csUnblockLock);
      return;
   }

   /* Now it is safe to destroy */
   CloseHandle(cvp->semBlockLock);
   CloseHandle(cvp->semBlockQueue);
   LeaveCriticalSection(&cvp->csUnblockLock);
   DeleteCriticalSection(&cvp->csUnblockLock);
}

int cv_wait(cv_t * cvp, mutex_t * mp)
{
   DWORD rv;
   int err = 0;
   int nSignalsWasLeft;

   /* s_assert(cvp && mp); */

   if (WaitForSingleObject(cvp->semBlockLock, INFINITE) != WAIT_OBJECT_0)
      return -1;

   ++cvp->waitersBlocked;
   if (!ReleaseSemaphore(cvp->semBlockLock, 1, 0))
      return -1;

   /* Now we can release 'mutex' and... */
   LeaveCriticalSection(mp);

   /* ...wait to be awakened by
                 pthread_cond_signal, or
                 pthread_cond_broadcast, or
                 timeout, or
                 thread cancellation
   
      Note:
           sem_timedwait is a cancellation point,
           hence providing the mechanism for making
           pthread_cond_wait a cancellation point.
           We use the cleanup mechanism to ensure we
           re-lock the mutex and adjust (to)unblock(ed) waiters
           counts if we are cancelled, timed out or signalled.
    */

   if ((rv = WaitForSingleObject(cvp->semBlockQueue, INFINITE)) != WAIT_OBJECT_0)
      err = rv == WAIT_TIMEOUT? -1: -2;

   /* Whether we got here as a result of signal/broadcast or because of
      timeout on wait or thread cancellation we indicate that we are no
      longer waiting. The waiter is responsible for adjusting waiters
      (to)unblock(ed) counts (protected by unblock lock). */

   EnterCriticalSection(&cvp->csUnblockLock);

   if ((nSignalsWasLeft = cvp->waitersToUnblock) != 0)
      --cvp->waitersToUnblock;
   else if (++cvp->waitersGone == (unsigned)(-1) / 2)
   {
      /* Use the non-cancellable version of sem_wait() */
      if (WaitForSingleObject(cvp->semBlockLock, INFINITE) != WAIT_OBJECT_0)
         return (int)GetLastError();

      cvp->waitersBlocked -= cvp->waitersGone;
      if (!ReleaseSemaphore(cvp->semBlockLock, 1, 0))
         return (int)GetLastError();

      cvp->waitersGone = 0;
   }
   LeaveCriticalSection(&cvp->csUnblockLock);
   if (nSignalsWasLeft == 1)
      if (!ReleaseSemaphore(cvp->semBlockLock, 1, 0))
         return (int)GetLastError();

   /* XSH: Upon successful return, the mutex has been locked and is owned
      by the calling thread */
   EnterCriticalSection(mp);
   return err;
}

int cv_signal(cv_t * cvp)
{
   /* s_assert(cvp); */
   return cv_unblock(cvp, 0);
}

int cv_broadcast(cv_t * cvp)
{
   /* s_assert(cvp); */
   return cv_unblock(cvp, 1);
}

#else

static void p_operation_cleanup(void * arg)
      { pthread_mutex_unlock(&((sem_t *)arg)->lock); }

int sem_create(sem_t * sp, int count)
{
   if (pthread_mutex_init(&sp->lock, 0) != 0)
      return -1;
   if (pthread_cond_init(&sp->cond, 0) != 0)
   {
      pthread_mutex_destroy(&sp->lock);
      return -1;
   }
   sp->count = count;
   return 0;
}

void sem_destroy(sem_t * sp)
{
   (void)pthread_mutex_destroy(&sp->lock);
   (void)pthread_cond_destroy(&sp->cond);
}

void sem_wait(sem_t * sp)
{
   pthread_mutex_lock(&sp->lock);
   pthread_cleanup_push(p_operation_cleanup, sp);

   while (sp->count <= 0)
      pthread_cond_wait(&sp->cond, &sp->lock);
   sp->count--;

   pthread_cleanup_pop(1);
}

void sem_signal(sem_t * sp, int count)
{
   pthread_mutex_lock(&sp->lock);

   sp->count += count;
   if (sp->count <= 0)
      pthread_cond_signal(&sp->cond);

   pthread_mutex_unlock(&sp->lock);
}

#endif /* ?_WIN32 */