Adjustments.

Author: asfernandes

src/common/ipc/IpcMessage.h

@@ -116,6 +116,7 @@ class IpcMessageObjectImpl final : public IpcObject
 	{
 		int32_t ownerPid;
 		int32_t ownerId;
+		std::atomic_uint8_t alive;
 #ifdef IPC_MESSAGE_USE_SHARED_SIGNAL
 		IpcSharedSignal receiverSignal;
 		IpcSharedSignal senderSignal;
@@ -140,6 +141,7 @@ class IpcMessageObjectImpl final : public IpcObject
 		{
 			header->ownerPid = (int) getpid();
 			header->ownerId = ++IPC_MESSAGE_COUNTER;
+			header->alive = 1;
 
 #ifdef IPC_MESSAGE_USE_SHARED_SIGNAL
 			new (&header->receiverSignal) IpcSharedSignal();
@@ -320,6 +322,9 @@ inline void IpcMessageReceiver<Message>::disconnect()
 	{
 		disconnected = true;
 		std::lock_guard mutexLock(mutex);
+
+		const auto header = ipc.sharedMemory.getHeader();
+		header->alive = 0;
 	}
 }
 
@@ -347,6 +352,7 @@ inline std::optional<Message> IpcMessageReceiver<Message>::receive(std::function
 	}
 
 	ipc.receiverSignal->reset();
+	header->receiverFlag.store(0, std::memory_order_release);
 
 	std::optional<Message> messageOpt;
 
@@ -375,10 +381,7 @@ inline std::optional<Message> IpcMessageReceiver<Message>::receive(std::function
 		memcpy(span.data(), header->messageBuffer, span.size());
 	}
 
-	header->receiverFlag.store(0, std::memory_order_release);
-
 	ipc.senderSignal->signal();
-
 	header->senderFlag.store(1, std::memory_order_release);
 
 	return messageOpt;
@@ -430,6 +433,9 @@ inline bool IpcMessageSender<Message>::send(const Message& message, std::functio
 
 	while (!guard.tryLock(IPC_MESSAGE_TIMEOUT))
 	{
+		if (!header->alive.load(std::memory_order_relaxed))
+			disconnected = true;
+
 		if (disconnected)
 			return false;
 
@@ -458,14 +464,16 @@ inline bool IpcMessageSender<Message>::send(const Message& message, std::functio
 		memcpy(header->messageBuffer, span.data(), span.size());
 	}
 
-	header->receiverFlag.store(1, std::memory_order_release);
-
 	ipc.receiverSignal->signal();
+	header->receiverFlag.store(1, std::memory_order_release);
 
 	while (header->senderFlag.load(std::memory_order_acquire) == 0)
 	{
 		if (!ipc.senderSignal->wait(IPC_MESSAGE_TIMEOUT))
 		{
+			if (!header->alive.load(std::memory_order_relaxed))
+				disconnected = true;
+
 			if (disconnected)
 				return false;
 
@@ -475,7 +483,6 @@ inline bool IpcMessageSender<Message>::send(const Message& message, std::functio
 	}
 
 	ipc.senderSignal->reset();
-
 	header->senderFlag.store(0, std::memory_order_release);
 
 	return true;

src/common/ipc/tests/IpcChatTest.cpp

@@ -80,11 +80,11 @@ BOOST_AUTO_TEST_CASE(IpcChatTest)
 	IpcChatServer<RequestMessage, ResponseMessage> server(parameters);
 	IpcChatClient<RequestMessage, ResponseMessage> client(parameters);
 
-	constexpr unsigned numMessages = 4'000;
+	constexpr unsigned NUM_MESSAGES = 4'000;
 	unsigned readCount = 0;
 
 	std::thread consumerThread([&]() {
-		for (readCount = 0; readCount < numMessages; ++readCount)
+		for (readCount = 0; readCount < NUM_MESSAGES; ++readCount)
 		{
 			const auto requestMessageOpt = server.receive();
 
@@ -104,7 +104,7 @@ BOOST_AUTO_TEST_CASE(IpcChatTest)
 		}
 	});
 
-	for (unsigned writeCount = 0; writeCount < numMessages; ++writeCount)
+	for (unsigned writeCount = 0; writeCount < NUM_MESSAGES; ++writeCount)
 	{
 		BOOST_CHECK(client.send(Request{ .clientAddress = client.getAddress(), .n = writeCount }));
 		const auto responseMessageOpt = client.receive();
@@ -116,7 +116,7 @@ BOOST_AUTO_TEST_CASE(IpcChatTest)
 
 	consumerThread.join();
 
-	BOOST_CHECK_EQUAL(readCount, numMessages);
+	BOOST_CHECK_EQUAL(readCount, NUM_MESSAGES);
 }
 
 

src/common/ipc/tests/IpcMessageTest.cpp

@@ -240,7 +240,7 @@ BOOST_AUTO_TEST_CASE(ServerDisconnectMessageTest)
 	std::thread producerThread([&]() {
 		try
 		{
-			while (!server.isDisconnected())
+			while (!client.isDisconnected())
 			{
 				if (client.send(Message{0}))
 					++produced;