Multiple Receive API Support

docs/Settings.md

@@ -64,6 +64,7 @@ The following settings are available via registry as well as via [QUIC_SETTINGS]
 | Stateless Operation Expiration     | uint16_t   | StatelessOperationExpirationMs |            100 | The time limit between operations for the same endpoint, in milliseconds.                                                     |
 | Congestion Control Algorithm       | uint16_t   | CongestionControlAlgorithm  |         0 (Cubic) | The congestion control algorithm used for the connection.                                                                     |
 | ECN                                | uint8_t    | EcnEnabled                  |         0 (FALSE) | Enable sender-side ECN support.                                                                                               |
+| Stream Multi Receive               | uint8_t    | StreamMultiReceiveEnabled   |         0 (FALSE) | Enable multi receive support                                                                                                  |
 
 The types map to registry types as follows:
   - `uint64_t` is a `REG_QWORD`.

docs/Streams.md

@@ -78,7 +78,7 @@ Typically, the buffer count is one, which means that most events will include a
 
 When the buffer count is 0, it signifies the reception of a QUIC frame with empty data, which also indicates the end of stream data.
 
-Currently, the maximum buffer count is 2 in the case of partial receive, where only a portion of the buffer data is consumed (as explained below). However, it is strongly advised not to assume in application code that the upper limit is always 2. This caution is important because future releases may incorporate multiple circular buffers to enhance performance, leading to potential changes in the buffer count limit.
+Currently, the maximum buffer count is 3 in the case of partial receive, where only a portion of the buffer data is consumed (as explained below). However, it is strongly advised not to assume in application code that the upper limit is always 3. This caution is important because future releases may change internal algorithm, leading to potential changes in the buffer count limit.
 
 The app then may respond to the event in a number of ways:
 
@@ -100,4 +100,12 @@ Any value less than or equal to the initial **TotalBufferLength** value is allow
 
 Whenever a receive isn't fully accepted by the app, additional receive events are immediately disabled. The app is assumed to be at capacity and not able to consume more until further indication. To re-enable receive callbacks, the app must call [StreamReceiveSetEnabled](api/StreamReceiveSetEnabled.md).
 
-There are cases where an app may want to partially accept the current data, but still immediately get a callback with the rest of the data. To do this (only works in the synchronous flow) the app must return `QUIC_STATUS_CONTINUE`.
+There are cases where an app may want to partially accept the current data, but still immediately get a callback with the rest of the data. To do this (only works in the synchronous flow) the app must return `QUIC_STATUS_CONTINUE`.
+
+## Multi Receive mode
+
+Setting [`StreamMultiReceiveEnabled`](./Settings.md) an app can continue getting indicated by `QUIC_STREAM_EVENT_RECEIVE` without returning `QUIC_STATUS_SUCCESS` nor calling [StreamReceiveComplete](api/StreamReceiveComplete.md).
+
+This changes internal receive buffer more efficient for continuous receiving.
+
+The app need to keep track of total `TotalBufferLength` to later call [StreamReceiveComplete](api/StreamReceiveComplete.md) appropriately.

docs/api/QUIC_SETTINGS.md

