Merge branch 'candidate-8.2.x' into candidate-8.4.x

Signed-off-by: Richard Chapman <rchapman@hpccsystems.com>

.devcontainer/devcontainer.json

@@ -15,7 +15,11 @@
 	],
 	// "postCreateCommand": "git config oh-my-zsh.hide-info 1",
 	// Set *default* container specific settings.json values on container create.
-	"settings": {},
+	"settings": {
+		"cmake.configureArgs": [
+			"-DUSE_SHLIBDEPS=ON"
+		]
+	},
 	// Add the IDs of extensions you want installed when the container is created.
 	"extensions": [
 		"ms-vscode.cpptools",

roxie/ccd/ccdqueue.cpp

@@ -2862,10 +2862,11 @@ public:
         if (udpResendLostPackets && udpMaxSlotsPerClient > TRACKER_BITS)
             udpMaxSlotsPerClient = TRACKER_BITS;
         unsigned serverFlowPort = topology->getPropInt("@serverFlowPort", CCD_SERVER_FLOW_PORT);
+        bool sendFlowOnDataPort = topology->getPropBool("@sendFlowOnDataPort", true);
         unsigned dataPort = topology->getPropInt("@dataPort", CCD_DATA_PORT);
         unsigned clientFlowPort = topology->getPropInt("@clientFlowPort", CCD_CLIENT_FLOW_PORT);
         receiveManager.setown(createReceiveManager(serverFlowPort, dataPort, clientFlowPort, udpQueueSize, udpMaxSlotsPerClient, encryptionInTransit));
-        sendManager.setown(createSendManager(serverFlowPort, dataPort, clientFlowPort, udpSendQueueSize, fastLaneQueue ? 3 : 2, bucket, encryptionInTransit));
+        sendManager.setown(createSendManager(sendFlowOnDataPort ? dataPort : serverFlowPort, dataPort, clientFlowPort, udpSendQueueSize, fastLaneQueue ? 3 : 2, bucket, encryptionInTransit));
     }
 
     virtual void abortPendingData(const SocketEndpoint &ep) override

roxie/udplib/udpmsgpk.cpp

