Worker thread is now operational

Source/Integration/IntegrationGameModeBase.cpp

@@ -3,7 +3,6 @@
 #include "IntegrationGameModeBase.h"
 #include "drvutil.hpp"
 #include "engineutil.hpp"
-#include "WorkerRunnable.hpp"
 #include <string>
 #include <string_view>
 
@@ -26,19 +25,24 @@ AIntegrationGameModeBase::AIntegrationGameModeBase()
 {
 	Thread = nullptr;
 	ThreadStopRequested = false;
+	EngineSeconds = 0.0;
+	NextThreadTrigger = 1.0;
 	//PrimaryActorTick.bCanEverTick = true; // Probably wrong
 	//PrimaryActorTick.bTickEvenWhenPaused = true; // Probably wrong
 	//PrimaryActorTick.TickGroup = TG_PrePhysics; // Probably wrong
 	SetActorTickEnabled(true);
-	SetActorTickInterval(1.0f);
+	SetActorTickInterval(0.05f);
+	ThreadEvent = FPlatformProcess::GetSynchEventFromPool(false);
 }
 
 AIntegrationGameModeBase::~AIntegrationGameModeBase()
 {
 	WaitForThread();
+	FPlatformProcess::ReturnSynchEventToPool(ThreadEvent);
+	ThreadEvent = nullptr;
 }
 
-void AIntegrationGameModeBase::HandleConsoleOutput()
+void AIntegrationGameModeBase::HandleLuprexConsoleOutput()
 {
 	uint32_t ndata; const char* data;
 	Luprex.get_outgoing(&Luprex, 0, &ndata, &data);
@@ -51,43 +55,35 @@ void AIntegrationGameModeBase::HandleConsoleOutput()
 	ConsoleOutput.Append(fs);
 }
 
-void AIntegrationGameModeBase::LaunchThread()
-{
-	if (Thread == nullptr) {
-		ThreadStopRequested = false;
-		Thread = FRunnableThread::Create(this, TEXT("Worker Thread"));
-	}
-}
-
-// Init routine called by the worker thread.
-bool AIntegrationGameModeBase::Init()
-{
-	engineutil::DPrint("WorkerRunnable::Init");
-	return true;
-}
-
 // Run routine called by the worker thread.
 uint32 AIntegrationGameModeBase::Run()
 {
-	while (!ThreadStopRequested)
+	while (true)
 	{
-		engineutil::DPrint("WorkerRunnable::Run");
+		bool triggered = ThreadEvent->Wait(3000);
-		FPlatformProcess::Sleep(1.0);
+		if (ThreadStopRequested) {
+			engineutil::DPrint("Thread stopping as requested");
+			break;
+		}
+		if (!triggered) {
+			engineutil::DPrint("Thread waiting a long time...");
+			continue;
+		}
+		engineutil::DPrint("Thread triggered.");
+		{
+			FScopeLock lk(&LuprexMutex);
+			Luprex.play_invoke_event_update(&Luprex, EngineSeconds);
+		}
 	}
-	engineutil::DPrint("WorkerRunnable Done");
 	return 0;
 }
 
-void AIntegrationGameModeBase::Stop()
-{
-	ThreadStopRequested = true;
-}
-
 #pragma optimize( "", off )
 void AIntegrationGameModeBase::WaitForThread()
 {
 	if (Thread == nullptr) return;
-	Stop(); // Notifies the thread to clean up and exit.
+	ThreadStopRequested = true;
+	ThreadEvent->Trigger();
 	delete Thread; // This waits for the thread to complete.
 	Thread = nullptr;
 }