@@ -54,7 +54,8 @@ typedef struct QUIC_SETTINGS {
             uint64_t ReliableResetEnabled                   : 1;
             uint64_t OneWayDelayEnabled                     : 1;
             uint64_t NetStatsEventEnabled                   : 1;
-            uint64_t RESERVED                               : 22;
+            uint64_t StreamMultiReceiveEnabled              : 1;
+            uint64_t RESERVED                               : 21;
 #else
             uint64_t RESERVED                               : 26;
 #endif
@@ -104,7 +105,8 @@ typedef struct QUIC_SETTINGS {
             uint64_t ReliableResetEnabled      : 1;
             uint64_t OneWayDelayEnabled        : 1;
             uint64_t NetStatsEventEnabled      : 1;
-            uint64_t ReservedFlags             : 59;
+            uint64_t StreamMultiReceiveEnabled : 1;
+            uint64_t ReservedFlags             : 58;
 #else
             uint64_t ReservedFlags             : 63;
 #endif
@@ -351,6 +353,12 @@ Initial stream receive flow control window size for remotely initiated unidirect
 
 **Default value:** 0 (no overwrite)
 
+`StreamMultiReceiveEnabled`
+
+Enable multi receive mode. An app can continue receiving stream data without calling `StreamReceiveComplete` for each `QUIC_STREAM_EVENT_RECEIVE` indication.
+
+**Default value:** 0 (`FALSE`)
+
 # Remarks
 
 When setting new values for the settings, the app must set the corresponding `.IsSet.*` parameter for each actual parameter that is being set or updated. For example:

docs/api/StreamReceiveComplete.md

@@ -22,8 +22,9 @@ void
 # Remarks
 
 This is an asynchronous API but can run inline if called in a callback.
-The application must ensure that one `StreamReceiveComplete` call corresponds to one `QUIC_STREAM_EVENT_RECEIVE` event.
+The application, without setting `StreamMultiReceiveEnabled`, must ensure that one `StreamReceiveComplete` call corresponds to one `QUIC_STREAM_EVENT_RECEIVE` event.
 Duplicate `StreamReceiveComplete` calls after one `QUIC_STREAM_EVENT_RECEIVE` event are ignored silently even with different `BufferLength`.
+The `StreamMultiReceiveEnabled` mode doesn't follow this rule. Multiple `QUIC_STREAM_EVENT_RECEIVE` events can be indicated at once by `StreamReceiveComplete`. The application needs to keep track of accumulated `TotalBufferLength` with this mode.
 
 # See Also
 

src/core/quicdef.h

@@ -527,6 +527,11 @@ CXPLAT_STATIC_ASSERT(
 //
 #define QUIC_DEFAULT_NET_STATS_EVENT_ENABLED         FALSE
 
+//
+// The default settings for using multiple parallel receives for streams.
+//
+#define QUIC_DEFAULT_STREAM_MULTI_RECEIVE_ENABLED    FALSE
+
 //
 // The number of rounds in Cubic Slow Start to sample RTT.
 //
@@ -633,6 +638,7 @@ CXPLAT_STATIC_ASSERT(
 #define QUIC_SETTING_RELIABLE_RESET_ENABLED         "ReliableResetEnabled"
 #define QUIC_SETTING_ONE_WAY_DELAY_ENABLED          "OneWayDelayEnabled"
 #define QUIC_SETTING_NET_STATS_EVENT_ENABLED        "NetStatsEventEnabled"
+#define QUIC_SETTING_STREAM_MULTI_RECEIVE_ENABLED   "StreamMultiReceiveEnabled"
 
 #define QUIC_SETTING_INITIAL_WINDOW_PACKETS         "InitialWindowPackets"
 #define QUIC_SETTING_SEND_IDLE_TIMEOUT_MS           "SendIdleTimeoutMs"

src/core/settings.c

@@ -165,6 +165,9 @@ QuicSettingsSetDefault(
     if (!Settings->IsSet.NetStatsEventEnabled) {
         Settings->NetStatsEventEnabled = QUIC_DEFAULT_NET_STATS_EVENT_ENABLED;
     }
+    if (!Settings->IsSet.StreamMultiReceiveEnabled) {
+        Settings->StreamMultiReceiveEnabled = QUIC_DEFAULT_STREAM_MULTI_RECEIVE_ENABLED;
+    }
 }
 
 _IRQL_requires_max_(PASSIVE_LEVEL)
@@ -330,6 +333,9 @@ QuicSettingsCopy(
     if (!Destination->IsSet.NetStatsEventEnabled) {
         Destination->NetStatsEventEnabled = Source->NetStatsEventEnabled;
     }
+    if (!Destination->IsSet.StreamMultiReceiveEnabled) {
+        Destination->StreamMultiReceiveEnabled = Source->StreamMultiReceiveEnabled;
+    }
 }
 
 _IRQL_requires_max_(PASSIVE_LEVEL)
@@ -700,6 +706,11 @@ QuicSettingApply(
         Destination->NetStatsEventEnabled = Source->NetStatsEventEnabled;
         Destination->IsSet.NetStatsEventEnabled = TRUE;
     }
+
+    if (Source->IsSet.StreamMultiReceiveEnabled && (!Destination->IsSet.StreamMultiReceiveEnabled || OverWrite)) {
+        Destination->StreamMultiReceiveEnabled = Source->StreamMultiReceiveEnabled;
+        Destination->IsSet.StreamMultiReceiveEnabled = TRUE;
+    }
     return TRUE;
 }
 
@@ -1358,6 +1369,16 @@ QuicSettingsLoad(
             &ValueLen);
         Settings->NetStatsEventEnabled = !!Value;
     }
+    if (!Settings->IsSet.StreamMultiReceiveEnabled) {
+        Value = QUIC_DEFAULT_STREAM_MULTI_RECEIVE_ENABLED;
+        ValueLen = sizeof(Value);
+        CxPlatStorageReadValue(
+            Storage,
+            QUIC_SETTING_STREAM_MULTI_RECEIVE_ENABLED,
+            (uint8_t*)&Value,
+            &ValueLen);
+        Settings->StreamMultiReceiveEnabled = !!Value;
+    }
 }
 
 _IRQL_requires_max_(PASSIVE_LEVEL)
@@ -1426,6 +1447,7 @@ QuicSettingsDump(
     QuicTraceLogVerbose(SettingReliableResetEnabled,        "[sett] ReliableResetEnabled   = %hhu", Settings->ReliableResetEnabled);
     QuicTraceLogVerbose(SettingOneWayDelayEnabled,          "[sett] OneWayDelayEnabled     = %hhu", Settings->OneWayDelayEnabled);
     QuicTraceLogVerbose(SettingNetStatsEventEnabled,        "[sett] NetStatsEventEnabled   = %hhu", Settings->NetStatsEventEnabled);
+    QuicTraceLogVerbose(SettingsStreamMultiReceiveEnabled,  "[sett] StreamMultiReceiveEnabled= %hhu", Settings->StreamMultiReceiveEnabled);
 }
 
 _IRQL_requires_max_(PASSIVE_LEVEL)
@@ -1587,6 +1609,9 @@ QuicSettingsDumpNew(
     if (Settings->IsSet.NetStatsEventEnabled) {
         QuicTraceLogVerbose(SettingNetStatsEventEnabled,            "[sett] NetStatsEventEnabled       = %hhu", Settings->NetStatsEventEnabled);
     }
+    if (Settings->IsSet.StreamMultiReceiveEnabled) {
+        QuicTraceLogVerbose(SettingStreamMultiReceiveEnabled,       "[sett] StreamMultiReceiveEnabled  = %hhu", Settings->StreamMultiReceiveEnabled);
+    }
 }
 
 #define SETTINGS_SIZE_THRU_FIELD(SettingsType, Field) \
@@ -1843,6 +1868,14 @@ QuicSettingsSettingsToInternal(
         SettingsSize,
         InternalSettings);
 
+    SETTING_COPY_FLAG_TO_INTERNAL_SIZED(
+        Flags,
+        StreamMultiReceiveEnabled,
+        QUIC_SETTINGS,
+        Settings,
+        SettingsSize,
+        InternalSettings);
+
     return QUIC_STATUS_SUCCESS;
 }
 
