ncbimtx.cpp

ncbi源码

#endif
    }
#endif
}




/////////////////////////////////////////////////////////////////////////////
//  SEMAPHORE
//


// Platform-specific representation (or emulation) of semaphore
struct SSemaphore
{
#if defined(NCBI_POSIX_THREADS)
    unsigned int     max_count;
    unsigned int     count; 
    unsigned int     wait_count;  // # of threads currently waiting on the sema
    pthread_mutex_t  mutex;
    pthread_cond_t   cond;

#elif defined(NCBI_WIN32_THREADS)
    HANDLE           sem;
#else
    unsigned int     max_count;
    unsigned int     count;
#endif
};


CSemaphore::CSemaphore(unsigned int init_count, unsigned int max_count)
{
    xncbi_Validate(max_count != 0,
                   "CSemaphore::CSemaphore() - max_count passed zero");
    xncbi_Validate(init_count <= max_count,
                   "CSemaphore::CSemaphore() - init_count "
                   "greater than max_count");

    m_Sem = new SSemaphore;
    auto_ptr<SSemaphore> auto_sem(m_Sem);

#if defined(NCBI_POSIX_THREADS)
    m_Sem->max_count = max_count;
    m_Sem->count     = init_count;
    m_Sem->wait_count = 0;

    xncbi_Validate(pthread_mutex_init(&m_Sem->mutex, 0) == 0,
                   "CSemaphore::CSemaphore() - pthread_mutex_init() failed");
    xncbi_Validate(pthread_cond_init(&m_Sem->cond, 0) == 0,
                   "CSemaphore::CSemaphore() - pthread_cond_init() failed");

#elif defined(NCBI_WIN32_THREADS)
    m_Sem->sem = CreateSemaphore(NULL, init_count, max_count, NULL);
    xncbi_Validate(m_Sem->sem != NULL,
                   "CSemaphore::CSemaphore() - CreateSemaphore() failed");

#else
    m_Sem->max_count = max_count;
    m_Sem->count     = init_count;
#endif

    auto_sem.release();
}


CSemaphore::~CSemaphore(void)
{
#if defined(NCBI_POSIX_THREADS)
    _ASSERT(m_Sem->wait_count == 0);
    xncbi_Verify(pthread_mutex_destroy(&m_Sem->mutex) == 0);
    xncbi_Verify(pthread_cond_destroy(&m_Sem->cond)  == 0);

#elif defined(NCBI_WIN32_THREADS)
    xncbi_Verify( CloseHandle(m_Sem->sem) );
#endif

    delete m_Sem;
}


void CSemaphore::Wait(void)
{
#if defined(NCBI_POSIX_THREADS)
    xncbi_Validate(pthread_mutex_lock(&m_Sem->mutex) == 0,
                   "CSemaphore::Wait() - pthread_mutex_lock() failed");

    if (m_Sem->count != 0) {
        m_Sem->count--;
    }
    else {
        m_Sem->wait_count++;
        do {
            if (pthread_cond_wait(&m_Sem->cond, &m_Sem->mutex) != 0) {
                xncbi_Validate(pthread_mutex_unlock(&m_Sem->mutex) == 0,
                               "CSemaphore::Wait() - "
                               "pthread_cond_wait() and "
                               "pthread_mutex_unlock() failed");
                xncbi_Validate(0,
                               "CSemaphore::Wait() - "
                               "pthread_cond_wait() failed");
            }
        } while (m_Sem->count == 0);
        m_Sem->wait_count--;
        m_Sem->count--;
    }

    xncbi_Validate(pthread_mutex_unlock(&m_Sem->mutex) == 0,
                   "CSemaphore::Wait() - pthread_mutex_unlock() failed");

#elif defined(NCBI_WIN32_THREADS)
    xncbi_Validate(WaitForSingleObject(m_Sem->sem, INFINITE) == WAIT_OBJECT_0,
                   "CSemaphore::Wait() - WaitForSingleObject() failed");

#else
    xncbi_Validate(m_Sem->count != 0,
                   "CSemaphore::Wait() - "
                   "wait with zero count in one-thread mode(?!)");
    m_Sem->count--;
#endif
}

#if defined(NCBI_NO_THREADS)
# define NCBI_THREADS_ARG(arg) 
#else
# define NCBI_THREADS_ARG(arg) arg
#endif

