birkeh/o3de
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[redcode/crythread-2nd-pass] remove dependency on cry threads in the remote console runtime

Browse Source 
Signed-off-by: AMZN-ScottR <24445312+AMZN-ScottR@users.noreply.github.com>
This commit is contained in:
AMZN-ScottR
2021-08-06 17:45:54 -07:00
parent 6b2c9cbede
commit 2f5927f1ac
2 changed files with 84 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Code/Tools/RemoteConsole/Core/RemoteConsoleCore.cpp
+52 -36
View File
@@ -93,40 +93,63 @@ bool RCON_IsRemoteAllowedToConnect(const AZ::AzSock::AzSocketAddress& connectee)
/////////////////////////////////////////////////////////////////////////////////////////////
/////////////////////////////////////////////////////////////////////////////////////////////
/////////////////////////////////////////////////////////////////////////////////////////////
void SRemoteThreadedObject::Start(const char* name)
{
AZStd::thread_desc desc;
desc.m_name = name;
auto function = AZStd::bind(&SRemoteThreadedObject::ThreadFunction, this);
m_thread = AZStd::thread(function, &desc);
}
void SRemoteThreadedObject::WaitForThread()
{
if (m_thread.joinable())
{
m_thread.join();
}
}
void SRemoteThreadedObject::ThreadFunction()
{
Run();
Terminate();
}
/////////////////////////////////////////////////////////////////////////////////////////////
/////////////////////////////////////////////////////////////////////////////////////////////
/////////////////////////////////////////////////////////////////////////////////////////////
void SRemoteServer::StartServer()
{
StopServer();
m_bAcceptClients = true;
Start(0, kServerThreadName);
Start(kServerThreadName);
}
/////////////////////////////////////////////////////////////////////////////////////////////
void SRemoteServer::StopServer()
{
Stop();
m_bAcceptClients = false;
AZ::AzSock::CloseSocket(m_socket);
m_socket = SOCKET_ERROR;
m_lock.Lock();
for (TClients::iterator it = m_clients.begin(); it != m_clients.end(); ++it)
{
it->pClient->StopClient();
AZStd::lock_guard<AZStd::recursive_mutex> lock(m_mutex);
for (TClients::iterator it = m_clients.begin(); it != m_clients.end(); ++it)
{
it->pClient->StopClient();
}
}
m_lock.Unlock();
m_stopEvent.Wait();
m_stopEvent.Set();
}
AZStd::unique_lock<AZStd::recursive_mutex> lock(m_mutex);
m_stopCondition.wait(lock, [this] { return m_clients.empty(); });}
/////////////////////////////////////////////////////////////////////////////////////////////
void SRemoteServer::ClientDone(SRemoteClient* pClient)
{
m_lock.Lock();
AZStd::lock_guard<AZStd::recursive_mutex> lock(m_mutex);
for (TClients::iterator it = m_clients.begin(); it != m_clients.end(); ++it)
{
if (it->pClient == pClient)
{
it->pClient->Stop();
delete it->pClient;
delete it->pEvents;
m_clients.erase(it);
@@ -136,9 +159,8 @@ void SRemoteServer::ClientDone(SRemoteClient* pClient)
if (m_clients.empty())
{
m_stopEvent.Set();
m_stopCondition.notify_all();
}
m_lock.Unlock();
}
/////////////////////////////////////////////////////////////////////////////////////////////
@@ -149,7 +171,6 @@ void SRemoteServer::Terminate()
/////////////////////////////////////////////////////////////////////////////////////////////
void SRemoteServer::Run()
{
SetName(kServerThreadName);
AZ_TRAIT_REMOTECONSOLE_SET_THREAD_AFFINITY
AZSOCKET sClient;
@@ -232,12 +253,10 @@ void SRemoteServer::Run()
continue;
}
m_lock.Lock();
m_stopEvent.Reset();
AZStd::lock_guard<AZStd::recursive_mutex> lock(m_mutex);
SRemoteClient* pClient = new SRemoteClient(this);
m_clients.push_back(SRemoteClientInfo(pClient));
pClient->StartClient(sClient);
m_lock.Unlock();
}
AZ::AzSock::CloseSocket(m_socket);
CryLog("Remote console terminating.\n");
@@ -247,43 +266,42 @@ void SRemoteServer::Run()
/////////////////////////////////////////////////////////////////////////////////////////////
void SRemoteServer::AddEvent(IRemoteEvent* pEvent)
{
m_lock.Lock();
AZStd::lock_guard<AZStd::recursive_mutex> lock(m_mutex);
for (TClients::iterator it = m_clients.begin(); it != m_clients.end(); ++it)
{
it->pEvents->push_back(pEvent->Clone());
}
m_lock.Unlock();
delete pEvent;
}
/////////////////////////////////////////////////////////////////////////////////////////////
void SRemoteServer::GetEvents(TEventBuffer& buffer)
{
m_lock.Lock();
AZStd::lock_guard<AZStd::recursive_mutex> lock(m_mutex);
buffer = m_eventBuffer;
m_eventBuffer.clear();
m_lock.Unlock();
}
/////////////////////////////////////////////////////////////////////////////////////////////
bool SRemoteServer::WriteBuffer(SRemoteClient* pClient, char* buffer, int& size)
{
m_lock.Lock();
IRemoteEvent* pEvent = nullptr;
for (TClients::iterator it = m_clients.begin(); it != m_clients.end(); ++it)
{
if (it->pClient == pClient)
AZStd::lock_guard<AZStd::recursive_mutex> lock(m_mutex);
for (TClients::iterator it = m_clients.begin(); it != m_clients.end(); ++it)
{
TEventBuffer* pEvents = it->pEvents;
if (!pEvents->empty())
if (it->pClient == pClient)
{
pEvent = pEvents->front();
pEvents->pop_front();
TEventBuffer* pEvents = it->pEvents;
if (!pEvents->empty())
{
pEvent = pEvents->front();
pEvents->pop_front();
}
break;
}
break;
}
}
m_lock.Unlock();
const bool res = (pEvent != nullptr);
if (pEvent)
{
@@ -297,7 +315,7 @@ bool SRemoteServer::WriteBuffer(SRemoteClient* pClient, char* buffer, int& size
bool SRemoteServer::ReadBuffer(const char* buffer, int data)
{
bool result = true;
// Sometimes multiple events can come in a single buffer, so make sure we look
// at the entire thing.
int bytesRemaining = data;
@@ -306,15 +324,14 @@ bool SRemoteServer::ReadBuffer(const char* buffer, int data)
{
// Create the event from the current sub string in the buffer.
IRemoteEvent* event = SRemoteEventFactory::GetInst()->CreateEventFromBuffer(curBuffer, bytesRemaining);
result &= (event != nullptr);
if (event)
{
if (event->GetType() != eCET_Noop)
{
m_lock.Lock();
AZStd::lock_guard<AZStd::recursive_mutex> lock(m_mutex);
m_eventBuffer.push_back(event);
m_lock.Unlock();
}
else
{
@@ -337,7 +354,7 @@ bool SRemoteServer::ReadBuffer(const char* buffer, int data)
void SRemoteClient::StartClient(AZSOCKET socket)
{
m_socket = socket;
Start(0, kClientThreadName);
Start(kClientThreadName);
}
/////////////////////////////////////////////////////////////////////////////////////////////
@@ -356,7 +373,6 @@ void SRemoteClient::Terminate()
/////////////////////////////////////////////////////////////////////////////////////////////
void SRemoteClient::Run()
{
SetName(kClientThreadName);
AZ_TRAIT_REMOTECONSOLE_SET_THREAD_AFFINITY
char szBuff[kDefaultBufferSize];
Code/Tools/RemoteConsole/Core/RemoteConsoleCore.h
+32 -10
View File
@@ -11,9 +11,11 @@
#include <AzCore/std/containers/map.h>
#include <AzCore/std/containers/list.h>
#include <AzCore/std/containers/vector.h>
#include <AzCore/std/parallel/condition_variable.h>
#include <AzCore/std/parallel/mutex.h>
#include <AzCore/std/parallel/thread.h>
#include <AzCore/std/string/string.h>
#include <CryThread.h>
extern const int defaultRemoteConsolePort;
@@ -146,6 +148,30 @@ private:
typedef AZStd::list<IRemoteEvent*> TEventBuffer;
/////////////////////////////////////////////////////////////////////////////////////////////
// SRemoteThreadedObject
//
// Simple runnable-like threaded object
//
/////////////////////////////////////////////////////////////////////////////////////////////
struct SRemoteThreadedObject
{
virtual ~SRemoteThreadedObject() = default;
void Start(const char* name);
void WaitForThread();
virtual void Run() = 0;
virtual void Terminate() = 0;
private:
void ThreadFunction();
AZStd::thread m_thread;
};
/////////////////////////////////////////////////////////////////////////////////////////////
// SRemoteServer
//
@@ -154,10 +180,10 @@ typedef AZStd::list<IRemoteEvent*> TEventBuffer;
/////////////////////////////////////////////////////////////////////////////////////////////
struct SRemoteClient;
struct SRemoteServer
: public CrySimpleThread<>
: public SRemoteThreadedObject
{
SRemoteServer()
: m_socket(AZ_SOCKET_INVALID) { m_stopEvent.Set(); }
: m_socket(AZ_SOCKET_INVALID) {}
void StartServer();
void StopServer();
@@ -165,10 +191,8 @@ struct SRemoteServer
void AddEvent(IRemoteEvent* pEvent);
void GetEvents(TEventBuffer& buffer);
// CrySimpleThread
void Terminate() override;
void Run() override;
// ~CrySimpleThread
private:
bool WriteBuffer(SRemoteClient* pClient, char* buffer, int& size);
@@ -189,9 +213,9 @@ private:
typedef AZStd::vector<SRemoteClientInfo> TClients;
TClients m_clients;
AZSOCKET m_socket;
CryMutex m_lock;
AZStd::recursive_mutex m_mutex;
TEventBuffer m_eventBuffer;
CryEvent m_stopEvent;
AZStd::condition_variable_any m_stopCondition;
volatile bool m_bAcceptClients;
friend struct SRemoteClient;
};
@@ -204,7 +228,7 @@ private:
//
/////////////////////////////////////////////////////////////////////////////////////////////
struct SRemoteClient
: public CrySimpleThread<>
: public SRemoteThreadedObject
{
SRemoteClient(SRemoteServer* pServer)
: m_pServer(pServer)
@@ -213,10 +237,8 @@ struct SRemoteClient
void StartClient(AZSOCKET socket);
void StopClient();
// CrySimpleThread
void Terminate() override;
void Run() override;
// ~CrySimpleThread
private:
bool RecvPackage(char* buffer, int& size);