@@ -95,29 +91,35 @@ void AIntegrationGameModeBase::WaitForThread()
 void AIntegrationGameModeBase::Tick(float DeltaSeconds)
 {
 	Super::Tick(DeltaSeconds);
-	if (!luprex_initialized()) {
+	{
-		return;
+		FScopeLock lk(&LuprexMutex);
-	}
-	HandleConsoleOutput();
-	if (Luprex.engine) {
 		EngineSeconds += DeltaSeconds;
-		Luprex.play_invoke_event_update(&Luprex, EngineSeconds);
+		HandleLuprexConsoleOutput();
 	}
 	TArray<FString> prints = engineutil::DPrintGetStored();
 	for (const FString& fs : prints) {
 		ConsoleOutput.AppendLine(fs);
 	}
-	if (ConsoleOutput.IsDirty()) {
+	if (ConsoleOutput.IsDirty())
+	{
 		ConsoleSetOutput(ConsoleOutput.Get());
 		ConsoleOutput.ClearDirty();
 	}
+	if ((Thread != nullptr) && (EngineSeconds >= NextThreadTrigger))
+	{
+		ThreadEvent->Trigger();
+		NextThreadTrigger += 1.0;
+	}
 }
 
 void AIntegrationGameModeBase::ConsoleSendInput(const FString& fs)
 {
-	if (Luprex.engine) {
+	if (Luprex.engine != nullptr)
+	{
+		FScopeLock lk(&LuprexMutex);
 		const TCHAR* fstchar = *fs;
-		if (sizeof(TCHAR) == 2) {
+		if (sizeof(TCHAR) == 2)
+		{
 			ConsoleOutput.AppendLine(FString("> ") + fs);
 			std::u16string_view fsview((const char16_t*)fstchar, fs.Len());
 			std::string utf8 = drvutil::utf16_to_utf8(fsview);
@@ -129,39 +131,62 @@ void AIntegrationGameModeBase::ConsoleSendInput(const FString& fs)
 
 void AIntegrationGameModeBase::BeginPlay()
 {
-	engineutil::DPrint("In BeginPlay");
 	Super::BeginPlay();
-	if (!luprex_initialized()) {
+
+	// There should be no thread at this point.
+	checkf(Thread == nullptr, TEXT("There should be no thread here."));
+
+	// Reinitialize simple state.
+	EngineSeconds = 0;
+	NextThreadTrigger = 1.0;
+	ThreadStopRequested = false;
+	ThreadEvent->Wait(0); // Clear the event if set.
+
+	// Try to initialize the wrapper.
+	if (Luprex.play_initialize == nullptr)
+	{
 		engineutil::init_wrapper(&Luprex);
-		if (!luprex_initialized()) {
+		if (Luprex.play_initialize == nullptr)
+		{
 			engineutil::DPrint("Luprex wrapper initialization failed");
-			return;
 		}
 	}
 
-	Luprex.release(&Luprex);
+	// If wrapper is initialized, try to initialize the luprex engine.
-	Luprex.hook_dprint(engineutil::DPrintHook);
+	if (Luprex.play_initialize != nullptr)
-	
+	{
-	drvutil::ostringstream srcpak;
+		Luprex.release(&Luprex);
-	std::string srcpakerr = drvutil::package_lua_source("c:\\Luprex", &srcpak);
+		Luprex.hook_dprint(engineutil::DPrintHook);
-	if (!srcpakerr.empty()) {
+
-		engineutil::DPrint(srcpakerr.c_str());
+		drvutil::ostringstream srcpak;
+		std::string srcpakerr = drvutil::package_lua_source("c:\\Luprex", &srcpak);
+		if (!srcpakerr.empty())
+		{
+			engineutil::DPrint(srcpakerr.c_str());
+		}
+		std::string_view srcpakv = srcpak.view();
+		char* argv[1];
+		argv[0] = const_cast<char*>("lpxserver");
+		Luprex.play_initialize(&Luprex, 1, argv, srcpakv.size(), srcpakv.data(), "");
+		if (Luprex.error[0])
+		{
+			engineutil::DPrint(Luprex.error);
+		}
+		else
+		{
+			engineutil::DPrint("Luprex initialize success");
+		}
 	}
-	std::string_view srcpakv = srcpak.view();
+
-	char* argv[1];
+	// If we successfully created a Luprex engine, create a worker thread.
-	argv[0] = const_cast<char*>("lpxserver");
+	if (Luprex.engine != nullptr)
-	Luprex.play_initialize(&Luprex, 1, argv, srcpakv.size(), srcpakv.data(), "");
+	{
-	if (Luprex.error[0]) {
+		Thread = FRunnableThread::Create(this, TEXT("Worker Thread"));
-		ConsoleOutput.AppendLine(FString(Luprex.error));
-	} else {
-		ConsoleOutput.AppendLine(FString("Initialize Luprex Success"));
 	}
-	EngineSeconds = 0;
-	LaunchThread();
 }
 
-void AIntegrationGameModeBase::EndPlay(const EEndPlayReason::Type EndPlayReason) {
+void AIntegrationGameModeBase::EndPlay(const EEndPlayReason::Type EndPlayReason)
-	engineutil::DPrint("In EndPlay");
+{
 	WaitForThread();
 }
 

Source/Integration/IntegrationGameModeBase.h

@@ -23,19 +23,11 @@ public:
 	virtual void Tick(float) override;
 	virtual void EndPlay(const EEndPlayReason::Type EndPlayReason);
 
-	// Thread start and shutdown.
+	// Thread shutdown and cleanup. Called by Blueprint thread.
-	void LaunchThread();
 	void WaitForThread();
 
-	// Methods of FRunnable, for the thread to use.
+	// Method of FRunnable, called by the Luprex thread.
-	virtual bool Init() override;
 	virtual uint32 Run() override;
-	virtual void Stop() override;
-
-	// Return true if luprex was successfully initialized.
-	inline bool luprex_initialized() {
-		return Luprex.play_initialize != nullptr;
-	}
 
 	// Set the entire contents of the console output box.
 	UFUNCTION(BlueprintImplementableEvent)
@@ -45,14 +37,32 @@ public:
 	UFUNCTION(BlueprintCallable)
 	void ConsoleSendInput(const FString& text);
 
-	void HandleConsoleOutput();
+	// Transfer console output from the Luprex engine to unreal.
+	void HandleLuprexConsoleOutput();
 
+	// This stores the entire text currently visible in the console.
+	engineutil::ConsoleOutput ConsoleOutput;
 
-	// The worker thread is responsible for networking and event_update
+	// The worker thread.
-	FRunnableThread* Thread;
+	FRunnableThread *Thread;
+
+	// Used to tell the worker thread to stop.
 	bool ThreadStopRequested;
 
-	engineutil::ConsoleOutput ConsoleOutput;
+	// This event is used to wake up the Luprex thread.  Normally,
+	// this means we want the worker to do one processing step.  But
+	// if ThreadStopRequested is true, it means we want the thread
+	// to exit.
+	FEvent* ThreadEvent;
+
+	// This critical section guards the use of the EngineWrapper.
+	FCriticalSection LuprexMutex;
+	
+	// The Luprex wrapper and engine.  You must claim the LuprexMutex
+	// before touching any of these variables.
 	static EngineWrapper Luprex;
 	float EngineSeconds;
+
+	// When do we next trigger the thread event (relative to EngineSeconds).
+	float NextThreadTrigger;
 };