Merge pull request #61 from MissouriMRDT/hotfix/data-corruption-bug

Rework Send and Receive of RoveComm_CPP to Improve Speed and Lower Latency

Author: ClayJay3

data/RoveComm

@@ -29,8 +29,10 @@ typedef int ssize_t;
 #undef _WIN32_WINNT
 #define _WIN32_WINNT 0x0600
 #endif
-#include <winsock2.h>
+#define _WINSOCKAPI_
+#include <mswsock.h>
 #include <windows.h>
+#include <winsock2.h>
 #include <ws2tcpip.h>
 #else
 #include <arpa/inet.h>

src/RoveComm/RoveCommManifest.h

@@ -192,9 +192,11 @@ namespace rovecomm
 
         // Pack the data
         RoveCommData stData = PackPacket(stPacket);
-
         // Get size of data not including the data not filled. (the null/zero data in RoveCommData)
         size_t siDataSize = ROVECOMM_PACKET_HEADER_SIZE + (sizeof(T) * stPacket.unDataCount);
+
+        // Acquire a write lock on the socket send mutex to protect the socket, which is shared between threads, but not thread-safe.
+        std::unique_lock<std::mutex> lkSocketSendLock(m_muSocketSendMutex);
         // Send the data
 #if defined(__ROVECOMM_WINDOWS_MODE__) && __ROVECOMM_WINDOWS_MODE__ == 1
         ssize_t siBytesSent = send(nClientSocket, reinterpret_cast<char*>(&stData), siDataSize, 0);

src/RoveComm/RoveCommPacket.h

@@ -56,6 +56,7 @@ namespace rovecomm
             std::atomic_int m_nCurrentTCPClientSocket;
             struct sockaddr_in m_saClientAddr;
             std::shared_mutex m_muCallbackMutex;
+            std::mutex m_muSocketSendMutex;
 
             // Packet processing functions
             template<typename T>

src/RoveComm/RoveCommTCP.cpp

@@ -50,11 +50,18 @@ namespace rovecomm
     class RoveCommUDP : AutonomyThread<void>
     {
         private:
+#if defined(__ROVECOMM_WINDOWS_MODE__) && __ROVECOMM_WINDOWS_MODE__ == 1
+            // Windows specific private member variables.
+            HANDLE m_stdIOCP;
+#endif
+
             // Private member variables
             std::atomic_int m_nUDPSocket;
             struct sockaddr_in m_saUDPServerAddr;
             std::vector<SubscriberInfo> vSubscribers;
             std::shared_mutex m_muCallbackMutex;
+            std::mutex m_muSocketSendMutex;
+            std::mutex m_muSocketReceiveMutex;
 
             // Packet processing functions
             template<typename T>

src/RoveComm/RoveCommTCP.h

@@ -132,6 +132,10 @@ int main()
     // Wait for packets to be processed.
     std::this_thread::sleep_for(std::chrono::milliseconds(500));
 
+    // Close the UDP and TCP sockets
+    pRoveCommUDP_Node.CloseUDPSocket();
+    pRoveCommTCP_Node.CloseTCPSocket();
+
     exit(0);
 }
 

src/RoveComm/RoveCommUDP.cpp

@@ -110,9 +110,14 @@ TEST(RoveCommUDP, SendUDPPacket)
 
                 // Send the packet to the localhost
                 ssize_t siBytesSent = pRoveCommUDP_Node.SendUDPPacket<uint8_t>(stPacket, "127.0.0.1", 11001);
-
-                // Check if the packet successfully sent
-                EXPECT_EQ(siBytesSent, ROVECOMM_PACKET_HEADER_SIZE + (sizeof(uint8_t) * stPacket.unDataCount));
+                // Check if the packet successfully sent.
+                EXPECT_EQ(siBytesSent, 1);
+
+                // Send two packets to the localhost
+                ssize_t siBytesSent1 = pRoveCommUDP_Node.SendUDPPacket<uint8_t>(stPacket, "127.0.0.1", 11001);
+                ssize_t siBytesSent2 = pRoveCommUDP_Node.SendUDPPacket<uint8_t>(stPacket, "127.0.0.1", 11001);
+                // Check if the packets successfully sent.
+                EXPECT_EQ(siBytesSent1 + siBytesSent2, 2);
 
                 // Close the socket
                 pRoveCommUDP_Node.CloseUDPSocket();