@@ -563,9 +563,15 @@ unsigned CMessageCollator::queryResends() const
 
 bool CMessageCollator::attach_databuffer(DataBuffer *dataBuff)
 {
-    activity = true;
     UdpPacketHeader *pktHdr = (UdpPacketHeader*) dataBuff->data;
     totalBytesReceived += pktHdr->length;
+    if (pktHdr->node.isNull())   // Indicates a packet that has been identified as a duplicate to be logged and discarded
+    {
+        noteDuplicate((pktHdr->pktSeq & UDP_PACKET_RESENT) != 0);
+        dataBuff->Release();
+        return true;
+    }
+    activity = true;
     if (memLimitExceeded || roxiemem::memPoolExhausted())
     {
         DBGLOG("UdpCollator: mem limit exceeded");

roxie/udplib/udpsha.hpp

@@ -40,7 +40,7 @@ struct UdpPacketHeader
 {
     unsigned short length;      // total length of packet including the header, data, and meta
     unsigned short metalength;  // length of metadata (comes after header and data)
-    ServerIdentifier  node;        // Node this message came from
+    ServerIdentifier  node;     // Node this message came from
     unsigned       msgSeq;      // sequence number of messages ever sent from given node, used with ruid to tell which packets are from same message
     unsigned       pktSeq;      // sequence number of this packet within the message (top bit signifies final packet)
     sequence_t     sendSeq;     // sequence number of this packet among all those send from this node to this target

roxie/udplib/udptrr.cpp

@@ -54,6 +54,9 @@ static unsigned lastFlowPermitsSent = 0;
 static unsigned lastFlowRequestsReceived = 0;
 static unsigned lastDataPacketsReceived = 0;
 
+// The code that redirects flow messages from data socket to flow socket relies on the assumption tested here
+static_assert(sizeof(UdpRequestToSendMsg) < sizeof(UdpPacketHeader), "Expected UDP rts size to be less than packet header");
+
 class CReceiveManager : implements IReceiveManager, public CInterface
 {
     /*
@@ -573,7 +576,8 @@ class CReceiveManager : implements IReceiveManager, public CInterface
     class receive_data : public Thread 
     {
         CReceiveManager &parent;
-        ISocket *receive_socket;
+        ISocket *receive_socket = nullptr;
+        ISocket *selfFlowSocket = nullptr;
         std::atomic<bool> running = { false };
         Semaphore started;
         
@@ -585,6 +589,7 @@ class CReceiveManager : implements IReceiveManager, public CInterface
             if (check_max_socket_read_buffer(ip_buffer) < 0) 
                 throw MakeStringException(ROXIE_UDP_ERROR, "System socket max read buffer is less than %u", ip_buffer);
             receive_socket = ISocket::udp_create(parent.data_port);
+            selfFlowSocket = ISocket::udp_connect(SocketEndpoint(parent.receive_flow_port, myNode.getIpAddress()));
             receive_socket->set_receive_buffer_size(ip_buffer);
             size32_t actualSize = receive_socket->get_receive_buffer_size();
             DBGLOG("UdpReceiver: rcv_data_socket created port=%d requested sockbuffsize=%d actual sockbuffsize=%d", parent.data_port, ip_buffer, actualSize);
@@ -603,8 +608,11 @@ class CReceiveManager : implements IReceiveManager, public CInterface
             running = false;
             if (receive_socket)
                 receive_socket->close();
+            if (selfFlowSocket)
+                selfFlowSocket->close();
             join();
             ::Release(receive_socket);
+            ::Release(selfFlowSocket);
         }
 
         virtual int run() 
@@ -625,7 +633,13 @@ class CReceiveManager : implements IReceiveManager, public CInterface
                 {
                     unsigned int res;
                     b = bufferManager->allocate();
-                    receive_socket->read(b->data, 1, DATA_PAYLOAD, res, 5);
+                    while (true)
+                    {
+                        receive_socket->read(b->data, 1, DATA_PAYLOAD, res, 5);
+                        if (res!=sizeof(UdpRequestToSendMsg))
+                            break;
+                        selfFlowSocket->write(b->data, res);
+                    }
                     dataPacketsReceived++;
                     UdpPacketHeader &hdr = *(UdpPacketHeader *) b->data;
                     assert(hdr.length == res && hdr.length > sizeof(hdr));
@@ -637,8 +651,7 @@ class CReceiveManager : implements IReceiveManager, public CInterface
                             StringBuffer s;
                             DBGLOG("UdpReceiver: discarding unwanted resent packet %" SEQF "u %x from %s", hdr.sendSeq, hdr.pktSeq, hdr.node.getTraceText(s).str());
                         }
-                        parent.noteDuplicate(b);
-                        ::Release(b);
+                        hdr.node.clear();  // Used to indicate a duplicate that collate thread should discard. We don't discard on this thread as don't want to do anything that requires locks...
                     }
                     else
                     {
@@ -647,8 +660,8 @@ class CReceiveManager : implements IReceiveManager, public CInterface
                             StringBuffer s;
                             DBGLOG("UdpReceiver: %u bytes received packet %" SEQF "u %x from %s", res, hdr.sendSeq, hdr.pktSeq, hdr.node.getTraceText(s).str());
                         }
-                        parent.input_queue->pushOwn(b);
                     }
+                    parent.input_queue->pushOwn(b);
                     b = NULL;
                 }
                 catch (IException *e) 
@@ -790,29 +803,6 @@ class CReceiveManager : implements IReceiveManager, public CInterface
             collatePacket(dataBuff);
         }
     }
-    void noteDuplicate(DataBuffer *dataBuff)
-    {
-        const UdpPacketHeader *pktHdr = (UdpPacketHeader*) dataBuff->data;
-        Linked <CMessageCollator> msgColl;
-        SpinBlock b(collatorsLock);
-        try
-        {
-            msgColl.set(collators[pktHdr->ruid]);
-        }
-        catch (IException *E)
-        {
-            EXCLOG(E);
-            E->Release();
-        }
-        catch (...)
-        {
-            IException *E = MakeStringException(ROXIE_INTERNAL_ERROR, "Unexpected exception caught in CPacketCollator::run");
-            EXCLOG(E);
-            E->Release();
-        }
-        if (msgColl)
-            msgColl->noteDuplicate((pktHdr->pktSeq & UDP_PACKET_RESENT) != 0);
-    }
 
     void collatePacket(DataBuffer *dataBuff)
     {

system/jlib/jmutex.hpp

@@ -31,7 +31,7 @@ extern jlib_decl void spinUntilReady(std::atomic_uint &value);
 
 #ifdef _DEBUG
 //#define SPINLOCK_USE_MUTEX // for testing
-//#define SPINLOCK_RR_CHECK     // checks for realtime threads
+#define SPINLOCK_RR_CHECK     // checks for realtime threads
 #define _ASSERT_LOCK_SUPPORT
 #endif