Debug concurrency problem exposed on macOS, use cond_wait instead of mutex

Co-authored-by: Caleb Herpin caleb.herpin@gmail.com

include/trick/MyCivetServer.hh

@@ -42,6 +42,8 @@ class MyCivetServer {
         pthread_t server_thread;                                                /* ** */
         bool sessionDataMarshalled;                                             /* ** */
         pthread_mutex_t lock_loop;                                              /* ** */
+	pthread_cond_t cond_loop;
+	bool service_connections = true;
         bool shutting_down;                                                     /* ** */
 
         std::map<std::string, WebSocketSessionMaker> WebSocketSessionMakerMap;  /* ** */
@@ -80,4 +82,4 @@ struct Data {
 };
 
 #endif
-#endif
+#endif

share/trick/pymods/trick/tests/civet_server/test_ws.py

@@ -34,23 +34,20 @@ class TestWebserverWs:
 	
 	@pytest.mark.asyncio
 	async def test_time(self, time_path):
-		if "macos" in platform.platform().lower():
+		if params.get_test_time():
-			logging.warning("Time endpoint is currently not working on MacOS.  Skipping this test.")
+			async with websockets.connect(time_path, ssl=TestWebserverWs.ssl_context) as websocket:
+				await websocket.send("LOCAL")
+				count = 0
+				while count < 2:
+					message = await websocket.recv()
+					test_format = "Time: %H:%M Date: %m/%d/%Y\n" #Not checking seconds.
+					time =  datetime.datetime.strftime(datetime.datetime.strptime(message, "Time: %H:%M:%S Date: %m/%d/%Y\n"), test_format)
+					test_time = datetime.datetime.now().strftime(test_format)
+					print("Checking:", time, "=", test_time)
+					assert time == test_time
+					count += 1
 		else:
-			if params.get_test_time():
+			raise RuntimeError("Parameter test_time is disabled.")
-				async with websockets.connect(time_path, ssl=TestWebserverWs.ssl_context) as websocket:
-					await websocket.send("LOCAL")
-					count = 0
-					while count < 2:
-						message = await websocket.recv()
-						test_format = "Time: %H:%M Date: %m/%d/%Y\n" #Not checking seconds.
-						time =  datetime.datetime.strftime(datetime.datetime.strptime(message, "Time: %H:%M:%S Date: %m/%d/%Y\n"), test_format)
-						test_time = datetime.datetime.now().strftime(test_format)
-						print("Checking:", time, "=", test_time)
-						assert time == test_time
-						count += 1
-			else:
-				raise RuntimeError("Parameter test_time is disabled.")
 
 	@pytest.mark.asyncio
 	async def test_variable_server_vars(self, variable_server_path):

trick_source/web/CivetServer/src/MyCivetServer.cpp

@@ -28,12 +28,14 @@ PURPOSE: (Represent the state and initial conditions for my server)
 
 void MyCivetServer::deleteWebSocketSession(struct mg_connection * nc) {
     std::map<mg_connection*, WebSocketSession*>::iterator iter;
+    pthread_mutex_lock(&WebSocketSessionMapLock);
     iter = webSocketSessionMap.find(nc);
     if (iter != webSocketSessionMap.end()) {
         WebSocketSession* session = iter->second;
         delete session;
         webSocketSessionMap.erase(iter);
     }
+    pthread_mutex_unlock(&WebSocketSessionMapLock);
 }
 
 static const char * style_css =
@@ -205,8 +207,13 @@ void* main_loop(void* S) {
 	MyCivetServer* server = (MyCivetServer*) S;
 	bool messageSent;
 
+
 	while(1) {
-		pthread_mutex_lock(&server->lock_loop);
+        pthread_mutex_lock(&server->lock_loop);
+        while (!server->service_connections)
+            pthread_cond_wait(&server->cond_loop, &server->lock_loop);
+		//pthread_mutex_lock(&server->lock_loop);
+        //pthread_cond_wait(&server->cond_loop, &server->lock_loop);
         if (server->shutting_down) {
             return NULL;
         }
@@ -219,13 +226,17 @@ void* main_loop(void* S) {
 		pthread_mutex_lock(&server->WebSocketSessionMapLock);
 		for (iter = server->webSocketSessionMap.begin(); iter != server->webSocketSessionMap.end(); iter++ ) {
 			WebSocketSession* session = iter->second;
+            message_publish(MSG_DEBUG, "Sending message...\n"); 
 			session->sendMessage();
+            message_publish(MSG_DEBUG, "Message sent.\n"); 
 			messageSent = true;
 		}
 		if (messageSent) { //If any  message was sent we say the data is now not marshalled.
 			server->sessionDataMarshalled = false;
 		}
 		pthread_mutex_unlock(&server->WebSocketSessionMapLock);
+        server->service_connections = false;
+        pthread_mutex_unlock(&server->lock_loop);
 	}
 }
 
@@ -315,13 +326,18 @@ int MyCivetServer::http_top_of_frame() {
         if (time_homogeneous) {
 		    marshallWebSocketSessionData();
         }
+        message_publish(MSG_DEBUG, "Top of frame.\n");
 		unlockConnections();
 	}
     return 0;
 }
 
 void MyCivetServer::unlockConnections() {
-	pthread_mutex_unlock(&lock_loop);
+	//pthread_mutex_unlock(&lock_loop);
+    pthread_mutex_lock(&lock_loop);
+    service_connections = true;
+    pthread_cond_signal(&cond_loop);
+    pthread_mutex_unlock(&lock_loop);
     // std::map<mg_connection*, WebSocketSession*>::iterator iter;
     // pthread_mutex_lock(&WebSocketSessionMapLock);
     // for (iter = webSocketSessionMap.begin(); iter != webSocketSessionMap.end(); iter++ ) {

trick_source/web/CivetServer/src/http_GET_handlers.cpp

@@ -92,6 +92,7 @@ int parent_http_handler(struct mg_connection* conn, void *data) {
             return http_send_error(conn, 405, msg.c_str(), msg.size(), 100);
         }
     }
+    // TODO add return value
 }
 
 void handle_HTTP_GET_vs_connections(struct mg_connection* conn, void *cbdata) {