serve.zone/remoteingress
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(remoteingress-core): add adaptive per-stream flow control based on active stream counts

Browse Source
This commit is contained in:
Jürgen Kunz
2026-03-16 19:37:06 +00:00
parent a808d4c9de
commit 0fdcdf566e
4 changed files with 50 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
rust/crates/remoteingress-core/src/hub.rs
View File

@@ -373,6 +373,9 @@ async fn handle_edge_connection(
);
}
// Per-edge active stream counter for adaptive flow control
let edge_stream_count = Arc::new(AtomicU32::new(0));
// QoS dual-channel tunnel writer: control frames (PING/PONG/WINDOW_UPDATE/CLOSE)
// have priority over data frames (DATA_BACK). This prevents PING starvation under load.
let (ctrl_tx, mut ctrl_rx) = mpsc::channel::<Vec<u8>>(64);
@@ -509,8 +512,10 @@ async fn handle_edge_connection(
}
// Spawn task: connect to SmartProxy, send PROXY header, pipe data
let stream_counter = Arc::clone(&edge_stream_count);
tokio::spawn(async move {
let _permit = permit; // hold semaphore permit until stream completes
stream_counter.fetch_add(1, Ordering::Relaxed);
let result = async {
// A2: Connect to SmartProxy with timeout
@@ -533,6 +538,7 @@ async fn handle_edge_connection(
// After writing to upstream, send WINDOW_UPDATE_BACK to edge
let writer_token = stream_token.clone();
let wub_tx = writer_tx.clone();
let stream_counter_w = Arc::clone(&stream_counter);
let writer_for_edge_data = tokio::spawn(async move {
let mut consumed_since_update: u32 = 0;
loop {
@@ -558,12 +564,18 @@ async fn handle_edge_connection(
break;
}
}
// Track consumption for flow control
// Track consumption for adaptive flow control.
// Increment capped to adaptive window to limit per-stream in-flight data.
consumed_since_update += len;
if consumed_since_update >= WINDOW_UPDATE_THRESHOLD {
let frame = encode_window_update(stream_id, FRAME_WINDOW_UPDATE_BACK, consumed_since_update);
let adaptive_window = remoteingress_protocol::compute_window_for_stream_count(
stream_counter_w.load(Ordering::Relaxed),
);
let threshold = adaptive_window / 2;
if consumed_since_update >= threshold {
let increment = consumed_since_update.min(adaptive_window);
let frame = encode_window_update(stream_id, FRAME_WINDOW_UPDATE_BACK, increment);
if wub_tx.try_send(frame).is_ok() {
consumed_since_update = 0;
consumed_since_update -= increment;
}
// If try_send fails, keep accumulating — retry on next threshold
}
@@ -610,7 +622,11 @@ async fn handle_edge_connection(
log::warn!("Stream {} download: window still 0 after stall timeout, closing", stream_id);
break;
}
let max_read = w.min(buf.len());
// Adaptive: cap read to current per-stream target window
let adaptive_cap = remoteingress_protocol::compute_window_for_stream_count(
stream_counter.load(Ordering::Relaxed),
) as usize;
let max_read = w.min(buf.len()).min(adaptive_cap);
tokio::select! {
read_result = up_read.read(&mut buf[..max_read]) => {
@@ -665,6 +681,7 @@ async fn handle_edge_connection(
stream_id,
});
}
stream_counter.fetch_sub(1, Ordering::Relaxed);
});
}
FRAME_DATA => {