bool CSemaphore::TryWait(unsigned int NCBI_THREADS_ARG(timeout_sec),
                         unsigned int NCBI_THREADS_ARG(timeout_nsec))
{
#if defined(NCBI_POSIX_THREADS)
    xncbi_Validate(pthread_mutex_lock(&m_Sem->mutex) == 0,
                   "CSemaphore::TryWait() - pthread_mutex_lock() failed");

    bool retval = false;
    if (m_Sem->count != 0) {
        m_Sem->count--;
        retval = true;
    }
    else if (timeout_sec > 0  ||  timeout_nsec > 0) {
# ifdef NCBI_OS_SOLARIS
        // arbitrary limit of 100Ms (~3.1 years) -- supposedly only for
        // native threads, but apparently also for POSIX threads :-/
        if (timeout_sec >= 100 * 1000 * 1000) {
            timeout_sec  = 100 * 1000 * 1000;
            timeout_nsec = 0;
        }
# endif
        static const unsigned int kBillion = 1000 * 1000 * 1000;
        struct timeval  now;
        struct timespec timeout = { 0, 0 };
        gettimeofday(&now, 0);
        // timeout_sec added below to avoid overflow
        timeout.tv_sec  = now.tv_sec;
        timeout.tv_nsec = now.tv_usec * 1000 + timeout_nsec;
        if ((unsigned int)timeout.tv_nsec >= kBillion) {
            timeout.tv_sec  += timeout.tv_nsec / kBillion;
            timeout.tv_nsec %= kBillion;
        }
        if (timeout_sec > (unsigned int)(kMax_Int - timeout.tv_sec)) {
            // Max out rather than overflowing
            timeout.tv_sec  = kMax_Int;
            timeout.tv_nsec = kBillion - 1;
        } else {
            timeout.tv_sec += timeout_sec;
        }
        
        m_Sem->wait_count++;
        do {
            int status = pthread_cond_timedwait(&m_Sem->cond, &m_Sem->mutex,
                                                &timeout);
            if (status == ETIMEDOUT) {
                break;
            } else if (status != 0  &&  status != EINTR) {
                // EINVAL, presumably?
                xncbi_Validate(pthread_mutex_unlock(&m_Sem->mutex) == 0,
                               "CSemaphore::TryWait() - "
                               "pthread_cond_timedwait() and "
                               "pthread_mutex_unlock() failed");
                xncbi_Validate(0, "CSemaphore::TryWait() - "
                               "pthread_cond_timedwait() failed");
            }
        } while (m_Sem->count == 0);
        m_Sem->wait_count--;
        m_Sem->count--;
        retval = true;
    }

    xncbi_Validate(pthread_mutex_unlock(&m_Sem->mutex) == 0,
                   "CSemaphore::TryWait() - pthread_mutex_unlock() failed");

    return retval;

#elif defined(NCBI_WIN32_THREADS)
    DWORD timeout_msec; // DWORD == unsigned long
    if (timeout_sec >= kMax_ULong / 1000) {
        timeout_msec = kMax_ULong;
    } else {
        timeout_msec = timeout_sec * 1000 + timeout_nsec / (1000 * 1000);
    }
    DWORD res = WaitForSingleObject(m_Sem->sem, timeout_msec);
    xncbi_Validate(res == WAIT_OBJECT_0  ||  res == WAIT_TIMEOUT,
                   "CSemaphore::TryWait() - WaitForSingleObject() failed");
    return (res == WAIT_OBJECT_0);

#else
    if (m_Sem->count == 0)
        return false;
    m_Sem->count--;
    return true;
#endif
}