@@ -2004,6 +2037,14 @@ QuicSettingsGetSettings(
         *SettingsLength,
         InternalSettings);
 
+    SETTING_COPY_FLAG_FROM_INTERNAL_SIZED(
+        Flags,
+        StreamMultiReceiveEnabled,
+        QUIC_SETTINGS,
+        Settings,
+        *SettingsLength,
+        InternalSettings);
+
     *SettingsLength = CXPLAT_MIN(*SettingsLength, sizeof(QUIC_SETTINGS));
 
     return QUIC_STATUS_SUCCESS;

src/core/settings.h

@@ -61,7 +61,8 @@ typedef struct QUIC_SETTINGS_INTERNAL {
             uint64_t ReliableResetEnabled                   : 1;
             uint64_t OneWayDelayEnabled                     : 1;
             uint64_t NetStatsEventEnabled                   : 1;
-            uint64_t RESERVED                               : 17;
+            uint64_t StreamMultiReceiveEnabled              : 1;
+            uint64_t RESERVED                               : 16;
         } IsSet;
     };
 
@@ -111,6 +112,7 @@ typedef struct QUIC_SETTINGS_INTERNAL {
     uint8_t ReliableResetEnabled            : 1;
     uint8_t OneWayDelayEnabled              : 1;
     uint8_t NetStatsEventEnabled            : 1;
+    uint8_t StreamMultiReceiveEnabled       : 1;
     uint8_t MtuDiscoveryMissingProbeCount;
 
 } QUIC_SETTINGS_INTERNAL;

