simplified event-queue, and fixed mutex-access-before-constructor-finished bug

Author: CodeByDrescher

VCellMessaging/include/VCELL/MessageEventManager.h

@@ -47,17 +47,18 @@ class MessageEventManager {
 		void processEvent(WorkerEvent* event);
 		void enqueue(WorkerEvent*);
 
-		std::mutex timeClockMutex;
+		std::thread eventQueueProcessingWorkerThread; //TODO: make a `std::jthread` once compilers catch up with standard
+		std::function<void(WorkerEvent*)> sendUpdateFunction;
 
+		// Critical Resources / Regions
+		// -> CR #1 - Whether stop has been requested or not
 		bool stopRequested;
 		std::condition_variable requestedStopForeman;
 		std::mutex stopRequestedMutex;
+		// -> CR #2 - The Event Queue
 		std::queue<WorkerEvent*> eventQueue;
+		std::condition_variable eventQueueForeman;
 		std::mutex queuetex;
-
-		std::thread eventQueueProcessingWorkerThread; //TODO: make a `std::jthread` once compilers catch up with standard
-		std::condition_variable needMessageProcessingForeman;
-		std::function<void(WorkerEvent*)> sendUpdateFunction;
 };
 
 #endif //VCELL_ODE_NUMERICS_MESSAGEEVENTQUEUE_H

VCellMessaging/include/VCELL/SimulationMessaging.h