void CSemaphore::Post(unsigned int count)
{
    if (count == 0)
        return;

#if defined (NCBI_POSIX_THREADS)
    xncbi_Validate(pthread_mutex_lock(&m_Sem->mutex) == 0,
                   "CSemaphore::Post() - pthread_mutex_lock() failed");

    if (m_Sem->count > kMax_UInt - count  ||
        m_Sem->count + count > m_Sem->max_count) {
        xncbi_Validate(pthread_mutex_unlock(&m_Sem->mutex) == 0,
                       "CSemaphore::Post() - "
                       "attempt to exceed max_count and "
                       "pthread_mutex_unlock() failed");
        xncbi_Validate(m_Sem->count <= kMax_UInt - count,
                       "CSemaphore::Post() - "
                       "would result in counter > MAX_UINT");
        xncbi_Validate(m_Sem->count + count <= m_Sem->max_count,
                       "CSemaphore::Post() - attempt to exceed max_count");
        _TROUBLE;
    }

    // Signal some (or all) of the threads waiting on this semaphore
    int err_code = 0;
    if (m_Sem->count + count >= m_Sem->wait_count) {
        err_code = pthread_cond_broadcast(&m_Sem->cond);
    } else {
        // Do not use broadcast here to avoid waking up more threads
        // than really needed...
        for (unsigned int n_sig = 0;  n_sig < count;  n_sig++) {
            err_code = pthread_cond_signal(&m_Sem->cond);
            if (err_code != 0) {
                err_code = pthread_cond_broadcast(&m_Sem->cond);
                break;
            }
        }
    }

    // Success
    if (err_code == 0) {
        m_Sem->count += count;
        xncbi_Validate(pthread_mutex_unlock(&m_Sem->mutex) == 0,
                       "CSemaphore::Post() - pthread_mutex_unlock() failed");
        return;
    }

    // Error
    xncbi_Validate(pthread_mutex_unlock(&m_Sem->mutex) == 0,
                   "CSemaphore::Post() - "
                   "pthread_cond_signal/broadcast() and "
                   "pthread_mutex_unlock() failed");
    xncbi_Validate(0,
                   "CSemaphore::Post() - "
                   "pthread_cond_signal/broadcast() failed");

#elif defined(NCBI_WIN32_THREADS)
    xncbi_Validate(ReleaseSemaphore(m_Sem->sem, count, NULL),
                   "CSemaphore::Post() - ReleaseSemaphore() failed");

#else
    xncbi_Validate(m_Sem->count + count <= m_Sem->max_count,
                   "CSemaphore::Post() - attempt to exceed max_count");
    m_Sem->count += count;
#endif
}

END_NCBI_SCOPE


/*
 * ===========================================================================
 * $Log: ncbimtx.cpp,v $
 * Revision 1000.2  2004/06/03 19:28:17  gouriano
 * PRODUCTION: UPGRADED [GCC34_MSVC7] Dev-tree R1.17
 *
 * Revision 1.17  2004/06/02 16:27:06  ucko
 * SSystemFastMutex::InitializeDynamic: drop gratuituous and potentially
 * error-prone check for double initialization after discussion with
 * Eugene Vasilchenko.
 *
 * Revision 1.16  2004/05/14 13:59:27  gorelenk
 * Added include of ncbi_pch.hpp
 *
 * Revision 1.15  2003/09/29 20:45:54  ivanov
 * CAutoInitializeStatic[Fast]Mutex:  changed method of initialization for static mutexes on MS Windows (by Eugene Vasilchenko)
 *
 * Revision 1.14  2003/09/17 17:56:47  vasilche
 * Fixed volatile methods of CThreadSystemID.
 *
 * Revision 1.13  2003/09/17 15:20:46  vasilche
 * Moved atomic counter swap functions to separate file.
 * Added CRef<>::AtomicResetFrom(), CRef<>::AtomicReleaseTo() methods.
 *
 * Revision 1.12  2003/09/02 19:03:59  vasilche
 * Removed debug abort().
 *
 * Revision 1.11  2003/09/02 16:08:49  vasilche
 * Fixed race condition with optimization on some compilers - added 'volatile'.
 * Moved mutex Lock/Unlock methods out of inline section - they are quite complex.
 *
 * Revision 1.10  2003/05/06 16:12:24  vasilche
 * Added check for mutexes located in stack.
 *
 * Revision 1.9  2002/09/24 18:29:53  vasilche
 * Removed TAB symbols. Removed unused arg warning in single thread mode.
 *
 * Revision 1.8  2002/09/23 13:47:23  vasilche
 * Made static mutex structures POD-types on Win32
 *
 * Revision 1.7  2002/09/20 18:46:24  vasilche
 * Fixed volatile incompatibility on Win32
 *
 * Revision 1.6  2002/09/19 20:24:08  vasilche
 * Replace missing std::count() by std::find()
 *
 * Revision 1.5  2002/09/19 20:05:42  vasilche
 * Safe initialization of static mutexes
 *
 * Revision 1.4  2002/07/11 14:18:27  gouriano
 * exceptions replaced by CNcbiException-type ones
 *
 * Revision 1.3  2002/04/11 21:08:02  ivanov
 * CVS log moved to end of the file
 *
 * Revision 1.2  2001/12/13 19:45:36  gouriano
 * added xxValidateAction functions
 *
 * Revision 1.1  2001/03/26 20:31:13  vakatov
 * Initial revision (moved code from "ncbithr.cpp")
 *
 * ===========================================================================
 */