One Level Up Top Level

src/event/quic/ngx_event_quic_streams.c - nginx source code

Functions defined

Macros defined

Source code


  2. /*
  3. * Copyright (C) Nginx, Inc.
  4. */


  7. #include <ngx_config.h>
  8. #include <ngx_core.h>
  9. #include <ngx_event.h>
  10. #include <ngx_event_quic_connection.h>


  13. ‌#define NGX_QUIC_STREAM_GONE     (void *) -1


  16. static ngx_int_t ngx_quic_do_reset_stream(ngx_quic_stream_t *qs,
  17.     ngx_uint_t err);
  18. static ngx_int_t ngx_quic_shutdown_stream_send(ngx_connection_t *c);
  19. static ngx_int_t ngx_quic_shutdown_stream_recv(ngx_connection_t *c);
  20. static ngx_quic_stream_t *ngx_quic_get_stream(ngx_connection_t *c, uint64_t id);
  21. static ngx_int_t ngx_quic_reject_stream(ngx_connection_t *c, uint64_t id);
  22. static void ngx_quic_init_stream_handler(ngx_event_t *ev);
  23. static void ngx_quic_init_streams_handler(ngx_connection_t *c);
  24. static ngx_int_t ngx_quic_do_init_streams(ngx_connection_t *c);
  25. static ngx_quic_stream_t *ngx_quic_create_stream(ngx_connection_t *c,
  26.     uint64_t id);
  27. static void ngx_quic_empty_handler(ngx_event_t *ev);
  28. static ssize_t ngx_quic_stream_recv(ngx_connection_t *c, u_char *buf,
  29.     size_t size);
  30. static ssize_t ngx_quic_stream_send(ngx_connection_t *c, u_char *buf,
  31.     size_t size);
  32. static ngx_chain_t *ngx_quic_stream_send_chain(ngx_connection_t *c,
  33.     ngx_chain_t *in, off_t limit);
  34. static ngx_int_t ngx_quic_stream_flush(ngx_quic_stream_t *qs);
  35. static void ngx_quic_stream_cleanup_handler(void *data);
  36. static ngx_int_t ngx_quic_close_stream(ngx_quic_stream_t *qs);
  37. static ngx_int_t ngx_quic_can_shutdown(ngx_connection_t *c);
  38. static ngx_int_t ngx_quic_control_flow(ngx_quic_stream_t *qs, uint64_t last);
  39. static ngx_int_t ngx_quic_update_flow(ngx_quic_stream_t *qs, uint64_t last);
  40. static ngx_int_t ngx_quic_update_max_stream_data(ngx_quic_stream_t *qs);
  41. static ngx_int_t ngx_quic_update_max_data(ngx_connection_t *c);
  42. static void ngx_quic_set_event(ngx_event_t *ev);


  45. ngx_connection_t *
  46. ‌ngx_quic_open_stream(ngx_connection_t *c, ngx_uint_t bidi)
  47. {
  48.     uint64_t                id;
  49.     ngx_connection_t       *pc, *sc;
  50.     ngx_quic_stream_t      *qs;
  51.     ngx_quic_connection_t  *qc;

  53.     pc = c->quic ? c->quic->parent : c;
  54.     qc = ngx_quic_get_connection(pc);

  56.     if (qc->closing) {
  57.         return NULL;
  58.     }

  60.     if (bidi) {
  61.         if (qc->streams.server_streams_bidi
  62.             >= qc->streams.server_max_streams_bidi)
  63.         {
  64.             ngx_log_debug1(NGX_LOG_DEBUG_EVENT, c->log, 0,
  65.                            "quic too many server bidi streams:%uL",
  66.                            qc->streams.server_streams_bidi);
  67.             return NULL;
  68.         }

  70.         id = (qc->streams.server_streams_bidi << 2)
  71.              | NGX_QUIC_STREAM_SERVER_INITIATED;

  73.         ngx_log_debug3(NGX_LOG_DEBUG_EVENT, c->log, 0,
  74.                        "quic creating server bidi stream"
  75.                        " streams:%uL max:%uL id:0x%xL",
  76.                        qc->streams.server_streams_bidi,
  77.                        qc->streams.server_max_streams_bidi, id);

  79.         qc->streams.server_streams_bidi++;

  81.     } else {
  82.         if (qc->streams.server_streams_uni
  83.             >= qc->streams.server_max_streams_uni)
  84.         {
  85.             ngx_log_debug1(NGX_LOG_DEBUG_EVENT, c->log, 0,
  86.                            "quic too many server uni streams:%uL",
  87.                            qc->streams.server_streams_uni);
  88.             return NULL;
  89.         }

  91.         id = (qc->streams.server_streams_uni << 2)
  92.              | NGX_QUIC_STREAM_SERVER_INITIATED
  93.              | NGX_QUIC_STREAM_UNIDIRECTIONAL;

  95.         ngx_log_debug3(NGX_LOG_DEBUG_EVENT, c->log, 0,
  96.                        "quic creating server uni stream"
  97.                        " streams:%uL max:%uL id:0x%xL",
  98.                        qc->streams.server_streams_uni,
  99.                        qc->streams.server_max_streams_uni, id);

  101.         qc->streams.server_streams_uni++;
  102.     }

  104.     qs = ngx_quic_create_stream(pc, id);
  105.     if (qs == NULL) {
  106.         return NULL;
  107.     }

  109.     sc = qs->connection;

  111.     sc->write->active = 1;
  112.     sc->write->ready = 1;

  114.     if (bidi) {
  115.         sc->read->active = 1;
  116.     }

  118.     return sc;
  119. }


  122. void
  123. ‌ngx_quic_rbtree_insert_stream(ngx_rbtree_node_t *temp,
  124.     ngx_rbtree_node_t *node, ngx_rbtree_node_t *sentinel)
  125. {
  126.     ngx_rbtree_node_t  **p;
  127.     ngx_quic_stream_t   *qn, *qnt;

  129.     for ( ;; ) {
  130.         qn = (ngx_quic_stream_t *) node;
  131.         qnt = (ngx_quic_stream_t *) temp;

  133.         p = (qn->id < qnt->id) ? &temp->left : &temp->right;

  135.         if (*p == sentinel) {
  136.             break;
  137.         }

  139.         temp = *p;
  140.     }

  142.     *p = node;
  143.     node->parent = temp;
  144.     node->left = sentinel;
  145.     node->right = sentinel;
  146.     ngx_rbt_red(node);
  147. }


  150. ngx_quic_stream_t *
  151. ‌ngx_quic_find_stream(ngx_rbtree_t *rbtree, uint64_t id)
  152. {
  153.     ngx_rbtree_node_t  *node, *sentinel;
  154.     ngx_quic_stream_t  *qn;

  156.     node = rbtree->root;
  157.     sentinel = rbtree->sentinel;

  159.     while (node != sentinel) {
  160.         qn = (ngx_quic_stream_t *) node;

  162.         if (id == qn->id) {
  163.             return qn;
  164.         }

  166.         node = (id < qn->id) ? node->left : node->right;
  167.     }

  169.     return NULL;
  170. }


  173. ngx_int_t
  174. ‌ngx_quic_close_streams(ngx_connection_t *c, ngx_quic_connection_t *qc)
  175. {
  176.     ngx_pool_t         *pool;
  177.     ngx_queue_t        *q;
  178.     ngx_rbtree_t       *tree;
  179.     ngx_connection_t   *sc;
  180.     ngx_rbtree_node_t  *node;
  181.     ngx_quic_stream_t  *qs;

  183.     while (!ngx_queue_empty(&qc->streams.uninitialized)) {
  184.         q = ngx_queue_head(&qc->streams.uninitialized);
  185.         ngx_queue_remove(q);

  187.         qs = ngx_queue_data(q, ngx_quic_stream_t, queue);
  188.         pool = qs->connection->pool;

  190.         ngx_close_connection(qs->connection);
  191.         ngx_destroy_pool(pool);
  192.     }

  194.     tree = &qc->streams.tree;

  196.     if (tree->root == tree->sentinel) {
  197.         return NGX_OK;
  198.     }

  200.     node = ngx_rbtree_min(tree->root, tree->sentinel);

  202.     while (node) {
  203.         qs = (ngx_quic_stream_t *) node;
  204.         node = ngx_rbtree_next(tree, node);
  205.         sc = qs->connection;

  207.         qs->recv_state = NGX_QUIC_STREAM_RECV_RESET_RECVD;
  208.         qs->send_state = NGX_QUIC_STREAM_SEND_RESET_SENT;

  210.         if (sc == NULL) {
  211.             ngx_quic_close_stream(qs);
  212.             continue;
  213.         }

  215.         sc->read->error = 1;
  216.         sc->write->error = 1;

  218.         ngx_quic_set_event(sc->read);
  219.         ngx_quic_set_event(sc->write);

  221.         sc->close = 1;
  222.         sc->read->handler(sc->read);
  223.     }

  225.     if (tree->root == tree->sentinel) {
  226.         return NGX_OK;
  227.     }

  229.     ngx_log_debug0(NGX_LOG_DEBUG_EVENT, c->log, 0,
  230.                    "quic connection has active streams");

  232.     return NGX_AGAIN;
  233. }


  236. ngx_int_t
  237. ‌ngx_quic_reset_stream(ngx_connection_t *c, ngx_uint_t err)
  238. {
  239.     return ngx_quic_do_reset_stream(c->quic, err);
  240. }


  243. static ngx_int_t
  244. ‌ngx_quic_do_reset_stream(ngx_quic_stream_t *qs, ngx_uint_t err)
  245. {
  246.     ngx_connection_t       *pc;
  247.     ngx_quic_frame_t       *frame;
  248.     ngx_quic_connection_t  *qc;

  250.     if (qs->send_state == NGX_QUIC_STREAM_SEND_DATA_RECVD
  251.         || qs->send_state == NGX_QUIC_STREAM_SEND_RESET_SENT
  252.         || qs->send_state == NGX_QUIC_STREAM_SEND_RESET_RECVD)
  253.     {
  254.         return NGX_OK;
  255.     }

  257.     qs->send_state = NGX_QUIC_STREAM_SEND_RESET_SENT;
  258.     qs->send_final_size = qs->send_offset;

  260.     if (qs->connection) {
  261.         qs->connection->write->error = 1;
  262.     }

  264.     pc = qs->parent;
  265.     qc = ngx_quic_get_connection(pc);

  267.     ngx_log_debug1(NGX_LOG_DEBUG_EVENT, pc->log, 0,
  268.                    "quic stream id:0x%xL reset", qs->id);

  270.     frame = ngx_quic_alloc_frame(pc);
  271.     if (frame == NULL) {
  272.         return NGX_ERROR;
  273.     }

  275.     frame->level = ssl_encryption_application;
  276.     frame->type = NGX_QUIC_FT_RESET_STREAM;
  277.     frame->u.reset_stream.id = qs->id;
  278.     frame->u.reset_stream.error_code = err;
  279.     frame->u.reset_stream.final_size = qs->send_offset;

  281.     ngx_quic_queue_frame(qc, frame);

  283.     ngx_quic_free_buffer(pc, &qs->send);

  285.     return NGX_OK;
  286. }


  289. ngx_int_t
  290. ‌ngx_quic_shutdown_stream(ngx_connection_t *c, int how)
  291. {
  292.     if (how == NGX_RDWR_SHUTDOWN || how == NGX_WRITE_SHUTDOWN) {
  293.         if (ngx_quic_shutdown_stream_send(c) != NGX_OK) {
  294.             return NGX_ERROR;
  295.         }
  296.     }

  298.     if (how == NGX_RDWR_SHUTDOWN || how == NGX_READ_SHUTDOWN) {
  299.         if (ngx_quic_shutdown_stream_recv(c) != NGX_OK) {
  300.             return NGX_ERROR;
  301.         }
  302.     }

  304.     return NGX_OK;
  305. }


  308. static ngx_int_t
  309. ‌ngx_quic_shutdown_stream_send(ngx_connection_t *c)
  310. {
  311.     ngx_quic_stream_t  *qs;

  313.     qs = c->quic;

  315.     if (qs->send_state != NGX_QUIC_STREAM_SEND_READY
  316.         && qs->send_state != NGX_QUIC_STREAM_SEND_SEND)
  317.     {
  318.         return NGX_OK;
  319.     }

  321.     qs->send_state = NGX_QUIC_STREAM_SEND_SEND;
  322.     qs->send_final_size = c->sent;

  324.     ngx_log_debug1(NGX_LOG_DEBUG_EVENT, qs->parent->log, 0,
  325.                    "quic stream id:0x%xL send shutdown", qs->id);

  327.     return ngx_quic_stream_flush(qs);
  328. }


  331. static ngx_int_t
  332. ‌ngx_quic_shutdown_stream_recv(ngx_connection_t *c)
  333. {
  334.     ngx_connection_t       *pc;
  335.     ngx_quic_frame_t       *frame;
  336.     ngx_quic_stream_t      *qs;
  337.     ngx_quic_connection_t  *qc;

  339.     qs = c->quic;

  341.     if (qs->recv_state != NGX_QUIC_STREAM_RECV_RECV
  342.         && qs->recv_state != NGX_QUIC_STREAM_RECV_SIZE_KNOWN)
  343.     {
  344.         return NGX_OK;
  345.     }

  347.     pc = qs->parent;
  348.     qc = ngx_quic_get_connection(pc);

  350.     if (qc->conf->stream_close_code == 0) {
  351.         return NGX_OK;
  352.     }

  354.     frame = ngx_quic_alloc_frame(pc);
  355.     if (frame == NULL) {
  356.         return NGX_ERROR;
  357.     }

  359.     ngx_log_debug1(NGX_LOG_DEBUG_EVENT, pc->log, 0,
  360.                    "quic stream id:0x%xL recv shutdown", qs->id);

  362.     frame->level = ssl_encryption_application;
  363.     frame->type = NGX_QUIC_FT_STOP_SENDING;
  364.     frame->u.stop_sending.id = qs->id;
  365.     frame->u.stop_sending.error_code = qc->conf->stream_close_code;

  367.     ngx_quic_queue_frame(qc, frame);

  369.     return NGX_OK;
  370. }


  373. static ngx_quic_stream_t *
  374. ‌ngx_quic_get_stream(ngx_connection_t *c, uint64_t id)
  375. {
  376.     uint64_t                min_id;
  377.     ngx_event_t            *rev;
  378.     ngx_quic_stream_t      *qs;
  379.     ngx_quic_connection_t  *qc;

  381.     qc = ngx_quic_get_connection(c);

  383.     qs = ngx_quic_find_stream(&qc->streams.tree, id);

  385.     if (qs) {
  386.         return qs;
  387.     }

  389.     if (qc->shutdown || qc->closing) {
  390.         return NGX_QUIC_STREAM_GONE;
  391.     }

  393.     ngx_log_debug1(NGX_LOG_DEBUG_EVENT, c->log, 0,
  394.                    "quic stream id:0x%xL is missing", id);

  396.     if (id & NGX_QUIC_STREAM_UNIDIRECTIONAL) {

  398.         if (id & NGX_QUIC_STREAM_SERVER_INITIATED) {
  399.             if ((id >> 2) < qc->streams.server_streams_uni) {
  400.                 return NGX_QUIC_STREAM_GONE;
  401.             }

  403.             qc->error = NGX_QUIC_ERR_STREAM_STATE_ERROR;
  404.             return NULL;
  405.         }

  407.         if ((id >> 2) < qc->streams.client_streams_uni) {
  408.             return NGX_QUIC_STREAM_GONE;
  409.         }

  411.         if ((id >> 2) >= qc->streams.client_max_streams_uni) {
  412.             qc->error = NGX_QUIC_ERR_STREAM_LIMIT_ERROR;
  413.             return NULL;
  414.         }

  416.         min_id = (qc->streams.client_streams_uni << 2)
  417.                  | NGX_QUIC_STREAM_UNIDIRECTIONAL;
  418.         qc->streams.client_streams_uni = (id >> 2) + 1;

  420.     } else {

  422.         if (id & NGX_QUIC_STREAM_SERVER_INITIATED) {
  423.             if ((id >> 2) < qc->streams.server_streams_bidi) {
  424.                 return NGX_QUIC_STREAM_GONE;
  425.             }

  427.             qc->error = NGX_QUIC_ERR_STREAM_STATE_ERROR;
  428.             return NULL;
  429.         }

  431.         if ((id >> 2) < qc->streams.client_streams_bidi) {
  432.             return NGX_QUIC_STREAM_GONE;
  433.         }

  435.         if ((id >> 2) >= qc->streams.client_max_streams_bidi) {
  436.             qc->error = NGX_QUIC_ERR_STREAM_LIMIT_ERROR;
  437.             return NULL;
  438.         }

  440.         min_id = (qc->streams.client_streams_bidi << 2);
  441.         qc->streams.client_streams_bidi = (id >> 2) + 1;
  442.     }

  444.     /*
  445.      * RFC 9000, 2.1.  Stream Types and Identifiers
  446.      *
  447.      * successive streams of each type are created with numerically increasing
  448.      * stream IDs.  A stream ID that is used out of order results in all
  449.      * streams of that type with lower-numbered stream IDs also being opened.
  450.      */

  452. #if (NGX_SUPPRESS_WARN)
  453.     qs = NULL;
  454. #endif

  456.     for ( /* void */ ; min_id <= id; min_id += 0x04) {

  458.         qs = ngx_quic_create_stream(c, min_id);

  460.         if (qs == NULL) {
  461.             if (ngx_quic_reject_stream(c, min_id) != NGX_OK) {
  462.                 return NULL;
  463.             }

  465.             continue;
  466.         }

  468.         ngx_queue_insert_tail(&qc->streams.uninitialized, &qs->queue);

  470.         rev = qs->connection->read;
  471.         rev->handler = ngx_quic_init_stream_handler;

  473.         if (qc->streams.initialized) {
  474.             ngx_post_event(rev, &ngx_posted_events);

  476.             if (qc->push.posted) {
  477.                 /*
  478.                  * The posted stream can produce output immediately.
  479.                  * By postponing the push event, we coalesce the stream
  480.                  * output with queued frames in one UDP datagram.
  481.                  */

  483.                 ngx_delete_posted_event(&qc->push);
  484.                 ngx_post_event(&qc->push, &ngx_posted_events);
  485.             }
  486.         }
  487.     }

  489.     if (qs == NULL) {
  490.         return NGX_QUIC_STREAM_GONE;
  491.     }

  493.     return qs;
  494. }


  497. static ngx_int_t
  498. ‌ngx_quic_reject_stream(ngx_connection_t *c, uint64_t id)
  499. {
  500.     uint64_t                code;
  501.     ngx_quic_frame_t       *frame;
  502.     ngx_quic_connection_t  *qc;

  504.     qc = ngx_quic_get_connection(c);

  506.     code = (id & NGX_QUIC_STREAM_UNIDIRECTIONAL)
  507.            ? qc->conf->stream_reject_code_uni
  508.            : qc->conf->stream_reject_code_bidi;

  510.     if (code == 0) {
  511.         return NGX_DECLINED;
  512.     }

  514.     ngx_log_debug2(NGX_LOG_DEBUG_EVENT, c->log, 0,
  515.                    "quic stream id:0x%xL reject err:0x%xL", id, code);

  517.     frame = ngx_quic_alloc_frame(c);
  518.     if (frame == NULL) {
  519.         return NGX_ERROR;
  520.     }

  522.     frame->level = ssl_encryption_application;
  523.     frame->type = NGX_QUIC_FT_RESET_STREAM;
  524.     frame->u.reset_stream.id = id;
  525.     frame->u.reset_stream.error_code = code;
  526.     frame->u.reset_stream.final_size = 0;

  528.     ngx_quic_queue_frame(qc, frame);

  530.     frame = ngx_quic_alloc_frame(c);
  531.     if (frame == NULL) {
  532.         return NGX_ERROR;
  533.     }

  535.     frame->level = ssl_encryption_application;
  536.     frame->type = NGX_QUIC_FT_STOP_SENDING;
  537.     frame->u.stop_sending.id = id;
  538.     frame->u.stop_sending.error_code = code;

  540.     ngx_quic_queue_frame(qc, frame);

  542.     return NGX_OK;
  543. }


  546. static void
  547. ‌ngx_quic_init_stream_handler(ngx_event_t *ev)
  548. {
  549.     ngx_connection_t   *c;
  550.     ngx_quic_stream_t  *qs;

  552.     c = ev->data;
  553.     qs = c->quic;

  555.     ngx_log_debug0(NGX_LOG_DEBUG_EVENT, c->log, 0, "quic init stream");

  557.     if ((qs->id & NGX_QUIC_STREAM_UNIDIRECTIONAL) == 0) {
  558.         c->write->active = 1;
  559.         c->write->ready = 1;
  560.     }

  562.     c->read->active = 1;

  564.     ngx_queue_remove(&qs->queue);

  566.     c->listening->handler(c);
  567. }


  570. ngx_int_t
  571. ‌ngx_quic_init_streams(ngx_connection_t *c)
  572. {
  573.     ngx_int_t               rc;
  574.     ngx_quic_connection_t  *qc;

  576.     qc = ngx_quic_get_connection(c);

  578.     if (qc->streams.initialized) {
  579.         return NGX_OK;
  580.     }

  582.     rc = ngx_ssl_ocsp_validate(c);

  584.     if (rc == NGX_ERROR) {
  585.         return NGX_ERROR;
  586.     }

  588.     if (rc == NGX_AGAIN) {
  589.         c->ssl->handler = ngx_quic_init_streams_handler;
  590.         return NGX_OK;
  591.     }

  593.     return ngx_quic_do_init_streams(c);
  594. }


  597. static void
  598. ‌ngx_quic_init_streams_handler(ngx_connection_t *c)
  599. {
  600.     if (ngx_quic_do_init_streams(c) != NGX_OK) {
  601.         ngx_quic_close_connection(c, NGX_ERROR);
  602.     }
  603. }


  606. static ngx_int_t
  607. ‌ngx_quic_do_init_streams(ngx_connection_t *c)
  608. {
  609.     ngx_queue_t            *q;
  610.     ngx_quic_stream_t      *qs;
  611.     ngx_quic_connection_t  *qc;

  613.     ngx_log_debug0(NGX_LOG_DEBUG_EVENT, c->log, 0, "quic init streams");

  615.     qc = ngx_quic_get_connection(c);

  617.     if (qc->conf->init) {
  618.         if (qc->conf->init(c) != NGX_OK) {
  619.             return NGX_ERROR;
  620.         }
  621.     }

  623.     for (q = ngx_queue_head(&qc->streams.uninitialized);
  624.          q != ngx_queue_sentinel(&qc->streams.uninitialized);
  625.          q = ngx_queue_next(q))
  626.     {
  627.         qs = ngx_queue_data(q, ngx_quic_stream_t, queue);
  628.         ngx_post_event(qs->connection->read, &ngx_posted_events);
  629.     }

  631.     qc->streams.initialized = 1;

  633.     if (!qc->closing && qc->close.timer_set) {
  634.         ngx_del_timer(&qc->close);
  635.     }

  637.     return NGX_OK;
  638. }


  641. static ngx_quic_stream_t *
  642. ‌ngx_quic_create_stream(ngx_connection_t *c, uint64_t id)
  643. {
  644.     ngx_str_t               addr_text;
  645.     ngx_log_t              *log;
  646.     ngx_pool_t             *pool;
  647.     ngx_uint_t              reusable;
  648.     ngx_queue_t            *q;
  649.     struct sockaddr        *sockaddr;
  650.     ngx_connection_t       *sc;
  651.     ngx_quic_stream_t      *qs;
  652.     ngx_pool_cleanup_t     *cln;
  653.     ngx_quic_connection_t  *qc;

  655.     ngx_log_debug1(NGX_LOG_DEBUG_EVENT, c->log, 0,
  656.                    "quic stream id:0x%xL create", id);

  658.     qc = ngx_quic_get_connection(c);

  660.     if (!ngx_queue_empty(&qc->streams.free)) {
  661.         q = ngx_queue_head(&qc->streams.free);
  662.         qs = ngx_queue_data(q, ngx_quic_stream_t, queue);
  663.         ngx_queue_remove(&qs->queue);

  665.     } else {
  666.         /*
  667.          * the number of streams is limited by transport
  668.          * parameters and application requirements
  669.          */

  671.         qs = ngx_palloc(c->pool, sizeof(ngx_quic_stream_t));
  672.         if (qs == NULL) {
  673.             return NULL;
  674.         }
  675.     }

  677.     ngx_memzero(qs, sizeof(ngx_quic_stream_t));

  679.     qs->node.key = id;
  680.     qs->parent = c;
  681.     qs->id = id;
  682.     qs->send_final_size = (uint64_t) -1;
  683.     qs->recv_final_size = (uint64_t) -1;

  685.     pool = ngx_create_pool(NGX_DEFAULT_POOL_SIZE, c->log);
  686.     if (pool == NULL) {
  687.         ngx_queue_insert_tail(&qc->streams.free, &qs->queue);
  688.         return NULL;
  689.     }

  691.     log = ngx_palloc(pool, sizeof(ngx_log_t));
  692.     if (log == NULL) {
  693.         ngx_destroy_pool(pool);
  694.         ngx_queue_insert_tail(&qc->streams.free, &qs->queue);
  695.         return NULL;
  696.     }

  698.     *log = *c->log;
  699.     pool->log = log;

  701.     sockaddr = ngx_palloc(pool, c->socklen);
  702.     if (sockaddr == NULL) {
  703.         ngx_destroy_pool(pool);
  704.         ngx_queue_insert_tail(&qc->streams.free, &qs->queue);
  705.         return NULL;
  706.     }

  708.     ngx_memcpy(sockaddr, c->sockaddr, c->socklen);

  710.     if (c->addr_text.data) {
  711.         addr_text.data = ngx_pnalloc(pool, c->addr_text.len);
  712.         if (addr_text.data == NULL) {
  713.             ngx_destroy_pool(pool);
  714.             ngx_queue_insert_tail(&qc->streams.free, &qs->queue);
  715.             return NULL;
  716.         }

  718.         ngx_memcpy(addr_text.data, c->addr_text.data, c->addr_text.len);
  719.         addr_text.len = c->addr_text.len;

  721.     } else {
  722.         addr_text.len = 0;
  723.         addr_text.data = NULL;
  724.     }

  726.     reusable = c->reusable;
  727.     ngx_reusable_connection(c, 0);

  729.     sc = ngx_get_connection(c->fd, log);
  730.     if (sc == NULL) {
  731.         ngx_destroy_pool(pool);
  732.         ngx_queue_insert_tail(&qc->streams.free, &qs->queue);
  733.         ngx_reusable_connection(c, reusable);
  734.         return NULL;
  735.     }

  737.     qs->connection = sc;

  739.     sc->quic = qs;
  740.     sc->shared = 1;
  741.     sc->type = SOCK_STREAM;
  742.     sc->pool = pool;
  743.     sc->ssl = c->ssl;
  744.     sc->sockaddr = sockaddr;
  745.     sc->socklen = c->socklen;
  746.     sc->listening = c->listening;
  747.     sc->addr_text = addr_text;
  748.     sc->local_sockaddr = c->local_sockaddr;
  749.     sc->local_socklen = c->local_socklen;
  750.     sc->number = ngx_atomic_fetch_add(ngx_connection_counter, 1);
  751.     sc->start_time = c->start_time;
  752.     sc->tcp_nodelay = NGX_TCP_NODELAY_DISABLED;

  754.     sc->recv = ngx_quic_stream_recv;
  755.     sc->send = ngx_quic_stream_send;
  756.     sc->send_chain = ngx_quic_stream_send_chain;

  758.     sc->read->log = log;
  759.     sc->write->log = log;

  761.     sc->read->handler = ngx_quic_empty_handler;
  762.     sc->write->handler = ngx_quic_empty_handler;

  764.     log->connection = sc->number;

  766.     if (id & NGX_QUIC_STREAM_UNIDIRECTIONAL) {
  767.         if (id & NGX_QUIC_STREAM_SERVER_INITIATED) {
  768.             qs->send_max_data = qc->ctp.initial_max_stream_data_uni;
  769.             qs->recv_state = NGX_QUIC_STREAM_RECV_DATA_READ;
  770.             qs->send_state = NGX_QUIC_STREAM_SEND_READY;

  772.         } else {
  773.             qs->recv_max_data = qc->tp.initial_max_stream_data_uni;
  774.             qs->recv_state = NGX_QUIC_STREAM_RECV_RECV;
  775.             qs->send_state = NGX_QUIC_STREAM_SEND_DATA_RECVD;
  776.         }

  778.     } else {
  779.         if (id & NGX_QUIC_STREAM_SERVER_INITIATED) {
  780.             qs->send_max_data = qc->ctp.initial_max_stream_data_bidi_remote;
  781.             qs->recv_max_data = qc->tp.initial_max_stream_data_bidi_local;

  783.         } else {
  784.             qs->send_max_data = qc->ctp.initial_max_stream_data_bidi_local;
  785.             qs->recv_max_data = qc->tp.initial_max_stream_data_bidi_remote;
  786.         }

  788.         qs->recv_state = NGX_QUIC_STREAM_RECV_RECV;
  789.         qs->send_state = NGX_QUIC_STREAM_SEND_READY;
  790.     }

  792.     qs->recv_window = qs->recv_max_data;

  794.     cln = ngx_pool_cleanup_add(pool, 0);
  795.     if (cln == NULL) {
  796.         ngx_close_connection(sc);
  797.         ngx_destroy_pool(pool);
  798.         ngx_queue_insert_tail(&qc->streams.free, &qs->queue);
  799.         ngx_reusable_connection(c, reusable);
  800.         return NULL;
  801.     }

  803.     cln->handler = ngx_quic_stream_cleanup_handler;
  804.     cln->data = sc;

  806.     ngx_rbtree_insert(&qc->streams.tree, &qs->node);

  808.     return qs;
  809. }


  812. void
  813. ‌ngx_quic_cancelable_stream(ngx_connection_t *c)
  814. {
  815.     ngx_connection_t       *pc;
  816.     ngx_quic_stream_t      *qs;
  817.     ngx_quic_connection_t  *qc;

  819.     qs = c->quic;
  820.     pc = qs->parent;
  821.     qc = ngx_quic_get_connection(pc);

  823.     if (!qs->cancelable) {
  824.         qs->cancelable = 1;

  826.         if (ngx_quic_can_shutdown(pc) == NGX_OK) {
  827.             ngx_reusable_connection(pc, 1);

  829.             if (qc->shutdown) {
  830.                 ngx_quic_shutdown_quic(pc);
  831.             }
  832.         }
  833.     }
  834. }


  837. static void
  838. ‌ngx_quic_empty_handler(ngx_event_t *ev)
  839. {
  840. }


  843. static ssize_t
  844. ‌ngx_quic_stream_recv(ngx_connection_t *c, u_char *buf, size_t size)
  845. {
  846.     ssize_t             len;
  847.     ngx_buf_t          *b;
  848.     ngx_chain_t        *cl, *in;
  849.     ngx_event_t        *rev;
  850.     ngx_connection_t   *pc;
  851.     ngx_quic_stream_t  *qs;

  853.     qs = c->quic;
  854.     pc = qs->parent;
  855.     rev = c->read;

  857.     if (qs->recv_state == NGX_QUIC_STREAM_RECV_RESET_RECVD
  858.         || qs->recv_state == NGX_QUIC_STREAM_RECV_RESET_READ)
  859.     {
  860.         qs->recv_state = NGX_QUIC_STREAM_RECV_RESET_READ;
  861.         return NGX_ERROR;
  862.     }

  864.     ngx_log_debug2(NGX_LOG_DEBUG_EVENT, pc->log, 0,
  865.                    "quic stream id:0x%xL recv buf:%uz", qs->id, size);

  867.     if (size == 0) {
  868.         return 0;
  869.     }

  871.     in = ngx_quic_read_buffer(pc, &qs->recv, size);
  872.     if (in == NGX_CHAIN_ERROR) {
  873.         return NGX_ERROR;
  874.     }

  876.     len = 0;

  878.     for (cl = in; cl; cl = cl->next) {
  879.         b = cl->buf;
  880.         len += b->last - b->pos;
  881.         buf = ngx_cpymem(buf, b->pos, b->last - b->pos);
  882.     }

  884.     ngx_quic_free_chain(pc, in);

  886.     if (len == 0) {
  887.         rev->ready = 0;

  889.         if (qs->recv_state == NGX_QUIC_STREAM_RECV_DATA_RECVD
  890.             && qs->recv_offset == qs->recv_final_size)
  891.         {
  892.             qs->recv_state = NGX_QUIC_STREAM_RECV_DATA_READ;
  893.         }

  895.         if (qs->recv_state == NGX_QUIC_STREAM_RECV_DATA_READ) {
  896.             rev->eof = 1;
  897.             return 0;
  898.         }

  900.         ngx_log_debug1(NGX_LOG_DEBUG_EVENT, c->log, 0,
  901.                        "quic stream id:0x%xL recv() not ready", qs->id);
  902.         return NGX_AGAIN;
  903.     }

  905.     ngx_log_debug2(NGX_LOG_DEBUG_EVENT, c->log, 0,
  906.                    "quic stream id:0x%xL recv len:%z", qs->id, len);

  908.     if (ngx_quic_update_flow(qs, qs->recv_offset + len) != NGX_OK) {
  909.         return NGX_ERROR;
  910.     }

  912.     return len;
  913. }


  916. static ssize_t
  917. ‌ngx_quic_stream_send(ngx_connection_t *c, u_char *buf, size_t size)
  918. {
  919.     ngx_buf_t    b;
  920.     ngx_chain_t  cl;

  922.     ngx_memzero(&b, sizeof(ngx_buf_t));

  924.     b.memory = 1;
  925.     b.pos = buf;
  926.     b.last = buf + size;

  928.     cl.buf = &b;
  929.     cl.next = NULL;

  931.     if (ngx_quic_stream_send_chain(c, &cl, 0) == NGX_CHAIN_ERROR) {
  932.         return NGX_ERROR;
  933.     }

  935.     if (b.pos == buf) {
  936.         return NGX_AGAIN;
  937.     }

  939.     return b.pos - buf;
  940. }


  943. static ngx_chain_t *
  944. ‌ngx_quic_stream_send_chain(ngx_connection_t *c, ngx_chain_t *in, off_t limit)
  945. {
  946.     uint64_t                n, flow;
  947.     ngx_event_t            *wev;
  948.     ngx_connection_t       *pc;
  949.     ngx_quic_stream_t      *qs;
  950.     ngx_quic_connection_t  *qc;

  952.     qs = c->quic;
  953.     pc = qs->parent;
  954.     qc = ngx_quic_get_connection(pc);
  955.     wev = c->write;

  957.     if (qs->send_state != NGX_QUIC_STREAM_SEND_READY
  958.         && qs->send_state != NGX_QUIC_STREAM_SEND_SEND)
  959.     {
  960.         wev->error = 1;
  961.         return NGX_CHAIN_ERROR;
  962.     }

  964.     qs->send_state = NGX_QUIC_STREAM_SEND_SEND;

  966.     flow = qs->acked + qc->conf->stream_buffer_size - qs->sent;

  968.     if (flow == 0) {
  969.         wev->ready = 0;
  970.         return in;
  971.     }

  973.     if (limit == 0 || limit > (off_t) flow) {
  974.         limit = flow;
  975.     }

  977.     n = qs->send.size;

  979.     in = ngx_quic_write_buffer(pc, &qs->send, in, limit, qs->sent);
  980.     if (in == NGX_CHAIN_ERROR) {
  981.         return NGX_CHAIN_ERROR;
  982.     }

  984.     n = qs->send.size - n;
  985.     c->sent += n;
  986.     qs->sent += n;
  987.     qc->streams.sent += n;

  989.     if (flow == n) {
  990.         wev->ready = 0;
  991.     }

  993.     ngx_log_debug1(NGX_LOG_DEBUG_EVENT, c->log, 0,
  994.                    "quic send_chain sent:%uL", n);

  996.     if (ngx_quic_stream_flush(qs) != NGX_OK) {
  997.         return NGX_CHAIN_ERROR;
  998.     }

  1000.     return in;
  1001. }


  1004. static ngx_int_t
  1005. ‌ngx_quic_stream_flush(ngx_quic_stream_t *qs)
  1006. {
  1007.     off_t                   limit, len;
  1008.     ngx_uint_t              last;
  1009.     ngx_chain_t            *out;
  1010.     ngx_quic_frame_t       *frame;
  1011.     ngx_connection_t       *pc;
  1012.     ngx_quic_connection_t  *qc;

  1014.     if (qs->send_state != NGX_QUIC_STREAM_SEND_SEND) {
  1015.         return NGX_OK;
  1016.     }

  1018.     pc = qs->parent;
  1019.     qc = ngx_quic_get_connection(pc);

  1021.     if (qc->streams.send_max_data == 0) {
  1022.         qc->streams.send_max_data = qc->ctp.initial_max_data;
  1023.     }

  1025.     limit = ngx_min(qc->streams.send_max_data - qc->streams.send_offset,
  1026.                     qs->send_max_data - qs->send_offset);

  1028.     ngx_log_debug2(NGX_LOG_DEBUG_EVENT, pc->log, 0,
  1029.                    "quic stream id:0x%xL flush limit:%O", qs->id, limit);

  1031.     len = qs->send.offset;

  1033.     out = ngx_quic_read_buffer(pc, &qs->send, limit);
  1034.     if (out == NGX_CHAIN_ERROR) {
  1035.         return NGX_ERROR;
  1036.     }

  1038.     len = qs->send.offset - len;
  1039.     last = 0;

  1041.     if (qs->send_final_size != (uint64_t) -1
  1042.         && qs->send_final_size == qs->send.offset)
  1043.     {
  1044.         qs->send_state = NGX_QUIC_STREAM_SEND_DATA_SENT;
  1045.         last = 1;
  1046.     }

  1048.     if (len == 0 && !last) {
  1049.         return NGX_OK;
  1050.     }

  1052.     frame = ngx_quic_alloc_frame(pc);
  1053.     if (frame == NULL) {
  1054.         return NGX_ERROR;
  1055.     }

  1057.     frame->level = ssl_encryption_application;
  1058.     frame->type = NGX_QUIC_FT_STREAM;
  1059.     frame->data = out;

  1061.     frame->u.stream.off = 1;
  1062.     frame->u.stream.len = 1;
  1063.     frame->u.stream.fin = last;

  1065.     frame->u.stream.stream_id = qs->id;
  1066.     frame->u.stream.offset = qs->send_offset;
  1067.     frame->u.stream.length = len;

  1069.     ngx_quic_queue_frame(qc, frame);

  1071.     qs->send_offset += len;
  1072.     qc->streams.send_offset += len;

  1074.     ngx_log_debug3(NGX_LOG_DEBUG_EVENT, pc->log, 0,
  1075.                    "quic stream id:0x%xL flush len:%O last:%ui",
  1076.                    qs->id, len, last);

  1078.     if (qs->connection == NULL) {
  1079.         return ngx_quic_close_stream(qs);
  1080.     }

  1082.     return NGX_OK;
  1083. }


  1086. static void
  1087. ‌ngx_quic_stream_cleanup_handler(void *data)
  1088. {
  1089.     ngx_connection_t *c = data;

  1091.     ngx_quic_stream_t      *qs;
  1092.     ngx_quic_connection_t  *qc;

  1094.     qs = c->quic;

  1096.     ngx_log_debug1(NGX_LOG_DEBUG_EVENT, qs->parent->log, 0,
  1097.                    "quic stream id:0x%xL cleanup", qs->id);

  1099.     if (ngx_quic_shutdown_stream(c, NGX_RDWR_SHUTDOWN) != NGX_OK) {
  1100.         qs->connection = NULL;
  1101.         goto failed;
  1102.     }

  1104.     qs->connection = NULL;

  1106.     if (ngx_quic_close_stream(qs) != NGX_OK) {
  1107.         goto failed;
  1108.     }

  1110.     return;

  1112. failed:

  1114.     qc = ngx_quic_get_connection(qs->parent);
  1115.     qc->error = NGX_QUIC_ERR_INTERNAL_ERROR;

  1117.     ngx_post_event(&qc->close, &ngx_posted_events);
  1118. }


  1121. static ngx_int_t
  1122. ‌ngx_quic_close_stream(ngx_quic_stream_t *qs)
  1123. {
  1124.     ngx_connection_t       *pc;
  1125.     ngx_quic_frame_t       *frame;
  1126.     ngx_quic_connection_t  *qc;

  1128.     pc = qs->parent;
  1129.     qc = ngx_quic_get_connection(pc);

  1131.     if (!qc->closing) {
  1132.         /* make sure everything is sent and final size is received */

  1134.         if (qs->recv_state == NGX_QUIC_STREAM_RECV_RECV) {
  1135.             return NGX_OK;
  1136.         }

  1138.         if (qs->send_state != NGX_QUIC_STREAM_SEND_DATA_RECVD
  1139.             && qs->send_state != NGX_QUIC_STREAM_SEND_RESET_RECVD)
  1140.         {
  1141.             return NGX_OK;
  1142.         }
  1143.     }

  1145.     ngx_log_debug1(NGX_LOG_DEBUG_EVENT, pc->log, 0,
  1146.                    "quic stream id:0x%xL close", qs->id);

  1148.     ngx_quic_free_buffer(pc, &qs->send);
  1149.     ngx_quic_free_buffer(pc, &qs->recv);

  1151.     ngx_rbtree_delete(&qc->streams.tree, &qs->node);
  1152.     ngx_queue_insert_tail(&qc->streams.free, &qs->queue);

  1154.     if (qc->closing) {
  1155.         /* schedule handler call to continue ngx_quic_close_connection() */
  1156.         ngx_post_event(&qc->close, &ngx_posted_events);
  1157.         return NGX_OK;
  1158.     }

  1160.     if (!pc->reusable && ngx_quic_can_shutdown(pc) == NGX_OK) {
  1161.         ngx_reusable_connection(pc, 1);
  1162.     }

  1164.     if (qc->shutdown) {
  1165.         ngx_quic_shutdown_quic(pc);
  1166.         return NGX_OK;
  1167.     }

  1169.     if ((qs->id & NGX_QUIC_STREAM_SERVER_INITIATED) == 0) {
  1170.         frame = ngx_quic_alloc_frame(pc);
  1171.         if (frame == NULL) {
  1172.             return NGX_ERROR;
  1173.         }

  1175.         frame->level = ssl_encryption_application;
  1176.         frame->type = NGX_QUIC_FT_MAX_STREAMS;

  1178.         if (qs->id & NGX_QUIC_STREAM_UNIDIRECTIONAL) {
  1179.             frame->u.max_streams.limit = ++qc->streams.client_max_streams_uni;
  1180.             frame->u.max_streams.bidi = 0;

  1182.         } else {
  1183.             frame->u.max_streams.limit = ++qc->streams.client_max_streams_bidi;
  1184.             frame->u.max_streams.bidi = 1;
  1185.         }

  1187.         ngx_quic_queue_frame(qc, frame);
  1188.     }

  1190.     return NGX_OK;
  1191. }


  1194. static ngx_int_t
  1195. ‌ngx_quic_can_shutdown(ngx_connection_t *c)
  1196. {
  1197.     ngx_rbtree_t           *tree;
  1198.     ngx_rbtree_node_t      *node;
  1199.     ngx_quic_stream_t      *qs;
  1200.     ngx_quic_connection_t  *qc;

  1202.     qc = ngx_quic_get_connection(c);

  1204.     tree = &qc->streams.tree;

  1206.     if (tree->root != tree->sentinel) {
  1207.         for (node = ngx_rbtree_min(tree->root, tree->sentinel);
  1208.              node;
  1209.              node = ngx_rbtree_next(tree, node))
  1210.         {
  1211.             qs = (ngx_quic_stream_t *) node;

  1213.             if (!qs->cancelable) {
  1214.                 return NGX_DECLINED;
  1215.             }
  1216.         }
  1217.     }

  1219.     return NGX_OK;
  1220. }


  1223. ngx_int_t
  1224. ‌ngx_quic_handle_stream_frame(ngx_connection_t *c, ngx_quic_header_t *pkt,
  1225.     ngx_quic_frame_t *frame)
  1226. {
  1227.     uint64_t                  last;
  1228.     ngx_quic_stream_t        *qs;
  1229.     ngx_quic_connection_t    *qc;
  1230.     ngx_quic_stream_frame_t  *f;

  1232.     qc = ngx_quic_get_connection(c);
  1233.     f = &frame->u.stream;

  1235.     if ((f->stream_id & NGX_QUIC_STREAM_UNIDIRECTIONAL)
  1236.         && (f->stream_id & NGX_QUIC_STREAM_SERVER_INITIATED))
  1237.     {
  1238.         qc->error = NGX_QUIC_ERR_STREAM_STATE_ERROR;
  1239.         return NGX_ERROR;
  1240.     }

  1242.     /* no overflow since both values are 62-bit */
  1243.     last = f->offset + f->length;

  1245.     qs = ngx_quic_get_stream(c, f->stream_id);

  1247.     if (qs == NULL) {
  1248.         return NGX_ERROR;
  1249.     }

  1251.     if (qs == NGX_QUIC_STREAM_GONE) {
  1252.         return NGX_OK;
  1253.     }

  1255.     if (qs->recv_state != NGX_QUIC_STREAM_RECV_RECV
  1256.         && qs->recv_state != NGX_QUIC_STREAM_RECV_SIZE_KNOWN)
  1257.     {
  1258.         return NGX_OK;
  1259.     }

  1261.     if (ngx_quic_control_flow(qs, last) != NGX_OK) {
  1262.         return NGX_ERROR;
  1263.     }

  1265.     if (qs->recv_final_size != (uint64_t) -1 && last > qs->recv_final_size) {
  1266.         qc->error = NGX_QUIC_ERR_FINAL_SIZE_ERROR;
  1267.         return NGX_ERROR;
  1268.     }

  1270.     if (last < qs->recv_offset) {
  1271.         return NGX_OK;
  1272.     }

  1274.     if (f->fin) {
  1275.         if (qs->recv_final_size != (uint64_t) -1 && qs->recv_final_size != last)
  1276.         {
  1277.             qc->error = NGX_QUIC_ERR_FINAL_SIZE_ERROR;
  1278.             return NGX_ERROR;
  1279.         }

  1281.         if (qs->recv_last > last) {
  1282.             qc->error = NGX_QUIC_ERR_FINAL_SIZE_ERROR;
  1283.             return NGX_ERROR;
  1284.         }

  1286.         qs->recv_final_size = last;
  1287.         qs->recv_state = NGX_QUIC_STREAM_RECV_SIZE_KNOWN;
  1288.     }

  1290.     if (ngx_quic_write_buffer(c, &qs->recv, frame->data, f->length, f->offset)
  1291.         == NGX_CHAIN_ERROR)
  1292.     {
  1293.         return NGX_ERROR;
  1294.     }

  1296.     if (qs->recv_state == NGX_QUIC_STREAM_RECV_SIZE_KNOWN
  1297.         && qs->recv.size == qs->recv_final_size)
  1298.     {
  1299.         qs->recv_state = NGX_QUIC_STREAM_RECV_DATA_RECVD;
  1300.     }

  1302.     if (qs->connection == NULL) {
  1303.         return ngx_quic_close_stream(qs);
  1304.     }

  1306.     if (f->offset <= qs->recv_offset) {
  1307.         ngx_quic_set_event(qs->connection->read);
  1308.     }

  1310.     return NGX_OK;
  1311. }


  1314. ngx_int_t
  1315. ‌ngx_quic_handle_max_data_frame(ngx_connection_t *c,
  1316.     ngx_quic_max_data_frame_t *f)
  1317. {
  1318.     ngx_rbtree_t           *tree;
  1319.     ngx_rbtree_node_t      *node;
  1320.     ngx_quic_stream_t      *qs;
  1321.     ngx_quic_connection_t  *qc;

  1323.     qc = ngx_quic_get_connection(c);
  1324.     tree = &qc->streams.tree;

  1326.     if (f->max_data <= qc->streams.send_max_data) {
  1327.         return NGX_OK;
  1328.     }

  1330.     if (tree->root == tree->sentinel
  1331.         || qc->streams.send_offset < qc->streams.send_max_data)
  1332.     {
  1333.         /* not blocked on MAX_DATA */
  1334.         qc->streams.send_max_data = f->max_data;
  1335.         return NGX_OK;
  1336.     }

  1338.     qc->streams.send_max_data = f->max_data;
  1339.     node = ngx_rbtree_min(tree->root, tree->sentinel);

  1341.     while (node && qc->streams.send_offset < qc->streams.send_max_data) {

  1343.         qs = (ngx_quic_stream_t *) node;
  1344.         node = ngx_rbtree_next(tree, node);

  1346.         if (ngx_quic_stream_flush(qs) != NGX_OK) {
  1347.             return NGX_ERROR;
  1348.         }
  1349.     }

  1351.     return NGX_OK;
  1352. }


  1355. ngx_int_t
  1356. ‌ngx_quic_handle_streams_blocked_frame(ngx_connection_t *c,
  1357.     ngx_quic_header_t *pkt, ngx_quic_streams_blocked_frame_t *f)
  1358. {
  1359.     return NGX_OK;
  1360. }


  1363. ngx_int_t
  1364. ‌ngx_quic_handle_data_blocked_frame(ngx_connection_t *c,
  1365.     ngx_quic_header_t *pkt, ngx_quic_data_blocked_frame_t *f)
  1366. {
  1367.     return ngx_quic_update_max_data(c);
  1368. }


  1371. ngx_int_t
  1372. ‌ngx_quic_handle_stream_data_blocked_frame(ngx_connection_t *c,
  1373.     ngx_quic_header_t *pkt, ngx_quic_stream_data_blocked_frame_t *f)
  1374. {
  1375.     ngx_quic_stream_t      *qs;
  1376.     ngx_quic_connection_t  *qc;

  1378.     qc = ngx_quic_get_connection(c);

  1380.     if ((f->id & NGX_QUIC_STREAM_UNIDIRECTIONAL)
  1381.         && (f->id & NGX_QUIC_STREAM_SERVER_INITIATED))
  1382.     {
  1383.         qc->error = NGX_QUIC_ERR_STREAM_STATE_ERROR;
  1384.         return NGX_ERROR;
  1385.     }

  1387.     qs = ngx_quic_get_stream(c, f->id);

  1389.     if (qs == NULL) {
  1390.         return NGX_ERROR;
  1391.     }

  1393.     if (qs == NGX_QUIC_STREAM_GONE) {
  1394.         return NGX_OK;
  1395.     }

  1397.     return ngx_quic_update_max_stream_data(qs);
  1398. }


  1401. ngx_int_t
  1402. ‌ngx_quic_handle_max_stream_data_frame(ngx_connection_t *c,
  1403.     ngx_quic_header_t *pkt, ngx_quic_max_stream_data_frame_t *f)
  1404. {
  1405.     ngx_quic_stream_t      *qs;
  1406.     ngx_quic_connection_t  *qc;

  1408.     qc = ngx_quic_get_connection(c);

  1410.     if ((f->id & NGX_QUIC_STREAM_UNIDIRECTIONAL)
  1411.         && (f->id & NGX_QUIC_STREAM_SERVER_INITIATED) == 0)
  1412.     {
  1413.         qc->error = NGX_QUIC_ERR_STREAM_STATE_ERROR;
  1414.         return NGX_ERROR;
  1415.     }

  1417.     qs = ngx_quic_get_stream(c, f->id);

  1419.     if (qs == NULL) {
  1420.         return NGX_ERROR;
  1421.     }

  1423.     if (qs == NGX_QUIC_STREAM_GONE) {
  1424.         return NGX_OK;
  1425.     }

  1427.     if (f->limit <= qs->send_max_data) {
  1428.         return NGX_OK;
  1429.     }

  1431.     if (qs->send_offset < qs->send_max_data) {
  1432.         /* not blocked on MAX_STREAM_DATA */
  1433.         qs->send_max_data = f->limit;
  1434.         return NGX_OK;
  1435.     }

  1437.     qs->send_max_data = f->limit;

  1439.     return ngx_quic_stream_flush(qs);
  1440. }


  1443. ngx_int_t
  1444. ‌ngx_quic_handle_reset_stream_frame(ngx_connection_t *c,
  1445.     ngx_quic_header_t *pkt, ngx_quic_reset_stream_frame_t *f)
  1446. {
  1447.     ngx_event_t            *rev;
  1448.     ngx_quic_stream_t      *qs;
  1449.     ngx_quic_connection_t  *qc;

  1451.     qc = ngx_quic_get_connection(c);

  1453.     if ((f->id & NGX_QUIC_STREAM_UNIDIRECTIONAL)
  1454.         && (f->id & NGX_QUIC_STREAM_SERVER_INITIATED))
  1455.     {
  1456.         qc->error = NGX_QUIC_ERR_STREAM_STATE_ERROR;
  1457.         return NGX_ERROR;
  1458.     }

  1460.     qs = ngx_quic_get_stream(c, f->id);

  1462.     if (qs == NULL) {
  1463.         return NGX_ERROR;
  1464.     }

  1466.     if (qs == NGX_QUIC_STREAM_GONE) {
  1467.         return NGX_OK;
  1468.     }

  1470.     if (qs->recv_state == NGX_QUIC_STREAM_RECV_RESET_RECVD
  1471.         || qs->recv_state == NGX_QUIC_STREAM_RECV_RESET_READ)
  1472.     {
  1473.         return NGX_OK;
  1474.     }

  1476.     qs->recv_state = NGX_QUIC_STREAM_RECV_RESET_RECVD;

  1478.     if (ngx_quic_control_flow(qs, f->final_size) != NGX_OK) {
  1479.         return NGX_ERROR;
  1480.     }

  1482.     if (qs->recv_final_size != (uint64_t) -1
  1483.         && qs->recv_final_size != f->final_size)
  1484.     {
  1485.         qc->error = NGX_QUIC_ERR_FINAL_SIZE_ERROR;
  1486.         return NGX_ERROR;
  1487.     }

  1489.     if (qs->recv_last > f->final_size) {
  1490.         qc->error = NGX_QUIC_ERR_FINAL_SIZE_ERROR;
  1491.         return NGX_ERROR;
  1492.     }

  1494.     qs->recv_final_size = f->final_size;

  1496.     if (ngx_quic_update_flow(qs, qs->recv_final_size) != NGX_OK) {
  1497.         return NGX_ERROR;
  1498.     }

  1500.     if (qs->connection == NULL) {
  1501.         return ngx_quic_close_stream(qs);
  1502.     }

  1504.     rev = qs->connection->read;
  1505.     rev->error = 1;

  1507.     ngx_quic_set_event(rev);

  1509.     return NGX_OK;
  1510. }


  1513. ngx_int_t
  1514. ‌ngx_quic_handle_stop_sending_frame(ngx_connection_t *c,
  1515.     ngx_quic_header_t *pkt, ngx_quic_stop_sending_frame_t *f)
  1516. {
  1517.     ngx_quic_stream_t      *qs;
  1518.     ngx_quic_connection_t  *qc;

  1520.     qc = ngx_quic_get_connection(c);

  1522.     if ((f->id & NGX_QUIC_STREAM_UNIDIRECTIONAL)
  1523.         && (f->id & NGX_QUIC_STREAM_SERVER_INITIATED) == 0)
  1524.     {
  1525.         qc->error = NGX_QUIC_ERR_STREAM_STATE_ERROR;
  1526.         return NGX_ERROR;
  1527.     }

  1529.     qs = ngx_quic_get_stream(c, f->id);

  1531.     if (qs == NULL) {
  1532.         return NGX_ERROR;
  1533.     }

  1535.     if (qs == NGX_QUIC_STREAM_GONE) {
  1536.         return NGX_OK;
  1537.     }

  1539.     if (ngx_quic_do_reset_stream(qs, f->error_code) != NGX_OK) {
  1540.         return NGX_ERROR;
  1541.     }

  1543.     if (qs->connection == NULL) {
  1544.         return ngx_quic_close_stream(qs);
  1545.     }

  1547.     ngx_quic_set_event(qs->connection->write);

  1549.     return NGX_OK;
  1550. }


  1553. ngx_int_t
  1554. ‌ngx_quic_handle_max_streams_frame(ngx_connection_t *c,
  1555.     ngx_quic_header_t *pkt, ngx_quic_max_streams_frame_t *f)
  1556. {
  1557.     ngx_quic_connection_t  *qc;

  1559.     qc = ngx_quic_get_connection(c);

  1561.     if (f->bidi) {
  1562.         if (qc->streams.server_max_streams_bidi < f->limit) {
  1563.             qc->streams.server_max_streams_bidi = f->limit;

  1565.             ngx_log_debug1(NGX_LOG_DEBUG_EVENT, c->log, 0,
  1566.                            "quic max_streams_bidi:%uL", f->limit);
  1567.         }

  1569.     } else {
  1570.         if (qc->streams.server_max_streams_uni < f->limit) {
  1571.             qc->streams.server_max_streams_uni = f->limit;

  1573.             ngx_log_debug1(NGX_LOG_DEBUG_EVENT, c->log, 0,
  1574.                            "quic max_streams_uni:%uL", f->limit);
  1575.         }
  1576.     }

  1578.     return NGX_OK;
  1579. }


  1582. void
  1583. ‌ngx_quic_handle_stream_ack(ngx_connection_t *c, ngx_quic_frame_t *f)
  1584. {
  1585.     uint64_t                acked;
  1586.     ngx_quic_stream_t      *qs;
  1587.     ngx_quic_connection_t  *qc;

  1589.     qc = ngx_quic_get_connection(c);

  1591.     switch (f->type) {

  1593.     case NGX_QUIC_FT_RESET_STREAM:

  1595.         qs = ngx_quic_find_stream(&qc->streams.tree, f->u.reset_stream.id);
  1596.         if (qs == NULL) {
  1597.             return;
  1598.         }

  1600.         qs->send_state = NGX_QUIC_STREAM_SEND_RESET_RECVD;

  1602.         ngx_log_debug2(NGX_LOG_DEBUG_EVENT, c->log, 0,
  1603.                        "quic stream id:0x%xL ack reset final_size:%uL",
  1604.                        qs->id, f->u.reset_stream.final_size);

  1606.         break;

  1608.     case NGX_QUIC_FT_STREAM:

  1610.         qs = ngx_quic_find_stream(&qc->streams.tree, f->u.stream.stream_id);
  1611.         if (qs == NULL) {
  1612.             return;
  1613.         }

  1615.         acked = qs->acked;
  1616.         qs->acked += f->u.stream.length;

  1618.         if (f->u.stream.fin) {
  1619.             qs->fin_acked = 1;
  1620.         }

  1622.         if (qs->send_state == NGX_QUIC_STREAM_SEND_DATA_SENT
  1623.             && qs->acked == qs->sent && qs->fin_acked)
  1624.         {
  1625.             qs->send_state = NGX_QUIC_STREAM_SEND_DATA_RECVD;
  1626.         }

  1628.         ngx_log_debug4(NGX_LOG_DEBUG_EVENT, c->log, 0,
  1629.                        "quic stream id:0x%xL ack len:%uL fin:%d unacked:%uL",
  1630.                        qs->id, f->u.stream.length, f->u.stream.fin,
  1631.                        qs->sent - qs->acked);

  1633.         if (qs->connection
  1634.             && qs->sent - acked == qc->conf->stream_buffer_size
  1635.             && f->u.stream.length > 0)
  1636.         {
  1637.             ngx_quic_set_event(qs->connection->write);
  1638.         }

  1640.         break;

  1642.     default:
  1643.         return;
  1644.     }

  1646.     if (qs->connection == NULL) {
  1647.         ngx_quic_close_stream(qs);
  1648.     }
  1649. }


  1652. static ngx_int_t
  1653. ‌ngx_quic_control_flow(ngx_quic_stream_t *qs, uint64_t last)
  1654. {
  1655.     uint64_t                len;
  1656.     ngx_connection_t       *pc;
  1657.     ngx_quic_connection_t  *qc;

  1659.     pc = qs->parent;
  1660.     qc = ngx_quic_get_connection(pc);

  1662.     if (last <= qs->recv_last) {
  1663.         return NGX_OK;
  1664.     }

  1666.     len = last - qs->recv_last;

  1668.     ngx_log_debug5(NGX_LOG_DEBUG_EVENT, pc->log, 0,
  1669.                    "quic stream id:0x%xL flow control msd:%uL/%uL md:%uL/%uL",
  1670.                    qs->id, last, qs->recv_max_data, qc->streams.recv_last + len,
  1671.                    qc->streams.recv_max_data);

  1673.     qs->recv_last += len;

  1675.     if (qs->recv_state == NGX_QUIC_STREAM_RECV_RECV
  1676.         && qs->recv_last > qs->recv_max_data)
  1677.     {
  1678.         qc->error = NGX_QUIC_ERR_FLOW_CONTROL_ERROR;
  1679.         return NGX_ERROR;
  1680.     }

  1682.     qc->streams.recv_last += len;

  1684.     if (qc->streams.recv_last > qc->streams.recv_max_data) {
  1685.         qc->error = NGX_QUIC_ERR_FLOW_CONTROL_ERROR;
  1686.         return NGX_ERROR;
  1687.     }

  1689.     return NGX_OK;
  1690. }


  1693. static ngx_int_t
  1694. ‌ngx_quic_update_flow(ngx_quic_stream_t *qs, uint64_t last)
  1695. {
  1696.     uint64_t                len;
  1697.     ngx_connection_t       *pc;
  1698.     ngx_quic_connection_t  *qc;

  1700.     pc = qs->parent;
  1701.     qc = ngx_quic_get_connection(pc);

  1703.     if (last <= qs->recv_offset) {
  1704.         return NGX_OK;
  1705.     }

  1707.     len = last - qs->recv_offset;

  1709.     ngx_log_debug2(NGX_LOG_DEBUG_EVENT, pc->log, 0,
  1710.                    "quic stream id:0x%xL flow update %uL", qs->id, last);

  1712.     qs->recv_offset += len;

  1714.     if (qs->recv_max_data <= qs->recv_offset + qs->recv_window / 2) {
  1715.         if (ngx_quic_update_max_stream_data(qs) != NGX_OK) {
  1716.             return NGX_ERROR;
  1717.         }
  1718.     }

  1720.     qc->streams.recv_offset += len;

  1722.     if (qc->streams.recv_max_data
  1723.         <= qc->streams.recv_offset + qc->streams.recv_window / 2)
  1724.     {
  1725.         if (ngx_quic_update_max_data(pc) != NGX_OK) {
  1726.             return NGX_ERROR;
  1727.         }
  1728.     }

  1730.     return NGX_OK;
  1731. }


  1734. static ngx_int_t
  1735. ‌ngx_quic_update_max_stream_data(ngx_quic_stream_t *qs)
  1736. {
  1737.     uint64_t                recv_max_data;
  1738.     ngx_connection_t       *pc;
  1739.     ngx_quic_frame_t       *frame;
  1740.     ngx_quic_connection_t  *qc;

  1742.     pc = qs->parent;
  1743.     qc = ngx_quic_get_connection(pc);

  1745.     if (qs->recv_state != NGX_QUIC_STREAM_RECV_RECV) {
  1746.         return NGX_OK;
  1747.     }

  1749.     recv_max_data = qs->recv_offset + qs->recv_window;

  1751.     if (qs->recv_max_data == recv_max_data) {
  1752.         return NGX_OK;
  1753.     }

  1755.     qs->recv_max_data = recv_max_data;

  1757.     ngx_log_debug2(NGX_LOG_DEBUG_EVENT, pc->log, 0,
  1758.                    "quic stream id:0x%xL flow update msd:%uL",
  1759.                    qs->id, qs->recv_max_data);

  1761.     frame = ngx_quic_alloc_frame(pc);
  1762.     if (frame == NULL) {
  1763.         return NGX_ERROR;
  1764.     }

  1766.     frame->level = ssl_encryption_application;
  1767.     frame->type = NGX_QUIC_FT_MAX_STREAM_DATA;
  1768.     frame->u.max_stream_data.id = qs->id;
  1769.     frame->u.max_stream_data.limit = qs->recv_max_data;

  1771.     ngx_quic_queue_frame(qc, frame);

  1773.     return NGX_OK;
  1774. }


  1777. static ngx_int_t
  1778. ‌ngx_quic_update_max_data(ngx_connection_t *c)
  1779. {
  1780.     uint64_t                recv_max_data;
  1781.     ngx_quic_frame_t       *frame;
  1782.     ngx_quic_connection_t  *qc;

  1784.     qc = ngx_quic_get_connection(c);

  1786.     recv_max_data = qc->streams.recv_offset + qc->streams.recv_window;

  1788.     if (qc->streams.recv_max_data == recv_max_data) {
  1789.         return NGX_OK;
  1790.     }

  1792.     qc->streams.recv_max_data = recv_max_data;

  1794.     ngx_log_debug1(NGX_LOG_DEBUG_EVENT, c->log, 0,
  1795.                    "quic flow update md:%uL", qc->streams.recv_max_data);

  1797.     frame = ngx_quic_alloc_frame(c);
  1798.     if (frame == NULL) {
  1799.         return NGX_ERROR;
  1800.     }

  1802.     frame->level = ssl_encryption_application;
  1803.     frame->type = NGX_QUIC_FT_MAX_DATA;
  1804.     frame->u.max_data.max_data = qc->streams.recv_max_data;

  1806.     ngx_quic_queue_frame(qc, frame);

  1808.     return NGX_OK;
  1809. }


  1812. static void
  1813. ‌ngx_quic_set_event(ngx_event_t *ev)
  1814. {
  1815.     ev->ready = 1;

  1817.     if (ev->active) {
  1818.         ngx_post_event(ev, &ngx_posted_events);
  1819.     }
  1820. }

One Level Up Top Level