lightweight_semaphore.hpp

Go to the documentation of this file.

#ifndef GENESIS_UTILS_THREADING_LIGHTWEIGHT_SEMAPHORE_H_
#define GENESIS_UTILS_THREADING_LIGHTWEIGHT_SEMAPHORE_H_
 
/*
    Genesis - A toolkit for working with phylogenetic data.
    Copyright (C) 2014-2024 Lucas Czech
 
    This program is free software: you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.
 
    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.
 
    You should have received a copy of the GNU General Public License
    along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
    Contact:
    Lucas Czech <lucas.czech@sund.ku.dk>
    University of Copenhagen, Globe Institute, Section for GeoGenetics
    Oster Voldgade 5-7, 1350 Copenhagen K, Denmark
*/
 
/*
    This code below is adapted from the excellent moodycamel::ConcurrentQueue
    (https://github.com/cameron314/concurrentqueue), using version v1.0.4,
    which was published under a simplified BSD license, and also dual-licensed
    under the Boost Software License. The full original license
    (https://github.com/cameron314/concurrentqueue/blob/master/LICENSE.md), is copied
    in our documentation as well, see `genesis/doc/manual/supplement/acknowledgements.md`, and
    @link supplement_acknowledgements_code_reuse_concurrent_queue Acknowledgements@endlink.
 
    See also ConcurrentQueue and BlockingConcurrentQueue for the other two classes
    that we adapted from the original repository.
 
    We adapted the original code by (roughly) formatting it to our formatting standard, as well as
    renaming the namespace from moodycamel to be contained within our namespace, to keep our
    documentation and usage consistent. Other than that, all functionality is kept as-is.
    It feels weird to rename the namespaces, as it might seem that we are trying to hide the fact
    that this code is not ours. That is not the case, and is merely based on our compulsion
    for orderly namespaces and documentation within genesis. Please note that the original code
    is excellet work that we would like to acknowledge here! Also, if anyone wants to do the same
    with genesis code, please feel free to do so ;-)
*/
 
// =================================================================================================
//     Lightweight Semaphore
// =================================================================================================
 
// Provides an efficient implementation of a semaphore (LightweightSemaphore).
// This is an extension of Jeff Preshing's sempahore implementation (licensed
// under the terms of its separate zlib license) that has been adapted and
// extended by Cameron Desrochers.
 
#include <atomic>
#include <cstddef> // For std::size_t
#include <type_traits> // For std::make_signed<T>
 
#if defined(_WIN32)
// Avoid including windows.h in a header; we only need a handful of
// items, so we'll redeclare them here (this is relatively safe since
// the API generally has to remain stable between Windows versions).
// I know this is an ugly hack but it still beats polluting the global
// namespace with thousands of generic names or adding a .cpp for nothing.
extern "C" {
struct _SECURITY_ATTRIBUTES;
__declspec(dllimport) void* __stdcall CreateSemaphoreW(_SECURITY_ATTRIBUTES* lpSemaphoreAttributes, long lInitialCount, long lMaximumCount, const wchar_t* lpName);
__declspec(dllimport) int __stdcall CloseHandle(void* hObject);
__declspec(dllimport) unsigned long __stdcall WaitForSingleObject(void* hHandle, unsigned long dwMilliseconds);
__declspec(dllimport) int __stdcall ReleaseSemaphore(void* hSemaphore, long lReleaseCount, long* lpPreviousCount);
}
#elif defined(__MACH__)
#include <mach/mach.h>
#elif defined(__MVS__)
#include <zos-semaphore.h>
#elif defined(__unix__)
#include <semaphore.h>
 
#if defined(__GLIBC_PREREQ) && defined(_GNU_SOURCE)
#if __GLIBC_PREREQ(2, 30)
#define MOODYCAMEL_LIGHTWEIGHTSEMAPHORE_MONOTONIC
#endif
#endif
#endif
 
