In resource-constrained environments, 10TB-scale 8-parallel processing in cloudberry may encounter specific anomalies related to Motion layer UDP communication. Below are four key scenarios and how the code modifications address them.

Four Anomaly Scenarios

1. Capacity Mismatch:
   The receiving end’s buffer becomes full, but the sender is unaware. As a result, the sender’s unacknowledged packet queue continues transmitting, leading to unnecessary retransmissions and packet drops.

2. False Deadlock Detection:
   The peer node processes heartbeat packets but fails to free up buffer capacity. This triggers a false deadlock judgment, incorrectly flagging network anomalies.

3. Unprocessed Packets Require Main Thread Wakeup:
   When the receive queue is full, incoming data packets are discarded. However, the main thread still needs to be awakened to process backlogged packets in the queue, preventing permanent stalling.

4. Execution Time Mismatch Across Nodes:
   Issues like data skew, computational performance gaps, or I/O bottlenecks cause significant differences in execution time between nodes. For example, in a hash join, if the inner table’s  is not ready, the node cannot process data from other nodes, leading to packet timeouts.
   *Example Plan*: Packets from  to  (via ) timeout because the  in  remains unready, blocking packet processing.

Code Modifications and Their Impact

The code changes target the above scenarios by enhancing UDP communication feedback, adjusting deadlock checks, and ensuring proper thread wakeup. Key modifications:

1. Addressing Capacity Mismatch:
   - Added  (256) to flag when the receive buffer is full.
   - When the receive queue is full (), a response with  is sent to the sender (). This notifies the sender to pause or adjust transmission, preventing blind retransmissions.

2. Fixing False Deadlock Detection:
   - Modified  to accept  as a parameter, enabling ACK polling during deadlock checks.
   - Extended the initial timeout for deadlock suspicion from  to 600 seconds, reducing premature network error reports.
   - If no response is received after 600 seconds, the buffer capacity is incrementally increased () to alleviate false bottlenecks, with detailed logging before triggering an error.

3. Ensuring Main Thread Wakeup on Full Queue:
   - In , even when packets are dropped due to a full queue, the main thread is awakened () if the packet matches the waiting query/node/route. This ensures backlogged packets in the queue are processed.

4. Mitigating Node Execution Mismatches:
   - Added logging for retransmissions after  attempts, providing visibility into prolonged packet delays (e.g., due to unready ).
   - Reset  after successful ACK polling, preventing excessive retry counts from triggering false timeouts.

Author: Zhao Xi

contrib/interconnect/udp/ic_udpifc.c

@@ -209,6 +209,7 @@ int
 #define UDPIC_FLAGS_DISORDER    		(32)
 #define UDPIC_FLAGS_DUPLICATE   		(64)
 #define UDPIC_FLAGS_CAPACITY    		(128)
+#define UDPIC_FLAGS_FULL        		(256)
 
 /*
  * ConnHtabBin
@@ -835,7 +836,7 @@ static void initUdpManager(mudp_manager_t mptr);
 static inline void checkNetworkTimeout(ICBuffer *buf, uint64 now, bool *networkTimeoutIsLogged);
 
 static void checkExpiration(ChunkTransportState *transportStates, ChunkTransportStateEntry *pEntry, MotionConn *triggerConn, uint64 now);
-static void checkDeadlock(ChunkTransportStateEntry *pChunkEntry, MotionConn *conn);
+static void checkDeadlock(ChunkTransportState *transportStates, ChunkTransportStateEntry *pChunkEntry, MotionConn *conn);
 
 static bool cacheFuturePacket(icpkthdr *pkt, struct sockaddr_storage *peer, int peer_len);
 static void cleanupStartupCache(void);
@@ -5220,6 +5221,12 @@ handleAcks(ChunkTransportState *transportStates, ChunkTransportStateEntry *pChun
 					shouldSendBuffers |= (handleAckForDisorderPkt(transportStates, &pEntry->entry, &ackConn->mConn, pkt));
 					break;
 				}
+				else if (pkt->flags & UDPIC_FLAGS_FULL)
+				{
+					if (DEBUG1 >= log_min_messages)
+						write_log("Recv buff is full [seq %d] from route %d; srcpid %d dstpid %d cmd %d flags 0x%x connseq %d", pkt->seq, ackConn->route, pkt->srcPid, pkt->dstPid, pkt->icId, pkt->flags, ackConn->conn_info.seq);
+					break;
+				}
 
 				/*
 				 * don't get out of the loop if pkt->seq equals to
@@ -6099,6 +6106,7 @@ checkExpiration(ChunkTransportState *transportStates,
 						if (pollAcks(transportStates, pEntryUdp->txfd, wait_time))
 						{
 							handleAcks(transportStates, pEntry, false);
+							curBuf->nRetry = 0;
 							break;
 						}
 
@@ -6120,6 +6128,12 @@ checkExpiration(ChunkTransportState *transportStates,
 					};
 				}
 
+				if (loop_ack > Gp_interconnect_min_retries_before_timeout / 5)
+					write_log("Resending packet (seq %d) to %s (pid %d cid %d) with %d retries in %lu seconds",
+							curBuf->pkt->seq, curBuf->conn->remoteHostAndPort,
+							curBuf->pkt->dstPid, curBuf->pkt->dstContentId, curBuf->nRetry,
+							(now - curBuf->sentTime) / 1000 / 1000);
+
 				currBuffConn = CONTAINER_OF(curBuf->conn, MotionConnUDP, mConn);
 
 				retransmits++;
@@ -6214,7 +6228,7 @@ checkExpiration(ChunkTransportState *transportStates,
  *
  */
 static void
