Added a ThreadDispatch Policy to the EBus code (#4405)
* Added a ThreadDispatch Policy to the EBus code The ThreadDispatch Policy can be configured by authors of an EBusTraits to invoke a callback function after an EBus has finished it's dispatching mechanism on a specific thread. It takes into account recursive calls as well and will only invoke the PostDispatch callback after all callstack entries for the current thread are cleared. Signed-off-by: lumberyard-employee-dm <56135373+lumberyard-employee-dm@users.noreply.github.com> * Clang build fix The Traits type is dependent on the template parameter, therefore the compiler needs to be told that the ThreadDispatchPolicy is a type and not a value. Signed-off-by: lumberyard-employee-dm <56135373+lumberyard-employee-dm@users.noreply.github.com> * Fixed DispatchLockGuard cxall in the TerrainWorldRendererComponent.cpp Signed-off-by: lumberyard-employee-dm <56135373+lumberyard-employee-dm@users.noreply.github.com> * Added EBusTrait for configuring the DispatchLockGuard Removed the ThreadPolicy trait, now that the DispatchLockGuard for the EBus Context can be configured. Used the DispatchLockGuard template along with the IsInDispatchThisThread function to determine when an EBus has finished dispatching on thread and released it's Context Mutex. Signed-off-by: lumberyard-employee-dm <56135373+lumberyard-employee-dm@users.noreply.github.com> * Tweaked comment format for the IsInDispatch function Signed-off-by: lumberyard-employee-dm <56135373+lumberyard-employee-dm@users.noreply.github.com> * Removed explicit GetContext call from ThreadDispatchTestBus. Signed-off-by: lumberyard-employee-dm <56135373+lumberyard-employee-dm@users.noreply.github.com> * Clang EBus Test fix for DispatchLockGuard trait Due to the clang compiler evalating constants within templates at the time of declaration, the LocklessDispatch value supplied to the template was always false resulting in the LocklessDispatch feature always locking. Signed-off-by: lumberyard-employee-dm <56135373+lumberyard-employee-dm@users.noreply.github.com>
This commit is contained in:
lumberyard-employee-dm
committed by
GitHub
GitHub
parent
73f8537030
commit
b7c2401056
@@ -160,8 +160,8 @@ namespace AZ
|
||||
/**
|
||||
* Locking primitive that is used when executing events in the event queue.
|
||||
*/
|
||||
using EventQueueMutexType = typename AZStd::Utils::if_c<AZStd::is_same<typename Traits::EventQueueMutexType, NullMutex>::value, // if EventQueueMutexType==NullMutex use MutexType otherwise EventQueueMutexType
|
||||
MutexType, typename Traits::EventQueueMutexType>::type;
|
||||
using EventQueueMutexType = AZStd::conditional_t<AZStd::is_same<typename Traits::EventQueueMutexType, NullMutex>::value, // if EventQueueMutexType==NullMutex use MutexType otherwise EventQueueMutexType
|
||||
MutexType, typename Traits::EventQueueMutexType>;
|
||||
|
||||
/**
|
||||
* Pointer to an address on the bus.
|
||||
@@ -180,14 +180,22 @@ namespace AZ
|
||||
* `<BusName>::ExecuteQueuedEvents()`.
|
||||
* By default, the event queue is disabled.
|
||||
*/
|
||||
static const bool EnableEventQueue = Traits::EnableEventQueue;
|
||||
static const bool EventQueueingActiveByDefault = Traits::EventQueueingActiveByDefault;
|
||||
static const bool EnableQueuedReferences = Traits::EnableQueuedReferences;
|
||||
static constexpr bool EnableEventQueue = Traits::EnableEventQueue;
|
||||
static constexpr bool EventQueueingActiveByDefault = Traits::EventQueueingActiveByDefault;
|
||||
static constexpr bool EnableQueuedReferences = Traits::EnableQueuedReferences;
|
||||
|
||||
/**
|
||||
* True if the EBus supports more than one address. Otherwise, false.
|
||||
*/
|
||||
static const bool HasId = Traits::AddressPolicy != EBusAddressPolicy::Single;
|
||||
static constexpr bool HasId = Traits::AddressPolicy != EBusAddressPolicy::Single;
|
||||
|
||||
/**
|
||||
* Template Lock Guard class that wraps around the Mutex
|
||||
* The EBus uses for Dispatching Events.
|
||||
* This is not the EBus Context Mutex if LocklessDispatch is true
|
||||
*/
|
||||
template <typename DispatchMutex>
|
||||
using DispatchLockGuard = typename Traits::template DispatchLockGuard<DispatchMutex, Traits::LocklessDispatch>;
|
||||
};
|
||||
|
||||
/**
|
||||
@@ -460,7 +468,7 @@ namespace AZ
|
||||
using BusPtr = typename Traits::BusPtr;
|
||||
|
||||
/**
|
||||
* Helper to queue an event by BusIdType only when function queueing is enabled
|
||||
* Helper to queue an event by BusIdType only when function queueing is enabled
|
||||
* @param id Address ID. Handlers that are connected to this ID will receive the event.
|
||||
* @param func Function pointer of the event to dispatch.
|
||||
* @param args Function arguments that are passed to each handler.
|
||||
@@ -581,7 +589,7 @@ namespace AZ
|
||||
, public EBusBroadcaster<Bus, Traits>
|
||||
, public EBusEventer<Bus, Traits>
|
||||
, public EBusEventEnumerator<Bus, Traits>
|
||||
, public AZStd::Utils::if_c<Traits::EnableEventQueue, EBusEventQueue<Bus, Traits>, EBusNullQueue>::type
|
||||
, public AZStd::conditional_t<Traits::EnableEventQueue, EBusEventQueue<Bus, Traits>, EBusNullQueue>
|
||||
{
|
||||
};
|
||||
|
||||
@@ -599,7 +607,7 @@ namespace AZ
|
||||
: public EventDispatcher<Bus, Traits>
|
||||
, public EBusBroadcaster<Bus, Traits>
|
||||
, public EBusBroadcastEnumerator<Bus, Traits>
|
||||
, public AZStd::Utils::if_c<Traits::EnableEventQueue, EBusBroadcastQueue<Bus, Traits>, EBusNullQueue>::type
|
||||
, public AZStd::conditional_t<Traits::EnableEventQueue, EBusBroadcastQueue<Bus, Traits>, EBusNullQueue>
|
||||
{
|
||||
};
|
||||
|
||||
|
||||
@@ -236,6 +236,17 @@ namespace AZ
|
||||
* code before or after an event.
|
||||
*/
|
||||
using EventProcessingPolicy = EBusEventProcessingPolicy;
|
||||
|
||||
/**
|
||||
* Template Lock Guard class that wraps around the Mutex
|
||||
* The EBus Context uses the LockGuard when dispatching
|
||||
* (either AZStd::scoped_lock<MutexType> or NullLockGuard<MutexType>)
|
||||
* The IsLocklessDispatch bool is there to defer evaluation of the LocklessDispatch constant
|
||||
* Otherwise the value above in EBusTraits.h is always used and not the value
|
||||
* that the derived trait class sets.
|
||||
*/
|
||||
template <typename DispatchMutex, bool IsLocklessDispatch>
|
||||
using DispatchLockGuard = AZStd::conditional_t<IsLocklessDispatch, AZ::Internal::NullLockGuard<DispatchMutex>, AZStd::scoped_lock<DispatchMutex>>;
|
||||
};
|
||||
|
||||
namespace Internal
|
||||
@@ -496,6 +507,14 @@ namespace AZ
|
||||
*/
|
||||
static const bool HasId = Traits::AddressPolicy != EBusAddressPolicy::Single;
|
||||
|
||||
/**
|
||||
* Template Lock Guard class that wraps around the Mutex
|
||||
* The EBus uses for Dispatching Events.
|
||||
* This is not EBus Context Mutex when LocklessDispatch is set
|
||||
*/
|
||||
template <typename DispatchMutex>
|
||||
using DispatchLockGuard = typename ImplTraits::template DispatchLockGuard<DispatchMutex>;
|
||||
|
||||
//////////////////////////////////////////////////////////////////////////
|
||||
// Check to help identify common mistakes
|
||||
/// @cond EXCLUDE_DOCS
|
||||
@@ -620,11 +639,11 @@ namespace AZ
|
||||
using ContextMutexType = AZStd::conditional_t<BusTraits::LocklessDispatch && AZStd::is_same_v<MutexType, AZ::NullMutex>, AZStd::shared_mutex, MutexType>;
|
||||
|
||||
/**
|
||||
* The scoped lock guard to use (either AZStd::scoped_lock<MutexType> or NullLockGuard<MutexType>
|
||||
* The scoped lock guard to use
|
||||
* during broadcast/event dispatch.
|
||||
* @see EBusTraits::LocklessDispatch
|
||||
*/
|
||||
using DispatchLockGuard = AZStd::conditional_t<BusTraits::LocklessDispatch, AZ::Internal::NullLockGuard<ContextMutexType>, AZStd::scoped_lock<ContextMutexType>>;
|
||||
using DispatchLockGuard = DispatchLockGuard<ContextMutexType>;
|
||||
|
||||
/**
|
||||
* The scoped lock guard to use during connection. Some specialized policies execute handler methods which
|
||||
@@ -704,6 +723,11 @@ namespace AZ
|
||||
static Context& GetOrCreateContext(bool trackCallstack=true);
|
||||
|
||||
static bool IsInDispatch(Context* context = GetContext(false));
|
||||
|
||||
/**
|
||||
* Returns whether the EBus context is in the middle of a dispatch on the current thread
|
||||
*/
|
||||
static bool IsInDispatchThisThread(Context* context = GetContext(false));
|
||||
/// @cond EXCLUDE_DOCS
|
||||
struct RouterCallstackEntry
|
||||
: public CallstackEntry
|
||||
@@ -1208,6 +1232,13 @@ AZ_POP_DISABLE_WARNING
|
||||
return context != nullptr && context->m_dispatches > 0;
|
||||
}
|
||||
|
||||
template<class Interface, class Traits>
|
||||
bool EBus<Interface, Traits>::IsInDispatchThisThread(Context* context)
|
||||
{
|
||||
return context != nullptr && context->s_callstack != nullptr
|
||||
&& context->s_callstack->m_prev != nullptr;
|
||||
}
|
||||
|
||||
//=========================================================================
|
||||
template<class Interface, class Traits>
|
||||
EBus<Interface, Traits>::RouterCallstackEntry::RouterCallstackEntry(Iterator it, const BusIdType* busId, bool isQueued, bool isReverse)
|
||||
|
||||
@@ -2088,7 +2088,7 @@ namespace UnitTest
|
||||
DisconnectNextHandlerByIdImpl multiHandler2;
|
||||
multiHandler2.BusConnect(DisconnectNextHandlerByIdImpl::firstBusAddress);
|
||||
multiHandler2.BusConnect(DisconnectNextHandlerByIdImpl::secondBusAddress);
|
||||
|
||||
|
||||
// Set the first handler m_nextHandler field to point to the second handler
|
||||
multiHandler1.m_nextHandler = &multiHandler2;
|
||||
|
||||
@@ -2807,7 +2807,7 @@ namespace UnitTest
|
||||
AZStd::this_thread::sleep_for(AZStd::chrono::milliseconds(m_val % m_maxSleep));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
void DoConnect() override
|
||||
{
|
||||
MyEventGroupBus::Handler::BusConnect(m_id);
|
||||
@@ -2854,7 +2854,7 @@ namespace UnitTest
|
||||
}
|
||||
|
||||
MyEventGroupBus::Event(id, &MyEventGroupBus::Events::Calculate, i, i * 2, i << 4);
|
||||
|
||||
|
||||
LocklessConnectorBus::Event(id, &LocklessConnectorBus::Events::DoDisconnect);
|
||||
|
||||
bool failed = (AZStd::find_if(&sentinel[0], end, [](char s) { return s != 0; }) != end);
|
||||
@@ -2891,7 +2891,7 @@ namespace UnitTest
|
||||
{
|
||||
MyEventGroupImpl()
|
||||
{
|
||||
|
||||
|
||||
}
|
||||
|
||||
~MyEventGroupImpl() override
|
||||
@@ -3614,7 +3614,7 @@ namespace UnitTest
|
||||
{
|
||||
AZStd::this_thread::yield();
|
||||
}
|
||||
|
||||
|
||||
EXPECT_GE(AZStd::chrono::system_clock::now(), endTime);
|
||||
};
|
||||
AZStd::thread connectThread([&connectHandler, &waitHandler]()
|
||||
@@ -3813,7 +3813,7 @@ namespace UnitTest
|
||||
struct LastHandlerDisconnectHandler
|
||||
: public LastHandlerDisconnectBus::Handler
|
||||
{
|
||||
void OnEvent() override
|
||||
void OnEvent() override
|
||||
{
|
||||
++m_numOnEvents;
|
||||
BusDisconnect();
|
||||
@@ -3854,7 +3854,7 @@ namespace UnitTest
|
||||
struct DisconnectAssertHandler
|
||||
: public DisconnectAssertBus::Handler
|
||||
{
|
||||
|
||||
|
||||
};
|
||||
|
||||
TEST_F(EBus, HandlerDestroyedWithoutDisconnect_Asserts)
|
||||
@@ -3995,6 +3995,191 @@ namespace UnitTest
|
||||
|
||||
idTestRequest.Disconnect();
|
||||
}
|
||||
|
||||
// IsInDispatchThisThread
|
||||
struct IsInThreadDispatchRequests
|
||||
: AZ::EBusTraits
|
||||
{
|
||||
using MutexType = AZStd::recursive_mutex;
|
||||
};
|
||||
|
||||
using IsInThreadDispatchBus = AZ::EBus<IsInThreadDispatchRequests>;
|
||||
|
||||
class IsInThreadDispatchHandler
|
||||
: public IsInThreadDispatchBus::Handler
|
||||
{};
|
||||
|
||||
TEST_F(EBus, InvokingIsInThisThread_ReturnsSuccess_OnlyIfThreadIsInDispatch)
|
||||
{
|
||||
IsInThreadDispatchHandler handler;
|
||||
handler.BusConnect();
|
||||
|
||||
auto ThreadDispatcher = [](IsInThreadDispatchRequests*)
|
||||
{
|
||||
EXPECT_TRUE(IsInThreadDispatchBus::IsInDispatchThisThread());
|
||||
auto PerThreadBusDispatch = []()
|
||||
{
|
||||
EXPECT_FALSE(IsInThreadDispatchBus::IsInDispatchThisThread());
|
||||
};
|
||||
AZStd::array threads{ AZStd::thread(PerThreadBusDispatch), AZStd::thread(PerThreadBusDispatch) };
|
||||
for (AZStd::thread& thread : threads)
|
||||
{
|
||||
thread.join();
|
||||
}
|
||||
};
|
||||
|
||||
static constexpr size_t ThreadDispatcherIterations = 4;
|
||||
for (size_t iteration = 0; iteration < ThreadDispatcherIterations; ++iteration)
|
||||
{
|
||||
EXPECT_FALSE(IsInThreadDispatchBus::IsInDispatchThisThread());
|
||||
IsInThreadDispatchBus::Broadcast(ThreadDispatcher);
|
||||
EXPECT_FALSE(IsInThreadDispatchBus::IsInDispatchThisThread());
|
||||
}
|
||||
}
|
||||
|
||||
// Thread Dispatch Policy
|
||||
struct ThreadDispatchTestBusTraits
|
||||
: AZ::EBusTraits
|
||||
{
|
||||
using MutexType = AZStd::recursive_mutex;
|
||||
|
||||
struct PostThreadDispatchTestInvoker
|
||||
{
|
||||
~PostThreadDispatchTestInvoker();
|
||||
};
|
||||
|
||||
template <typename DispatchMutex>
|
||||
struct ThreadDispatchTestLockGuard
|
||||
{
|
||||
ThreadDispatchTestLockGuard(DispatchMutex& contextMutex)
|
||||
: m_lock{ contextMutex }
|
||||
{}
|
||||
ThreadDispatchTestLockGuard(DispatchMutex& contextMutex, AZStd::adopt_lock_t adopt_lock)
|
||||
: m_lock{ contextMutex, adopt_lock }
|
||||
{}
|
||||
ThreadDispatchTestLockGuard(const ThreadDispatchTestLockGuard&) = delete;
|
||||
ThreadDispatchTestLockGuard& operator=(const ThreadDispatchTestLockGuard&) = delete;
|
||||
private:
|
||||
PostThreadDispatchTestInvoker m_threadPolicyInvoker;
|
||||
using LockType = AZStd::conditional_t<LocklessDispatch, AZ::Internal::NullLockGuard<DispatchMutex>, AZStd::scoped_lock<DispatchMutex>>;
|
||||
LockType m_lock;
|
||||
};
|
||||
|
||||
template <typename DispatchMutex, bool IsLocklessDispatch>
|
||||
using DispatchLockGuard = ThreadDispatchTestLockGuard<DispatchMutex>;
|
||||
|
||||
static inline AZStd::atomic<int32_t> s_threadPostDispatchCalls;
|
||||
};
|
||||
|
||||
class ThreadDispatchTestRequests
|
||||
{
|
||||
public:
|
||||
virtual void FirstCall() = 0;
|
||||
virtual void SecondCall() = 0;
|
||||
virtual void ThirdCall() = 0;
|
||||
};
|
||||
|
||||
using ThreadDispatchTestBus = AZ::EBus<ThreadDispatchTestRequests, ThreadDispatchTestBusTraits>;
|
||||
|
||||
ThreadDispatchTestBusTraits::PostThreadDispatchTestInvoker::~PostThreadDispatchTestInvoker()
|
||||
{
|
||||
if (!ThreadDispatchTestBus::IsInDispatchThisThread())
|
||||
{
|
||||
++s_threadPostDispatchCalls;
|
||||
}
|
||||
}
|
||||
|
||||
class ThreadDispatchTestHandler
|
||||
: public ThreadDispatchTestBus::Handler
|
||||
{
|
||||
public:
|
||||
void Connect()
|
||||
{
|
||||
ThreadDispatchTestBus::Handler::BusConnect();
|
||||
}
|
||||
void Disconnect()
|
||||
{
|
||||
ThreadDispatchTestBus::Handler::BusDisconnect();
|
||||
}
|
||||
|
||||
void FirstCall() override
|
||||
{
|
||||
ThreadDispatchTestBus::Broadcast(&ThreadDispatchTestBus::Events::SecondCall);
|
||||
}
|
||||
void SecondCall() override
|
||||
{
|
||||
ThreadDispatchTestBus::Broadcast(&ThreadDispatchTestBus::Events::ThirdCall);
|
||||
}
|
||||
void ThirdCall() override
|
||||
{
|
||||
}
|
||||
};
|
||||
|
||||
template <typename ParamType>
|
||||
class EBusParamFixture
|
||||
: public ScopedAllocatorSetupFixture
|
||||
, public ::testing::WithParamInterface<ParamType>
|
||||
{};
|
||||
|
||||
struct ThreadDispatchParams
|
||||
{
|
||||
size_t m_threadCount{};
|
||||
size_t m_handlerCount{};
|
||||
};
|
||||
|
||||
using ThreadDispatchParamFixture = EBusParamFixture<ThreadDispatchParams>;
|
||||
|
||||
INSTANTIATE_TEST_CASE_P(
|
||||
ThreadDispatch,
|
||||
ThreadDispatchParamFixture,
|
||||
::testing::Values(
|
||||
ThreadDispatchParams{ 1, 1 },
|
||||
ThreadDispatchParams{ 2, 1 },
|
||||
ThreadDispatchParams{ 1, 2 },
|
||||
ThreadDispatchParams{ 2, 2 },
|
||||
ThreadDispatchParams{ 16, 8 }
|
||||
)
|
||||
);
|
||||
|
||||
TEST_P(ThreadDispatchParamFixture, CustomDispatchLockGuard_InvokesPostDispatchFunction_AfterThreadHasFinishedDispatch)
|
||||
{
|
||||
ThreadDispatchTestBusTraits::s_threadPostDispatchCalls = 0;
|
||||
ThreadDispatchParams threadDispatchParams = GetParam();
|
||||
AZStd::vector<AZStd::thread> testThreads;
|
||||
AZStd::vector<ThreadDispatchTestHandler> testHandlers(threadDispatchParams.m_handlerCount);
|
||||
for (ThreadDispatchTestHandler& testHandler : testHandlers)
|
||||
{
|
||||
testHandler.Connect();
|
||||
}
|
||||
|
||||
static constexpr size_t DispatchThreadCalls = 3;
|
||||
const size_t totalThreadDispatchCalls = threadDispatchParams.m_threadCount * DispatchThreadCalls;
|
||||
|
||||
auto DispatchThreadWorker = []()
|
||||
{
|
||||
ThreadDispatchTestBus::Broadcast(&ThreadDispatchTestBus::Events::FirstCall);
|
||||
ThreadDispatchTestBus::Broadcast(&ThreadDispatchTestBus::Events::SecondCall);
|
||||
ThreadDispatchTestBus::Broadcast(&ThreadDispatchTestBus::Events::ThirdCall);
|
||||
};
|
||||
|
||||
for (size_t threadIndex = 0; threadIndex < threadDispatchParams.m_threadCount; ++threadIndex)
|
||||
{
|
||||
testThreads.emplace_back(DispatchThreadWorker);
|
||||
}
|
||||
|
||||
for (AZStd::thread& thread : testThreads)
|
||||
{
|
||||
thread.join();
|
||||
}
|
||||
|
||||
for (ThreadDispatchTestHandler& testHandler : testHandlers)
|
||||
{
|
||||
testHandler.Disconnect();
|
||||
}
|
||||
|
||||
EXPECT_EQ(totalThreadDispatchCalls, ThreadDispatchTestBusTraits::s_threadPostDispatchCalls);
|
||||
ThreadDispatchTestBusTraits::s_threadPostDispatchCalls = 0;
|
||||
}
|
||||
} // namespace UnitTest
|
||||
|
||||
#if defined(HAVE_BENCHMARK)
|
||||
@@ -4370,7 +4555,7 @@ namespace Benchmark
|
||||
Bus::ExecuteQueuedEvents();
|
||||
}
|
||||
s_benchmarkEBusEnv<Bus>.Disconnect(state);
|
||||
|
||||
|
||||
}
|
||||
BUS_BENCHMARK_REGISTER_ALL(BM_EBus_ExecuteBroadcast);
|
||||
|
||||
|
||||
Reference in New Issue
Block a user