namespace genesis {
namespace utils {
namespace details {
 
// Code in the mpmc_sema namespace below is an adaptation of Jeff Preshing's
// portable + lightweight semaphore implementations, originally from
// https://github.com/preshing/cpp11-on-multicore/blob/master/common/sema.h
// LICENSE:
// Copyright (c) 2015 Jeff Preshing
//
// This software is provided 'as-is', without any express or implied
// warranty. In no event will the authors be held liable for any damages
// arising from the use of this software.
//
// Permission is granted to anyone to use this software for any purpose,
// including commercial applications, and to alter it and redistribute it
// freely, subject to the following restrictions:
//
// 1. The origin of this software must not be misrepresented; you must not
//  claim that you wrote the original software. If you use this software
//  in a product, an acknowledgement in the product documentation would be
//  appreciated but is not required.
// 2. Altered source versions must be plainly marked as such, and must not be
//  misrepresented as being the original software.
// 3. This notice may not be removed or altered from any source distribution.
#if defined(_WIN32)
    class Semaphore {
    private:
        void* m_hSema;
 
        Semaphore(const Semaphore& other) MOODYCAMEL_DELETE_FUNCTION;
        Semaphore& operator=(const Semaphore& other) MOODYCAMEL_DELETE_FUNCTION;
 
    public:
        Semaphore(int initialCount = 0)
        {
            assert(initialCount >= 0);
            const long maxLong = 0x7fffffff;
            m_hSema = CreateSemaphoreW(nullptr, initialCount, maxLong, nullptr);
            assert(m_hSema);
        }
 
        ~Semaphore()
        {
            CloseHandle(m_hSema);
        }
 
        bool wait()
        {
            const unsigned long infinite = 0xffffffff;
            return WaitForSingleObject(m_hSema, infinite) == 0;
        }
 
        bool try_wait()
        {
            return WaitForSingleObject(m_hSema, 0) == 0;
        }
 
        bool timed_wait(std::uint64_t usecs)
        {
            return WaitForSingleObject(m_hSema, (unsigned long)(usecs / 1000)) == 0;
        }
 
        void signal(int count = 1)
        {
            while (!ReleaseSemaphore(m_hSema, count, nullptr))
                ;
        }
    };
#elif defined(__MACH__)
    //---------------------------------------------------------
    // Semaphore (Apple iOS and OSX)
    // Can't use POSIX semaphores due to http://lists.apple.com/archives/darwin-kernel/2009/Apr/msg00010.html
    //---------------------------------------------------------
    class Semaphore {
    private:
        semaphore_t m_sema;
 
        Semaphore(const Semaphore& other) MOODYCAMEL_DELETE_FUNCTION;
        Semaphore& operator=(const Semaphore& other) MOODYCAMEL_DELETE_FUNCTION;
 
    public:
        Semaphore(int initialCount = 0)
        {
            assert(initialCount >= 0);
            kern_return_t rc = semaphore_create(mach_task_self(), &m_sema, SYNC_POLICY_FIFO, initialCount);
            assert(rc == KERN_SUCCESS);
            (void)rc;
        }
 
        ~Semaphore()
        {
            semaphore_destroy(mach_task_self(), m_sema);
        }
 
        bool wait()
        {
            return semaphore_wait(m_sema) == KERN_SUCCESS;
        }
 
        bool try_wait()
        {
            return timed_wait(0);
        }
 
        bool timed_wait(std::uint64_t timeout_usecs)
        {
            mach_timespec_t ts;
            ts.tv_sec = static_cast<unsigned int>(timeout_usecs / 1000000);
            ts.tv_nsec = static_cast<int>((timeout_usecs % 1000000) * 1000);
 
            // added in OSX 10.10: https://developer.apple.com/library/prerelease/mac/documentation/General/Reference/APIDiffsMacOSX10_10SeedDiff/modules/Darwin.html
            kern_return_t rc = semaphore_timedwait(m_sema, ts);
            return rc == KERN_SUCCESS;
        }
 
        void signal()
        {
            while (semaphore_signal(m_sema) != KERN_SUCCESS)
                ;
        }
 
        void signal(int count)
        {
            while (count-- > 0) {
                while (semaphore_signal(m_sema) != KERN_SUCCESS)
                    ;
            }
        }
    };
#elif defined(__unix__) || defined(__MVS__)
    //---------------------------------------------------------
    // Semaphore (POSIX, Linux, zOS)
    //---------------------------------------------------------
    class Semaphore {
    private:
        sem_t m_sema;
 