src/core/stream.c

@@ -66,6 +66,7 @@ QuicStreamInitialize(
     Stream->Flags.Allocated = TRUE;
     Stream->Flags.SendEnabled = TRUE;
     Stream->Flags.ReceiveEnabled = TRUE;
+    Stream->Flags.ReceiveMultiple = Connection->Settings.StreamMultiReceiveEnabled;
     Stream->RecvMaxLength = UINT64_MAX;
     Stream->RefCount = 1;
     Stream->SendRequestsTail = &Stream->SendRequests;
@@ -131,7 +132,8 @@ QuicStreamInitialize(
             &Stream->RecvBuffer,
             InitialRecvBufferLength,
             FlowControlWindowSize,
-            QUIC_RECV_BUF_MODE_CIRCULAR,
+            Stream->Flags.ReceiveMultiple ?
+                QUIC_RECV_BUF_MODE_MULTIPLE : QUIC_RECV_BUF_MODE_CIRCULAR,
             PreallocatedRecvChunk);
     if (QUIC_FAILED(Status)) {
         goto Exit;

src/core/stream.h

@@ -139,6 +139,7 @@ typedef union QUIC_STREAM_FLAGS {
 
         BOOLEAN SendEnabled             : 1;    // Application is allowed to send data.
         BOOLEAN ReceiveEnabled          : 1;    // Application is ready for receive callbacks.
+        BOOLEAN ReceiveMultiple         : 1;    // The app supports multiple parallel receive indications.
         BOOLEAN ReceiveFlushQueued      : 1;    // The receive flush operation is queued.
         BOOLEAN ReceiveDataPending      : 1;    // Data (or FIN) is queued and ready for delivery.
         BOOLEAN ReceiveCallActive       : 1;    // There is an active receive to the app.

src/core/stream_recv.c

@@ -129,10 +129,7 @@ QuicStreamRecvQueueFlush(
     // The caller has indicated data is ready to be indicated to the
     // application. Queue a FLUSH_RECV if one isn't already queued.
     //
-
-    if (Stream->Flags.ReceiveEnabled &&
-        Stream->Flags.ReceiveDataPending &&
-        Stream->RecvPendingLength == 0) {
+    if (Stream->Flags.ReceiveEnabled && Stream->Flags.ReceiveDataPending) {
 
         if (AllowInlineFlush) {
             QuicStreamRecvFlush(Stream);
@@ -541,7 +538,9 @@ QuicStreamProcessStreamFrame(
         }
     }
 
-    if (ReadyToDeliver) {
+    if (ReadyToDeliver &&
+        (Stream->RecvBuffer.RecvMode == QUIC_RECV_BUF_MODE_MULTIPLE ||
+         Stream->RecvBuffer.ReadPendingLength == 0)) {
         Stream->Flags.ReceiveDataPending = TRUE;
         QuicStreamRecvQueueFlush(
             Stream,
@@ -870,8 +869,6 @@ QuicStreamRecvFlush(
         return;
     }
 
-    CXPLAT_TEL_ASSERT(!Stream->RecvPendingLength); // N.B. - Will be an invalid assert once we support multiple receives
-
     BOOLEAN FlushRecv = TRUE;
     while (FlushRecv) {
         CXPLAT_DBG_ASSERT(!Stream->Flags.SentStopSending);
@@ -924,9 +921,10 @@ QuicStreamRecvFlush(
             Event.RECEIVE.Flags |= QUIC_RECEIVE_FLAG_FIN; // TODO - 0-RTT flag?
         }
 
-        Stream->Flags.ReceiveEnabled = FALSE;
+        Stream->Flags.ReceiveEnabled = Stream->Flags.ReceiveMultiple;
         Stream->Flags.ReceiveCallActive = TRUE;
         Stream->RecvPendingLength += Event.RECEIVE.TotalBufferLength;
+        CXPLAT_DBG_ASSERT(Stream->RecvPendingLength <= Stream->RecvBuffer.ReadPendingLength);
 
         QuicTraceEvent(
             StreamAppReceive,
@@ -1056,7 +1054,7 @@ QuicStreamReceiveComplete(
         //
         Stream->Flags.ReceiveEnabled = TRUE;
 
-    } else {
+    } else if (!Stream->Flags.ReceiveMultiple) {
         //
         // The app didn't drain all the data, so we will need to wait for them
         // to request a new receive.
@@ -1080,9 +1078,10 @@ QuicStreamReceiveComplete(
     if (Stream->Flags.ReceiveDataPending) {
         //
         // There is still more data for the app to process and it still has
-        // receive callbacks enabled, so do another recv flush.
+        // receive callbacks enabled, so do another recv flush (if not already
+        // doing multi-receive mode).
         //
-        return TRUE;
+        return !Stream->Flags.ReceiveMultiple;
     }
 
     if (Stream->RecvBuffer.BaseOffset == Stream->RecvMaxLength) {
@@ -1157,7 +1156,9 @@ QuicStreamRecvSetEnabledState(
         CXPLAT_DBG_ASSERT(!Stream->Flags.SentStopSending);
         Stream->Flags.ReceiveEnabled = NewRecvEnabled;
 
-        if (Stream->Flags.Started && NewRecvEnabled) {
+        if (Stream->Flags.Started && NewRecvEnabled &&
+            (Stream->RecvBuffer.RecvMode == QUIC_RECV_BUF_MODE_MULTIPLE ||
+            Stream->RecvBuffer.ReadPendingLength == 0)) {
             //
             // The application just resumed receive callbacks. Queue a
             // flush receive operation to start draining the receive buffer.

src/core/unittest/SettingsTest.cpp

@@ -124,6 +124,7 @@ TEST(SettingsTest, TestAllSettingsFieldsSet)
     SETTINGS_FEATURE_SET_TEST(ReliableResetEnabled, QuicSettingsSettingsToInternal);
     SETTINGS_FEATURE_SET_TEST(OneWayDelayEnabled, QuicSettingsSettingsToInternal);
     SETTINGS_FEATURE_SET_TEST(NetStatsEventEnabled, QuicSettingsSettingsToInternal);
+    SETTINGS_FEATURE_SET_TEST(StreamMultiReceiveEnabled, QuicSettingsSettingsToInternal);
 
     Settings.IsSetFlags = 0;
     Settings.IsSet.RESERVED = ~Settings.IsSet.RESERVED;
@@ -209,6 +210,7 @@ TEST(SettingsTest, TestAllSettingsFieldsGet)
     SETTINGS_FEATURE_GET_TEST(ReliableResetEnabled, QuicSettingsGetSettings);
     SETTINGS_FEATURE_GET_TEST(OneWayDelayEnabled, QuicSettingsGetSettings);
     SETTINGS_FEATURE_GET_TEST(NetStatsEventEnabled, QuicSettingsGetSettings);
+    SETTINGS_FEATURE_GET_TEST(StreamMultiReceiveEnabled, QuicSettingsGetSettings);
 
     Settings.IsSetFlags = 0;
     Settings.IsSet.RESERVED = ~Settings.IsSet.RESERVED;