Release 2.23.3

- [BUGFIX] Update packetization threshold when writing to stream
  after packet size is reduced following an RTO.

CHANGELOG

@@ -1,3 +1,8 @@
+2020-10-22
+    - 2.23.3
+    - [BUGFIX] Update packetization threshold when writing to stream
+      after packet size is reduced following an RTO.
+
 2020-10-21
     - 2.23.2
     - Add QPACK stats collection and experimentation mode, see the new

docs/conf.py

@@ -26,7 +26,7 @@ author = u'LiteSpeed Technologies'
 # The short X.Y version
 version = u'2.23'
 # The full version, including alpha/beta/rc tags
-release = u'2.23.2'
+release = u'2.23.3'
 
 
 # -- General configuration ---------------------------------------------------

include/lsquic.h

@@ -25,7 +25,7 @@ extern "C" {
 
 #define LSQUIC_MAJOR_VERSION 2
 #define LSQUIC_MINOR_VERSION 23
-#define LSQUIC_PATCH_VERSION 2
+#define LSQUIC_PATCH_VERSION 3
 
 /**
  * Engine flags:

src/liblsquic/lsquic_stream.c

@@ -2538,6 +2538,7 @@ struct frame_gen_ctx
     size_t              (*fgc_size) (void *ctx);
     int                 (*fgc_fin) (void *ctx);
     gsf_read_f            fgc_read;
+    size_t                fgc_thresh;
 };
 
 
@@ -2703,6 +2704,22 @@ incr_sm_payload (struct lsquic_stream *stream, size_t incr)
 }
 
 
+static void
+maybe_resize_threshold (struct frame_gen_ctx *fg_ctx)
+{
+    struct lsquic_stream *stream = fg_ctx->fgc_stream;
+    size_t old;
+
+    if (fg_ctx->fgc_thresh)
+    {
+        old = fg_ctx->fgc_thresh;
+        fg_ctx->fgc_thresh
+            = lsquic_stream_flush_threshold(stream, fg_ctx->fgc_size(fg_ctx));
+        LSQ_DEBUG("changed threshold from %zd to %zd", old, fg_ctx->fgc_thresh);
+    }
+}
+
+
 static size_t
 frame_std_gen_read (void *ctx, void *begin_buf, size_t len, int *fin)
 {
@@ -2721,7 +2738,10 @@ frame_std_gen_read (void *ctx, void *begin_buf, size_t len, int *fin)
                                                 stream->sm_n_buffered - len);
             stream->sm_n_buffered -= len;
             if (0 == stream->sm_n_buffered)
+            {
                 maybe_resize_stream_buffer(stream);
+                maybe_resize_threshold(fg_ctx);
+            }
             assert(stream->max_send_off >= stream->tosend_off + stream->sm_n_buffered);
             incr_sm_payload(stream, len);
             *fin = fg_ctx->fgc_fin(fg_ctx);
@@ -2731,6 +2751,7 @@ frame_std_gen_read (void *ctx, void *begin_buf, size_t len, int *fin)
         p += stream->sm_n_buffered;
         stream->sm_n_buffered = 0;
         maybe_resize_stream_buffer(stream);
+        maybe_resize_threshold(fg_ctx);
     }
 
     available = lsquic_stream_write_avail(fg_ctx->fgc_stream);
@@ -3297,6 +3318,7 @@ stream_write_to_packets (lsquic_stream_t *stream, struct lsquic_reader *reader,
         .fgc_stream = stream,
         .fgc_reader = reader,
         .fgc_nread_from_reader = 0,
+        .fgc_thresh = thresh,
     };
 
 #if LSQUIC_EXTRA_CHECKS
@@ -3319,7 +3341,9 @@ stream_write_to_packets (lsquic_stream_t *stream, struct lsquic_reader *reader,
     }
 
     seen_ok = 0;
-    while ((size = fg_ctx.fgc_size(&fg_ctx), thresh ? size >= thresh : size > 0)
+    while ((size = fg_ctx.fgc_size(&fg_ctx),
+                            fg_ctx.fgc_thresh
+                          ? size >= fg_ctx.fgc_thresh : size > 0)
            || fg_ctx.fgc_fin(&fg_ctx))
     {
         switch (stream->sm_write_to_packet(&fg_ctx, size))
@@ -3354,9 +3378,9 @@ stream_write_to_packets (lsquic_stream_t *stream, struct lsquic_reader *reader,
     if (use_framing && seen_ok)
         maybe_close_varsize_hq_frame(stream);
 
-    if (thresh && (swo & SWO_BUFFER))
+    if (fg_ctx.fgc_thresh && (swo & SWO_BUFFER))
     {
-        assert(size < thresh);
+        assert(size < fg_ctx.fgc_thresh);
         assert(size >= stream->sm_n_buffered);
         size -= stream->sm_n_buffered;
         if (size > 0)

tests/test_stream.c

@@ -2469,6 +2469,80 @@ test_changing_pack_size (void)
 }
 
 
+/* This tests what happens when a stream data is buffered using one packet
+ * size, but then packet size get smaller (which is what happens when an RTO
+ * occurs), and then more data is written.
+ *
+ * In particular, the write sizes in this tests are structured to make
+ * maybe_resize_threshold() change the threshold.
+ */
+static void
+test_reducing_pack_size (void)
+{
+    ssize_t nw;
+    struct test_objs tobjs;
+    struct lsquic_conn *lconn = &tobjs.lconn;
+    struct lsquic_stream *stream;
+    int s;
+    unsigned i;
+    unsigned char buf[0x4000];
+
+    init_buf(buf, sizeof(buf));
+
+    enum lsquic_version versions_to_test[] =
+    {
+        LSQVER_050,
+        LSQVER_ID29,
+    };
+
+    /* Particular versions should not matter as this is tests the logic in
+     * stream only, but we do it for completeness.
+     */
+    for (i = 0; i < sizeof(versions_to_test) / sizeof(versions_to_test[0]); i++)
+    {
+        g_pf = select_pf_by_ver(versions_to_test[i]);
+
+        init_test_ctl_settings(&g_ctl_settings);
+        g_ctl_settings.tcs_schedule_stream_packets_immediately = 1;
+        g_ctl_settings.tcs_bp_type = BPT_OTHER_PRIO;
+        init_test_objs(&tobjs, 0x4000, 0x4000, NULL);
+        n_closed = 0;
+        if ((1 << versions_to_test[i]) & LSQUIC_IETF_VERSIONS)
+        {
+            tobjs.ctor_flags |= SCF_IETF;
+            lconn->cn_flags |= LSCONN_IETF;
+        }
+        network_path.np_pack_size = 2000;
+        stream = new_stream(&tobjs, 5);
+        assert(("Stream initialized", stream));
+        const struct test_ctx *const test_ctx_local = tobjs.stream_if_ctx;
+        assert(("on_new_stream called correctly", stream == test_ctx_local->stream));
+
+        nw = lsquic_stream_write(stream, buf, 1400);
+        assert(stream->sm_n_allocated <= 2000);
+        assert(stream->sm_n_buffered > 0);
+        assert(("n bytes written correctly", (size_t)nw == 1400));
+
+        /* Shrink packet size */
+        network_path.np_pack_size = 1300;
+
+        nw = lsquic_stream_write(stream, buf, 3000);
+        assert(stream->sm_n_allocated <= 1300);
+        assert(stream->sm_n_buffered > 0);
+        assert(("n bytes written correctly", (size_t)nw == 3000));
+
+        s = lsquic_stream_flush(stream);
+        assert(stream->sm_n_buffered == 0);
+        assert(0 == s);
+
+        lsquic_stream_destroy(stream);
+        assert(("on_close called", 1 == n_closed));
+        deinit_test_objs(&tobjs);
+    }
+    g_pf = select_pf_by_ver(LSQVER_043);
+}
+
+
 /* Test window update logic, connection-limited */
 static void
 test_window_update1 (void)
@@ -3409,6 +3483,7 @@ main (int argc, char **argv)
     test_writing_to_stream_outside_callback();
     test_stealing_ack();
     test_changing_pack_size();
+    test_reducing_pack_size();
     test_window_update1();
     test_window_update2();
     test_forced_flush_when_conn_blocked();