        Semaphore(const Semaphore& other) MOODYCAMEL_DELETE_FUNCTION;
        Semaphore& operator=(const Semaphore& other) MOODYCAMEL_DELETE_FUNCTION;
 
    public:
        Semaphore(int initialCount = 0)
        {
            assert(initialCount >= 0);
            int rc = sem_init(&m_sema, 0, static_cast<unsigned int>(initialCount));
            assert(rc == 0);
            (void)rc;
        }
 
        ~Semaphore()
        {
            sem_destroy(&m_sema);
        }
 
        bool wait()
        {
            // http://stackoverflow.com/questions/2013181/gdb-causes-sem-wait-to-fail-with-eintr-error
            int rc;
            do {
                rc = sem_wait(&m_sema);
            } while (rc == -1 && errno == EINTR);
            return rc == 0;
        }
 
        bool try_wait()
        {
            int rc;
            do {
                rc = sem_trywait(&m_sema);
            } while (rc == -1 && errno == EINTR);
            return rc == 0;
        }
 
        bool timed_wait(std::uint64_t usecs)
        {
            struct timespec ts;
            const int usecs_in_1_sec = 1000000;
            const int nsecs_in_1_sec = 1000000000;
#ifdef MOODYCAMEL_LIGHTWEIGHTSEMAPHORE_MONOTONIC
            clock_gettime(CLOCK_MONOTONIC, &ts);
#else
            clock_gettime(CLOCK_REALTIME, &ts);
#endif
            ts.tv_sec += (time_t)(usecs / usecs_in_1_sec);
            ts.tv_nsec += (long)(usecs % usecs_in_1_sec) * 1000;
            // sem_timedwait bombs if you have more than 1e9 in tv_nsec
            // so we have to clean things up before passing it in
            if (ts.tv_nsec >= nsecs_in_1_sec) {
                ts.tv_nsec -= nsecs_in_1_sec;
                ++ts.tv_sec;
            }
 
            int rc;
            do {
#ifdef MOODYCAMEL_LIGHTWEIGHTSEMAPHORE_MONOTONIC
                rc = sem_clockwait(&m_sema, CLOCK_MONOTONIC, &ts);
#else
                rc = sem_timedwait(&m_sema, &ts);
#endif
            } while (rc == -1 && errno == EINTR);
            return rc == 0;
        }
 
        void signal()
        {
            while (sem_post(&m_sema) == -1)
                ;
        }
 
        void signal(int count)
        {
            while (count-- > 0) {
                while (sem_post(&m_sema) == -1)
                    ;
            }
        }
    };
#else
#error Unsupported platform! (No semaphore wrapper available)
#endif
 
} // end namespace details
 
//---------------------------------------------------------
// LightweightSemaphore
//---------------------------------------------------------
class LightweightSemaphore {
public:
    typedef std::make_signed<std::size_t>::type ssize_t;
 
private:
    std::atomic<ssize_t> m_count;
    details::Semaphore m_sema;
    int m_maxSpins;
 
    bool waitWithPartialSpinning(std::int64_t timeout_usecs = -1)
    {
        ssize_t oldCount;
        int spin = m_maxSpins;
        while (--spin >= 0) {
            oldCount = m_count.load(std::memory_order_relaxed);
            if ((oldCount > 0) && m_count.compare_exchange_strong(oldCount, oldCount - 1, std::memory_order_acquire, std::memory_order_relaxed))
                return true;
            std::atomic_signal_fence(std::memory_order_acquire); // Prevent the compiler from collapsing the loop.
        }
        oldCount = m_count.fetch_sub(1, std::memory_order_acquire);
        if (oldCount > 0)
            return true;
        if (timeout_usecs < 0) {
            if (m_sema.wait())
                return true;
        }
        if (timeout_usecs > 0 && m_sema.timed_wait((std::uint64_t)timeout_usecs))
            return true;
        // At this point, we've timed out waiting for the semaphore, but the
        // count is still decremented indicating we may still be waiting on
        // it. So we have to re-adjust the count, but only if the semaphore
        // wasn't signaled enough times for us too since then. If it was, we
        // need to release the semaphore too.
        while (true) {
            oldCount = m_count.load(std::memory_order_acquire);
            if (oldCount >= 0 && m_sema.try_wait())
                return true;
            if (oldCount < 0 && m_count.compare_exchange_strong(oldCount, oldCount + 1, std::memory_order_relaxed, std::memory_order_relaxed))
                return false;
        }
    }
 
