mirror of
https://github.com/XRPLF/rippled.git
synced 2025-11-21 19:45:53 +00:00
Add beast_basics module
This commit is contained in:
166
modules/beast_basics/threads/beast_CallQueue.cpp
Normal file
166
modules/beast_basics/threads/beast_CallQueue.cpp
Normal file
@@ -0,0 +1,166 @@
|
||||
/*============================================================================*/
|
||||
/*
|
||||
VFLib: https://github.com/vinniefalco/VFLib
|
||||
|
||||
Copyright (C) 2008 by Vinnie Falco <vinnie.falco@gmail.com>
|
||||
|
||||
This library contains portions of other open source products covered by
|
||||
separate licenses. Please see the corresponding source files for specific
|
||||
terms.
|
||||
|
||||
VFLib is provided under the terms of The MIT License (MIT):
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
of this software and associated documentation files (the "Software"), to deal
|
||||
in the Software without restriction, including without limitation the rights
|
||||
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
copies of the Software, and to permit persons to whom the Software is
|
||||
furnished to do so, subject to the following conditions:
|
||||
|
||||
The above copyright notice and this permission notice shall be included in
|
||||
all copies or substantial portions of the Software.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
|
||||
FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
|
||||
IN THE SOFTWARE.
|
||||
*/
|
||||
/*============================================================================*/
|
||||
|
||||
CallQueue::CallQueue (String name)
|
||||
: m_name (name)
|
||||
{
|
||||
}
|
||||
|
||||
CallQueue::~CallQueue ()
|
||||
{
|
||||
// Someone forget to close the queue.
|
||||
bassert (m_closed.isSignaled ());
|
||||
|
||||
// Can't destroy queue with unprocessed calls.
|
||||
bassert (m_queue.empty ());
|
||||
}
|
||||
|
||||
bool CallQueue::isAssociatedWithCurrentThread () const
|
||||
{
|
||||
return Thread::getCurrentThreadId () == m_id;
|
||||
}
|
||||
|
||||
// Adds a call to the queue of execution.
|
||||
void CallQueue::queuep (Work* c)
|
||||
{
|
||||
// If this goes off it means calls are being made after the
|
||||
// queue is closed, and probably there is no one around to
|
||||
// process it.
|
||||
bassert (!m_closed.isSignaled ());
|
||||
|
||||
if (m_queue.push_back (c))
|
||||
signal ();
|
||||
}
|
||||
|
||||
// Append the Work to the queue. If this call is made from the same
|
||||
// thread as the last thread that called synchronize(), then the call
|
||||
// will execute synchronously.
|
||||
//
|
||||
void CallQueue::callp (Work* c)
|
||||
{
|
||||
queuep (c);
|
||||
|
||||
// If we are called on the process thread and we are not
|
||||
// recursed into doSynchronize, then process the queue. This
|
||||
// makes calls from the process thread synchronous.
|
||||
//
|
||||
// NOTE: The value of isBeingSynchronized is invalid/volatile unless
|
||||
// this thread is the last process thread.
|
||||
//
|
||||
// NOTE: There is a small window of opportunity where we
|
||||
// might get an undesired synchronization if new thread
|
||||
// calls synchronize() concurrently.
|
||||
//
|
||||
if (isAssociatedWithCurrentThread () &&
|
||||
m_isBeingSynchronized.trySignal ())
|
||||
{
|
||||
doSynchronize ();
|
||||
|
||||
m_isBeingSynchronized.reset ();
|
||||
}
|
||||
}
|
||||
|
||||
bool CallQueue::synchronize ()
|
||||
{
|
||||
bool did_something;
|
||||
|
||||
// Detect recursion into doSynchronize(), and
|
||||
// break ties for concurrent calls atomically.
|
||||
//
|
||||
if (m_isBeingSynchronized.trySignal ())
|
||||
{
|
||||
// Remember this thread.
|
||||
m_id = Thread::getCurrentThreadId ();
|
||||
|
||||
did_something = doSynchronize ();
|
||||
|
||||
m_isBeingSynchronized.reset ();
|
||||
}
|
||||
else
|
||||
{
|
||||
did_something = false;
|
||||
}
|
||||
|
||||
return did_something;
|
||||
}
|
||||
|
||||
// Can still have pending calls, just can't put new ones in.
|
||||
void CallQueue::close ()
|
||||
{
|
||||
m_closed.signal ();
|
||||
|
||||
synchronize ();
|
||||
}
|
||||
|
||||
// Process everything in the queue. The list of pending calls is
|
||||
// acquired atomically. New calls may enter the queue while we are
|
||||
// processing.
|
||||
//
|
||||
// Returns true if any functors were called.
|
||||
//
|
||||
bool CallQueue::doSynchronize ()
|
||||
{
|
||||
bool did_something;
|
||||
|
||||
// Reset since we are emptying the queue. Since we loop
|
||||
// until the queue is empty, it is possible for us to exit
|
||||
// this function with an empty queue and signaled state.
|
||||
//
|
||||
reset ();
|
||||
|
||||
Work* call = m_queue.pop_front ();
|
||||
|
||||
if (call)
|
||||
{
|
||||
did_something = true;
|
||||
|
||||
// This method of processing one at a time has the desired
|
||||
// side effect of synchronizing nested calls to us from a functor.
|
||||
//
|
||||
for (;;)
|
||||
{
|
||||
call->operator () ();
|
||||
delete call;
|
||||
|
||||
call = m_queue.pop_front ();
|
||||
|
||||
if (call == 0)
|
||||
break;
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
did_something = false;
|
||||
}
|
||||
|
||||
return did_something;
|
||||
}
|
||||
Reference in New Issue
Block a user