@@ -33,10 +33,8 @@ static constexpr int Message_DEFAULT_PRIORITY = 0;
 //TODO: Re-write so that there is a messaging handler abstract class with two children: stdOut and CUrl
 class SimulationMessaging {
 	public:
-	    virtual ~SimulationMessaging() noexcept;
-		static void initialize();
-		static SimulationMessaging* create();
 		static SimulationMessaging* getInstVar();
+		static void cleanupInstanceVar();
 		void setWorkerEvent(JobEvent::Status status, const char *eventMessage);
 		void setWorkerEvent(JobEvent::Status status, double progress, double timepoint);
 		void setWorkerEvent(JobEvent::Status status, double progress, double timepoint, const char *eventMessage);
@@ -54,11 +52,12 @@ class SimulationMessaging {
 		}
 		void waitUntilFinished();
 		#ifdef USE_MESSAGING
-		static SimulationMessaging* create(const char* broker, const char* smqusername, const char* passwd, const char* qname, const char* tname,
-				const char* vcusername, int simKey, int jobIndex, int taskID, int ttl_low=DEFAULT_TTL_LOW, int ttl_high=DEFAULT_TTL_HIGH);
-		friend void* startMessagingThread(void* param);
+		void initialize_curl_messaging(bool alsoPrintToStdOut, const char* broker, const char* vcusername, int simKey, int jobIndex, int taskID, int ttl_low=DEFAULT_TTL_LOW, int ttl_high=DEFAULT_TTL_HIGH);
 		#endif
 
+	protected:
+		virtual ~SimulationMessaging() noexcept;
+
 	private:
 		// Statics
 		static void sendStdOutStatus(WorkerEvent*);
@@ -78,9 +77,6 @@ class SimulationMessaging {
 		static bool isInitialized;
 
 		SimulationMessaging();
-		#ifdef USE_MESSAGING
-		SimulationMessaging(const char* broker, const char* vcusername, int simKey, int jobIndex,  int taskID, int ttl_low=DEFAULT_TTL_LOW, int ttl_high=DEFAULT_TTL_HIGH);
-		#endif
 
 		void sendStatus(WorkerEvent*);
 		void keepAlive();

VCellMessaging/src/MessageEventManager.cpp

@@ -10,10 +10,9 @@
 
 MessageEventManager::MessageEventManager(std::function<void(WorkerEvent*)> sendUpdateFunction):
 		stopRequested{false},
-		sendUpdateFunction{std::move(sendUpdateFunction)},
-		eventQueueProcessingWorkerThread([this]() {this->performQueueProcessing();}) // Should auto-start thread
+		sendUpdateFunction{std::move(sendUpdateFunction)}
 {
-
+	this->eventQueueProcessingWorkerThread = std::thread([this]() {this->performQueueProcessing();}); // Should auto-start thread
 }
 
 MessageEventManager::~MessageEventManager() {
@@ -37,14 +36,16 @@ void MessageEventManager::requestStopAndWaitForIt() {
 	std::unique_lock stopRequestedLock{this->stopRequestedMutex};
 	if (this->stopRequested) return;
 	this->stopRequested = true;
-	this->needMessageProcessingForeman.notify_all();
-	this->requestedStopForeman.wait(stopRequestedLock, [this]()->bool{return this->eventQueue.empty();});
+	this->eventQueueForeman.notify_all(); // We want any sleeping workers to wake up, so they check if a stop was requested.
+	this->requestedStopForeman.wait(stopRequestedLock, [this]()->bool {
+		std::unique_lock queueLock{this->queuetex};
+		return this->eventQueue.empty();
+	});
 }
 
 bool MessageEventManager::stopWasCalled() {
 	std::unique_lock stopRequestedLock{this->stopRequestedMutex};
 	return this->stopRequested;
-
 }
 
 ///////////////////////////////////////////////////
@@ -66,35 +67,27 @@ void MessageEventManager::performQueueProcessing() {
 void MessageEventManager::processQueue() {
 	while (true) {
 		WorkerEvent* event;
-		{// START Clock-out Scope //
-			// The order of the locks is important! See Header File
-			std::unique_lock checkIfTheresWorkLock{this->timeClockMutex}; // This is to prevent loop processing while `enqueue` is being called
+		{// START Queuetex Scope //
 			std::unique_lock shouldBeActiveLock(this->queuetex);
 			if (this->eventQueue.empty()) {
 				{ // START stop-requested Scope
 					std::unique_lock stopRequestedLock{this->stopRequestedMutex};
 					if (this->stopRequested) {
-						this->requestedStopForeman.notify_all();
-						return; // We're all done; since this should be done on a jthread, this should also trigger a join.
+						this->requestedStopForeman.notify_all(); // Tell all stop-requesters that we've registered a stop.
+						return; // We're all done
 					}
 				} // END stop-requested Scope
-				checkIfTheresWorkLock.unlock(); // Need to allow enqueuing, can't do that with this locked, and we can't rescope but unique_lock order matters above!
-				// The worker can stop working, and "get some sleep"
-				// Wait for the worker to be "prodded" (via `needMessagingForeman.notify_one()`), AND the event queue is not empty
-				// Note that this `wait()` call, by design, unlocks the queue mutex, until it is "prodded".
-				this->needMessageProcessingForeman.wait(shouldBeActiveLock);
-				std::unique_lock stopRequestedLock{this->stopRequestedMutex};
+				// If the code gets to here, the worker can stop working, and "get some sleep"
+				this->eventQueueForeman.wait(shouldBeActiveLock); // Note that this `wait()` call, by design, unlocks the queue mutex, until it is "prodded".
 				if (this->eventQueue.empty()) continue; // Probably means we need to check if stop was requested again
 			}
-			std::unique_lock stopRequestedLock{this->stopRequestedMutex};
 			event = this->eventQueue.front();
 			this->eventQueue.pop();
-		}// END Clock-out Scope  //
+		}// END Queuetex Scope  //
 
 		// Process Event
 		this->processEvent(event);
-		// Remember to delete the event! We need the memory back!
-		delete event;
+		delete event; // Remember to delete the event! We need the memory back!
 	}
 }
 
@@ -103,19 +96,13 @@ void MessageEventManager::processEvent(WorkerEvent* event) {
 }
 
 void MessageEventManager::enqueue(WorkerEvent* event) {
-	// The order of the locks is important! See Header File
-	std::lock_guard workerIsActiveLock{this->timeClockMutex}; // Need to lock to prevent worker from "clocking out" while we set this up
+	std::lock_guard workerIsActiveLock{this->queuetex}; // Need to lock to prevent worker from checking if it has more work while we make the request.
 	std::lock_guard stopRequestedLock{this->stopRequestedMutex}; // We scope this lock to the whole function; "last in the door" policy
 	if (this->stopRequested) {
 		std::cerr << "A new event was added to the messaging queue, but this queue has had `stop` requested!" << std::endl;
 		delete event;
 		return; // note: on this return function scope ends, and thus so too does out function-scoped locks
 	}
-
-	{// START Emplace Scope //
-		std::lock_guard queueLock{this->queuetex};
-		this->eventQueue.emplace(event);
-	}// END Emplace Scope  //
-
-	this->needMessageProcessingForeman.notify_one(); // Nudges one sleeping worker awake. If the worker is awake...this does nothing; as it should.
+	this->eventQueue.emplace(event);
+	this->eventQueueForeman.notify_one(); // Nudges one sleeping worker awake. If the worker is awake...this does nothing; as it should.
 }

VCellMessaging/src/SimulationMessaging.cpp

@@ -4,50 +4,37 @@
 #include <string>
 #include <format>
 
-
-
-static const double WORKEREVENT_MESSAGE_MIN_TIME_SECONDS = 15.0;
+static constexpr double WORKEREVENT_MESSAGE_MIN_TIME_SECONDS = 15.0;
 bool SimulationMessaging::isInitialized = false;
 
 
-SimulationMessaging *SimulationMessaging::m_inst = NULL;
+SimulationMessaging *SimulationMessaging::m_inst = nullptr;
 
 SimulationMessaging::SimulationMessaging():
 	eventHandler(std::bind(&SimulationMessaging::sendStatus, this, std::placeholders::_1)),
+	workerEventOutputMode{WORKEREVENT_OUTPUT_MODE_STDOUT},
 	taskID{-1},
-	m_jobIndex{-1}
-{
-	this->taskID = -1;
-	this->workerEventOutputMode = WORKEREVENT_OUTPUT_MODE_STDOUT;
-	this->curlHandler = new NullCurlProxy();
-
-}
-
-#ifdef USE_MESSAGING
-SimulationMessaging::SimulationMessaging(const char* broker, const char* vcusername, int simKey, int jobIndex, int taskID, int ttl_low, int ttl_high):
-	eventHandler(std::bind(&SimulationMessaging::sendStatus, this, std::placeholders::_1)),
-	workerEventOutputMode{WORKEREVENT_OUTPUT_MODE_MESSAGING},
-	taskID{taskID},
-	m_jobIndex{jobIndex},
+	m_jobIndex{-1},
 	bNewWorkerEvent{false}
 {
-	this->curlHandler = new CurlProxy(simKey, taskID, jobIndex, vcusername, broker, ttl_low, ttl_high);
+	this->curlHandler = new NullCurlProxy();
 	time(&this->lastSentEventTime);
 }
-#endif
 
 SimulationMessaging::~SimulationMessaging() noexcept{
 	delete this->curlHandler;
 }
 
 SimulationMessaging* SimulationMessaging::getInstVar() {
+	if (nullptr == SimulationMessaging::m_inst) SimulationMessaging::m_inst = new SimulationMessaging();
 	return SimulationMessaging::m_inst;
 }
 
-SimulationMessaging* SimulationMessaging::create(){
-	if (SimulationMessaging::m_inst == NULL) SimulationMessaging::m_inst = new SimulationMessaging();
-
-    return SimulationMessaging::m_inst;
+void SimulationMessaging::cleanupInstanceVar() {
+	if (nullptr == SimulationMessaging::m_inst) return;
+	SimulationMessaging::m_inst->waitUntilFinished();
+	delete SimulationMessaging::m_inst;
+	SimulationMessaging::m_inst = nullptr;
 }
 
 void SimulationMessaging::sendStatus(WorkerEvent* event) {
@@ -132,26 +119,16 @@ void SimulationMessaging::setWorkerEvent(JobEvent::Status status, const double p
 
 void SimulationMessaging::waitUntilFinished() {
 	this->eventHandler.requestStopAndWaitForIt();
-	// if (workerEventOutputMode == WORKEREVENT_OUTPUT_MODE_STDOUT) return;
-	// #ifdef USE_MESSAGING
-	// std::cout << "!!!waiting for thread to exit" << std::endl;
-	// pthread_join(newWorkerEventThread, NULL);
-	// std::cout << "!!Threads joined successfully" << std::endl;
-	// #endif
 }
 
 #ifdef USE_MESSAGING
 
-SimulationMessaging* SimulationMessaging::create(const char* broker, const char* smqusername, const char* passwd, const char* qname, const char* tname, const char* vcusername, int simKey, int jobIndex, int taskID, int ttl_low, int ttl_high){
-	if (m_inst != NULL && m_inst->workerEventOutputMode == WORKEREVENT_OUTPUT_MODE_STDOUT) {
-		delete m_inst;
-		m_inst = NULL;
-	}
-	if (m_inst == NULL){    
-        m_inst = new SimulationMessaging(broker, vcusername, simKey, jobIndex, taskID, ttl_low, ttl_high);
-	}
-
-    return(m_inst);
+void SimulationMessaging::initialize_curl_messaging(bool alsoPrintToStdOut, const char* broker, const char* vcusername, int simKey, int jobIndex, int givenTaskID, int ttl_low, int ttl_high){
+	this->workerEventOutputMode = alsoPrintToStdOut ? WORKEREVENT_OUTPUT_MODE_ALL : WORKEREVENT_OUTPUT_MODE_MESSAGING;
+	this->taskID = givenTaskID;
+	this->m_jobIndex = jobIndex;
+	delete this->curlHandler; // get rid of the null one we make by default
+	this->curlHandler = new CurlProxy(simKey, taskID, jobIndex, vcusername, broker, ttl_low, ttl_high);
 }
 
 // void SimulationMessaging::start() {