Keepalive (#272)

* add udx_stream_set_keepalive() / udx_stream_clear_keepalive, fixed a bug where a TTL meant for a specific packet could be applied to the wrong packet

* remove udx_stream_clear_keepalive. use udx_stream_set_keepalive(stream, 0) instead

Author: jthomas43

examples/udxperf.c

@@ -278,6 +278,11 @@ print_interval (udxperf_client_t *client, uint64_t bytes, uint64_t start, uint64
   if (is_client && extra_wanted) {
     printf(" cwnd=%d ssthresh=%d pacing_rate=%u fast_recovery_count=%d rto_count=%d rtx_count=%d", stream->cwnd, stream->pacing_bytes_per_ms, stream->ssthresh, stream->fast_recovery_count, stream->rto_count, stream->retransmit_count);
   }
+
+  uint64_t bw;
+  udx_stream_get_bw(stream, &bw);
+
+  printf(" max_bw=%ld", bw);
   printf("\n");
 }
 

include/udx.h

@@ -51,11 +51,12 @@ extern "C" {
 #define UDX_BBR_STATE_PROBE_BW  2
 #define UDX_BBR_STATE_PROBE_RTT 3
 
-#define UDX_HEADER_DATA    0b00001
-#define UDX_HEADER_END     0b00010
-#define UDX_HEADER_SACK    0b00100
-#define UDX_HEADER_MESSAGE 0b01000
-#define UDX_HEADER_DESTROY 0b10000
+#define UDX_HEADER_DATA      0b000001
+#define UDX_HEADER_END       0b000010
+#define UDX_HEADER_SACK      0b000100
+#define UDX_HEADER_MESSAGE   0b001000
+#define UDX_HEADER_DESTROY   0b010000
+#define UDX_HEADER_HEARTBEAT 0b100000
 
 #define UDX_DEBUG_FORCE_RELAY_SLOW_PATH 0x01
 #define UDX_DEBUG_FORCE_DROP_PROBES     0x02
@@ -294,6 +295,7 @@ struct udx_stream_s {
   uint32_t srtt;
   uint32_t rttvar;
   uint32_t rto;
+  uint32_t keepalive_timeout_ms;
 
   win_filter_t rtt_min;
 
@@ -361,7 +363,7 @@ struct udx_stream_s {
   int nrefs;
   uv_timer_t rto_timer;
   uv_timer_t rack_reo_timer;
-  uv_timer_t tlp_timer;
+  uv_timer_t tlp_and_keepalive_timer;
   uv_timer_t zwp_timer;
   uv_timer_t refill_pacing_timer;
 
@@ -601,6 +603,12 @@ udx_stream_get_seq (udx_stream_t *stream, uint32_t *seq);
 int
 udx_stream_set_seq (udx_stream_t *stream, uint32_t seq);
 
+int
+udx_stream_set_keepalive (udx_stream_t *stream, uint32_t keepalive_timeout_ms);
+
+int
+udx_stream_clear_keepalive (udx_stream_t *stream);
+
 int
 udx_stream_get_ack (udx_stream_t *stream, uint32_t *ack);
 

src/udx.c

@@ -425,13 +425,13 @@ close_stream (udx_stream_t *stream, int err) {
 
   uv_timer_stop(&stream->rto_timer);
   uv_timer_stop(&stream->rack_reo_timer);
-  uv_timer_stop(&stream->tlp_timer);
+  uv_timer_stop(&stream->tlp_and_keepalive_timer);
   uv_timer_stop(&stream->zwp_timer);
   uv_timer_stop(&stream->refill_pacing_timer);
 
   uv_close((uv_handle_t *) &stream->rto_timer, finalize_maybe);
   uv_close((uv_handle_t *) &stream->rack_reo_timer, finalize_maybe);
-  uv_close((uv_handle_t *) &stream->tlp_timer, finalize_maybe);
+  uv_close((uv_handle_t *) &stream->tlp_and_keepalive_timer, finalize_maybe);
   uv_close((uv_handle_t *) &stream->zwp_timer, finalize_maybe);
   uv_close((uv_handle_t *) &stream->refill_pacing_timer, finalize_maybe);
   uv_close((uv_handle_t *) &stream->pending_packet_prepare, finalize_maybe);
@@ -446,37 +446,8 @@ close_stream (udx_stream_t *stream, int err) {
 static void
 udx_rto_timeout (uv_timer_t *handle);
 
-static void
-udx_tlp_timeout (uv_timer_t *handle);
-
-static void
-schedule_loss_probe (udx_stream_t *stream);
-
-// rack recovery implemented using https://datatracker.ietf.org/doc/rfc8985/
-
-uint32_t
-udx_rtt_min (udx_stream_t *stream) {
-  return win_filter_get(&stream->rtt_min);
-}
-
-static inline uint32_t
-rack_update_reo_wnd (udx_stream_t *stream) {
-  // rack 6.2.4
-  // TODO: add the DSACK logic also (skipped for now as we didnt impl and only recommended...)
-
-  if (!stream->reordering_seen) {
-    if (stream->ca_state == UDX_CA_RECOVERY || stream->ca_state == UDX_CA_LOSS) return 0;
-    if (stream->sacks >= 3) return 0;
-  }
-
-  uint32_t r = udx_rtt_min(stream) / 4;
-  return r < stream->srtt ? r : stream->srtt;
-}
-
-// if the next packet in the queue requires a specific TTL
-// then we set it here
 static bool
-maybe_adjust_ttl (udx_socket_t *socket) {
+_maybe_adjust_ttl (udx_socket_t *socket) {
 
   if (socket->specific_ttl_send_queue.len == 0) {
     return false;
@@ -491,23 +462,97 @@ maybe_adjust_ttl (udx_socket_t *socket) {
   return false;
 }
 
-// called every sent packet, check to see if the next packet needs the TTL specified
+// every packet sent via uv_udp_send() must call this function as part of their callback
 static void
-on_slow_send (uv_udp_t *udp) {
+maybe_adjust_ttl (uv_udp_t *udp) {
   udx_socket_t *socket = (udx_socket_t *) udp; // todo: use offsetof instead?
   socket->packets_sent_via_uv_send_queue++;
 
-  maybe_adjust_ttl(socket);
+  _maybe_adjust_ttl(socket);
 }
 
-static void
-on_ack_send_slow (uv_udp_send_t *req, int status) {
-  on_slow_send(req->handle);
+// used to free simple (ack, probe, and relay) memory
+// stream-write, stream-send and stream-destroy packets have their own callbacks
+void
+on_packet_send_slow (uv_udp_send_t *req, int status) {
+  maybe_adjust_ttl(req->handle);
 
   UDX_UNUSED(status);
   free(req);
 }
 
+static void
+send_probe (udx_stream_t *stream) {
+  if (!stream->socket) {
+    return;
+  }
+
+  // todo: send a data packet with seq=remote_acked-1 instead
+
+  uint8_t header[20];
+  udx_write_header(header, stream, UDX_HEADER_HEARTBEAT);
+
+  // fast path
+  uv_buf_t buf = uv_buf_init((char *) header, sizeof(header));
+  int err = uv_udp_try_send(&stream->socket->uv_udp, &buf, 1, (struct sockaddr *) &stream->remote_addr);
+
+  if (err == UV_EAGAIN) {
+    // slow path
+    uv_udp_send_t *req = malloc(sizeof(uv_udp_send_t) + buf.len);
+    char *data = (char *) (req + 1);
+    memcpy(data, buf.base, buf.len);
+    buf.base = data;
+    req->data = stream;
+    int err = uv_udp_send(req, &stream->socket->uv_udp, &buf, 1, (struct sockaddr *) &stream->remote_addr, on_packet_send_slow);
+    if (err) {
+      debug_printf("uv_udp_send error: %s\n", uv_strerror(err));
+    }
+  }
+
+  // consider the probe to be sent, even on slow path.
+  stream->packets_tx++;
+  stream->bytes_tx += buf.len;
+
+  stream->socket->packets_tx++;
+  stream->socket->bytes_tx += buf.len;
+  stream->udx->packets_tx++;
+  stream->udx->bytes_tx += buf.len;
+}
+
+static void
+udx_keepalive_timeout (uv_timer_t *timer) {
+  udx_stream_t *stream = timer->data;
+  assert(stream->seq == stream->remote_acked);
+
+  send_probe(stream);
+
+  uv_timer_start(&stream->tlp_and_keepalive_timer, udx_keepalive_timeout, stream->keepalive_timeout_ms, 0);
+}
+
+static void
+schedule_loss_probe (udx_stream_t *stream);
+
+// rack recovery implemented using https://datatracker.ietf.org/doc/rfc8985/
+
+uint32_t
+udx_rtt_min (udx_stream_t *stream) {
+  return win_filter_get(&stream->rtt_min);
+}
+
+static inline uint32_t
+rack_update_reo_wnd (udx_stream_t *stream) {
+  // rack 6.2.4
+  // TODO: add the DSACK logic also (skipped for now as we didnt impl and only recommended...)
+
+  if (!stream->reordering_seen) {
+    if (stream->ca_state == UDX_CA_RECOVERY || stream->ca_state == UDX_CA_LOSS) return 0;
+    if (stream->sacks >= 3) return 0;
+  }
+
+  uint32_t r = udx_rtt_min(stream) / 4;
+  return r < stream->srtt ? r : stream->srtt;
+}
+
 static void
 send_ack (udx_stream_t *stream) {
   if (!stream->socket) {
@@ -574,7 +619,7 @@ send_ack (udx_stream_t *stream) {
     buf.base = data;
     req->data = stream;
 
-    int err = uv_udp_send(req, &stream->socket->uv_udp, &buf, 1, (struct sockaddr *) &stream->remote_addr, on_ack_send_slow);
+    int err = uv_udp_send(req, &stream->socket->uv_udp, &buf, 1, (struct sockaddr *) &stream->remote_addr, on_packet_send_slow);
     if (err) {
       debug_printf("uv_udp_send: err=%s\n", uv_strerror(err));
     }
@@ -624,6 +669,7 @@ on_stream_data_write (uv_udp_send_t *send, int status) {
     debug_printf("sendmsg: %s\n", uv_strerror(status));
   }
 
+  maybe_adjust_ttl(send->handle); // send is freed with packet
   deref_packet(pkt);
 }
 
@@ -951,7 +997,7 @@ schedule_loss_probe (udx_stream_t *stream) {
     pto = uv_timer_get_due_in(&stream->rto_timer);
   }
 
-  uv_timer_start(&stream->tlp_timer, udx_tlp_timeout, pto, 0);
+  uv_timer_start(&stream->tlp_and_keepalive_timer, udx_tlp_timeout, pto, 0);
 }
 
 static uint32_t
@@ -1310,12 +1356,6 @@ process_data_packet (udx_stream_t *stream, int type, uint32_t seq, char *data, s
   udx__cirbuf_set(&(stream->incoming), (udx_cirbuf_val_t *) pkt);
 }
 
-static void
-on_relay_send_slow (uv_udp_send_t *req, int status) {
-  UDX_UNUSED(status); // todo: log or assert?
-  free(req);
-}
-
 static int
 relay_packet (udx_stream_t *stream, char *buf, ssize_t buf_len, int type, uint32_t seq) {
 
@@ -1347,7 +1387,7 @@ relay_packet (udx_stream_t *stream, char *buf, ssize_t buf_len, int type, uint32
       memcpy(data, buf, buf_len);
       b = uv_buf_init(data, b.len);
 
-      err = uv_udp_send(req, &stream->socket->uv_udp, &b, 1, (struct sockaddr *) &relay->remote_addr, on_relay_send_slow);
+      err = uv_udp_send(req, &stream->socket->uv_udp, &b, 1, (struct sockaddr *) &relay->remote_addr, on_packet_send_slow);
     }
   }
 
@@ -1432,6 +1472,13 @@ process_packet (udx_socket_t *socket, char *buf, ssize_t buf_len, struct sockadd
   uint32_t prior_remote_acked = stream->remote_acked;
   bool ack_advanced = seq_diff(ack, prior_remote_acked) > 0;
   bool data_inflight = stream->remote_acked != stream->seq;
+  // todo: send data packet with seq=remote_acked-1
+  bool is_probe = type & UDX_HEADER_HEARTBEAT;
+
+  if (is_probe) {
+    send_ack(stream);
+    return 1;
+  }
 
   buf += UDX_HEADER_SIZE;
   buf_len -= UDX_HEADER_SIZE;
@@ -1565,7 +1612,12 @@ process_packet (udx_socket_t *socket, char *buf, ssize_t buf_len, struct sockadd
 
     if (stream->remote_acked == stream->seq) {
       uv_timer_stop(&stream->rto_timer);
-      uv_timer_stop(&stream->tlp_timer);
+
+      if (stream->keepalive_timeout_ms) {
+        uv_timer_start(&stream->tlp_and_keepalive_timer, udx_keepalive_timeout, stream->keepalive_timeout_ms, 0);
+      } else {
+        uv_timer_stop(&stream->tlp_and_keepalive_timer);
+      }
     }
 
     if ((stream->status & UDX_STREAM_ALL_ENDED) == UDX_STREAM_ALL_ENDED) {
@@ -1612,7 +1664,12 @@ process_packet (udx_socket_t *socket, char *buf, ssize_t buf_len, struct sockadd
     if (stream->remote_acked == stream->seq) {
       assert(stream->inflight_queue.len == 0 && stream->retransmit_queue.len == 0);
       uv_timer_stop(&stream->rto_timer);
-      uv_timer_stop(&stream->tlp_timer);
+
+      if (stream->keepalive_timeout_ms) {
+        uv_timer_start(&stream->tlp_and_keepalive_timer, udx_keepalive_timeout, stream->keepalive_timeout_ms, 0);
+      } else {
+        uv_timer_stop(&stream->tlp_and_keepalive_timer);
+      }
     } else {
       assert(!(stream->status & UDX_STREAM_CLOSED));
       uv_timer_start(&stream->rto_timer, udx_rto_timeout, stream->rto, 0);
@@ -1678,10 +1735,12 @@ pacing_timer_timeout (uv_timer_t *timer) {
 }
 
 // arms the retransmit timers (RTO, TLP, ZWP), called after data is transmitted
-// or retransmitted. the current timers armed are the
-// optimize: use a single timer and a 'meaning' flag
+// or retransmitted.
 static void
 arm_stream_timers (udx_stream_t *stream, bool sent_tlp) {
+  assert(stream->inflight_queue.len > 0);
+  assert(stream->remote_acked != stream->seq);
+
   if (!uv_is_active((uv_handle_t *) &stream->rto_timer)) {
     assert(stream->rto >= 1);
     assert(stream->status != UDX_STREAM_CLOSED);
@@ -1691,7 +1750,7 @@ arm_stream_timers (udx_stream_t *stream, bool sent_tlp) {
   // rack 7.2 rearm tlp timer
 
   if (stream->ca_state != UDX_CA_OPEN || stream->sacks) {
-    uv_timer_stop(&stream->tlp_timer);
+    uv_timer_stop(&stream->tlp_and_keepalive_timer);
   } else {
     if (!sent_tlp) {
       schedule_loss_probe(stream);
@@ -1990,7 +2049,7 @@ on_socket_send_slow (uv_udp_send_t *_req, int status) {
   }
 
   // 2. if next packet is also a specific ttl it will be re-set here
-  on_slow_send(_req->handle);
+  maybe_adjust_ttl(_req->handle);
   if (req->on_send) {
     req->on_send(req, status);
   }
@@ -2044,7 +2103,7 @@ udx_socket_send_ttl (udx_socket_send_t *req, udx_socket_t *socket, const uv_buf_
     }
 
     err = uv_udp_send(&req->uv_udp_send, &socket->uv_udp, bufs, bufs_len, dest, on_socket_send_slow);
-    maybe_adjust_ttl(socket); // edge case: queue was empty
+    _maybe_adjust_ttl(socket); // edge case: queue was empty
 
     return err;
   }
@@ -2139,8 +2198,8 @@ udx_stream_init (udx_t *udx, udx_stream_t *stream, uint32_t local_id, udx_stream
   uv_timer_init(udx->loop, &stream->rack_reo_timer);
   stream->rack_reo_timer.data = stream;
 
-  uv_timer_init(udx->loop, &stream->tlp_timer);
-  stream->tlp_timer.data = stream;
+  uv_timer_init(udx->loop, &stream->tlp_and_keepalive_timer);
+  stream->tlp_and_keepalive_timer.data = stream;
 
   uv_timer_init(udx->loop, &stream->zwp_timer);
   stream->zwp_timer.data = stream;
@@ -2200,6 +2259,18 @@ udx_stream_set_seq (udx_stream_t *stream, uint32_t seq) {
   return 0;
 }
 
+int
+udx_stream_set_keepalive (udx_stream_t *stream, uint32_t keepalive_timeout_ms) {
+
+  stream->keepalive_timeout_ms = keepalive_timeout_ms;
+
+  if (stream->remote_acked == stream->seq && keepalive_timeout_ms && stream->status & UDX_STREAM_CONNECTED) {
+    uv_timer_start(&stream->tlp_and_keepalive_timer, udx_keepalive_timeout, stream->keepalive_timeout_ms, 0);
+  }
+
+  return 0;
+}
+
 int
 udx_stream_get_ack (udx_stream_t *stream, uint32_t *ack) {
   *ack = stream->ack;
@@ -2399,6 +2470,10 @@ udx_stream_connect (udx_stream_t *stream, udx_socket_t *socket, uint32_t remote_
     stream->ack_needed = false;
   }
 
+  if (stream->keepalive_timeout_ms) {
+    uv_timer_start(&stream->tlp_and_keepalive_timer, udx_keepalive_timeout, stream->keepalive_timeout_ms, 0);
+  }
+
   return 0;
 }
 

test/CMakeLists.txt

@@ -25,6 +25,7 @@ list(APPEND tests
   win-filter
   stream-bbr-state
   stream-combined-write
+  stream-keepalive
 )
 
 list(APPEND skipped_tests