    ssize_t waitManyWithPartialSpinning(ssize_t max, std::int64_t timeout_usecs = -1)
    {
        assert(max > 0);
        ssize_t oldCount;
        int spin = m_maxSpins;
        while (--spin >= 0) {
            oldCount = m_count.load(std::memory_order_relaxed);
            if (oldCount > 0) {
                ssize_t newCount = oldCount > max ? oldCount - max : 0;
                if (m_count.compare_exchange_strong(oldCount, newCount, std::memory_order_acquire, std::memory_order_relaxed))
                    return oldCount - newCount;
            }
            std::atomic_signal_fence(std::memory_order_acquire);
        }
        oldCount = m_count.fetch_sub(1, std::memory_order_acquire);
        if (oldCount <= 0) {
            if ((timeout_usecs == 0) || (timeout_usecs < 0 && !m_sema.wait()) || (timeout_usecs > 0 && !m_sema.timed_wait((std::uint64_t)timeout_usecs))) {
                while (true) {
                    oldCount = m_count.load(std::memory_order_acquire);
                    if (oldCount >= 0 && m_sema.try_wait())
                        break;
                    if (oldCount < 0 && m_count.compare_exchange_strong(oldCount, oldCount + 1, std::memory_order_relaxed, std::memory_order_relaxed))
                        return 0;
                }
            }
        }
        if (max > 1)
            return 1 + tryWaitMany(max - 1);
        return 1;
    }
 
public:
    LightweightSemaphore(ssize_t initialCount = 0, int maxSpins = 10000)
        : m_count(initialCount)
        , m_maxSpins(maxSpins)
    {
        assert(initialCount >= 0);
        assert(maxSpins >= 0);
    }
 
    bool tryWait()
    {
        ssize_t oldCount = m_count.load(std::memory_order_relaxed);
        while (oldCount > 0) {
            if (m_count.compare_exchange_weak(oldCount, oldCount - 1, std::memory_order_acquire, std::memory_order_relaxed))
                return true;
        }
        return false;
    }
 
    bool wait()
    {
        return tryWait() || waitWithPartialSpinning();
    }
 
    bool wait(std::int64_t timeout_usecs)
    {
        return tryWait() || waitWithPartialSpinning(timeout_usecs);
    }
 
    // Acquires between 0 and (greedily) max, inclusive
    ssize_t tryWaitMany(ssize_t max)
    {
        assert(max >= 0);
        ssize_t oldCount = m_count.load(std::memory_order_relaxed);
        while (oldCount > 0) {
            ssize_t newCount = oldCount > max ? oldCount - max : 0;
            if (m_count.compare_exchange_weak(oldCount, newCount, std::memory_order_acquire, std::memory_order_relaxed))
                return oldCount - newCount;
        }
        return 0;
    }
 
    // Acquires at least one, and (greedily) at most max
    ssize_t waitMany(ssize_t max, std::int64_t timeout_usecs)
    {
        assert(max >= 0);
        ssize_t result = tryWaitMany(max);
        if (result == 0 && max > 0)
            result = waitManyWithPartialSpinning(max, timeout_usecs);
        return result;
    }
 
    ssize_t waitMany(ssize_t max)
    {
        ssize_t result = waitMany(max, -1);
        assert(result > 0);
        return result;
    }
 
    void signal(ssize_t count = 1)
    {
        assert(count >= 0);
        ssize_t oldCount = m_count.fetch_add(count, std::memory_order_release);
        ssize_t toRelease = -oldCount < count ? -oldCount : count;
        if (toRelease > 0) {
            m_sema.signal((int)toRelease);
        }
    }
 
    std::size_t availableApprox() const
    {
        ssize_t count = m_count.load(std::memory_order_relaxed);
        return count > 0 ? static_cast<std::size_t>(count) : 0;
    }
};
 
} // namespace utils
} // namespace genesis
 
#endif // include guard