Optimize motion timeout transmission (#1365)

* The primary goal is to address various issues currently encountered during concurrent processes,
such as excessive motion retries, congestion, retransmission storms, and network skew.
The code addresses inefficient network retransmission handling in
unreliable network environments. Specifically:

Fixed Timeout Thresholds: Traditional TCP-style Retransmission Timeout
(RTTVAR.RTO) calculations may be too rigid for networks with volatile
latency (e.g., satellite links, wireless networks). This leads to:
• Premature Retransmissions: Unnecessary data resends during temporary
latency spikes, wasting bandwidth.
• Delayed Recovery: Slow reaction to actual packet loss when RTO is
overly conservative.

Lack of Context Awareness: Static RTO ignores real-time network behavior
patterns, reducing throughput and responsiveness.

Solution: Dynamic Timeout Threshold Adjustment
Implements an adaptive timeout mechanism to optimize retransmission:
if (now < (curBuf->sentTime + conn->rttvar.rto)) {
uint32_t diff = (curBuf->sentTime + conn->rttvar.rto) - now;
// ... (statistical tracking and threshold adjustment)
}

Key Components:
• Statistical Tracking:
\- min/max: Tracks observed minimum/maximum residual time (time
left until RTO expiry).
\- retrans_count/no_retrans_count: Counts retransmission vs.
non-retransmission events.

• Weighted Threshold Calculation:
unack_queue_ring.time_difference = (uint32_t)(
unack_queue_ring.max * weight_no_retrans +
unack_queue_ring.min * weight_retrans
);
Weights derived from historical ratios of retransmissions
(weight_retrans) vs. successful deliveries (weight_no_retrans).

How It Solves the Problem:
• Temporary Latency Spike: Uses max (conservative) to avoid false
retransmits, reducing bandwidth waste (vs. traditional mistaken
retransmissions).
• Persistent Packet Loss: Prioritizes min (aggressive) via
weight_retrans, accelerating recovery (vs. slow fixed-RTO reaction).
• Stable Network: Balances weights for equilibrium throughput (vs.
static RTO limitations).

EstimateRTT - Dynamically estimates the Round-Trip Time (RTT) and adjusts Retransmission Timeout (RTO)

This function implements a variant of the Jacobson/Karels algorithm for RTT estimation, adapted for UDP-based
motion control connections. It updates smoothed RTT (srtt), mean deviation (mdev), and RTO values based on
newly measured RTT samples (mrtt). The RTO calculation ensures reliable data transmission over unreliable networks.

Key Components:

* srtt:   Smoothed Round-Trip Time (weighted average of historical RTT samples)
* mdev:   Mean Deviation (measure of RTT variability)
* rttvar: Adaptive RTT variation bound (used to clamp RTO updates)
* rto:    Retransmission Timeout (dynamically adjusted based on srtt + rttvar)

Algorithm Details:

1. For the first RTT sample:
   srtt    = mrtt << 3   (scaled by 8 for fixed-point arithmetic)
   mdev    = mrtt << 1   (scaled by 2)
   rttvar  = max(mdev, rto_min)
2. For subsequent samples:
   Delta   = mrtt - (srtt >> 3)  (difference between new sample and smoothed RTT)
   srtt   += Delta               (update srtt with 1/8 weight of new sample)
   Delta   = abs(Delta) - (mdev >> 2)
   mdev   += Delta               (update mdev with 1/4 weight)
3. rttvar bounds the maximum RTT variation:
   If mdev > mdev_max, update mdev_max and rttvar
   On new ACKs (snd_una > rtt_seq), decay rttvar toward mdev\_max
4. Final RTO calculation:
   rto = (srtt >> 3) + rttvar   (clamped to RTO_MAX)

Network Latency Filtering and RTO Optimization

This logic mitigates RTO distortion caused by non-network delays in database
execution pipelines. Key challenges addressed:

* Operator processing delays (non-I/O wait) inflate observed ACK times
* Spurious latency amplification in lossy networks triggers excessive RTO_MAX waits
* Congestion collapse from synchronized retransmissions

Core Mechanisms:

1. Valid RTT Sampling Filter:
   Condition: 4 * (pkt->recv_time - pkt->send_time) > ackTime && pkt->retry_times != Gp_interconnect_min_retries_before_timeout
   Rationale:

   * Filters packets exceeding 2x expected round-trips (4x one-way)
   * Excludes artificial retries (retry_times=Gp_interconnect_min_retries_before_timeout) to avoid sampling bias
     Action: Update RTT estimation only with valid samples via EstimateRTT()

2. Randomized Backoff:
   Condition: buf->nRetry > 0
   Algorithm:
   rto += (rto >> (4 * buf->nRetry))
   Benefits:

   * Exponential decay: Shifts create geometrically decreasing increments
   * Connection-specific randomization: Prevents global synchronization
   * Dynamic scaling: Adapts to retry depth (nRetry)

3. Timer List Management (NEW_TIMER):
   Operations:
   RemoveFromRTOList(&mudp, bufConn) → Detaches from monitoring
   AddtoRTOList(\&mudp, bufConn)       → Reinserts with updated rto
   Purpose: Maintains real-time ordering of expiration checks

We conducted multiple full-scale TPCDS benchmarks using both a single physical machine with 48 nodes and four physical machines with 96 nodes, testing with MTU values of 1500 and 9000.
In the single-node environment with no network bottlenecks, there were no significant performance differences between using MTU 1500 and 9000. In the 96-node environment, under single-threaded execution, there were no significant performance differences.
However, under multi-threaded execution (4 threads), SQL statements with a high percentage of data movement showed significant performance variations, ranging from 5 to 10 times, especially with MTU 1500.

* Cleaning up the code

---------

Co-authored-by: zhaoxi <oracleloyal@gmail.com>
Co-authored-by: zhaoxi <zhaoxi@hashdata.cn>

Author: oracleloyall

contrib/interconnect/ic_internal.h

@@ -10,6 +10,11 @@
  */
 #ifndef INTER_CONNECT_INTERNAL_H
 #define INTER_CONNECT_INTERNAL_H
+#include <stdint.h>
+#include <stdlib.h>
+#include <sys/time.h>
+#include <sys/queue.h>
+#include <pthread.h>
 
 #include "tcp/ic_tcp.h"
 #include "udp/ic_udpifc.h"
@@ -33,6 +38,27 @@ typedef enum MotionConnState
 	mcsEosSent
 }			MotionConnState;
 
