QUIC APL: Implement backpressure on stream creation

Reviewed-by: Matt Caswell <matt@openssl.org>
Reviewed-by: Tomas Mraz <tomas@openssl.org>
(Merged from https://github.com/openssl/openssl/pull/21811)

4 files changed, 92 insertions, 13 deletions

diff --git a/ssl/quic/quic_channel.c b/ssl/quic/quic_channel.c
index 5cee9f7532..98e1a0110f 100644
--- a/ssl/quic/quic_channel.c
+++ b/ssl/quic/quic_channel.c
@@ -3434,6 +3434,23 @@ err:
     return 0;
 }
 
+static uint64_t *ch_get_local_stream_next_ordinal_ptr(QUIC_CHANNEL *ch,
+                                                      int is_uni)
+{
+    return is_uni ? &ch->next_local_stream_ordinal_uni
+                  : &ch->next_local_stream_ordinal_bidi;
+}
+
+int ossl_quic_channel_is_new_local_stream_admissible(QUIC_CHANNEL *ch,
+                                                     int is_uni)
+{
+    uint64_t *p_next_ordinal = ch_get_local_stream_next_ordinal_ptr(ch, is_uni);
+
+    return ossl_quic_stream_map_is_local_allowed_by_stream_limit(&ch->qsm,
+                                                                 *p_next_ordinal,
+                                                                 is_uni);
+}
+
 QUIC_STREAM *ossl_quic_channel_new_stream_local(QUIC_CHANNEL *ch, int is_uni)
 {
     QUIC_STREAM *qs;
@@ -3443,13 +3460,12 @@ QUIC_STREAM *ossl_quic_channel_new_stream_local(QUIC_CHANNEL *ch, int is_uni)
     type = ch->is_server ? QUIC_STREAM_INITIATOR_SERVER
                          : QUIC_STREAM_INITIATOR_CLIENT;
 
-    if (is_uni) {
-        p_next_ordinal = &ch->next_local_stream_ordinal_uni;
+    p_next_ordinal = ch_get_local_stream_next_ordinal_ptr(ch, is_uni);
+
+    if (is_uni)
         type |= QUIC_STREAM_DIR_UNI;
-    } else {
-        p_next_ordinal = &ch->next_local_stream_ordinal_bidi;
+    else
         type |= QUIC_STREAM_DIR_BIDI;
-    }
 
     if (*p_next_ordinal >= ((uint64_t)1) << 62)
         return NULL;
diff --git a/ssl/quic/quic_impl.c b/ssl/quic/quic_impl.c
index 6bb6b46583..5073bb1e6f 100644
--- a/ssl/quic/quic_impl.c
+++ b/ssl/quic/quic_impl.c
@@ -1717,13 +1717,35 @@ err:
     return NULL;
 }
 
+struct quic_new_stream_wait_args {
+    QUIC_CONNECTION *qc;
+    int is_uni;
+};
+
+static int quic_new_stream_wait(void *arg)
+{
+    struct quic_new_stream_wait_args *args = arg;
+    QUIC_CONNECTION *qc = args->qc;
+
+    if (!quic_mutation_allowed(qc, /*req_active=*/1))
+        return -1;
+
+    if (ossl_quic_channel_is_new_local_stream_admissible(qc->ch, args->is_uni))
+        return 1;
+
+    return 0;
+}
+
 /* locking depends on need_lock */
 static SSL *quic_conn_stream_new(QCTX *ctx, uint64_t flags, int need_lock)
 {
+    int ret;
     QUIC_CONNECTION *qc = ctx->qc;
     QUIC_XSO *xso = NULL;
     QUIC_STREAM *qs = NULL;
     int is_uni = ((flags & SSL_STREAM_FLAG_UNI) != 0);
+    int no_blocking = ((flags & SSL_STREAM_FLAG_NO_BLOCK) != 0);
+    int advance = ((flags & SSL_STREAM_FLAG_ADVANCE) != 0);
 
     if (need_lock)
         quic_lock(qc);
@@ -1733,6 +1755,33 @@ static SSL *quic_conn_stream_new(QCTX *ctx, uint64_t flags, int need_lock)
         goto err;
     }
 
