move to std counting semaphore

This commit is contained in:
Valentin Balaschenko
2025-06-25 19:31:27 +01:00
parent 8f2f5310e2
commit 2966efd025
3 changed files with 8 additions and 92 deletions

View File

@@ -104,7 +104,7 @@ Workers::setNumberOfThreads(int numberOfThreads)
++m_pauseCount; ++m_pauseCount;
// Pausing a thread counts as one "internal task" // Pausing a thread counts as one "internal task"
m_semaphore.notify(); m_semaphore.release();
} }
} }
@@ -128,7 +128,7 @@ Workers::stop()
void void
Workers::addTask() Workers::addTask()
{ {
m_semaphore.notify(); m_semaphore.release();
} }
int int
@@ -213,7 +213,7 @@ Workers::Worker::run()
// Acquire a task or "internal task." // Acquire a task or "internal task."
// //
m_workers.m_semaphore.wait(); m_workers.m_semaphore.acquire();
// See if there's a pause request. This // See if there's a pause request. This
// counts as an "internal task." // counts as an "internal task."

View File

@@ -20,13 +20,12 @@
#ifndef RIPPLE_CORE_WORKERS_H_INCLUDED #ifndef RIPPLE_CORE_WORKERS_H_INCLUDED
#define RIPPLE_CORE_WORKERS_H_INCLUDED #define RIPPLE_CORE_WORKERS_H_INCLUDED
#include <xrpld/core/detail/semaphore.h>
#include <xrpl/beast/core/LockFreeStack.h> #include <xrpl/beast/core/LockFreeStack.h>
#include <atomic> #include <atomic>
#include <condition_variable> #include <condition_variable>
#include <mutex> #include <mutex>
#include <semaphore>
#include <string> #include <string>
#include <thread> #include <thread>
@@ -223,10 +222,10 @@ private:
std::condition_variable m_cv; // signaled when all threads paused std::condition_variable m_cv; // signaled when all threads paused
std::mutex m_mut; std::mutex m_mut;
bool m_allPaused; bool m_allPaused;
semaphore m_semaphore; // each pending task is 1 resource std::counting_semaphore<> m_semaphore; // each pending task is 1 resource
int m_numberOfThreads; // how many we want active now int m_numberOfThreads; // how many we want active now
std::atomic<int> m_activeCount; // to know when all are paused std::atomic<int> m_activeCount; // to know when all are paused
std::atomic<int> m_pauseCount; // how many threads need to pause now std::atomic<int> m_pauseCount; // how many threads need to pause now
std::atomic<int> std::atomic<int>
m_runningTaskCount; // how many calls to processTask() active m_runningTaskCount; // how many calls to processTask() active
beast::LockFreeStack<Worker> m_everyone; // holds all created workers beast::LockFreeStack<Worker> m_everyone; // holds all created workers

View File

@@ -1,83 +0,0 @@
//------------------------------------------------------------------------------
/*
This file is part of rippled: https://github.com/ripple/rippled
Copyright (c) 2012, 2013 Ripple Labs Inc.
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL , DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
#ifndef RIPPLE_CORE_SEMAPHORE_H_INCLUDED
#define RIPPLE_CORE_SEMAPHORE_H_INCLUDED
#include <condition_variable>
#include <mutex>
namespace ripple {
template <class Mutex, class CondVar>
class basic_semaphore
{
private:
Mutex m_mutex;
CondVar m_cond;
std::size_t m_count;
public:
using size_type = std::size_t;
/** Create the semaphore, with an optional initial count.
If unspecified, the initial count is zero.
*/
explicit basic_semaphore(size_type count = 0) : m_count(count)
{
}
/** Increment the count and unblock one waiting thread. */
void
notify()
{
std::lock_guard lock{m_mutex};
++m_count;
m_cond.notify_one();
}
/** Block until notify is called. */
void
wait()
{
std::unique_lock lock{m_mutex};
while (m_count == 0)
m_cond.wait(lock);
--m_count;
}
/** Perform a non-blocking wait.
@return `true` If the wait would be satisfied.
*/
bool
try_wait()
{
std::lock_guard lock{m_mutex};
if (m_count == 0)
return false;
--m_count;
return true;
}
};
using semaphore = basic_semaphore<std::mutex, std::condition_variable>;
} // namespace ripple
#endif