+struct udp_send_vars
+{
+	/* send sequence variables */
+	uint32_t snd_una;		/* send unacknoledged */
+	uint32_t snd_wnd;		/* send window (unscaled) */
+
+	/* retransmission timeout variables */
+	uint8_t nrtx;			/* number of retransmission */
+	uint8_t max_nrtx;		/* max number of retransmission */
+	uint32_t rto;			/* retransmission timeout */
+	uint32_t ts_rto;		/* timestamp for retransmission timeout */
+
+	/* congestion control variables */
+	uint32_t cwnd;				/* congestion window */
+	uint32_t ssthresh;			/* slow start threshold */
+
+	TAILQ_ENTRY(MotionConnUDP) send_link;
+	TAILQ_ENTRY(MotionConnUDP) timer_link;		/* timer link (rto list) */
+
+};
+
 /*
  * Structure used for keeping track of a pt-to-pt connection between two
  * Cdb Entities (either QE or QD).
@@ -153,6 +179,32 @@ typedef struct MotionConnUDP
 	uint64		stat_count_resent;
 	uint64		stat_max_resent;
 	uint64		stat_count_dropped;
+
+	struct {
+			uint32_t ts_rto;
+			uint32_t rto;
+			uint32_t srtt;
+			uint32_t rttvar;
+			uint32_t snd_una;
+			uint16_t nrtx;
+			uint16_t max_nrtx;
+			uint32_t mss;
+			uint32_t cwnd;
+			uint32_t ssthresh;
+			uint32_t fss;
+			uint8_t loss_count;
+			uint32_t mdev;
+			uint32_t mdev_max;
+			uint32_t rtt_seq;		/* sequence number to update rttvar */
+			uint32_t ts_all_rto;
+			bool karn_mode;
+	} rttvar;
+	
+	uint8_t on_timewait_list;
+	int16_t on_rto_idx;
+
+	uint32_t snd_nxt;		/* send next */
+	struct udp_send_vars sndvar;
 }			MotionConnUDP;
 
 typedef struct MotionConnTCP

contrib/interconnect/test/ic_test_env.c

@@ -330,6 +330,7 @@ client_side_global_var_init(MotionIPCLayer * motion_ipc_layer, pid_t *ic_proxy_p
 
 	Gp_interconnect_queue_depth = 800;
 	Gp_interconnect_snd_queue_depth = 600;
+	Gp_interconnect_mem_size = 20;
 	Gp_interconnect_timer_period = 1;
 	Gp_interconnect_timer_checking_period = 2;
 	InitializeLatchSupport();
@@ -374,6 +375,7 @@ server_side_global_var_init(MotionIPCLayer * motion_ipc_layer, pid_t *ic_proxy_p
 
 	Gp_interconnect_queue_depth = 800;
 	Gp_interconnect_snd_queue_depth = 600;
+	Gp_interconnect_mem_size = 20;
 	Gp_interconnect_timer_period = 1;
 	Gp_interconnect_timer_checking_period = 2;
 	InitializeLatchSupport();