+    if (!advance
+        && !ossl_quic_channel_is_new_local_stream_admissible(qc->ch, is_uni)) {
+        struct quic_new_stream_wait_args args;
+
+        /*
+         * Stream count flow control currently doesn't permit this stream to be
+         * opened.
+         */
+        if (no_blocking || !qc_blocking_mode(qc)) {
+            QUIC_RAISE_NON_NORMAL_ERROR(ctx, SSL_R_STREAM_COUNT_LIMITED, NULL);
+            goto err;
+        }
+
+        args.qc     = qc;
+        args.is_uni = is_uni;
+
+        /* Blocking mode - wait until we can get a stream. */
+        ret = block_until_pred(ctx->qc, quic_new_stream_wait, &args, 0);
+        if (!quic_mutation_allowed(qc, /*req_active=*/1)) {
+            QUIC_RAISE_NON_NORMAL_ERROR(ctx, SSL_R_PROTOCOL_IS_SHUTDOWN, NULL);
+            goto err; /* Shutdown before completion */
+        } else if (ret <= 0) {
+            QUIC_RAISE_NON_NORMAL_ERROR(ctx, ERR_R_INTERNAL_ERROR, NULL);
+            goto err; /* Non-protocol error */
+        }
+    }
+
     qs = ossl_quic_channel_new_stream_local(qc->ch, is_uni);
     if (qs == NULL) {
         QUIC_RAISE_NON_IO_ERROR(ctx, ERR_R_INTERNAL_ERROR, NULL);
diff --git a/ssl/quic/quic_stream_map.c b/ssl/quic/quic_stream_map.c
index 5d4354a2df..4b595f50e7 100644
--- a/ssl/quic/quic_stream_map.c
+++ b/ssl/quic/quic_stream_map.c
@@ -311,19 +311,31 @@ static int qsm_ready_for_gc(QUIC_STREAM_MAP *qsm, QUIC_STREAM *qs)
             || qs->send_state == QUIC_SSTREAM_STATE_RESET_RECVD);
 }
 
+int ossl_quic_stream_map_is_local_allowed_by_stream_limit(QUIC_STREAM_MAP *qsm,
+                                                          uint64_t stream_ordinal,
+                                                          int is_uni)
+{
+    uint64_t stream_limit;
+
+    if (qsm->get_stream_limit_cb == NULL)
+        return 1;
+
+    stream_limit = qsm->get_stream_limit_cb(is_uni, qsm->get_stream_limit_cb_arg);
+    return stream_ordinal < stream_limit;
+}
+
 void ossl_quic_stream_map_update_state(QUIC_STREAM_MAP *qsm, QUIC_STREAM *s)
 {
     int should_be_active, allowed_by_stream_limit = 1;
 
-    if (qsm->get_stream_limit_cb != NULL
-        && ossl_quic_stream_is_server_init(s) == qsm->is_server) {
-        int uni = !ossl_quic_stream_is_bidi(s);
-        uint64_t stream_limit, stream_ordinal = s->id >> 2;
-
-        stream_limit
-            = qsm->get_stream_limit_cb(uni, qsm->get_stream_limit_cb_arg);
+    if (ossl_quic_stream_is_server_init(s) == qsm->is_server) {
+        int is_uni = !ossl_quic_stream_is_bidi(s);
+        uint64_t stream_ordinal = s->id >> 2;
 
-        allowed_by_stream_limit = (stream_ordinal < stream_limit);
+        allowed_by_stream_limit
+            = ossl_quic_stream_map_is_local_allowed_by_stream_limit(qsm,
+                                                                    stream_ordinal,
+                                                                    is_uni);
     }
 
     if (s->send_state == QUIC_SSTREAM_STATE_DATA_SENT
diff --git a/ssl/ssl_err.c b/ssl/ssl_err.c
index 7bbc55dc69..1cda4a0aef 100644
--- a/ssl/ssl_err.c
+++ b/ssl/ssl_err.c
@@ -466,6 +466,8 @@ static const ERR_STRING_DATA SSL_str_reasons[] = {
     {ERR_PACK(ERR_LIB_SSL, 0, SSL_R_SSL_SESSION_VERSION_MISMATCH),
     "ssl session version mismatch"},
     {ERR_PACK(ERR_LIB_SSL, 0, SSL_R_STILL_IN_INIT), "still in init"},
+    {ERR_PACK(ERR_LIB_SSL, 0, SSL_R_STREAM_COUNT_LIMITED),
+    "stream count limited"},
     {ERR_PACK(ERR_LIB_SSL, 0, SSL_R_STREAM_FINISHED), "stream finished"},
     {ERR_PACK(ERR_LIB_SSL, 0, SSL_R_STREAM_RECV_ONLY), "stream recv only"},
     {ERR_PACK(ERR_LIB_SSL, 0, SSL_R_STREAM_RESET), "stream reset"},