-checkDeadlock(ChunkTransportStateEntry *pChunkEntry, MotionConn *mConn)
+checkDeadlock(ChunkTransportState *transportStates, ChunkTransportStateEntry *pChunkEntry, MotionConn *mConn)
 {
 	uint64		deadlockCheckTime;
 	ChunkTransportStateEntryUDP *pEntry = NULL;
@@ -6251,17 +6265,31 @@ checkDeadlock(ChunkTransportStateEntry *pChunkEntry, MotionConn *mConn)
 			ic_control_info.lastDeadlockCheckTime = now;
 			ic_statistics.statusQueryMsgNum++;
 
+			if (Gp_interconnect_fc_method == INTERCONNECT_FC_METHOD_LOSS_ADVANCE && pollAcks(transportStates, pEntry->txfd, 50))
+			{
+				handleAcks(transportStates, pChunkEntry, false);
+				conn->deadlockCheckBeginTime = now;
+			}
+
 			/* check network error. */
-			if ((now - conn->deadlockCheckBeginTime) > ((uint64) Gp_interconnect_transmit_timeout * 1000 * 1000))
+			if ((now - conn->deadlockCheckBeginTime) > ((uint64) 600 * 1000 * 1000))
 			{
-				ereport(ERROR,
-						(errcode(ERRCODE_GP_INTERCONNECTION_ERROR),
-						 errmsg("interconnect encountered a network error, please check your network"),
-						 errdetail("Did not get any response from %s (pid %d cid %d) in %d seconds.",
-								   conn->mConn.remoteHostAndPort,
-								   conn->conn_info.dstPid,
-								   conn->conn_info.dstContentId,
-								   Gp_interconnect_transmit_timeout)));
+				write_log("Did not get any response from %s (pid %d cid %d) in 600 seconds.",conn->mConn.remoteHostAndPort,
+									  conn->conn_info.dstPid,
+									  conn->conn_info.dstContentId);
+
+				if (Gp_interconnect_fc_method == INTERCONNECT_FC_METHOD_LOSS_TIMER)
+					conn->capacity += 1;
+
+				if ((now - conn->deadlockCheckBeginTime) > ((uint64) Gp_interconnect_transmit_timeout * 1000 * 1000))
+					ereport(ERROR,
+							(errcode(ERRCODE_GP_INTERCONNECTION_ERROR),
+							errmsg("interconnect encountered a network error, please check your network"),
+							errdetail("Did not get any response from %s (pid %d cid %d) in %d seconds.",
+									conn->mConn.remoteHostAndPort,
+									conn->conn_info.dstPid,
+									conn->conn_info.dstContentId,
+									Gp_interconnect_transmit_timeout)));
 			}
 		}
 	}
@@ -6393,7 +6421,7 @@ checkExceptions(ChunkTransportState *transportStates,
 
 	if ((retry & 0x3) == 2)
 	{
-		checkDeadlock(pEntry, conn);
+		checkDeadlock(transportStates, pEntry, conn);
 		checkRxThreadError();
 		ML_CHECK_FOR_INTERRUPTS(transportStates->teardownActive);
 	}
@@ -6543,6 +6571,9 @@ SendChunkUDPIFC(ChunkTransportState *transportStates,
 		}
 		checkExceptions(transportStates, &pEntry->entry, &conn->mConn, retry++, timeout);
 		doCheckExpiration = false;
+		if (!doCheckExpiration && icBufferListLength(&conn->unackQueue) == 0 && conn->capacity > 0 && icBufferListLength(&conn->sndQ
+ueue) > 0)
+			sendBuffers(transportStates, &pEntry->entry, &conn->mConn);
 	}
 
 	conn->mConn.pBuff = (uint8 *) conn->curBuff->pkt;
@@ -7136,6 +7167,30 @@ handleDataPacket(MotionConn *mConn, icpkthdr *pkt, struct sockaddr_storage *peer
 		logPkt("Interconnect error: received a packet when the queue is full ", pkt);
 		ic_statistics.disorderedPktNum++;
 		conn->stat_count_dropped++;
+
+		if (Gp_interconnect_fc_method == INTERCONNECT_FC_METHOD_LOSS_TIMER && rx_control_info.mainWaitingState.waiting &&
+				rx_control_info.mainWaitingState.waitingNode == pkt->motNodeId &&
+				rx_control_info.mainWaitingState.waitingQuery == pkt->icId)
+		{
+				if (rx_control_info.mainWaitingState.waitingRoute == ANY_ROUTE)
+				{
+					if (rx_control_info.mainWaitingState.reachRoute == ANY_ROUTE)
+						rx_control_info.mainWaitingState.reachRoute = conn->route;
+				}
+				else if (rx_control_info.mainWaitingState.waitingRoute == conn->route)
+				{
+					if (DEBUG2 >= log_min_messages)
+						write_log("rx thread: main_waiting waking it route %d", rx_control_info.mainWaitingState.waitingRoute);
+					rx_control_info.mainWaitingState.reachRoute = conn->route;
+				}
+				/* WAKE MAIN THREAD HERE */
+				*wakeup_mainthread = true;
+		}
+
+		if (Gp_interconnect_fc_method == INTERCONNECT_FC_METHOD_LOSS_ADVANCE)
+		{
+			setAckSendParam(param, &conn->mConn, UDPIC_FLAGS_FULL, conn->conn_info.seq - 1, conn->conn_info.extraSeq);
+		}
 		return false;
 	}