One Level Up Top Level

src/event/quic/ngx_event_quic_streams.c - nginx

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, posted_events;
  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.     ngx_queue_init(&posted_events);

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

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

  209.         qs->recv_state = NGX_QUIC_STREAM_RECV_RESET_RECVD;
  210.         qs->send_state = NGX_QUIC_STREAM_SEND_RESET_SENT;

  212.         if (sc == NULL) {
  213.             ngx_quic_close_stream(qs);
  214.             continue;
  215.         }

  217.         sc->read->error = 1;
  218.         sc->read->ready = 1;
  219.         sc->write->error = 1;
  220.         sc->write->ready = 1;

  222.         sc->close = 1;

  224.         if (sc->read->posted) {
  225.             ngx_delete_posted_event(sc->read);
  226.         }

  228.         ngx_post_event(sc->read, &posted_events);
  229.     }

  231.     ngx_event_process_posted((ngx_cycle_t *) ngx_cycle, &posted_events);

  233.     if (tree->root == tree->sentinel) {
  234.         return NGX_OK;
  235.     }

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

  240.     return NGX_AGAIN;
  241. }


  244. ngx_int_t
  245. ‌ngx_quic_reset_stream(ngx_connection_t *c, ngx_uint_t err)
  246. {
  247.     return ngx_quic_do_reset_stream(c->quic, err);
  248. }


  251. static ngx_int_t
  252. ‌ngx_quic_do_reset_stream(ngx_quic_stream_t *qs, ngx_uint_t err)
  253. {
  254.     ngx_connection_t       *pc;
  255.     ngx_quic_frame_t       *frame;
  256.     ngx_quic_connection_t  *qc;

  258.     if (qs->send_state == NGX_QUIC_STREAM_SEND_DATA_RECVD
  259.         || qs->send_state == NGX_QUIC_STREAM_SEND_RESET_SENT
  260.         || qs->send_state == NGX_QUIC_STREAM_SEND_RESET_RECVD)
  261.     {
  262.         return NGX_OK;
  263.     }

  265.     qs->send_state = NGX_QUIC_STREAM_SEND_RESET_SENT;
  266.     qs->send_final_size = qs->send_offset;

  268.     if (qs->connection) {
  269.         qs->connection->write->error = 1;
  270.     }

  272.     pc = qs->parent;
  273.     qc = ngx_quic_get_connection(pc);

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

  278.     frame = ngx_quic_alloc_frame(pc);
  279.     if (frame == NULL) {
  280.         return NGX_ERROR;
  281.     }

  283.     frame->level = NGX_QUIC_ENCRYPTION_APPLICATION;
  284.     frame->type = NGX_QUIC_FT_RESET_STREAM;
  285.     frame->u.reset_stream.id = qs->id;
  286.     frame->u.reset_stream.error_code = err;
  287.     frame->u.reset_stream.final_size = qs->send_offset;

  289.     ngx_quic_queue_frame(qc, frame);

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

  293.     return NGX_OK;
  294. }


  297. ngx_int_t
  298. ‌ngx_quic_shutdown_stream(ngx_connection_t *c, int how)
  299. {
  300.     if (how == NGX_RDWR_SHUTDOWN || how == NGX_WRITE_SHUTDOWN) {
  301.         if (ngx_quic_shutdown_stream_send(c) != NGX_OK) {
  302.             return NGX_ERROR;
  303.         }
  304.     }

  306.     if (how == NGX_RDWR_SHUTDOWN || how == NGX_READ_SHUTDOWN) {
  307.         if (ngx_quic_shutdown_stream_recv(c) != NGX_OK) {
  308.             return NGX_ERROR;
  309.         }
  310.     }

  312.     return NGX_OK;
  313. }


  316. static ngx_int_t
  317. ‌ngx_quic_shutdown_stream_send(ngx_connection_t *c)
  318. {
  319.     ngx_quic_stream_t  *qs;

  321.     qs = c->quic;

  323.     if (qs->send_state != NGX_QUIC_STREAM_SEND_READY
  324.         && qs->send_state != NGX_QUIC_STREAM_SEND_SEND)
  325.     {
  326.         return NGX_OK;
  327.     }

  329.     qs->send_state = NGX_QUIC_STREAM_SEND_SEND;
  330.     qs->send_final_size = c->sent;

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

  335.     return ngx_quic_stream_flush(qs);
  336. }


  339. static ngx_int_t
  340. ‌ngx_quic_shutdown_stream_recv(ngx_connection_t *c)
  341. {
  342.     ngx_connection_t       *pc;
  343.     ngx_quic_frame_t       *frame;
  344.     ngx_quic_stream_t      *qs;
  345.     ngx_quic_connection_t  *qc;

  347.     qs = c->quic;

  349.     if (qs->recv_state != NGX_QUIC_STREAM_RECV_RECV
  350.         && qs->recv_state != NGX_QUIC_STREAM_RECV_SIZE_KNOWN)
  351.     {
  352.         return NGX_OK;
  353.     }

  355.     pc = qs->parent;
  356.     qc = ngx_quic_get_connection(pc);

  358.     if (qc->conf->stream_close_code == 0) {
  359.         return NGX_OK;
  360.     }

  362.     frame = ngx_quic_alloc_frame(pc);
  363.     if (frame == NULL) {
  364.         return NGX_ERROR;
  365.     }

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

  370.     frame->level = NGX_QUIC_ENCRYPTION_APPLICATION;
  371.     frame->type = NGX_QUIC_FT_STOP_SENDING;
  372.     frame->u.stop_sending.id = qs->id;
  373.     frame->u.stop_sending.error_code = qc->conf->stream_close_code;

  375.     ngx_quic_queue_frame(qc, frame);

  377.     return NGX_OK;
  378. }


  381. static ngx_quic_stream_t *
  382. ‌ngx_quic_get_stream(ngx_connection_t *c, uint64_t id)
  383. {
  384.     uint64_t                min_id;
  385.     ngx_event_t            *rev;
  386.     ngx_quic_stream_t      *qs;
  387.     ngx_quic_connection_t  *qc;

  389.     qc = ngx_quic_get_connection(c);

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

  393.     if (qs) {
  394.         return qs;
  395.     }

  397.     if (qc->shutdown || qc->closing) {
  398.         return NGX_QUIC_STREAM_GONE;
  399.     }

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

  404.     if (id & NGX_QUIC_STREAM_UNIDIRECTIONAL) {

  406.         if (id & NGX_QUIC_STREAM_SERVER_INITIATED) {
  407.             if ((id >> 2) < qc->streams.server_streams_uni) {
  408.                 return NGX_QUIC_STREAM_GONE;
  409.             }

  411.             qc->error = NGX_QUIC_ERR_STREAM_STATE_ERROR;
  412.             return NULL;
  413.         }

  415.         if ((id >> 2) < qc->streams.client_streams_uni) {
  416.             return NGX_QUIC_STREAM_GONE;
  417.         }

  419.         if ((id >> 2) >= qc->streams.client_max_streams_uni) {
  420.             qc->error = NGX_QUIC_ERR_STREAM_LIMIT_ERROR;
  421.             return NULL;
  422.         }

  424.         min_id = (qc->streams.client_streams_uni << 2)
  425.                  | NGX_QUIC_STREAM_UNIDIRECTIONAL;
  426.         qc->streams.client_streams_uni = (id >> 2) + 1;

  428.     } else {

  430.         if (id & NGX_QUIC_STREAM_SERVER_INITIATED) {
  431.             if ((id >> 2) < qc->streams.server_streams_bidi) {
  432.                 return NGX_QUIC_STREAM_GONE;
  433.             }

  435.             qc->error = NGX_QUIC_ERR_STREAM_STATE_ERROR;
  436.             return NULL;
  437.         }

  439.         if ((id >> 2) < qc->streams.client_streams_bidi) {
  440.             return NGX_QUIC_STREAM_GONE;
  441.         }

  443.         if ((id >> 2) >= qc->streams.client_max_streams_bidi) {
  444.             qc->error = NGX_QUIC_ERR_STREAM_LIMIT_ERROR;
  445.             return NULL;
  446.         }

  448.         min_id = (qc->streams.client_streams_bidi << 2);
  449.         qc->streams.client_streams_bidi = (id >> 2) + 1;
  450.     }

  452.     /*
  453.      * RFC 9000, 2.1.  Stream Types and Identifiers
  454.      *
  455.      * successive streams of each type are created with numerically increasing
  456.      * stream IDs.  A stream ID that is used out of order results in all
  457.      * streams of that type with lower-numbered stream IDs also being opened.
  458.      */

  460. #if (NGX_SUPPRESS_WARN)
  461.     qs = NULL;
  462. #endif

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

  466.         qs = ngx_quic_create_stream(c, min_id);

  468.         if (qs == NULL) {
  469.             if (ngx_quic_reject_stream(c, min_id) != NGX_OK) {
  470.                 return NULL;
  471.             }

  473.             continue;
  474.         }

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

  478.         rev = qs->connection->read;
  479.         rev->handler = ngx_quic_init_stream_handler;

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

  484.             if (qc->push.posted) {
  485.                 /*
  486.                  * The posted stream can produce output immediately.
  487.                  * By postponing the push event, we coalesce the stream
  488.                  * output with queued frames in one UDP datagram.
  489.                  */

  491.                 ngx_delete_posted_event(&qc->push);
  492.                 ngx_post_event(&qc->push, &ngx_posted_events);
  493.             }
  494.         }
  495.     }

  497.     if (qs == NULL) {
  498.         return NGX_QUIC_STREAM_GONE;
  499.     }

  501.     return qs;
  502. }


  505. static ngx_int_t
  506. ‌ngx_quic_reject_stream(ngx_connection_t *c, uint64_t id)
  507. {
  508.     uint64_t                code;
  509.     ngx_quic_frame_t       *frame;
  510.     ngx_quic_connection_t  *qc;

  512.     qc = ngx_quic_get_connection(c);

  514.     code = (id & NGX_QUIC_STREAM_UNIDIRECTIONAL)
  515.            ? qc->conf->stream_reject_code_uni
  516.            : qc->conf->stream_reject_code_bidi;

  518.     if (code == 0) {
  519.         return NGX_DECLINED;
  520.     }

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

  525.     frame = ngx_quic_alloc_frame(c);
  526.     if (frame == NULL) {
  527.         return NGX_ERROR;
  528.     }

  530.     frame->level = NGX_QUIC_ENCRYPTION_APPLICATION;
  531.     frame->type = NGX_QUIC_FT_RESET_STREAM;
  532.     frame->u.reset_stream.id = id;
  533.     frame->u.reset_stream.error_code = code;
  534.     frame->u.reset_stream.final_size = 0;

  536.     ngx_quic_queue_frame(qc, frame);

  538.     frame = ngx_quic_alloc_frame(c);
  539.     if (frame == NULL) {
  540.         return NGX_ERROR;
  541.     }

  543.     frame->level = NGX_QUIC_ENCRYPTION_APPLICATION;
  544.     frame->type = NGX_QUIC_FT_STOP_SENDING;
  545.     frame->u.stop_sending.id = id;
  546.     frame->u.stop_sending.error_code = code;

  548.     ngx_quic_queue_frame(qc, frame);

  550.     return NGX_OK;
  551. }


  554. static void
  555. ‌ngx_quic_init_stream_handler(ngx_event_t *ev)
  556. {
  557.     ngx_connection_t   *c;
  558.     ngx_quic_stream_t  *qs;

  560.     c = ev->data;
  561.     qs = c->quic;

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

  565.     if ((qs->id & NGX_QUIC_STREAM_UNIDIRECTIONAL) == 0) {
  566.         c->write->active = 1;
  567.         c->write->ready = 1;
  568.     }

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

  572.     ngx_queue_remove(&qs->queue);

  574.     c->listening->handler(c);
  575. }


  578. ngx_int_t
  579. ‌ngx_quic_init_streams(ngx_connection_t *c)
  580. {
  581.     ngx_int_t               rc;
  582.     ngx_quic_connection_t  *qc;

  584.     qc = ngx_quic_get_connection(c);

  586.     if (qc->streams.initialized) {
  587.         return NGX_OK;
  588.     }

  590.     rc = ngx_ssl_ocsp_validate(c);

  592.     if (rc == NGX_ERROR) {
  593.         return NGX_ERROR;
  594.     }

  596.     if (rc == NGX_AGAIN) {
  597.         c->ssl->handler = ngx_quic_init_streams_handler;
  598.         return NGX_OK;
  599.     }

  601.     return ngx_quic_do_init_streams(c);
  602. }


  605. static void
  606. ‌ngx_quic_init_streams_handler(ngx_connection_t *c)
  607. {
  608.     if (ngx_quic_do_init_streams(c) != NGX_OK) {
  609.         ngx_quic_close_connection(c, NGX_ERROR);
  610.     }
  611. }


  614. static ngx_int_t
  615. ‌ngx_quic_do_init_streams(ngx_connection_t *c)
  616. {
  617.     ngx_queue_t            *q;
  618.     ngx_quic_stream_t      *qs;
  619.     ngx_quic_connection_t  *qc;

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

  623.     qc = ngx_quic_get_connection(c);

  625.     if (qc->conf->init) {
  626.         if (qc->conf->init(c) != NGX_OK) {
  627.             return NGX_ERROR;
  628.         }
  629.     }

  631.     for (q = ngx_queue_head(&qc->streams.uninitialized);
  632.          q != ngx_queue_sentinel(&qc->streams.uninitialized);
  633.          q = ngx_queue_next(q))
  634.     {
  635.         qs = ngx_queue_data(q, ngx_quic_stream_t, queue);
  636.         ngx_post_event(qs->connection->read, &ngx_posted_events);
  637.     }

  639.     qc->streams.initialized = 1;

  641.     if (!qc->closing && qc->close.timer_set) {
  642.         ngx_del_timer(&qc->close);
  643.     }

  645.     return NGX_OK;
  646. }


  649. static ngx_quic_stream_t *
  650. ‌ngx_quic_create_stream(ngx_connection_t *c, uint64_t id)
  651. {
  652.     ngx_str_t               addr_text;
  653.     ngx_log_t              *log;
  654.     ngx_pool_t             *pool;
  655.     ngx_uint_t              reusable;
  656.     ngx_queue_t            *q;
  657.     struct sockaddr        *sockaddr;
  658.     ngx_connection_t       *sc;
  659.     ngx_quic_stream_t      *qs;
  660.     ngx_pool_cleanup_t     *cln;
  661.     ngx_quic_connection_t  *qc;

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

  666.     qc = ngx_quic_get_connection(c);

  668.     if (!ngx_queue_empty(&qc->streams.free)) {
  669.         q = ngx_queue_head(&qc->streams.free);
  670.         qs = ngx_queue_data(q, ngx_quic_stream_t, queue);
  671.         ngx_queue_remove(&qs->queue);

  673.     } else {
  674.         /*
  675.          * the number of streams is limited by transport
  676.          * parameters and application requirements
  677.          */

  679.         qs = ngx_palloc(c->pool, sizeof(ngx_quic_stream_t));
  680.         if (qs == NULL) {
  681.             return NULL;
  682.         }
  683.     }

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

  687.     qs->node.key = id;
  688.     qs->parent = c;
  689.     qs->id = id;
  690.     qs->send_final_size = (uint64_t) -1;
  691.     qs->recv_final_size = (uint64_t) -1;

  693.     pool = ngx_create_pool(NGX_DEFAULT_POOL_SIZE, c->log);
  694.     if (pool == NULL) {
  695.         ngx_queue_insert_tail(&qc->streams.free, &qs->queue);
  696.         return NULL;
  697.     }

  699.     log = ngx_palloc(pool, sizeof(ngx_log_t));
  700.     if (log == NULL) {
  701.         ngx_destroy_pool(pool);
  702.         ngx_queue_insert_tail(&qc->streams.free, &qs->queue);
  703.         return NULL;
  704.     }

  706.     *log = *c->log;
  707.     pool->log = log;

  709.     sockaddr = ngx_palloc(pool, c->socklen);
  710.     if (sockaddr == NULL) {
  711.         ngx_destroy_pool(pool);
  712.         ngx_queue_insert_tail(&qc->streams.free, &qs->queue);
  713.         return NULL;
  714.     }

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

  718.     if (c->addr_text.data) {
  719.         addr_text.data = ngx_pnalloc(pool, c->addr_text.len);
  720.         if (addr_text.data == NULL) {
  721.             ngx_destroy_pool(pool);
  722.             ngx_queue_insert_tail(&qc->streams.free, &qs->queue);
  723.             return NULL;
  724.         }

  726.         ngx_memcpy(addr_text.data, c->addr_text.data, c->addr_text.len);
  727.         addr_text.len = c->addr_text.len;

  729.     } else {
  730.         addr_text.len = 0;
  731.         addr_text.data = NULL;
  732.     }

  734.     reusable = c->reusable;
  735.     ngx_reusable_connection(c, 0);

  737.     sc = ngx_get_connection(c->fd, log);
  738.     if (sc == NULL) {
  739.         ngx_destroy_pool(pool);
  740.         ngx_queue_insert_tail(&qc->streams.free, &qs->queue);
  741.         ngx_reusable_connection(c, reusable);
  742.         return NULL;
  743.     }

  745.     qs->connection = sc;

  747.     sc->quic = qs;
  748.     sc->shared = 1;
  749.     sc->type = SOCK_STREAM;
  750.     sc->pool = pool;
  751.     sc->ssl = c->ssl;
  752.     sc->sockaddr = sockaddr;
  753.     sc->socklen = c->socklen;
  754.     sc->listening = c->listening;
  755.     sc->addr_text = addr_text;
  756.     sc->local_sockaddr = c->local_sockaddr;
  757.     sc->local_socklen = c->local_socklen;
  758.     sc->number = ngx_atomic_fetch_add(ngx_connection_counter, 1);
  759.     sc->start_time = c->start_time;
  760.     sc->tcp_nodelay = NGX_TCP_NODELAY_DISABLED;

  762.     sc->recv = ngx_quic_stream_recv;
  763.     sc->send = ngx_quic_stream_send;
  764.     sc->send_chain = ngx_quic_stream_send_chain;

  766.     sc->read->log = log;
  767.     sc->write->log = log;

  769.     sc->read->handler = ngx_quic_empty_handler;
  770.     sc->write->handler = ngx_quic_empty_handler;

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

  774.     if (id & NGX_QUIC_STREAM_UNIDIRECTIONAL) {
  775.         if (id & NGX_QUIC_STREAM_SERVER_INITIATED) {
  776.             qs->send_max_data = qc->ctp.initial_max_stream_data_uni;
  777.             qs->recv_state = NGX_QUIC_STREAM_RECV_DATA_READ;
  778.             qs->send_state = NGX_QUIC_STREAM_SEND_READY;

  780.         } else {
  781.             qs->recv_max_data = qc->tp.initial_max_stream_data_uni;
  782.             qs->recv_state = NGX_QUIC_STREAM_RECV_RECV;
  783.             qs->send_state = NGX_QUIC_STREAM_SEND_DATA_RECVD;
  784.         }

  786.     } else {
  787.         if (id & NGX_QUIC_STREAM_SERVER_INITIATED) {
  788.             qs->send_max_data = qc->ctp.initial_max_stream_data_bidi_remote;
  789.             qs->recv_max_data = qc->tp.initial_max_stream_data_bidi_local;

  791.         } else {
  792.             qs->send_max_data = qc->ctp.initial_max_stream_data_bidi_local;
  793.             qs->recv_max_data = qc->tp.initial_max_stream_data_bidi_remote;
  794.         }

  796.         qs->recv_state = NGX_QUIC_STREAM_RECV_RECV;
  797.         qs->send_state = NGX_QUIC_STREAM_SEND_READY;
  798.     }

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

  802.     cln = ngx_pool_cleanup_add(pool, 0);
  803.     if (cln == NULL) {
  804.         ngx_close_connection(sc);
  805.         ngx_destroy_pool(pool);
  806.         ngx_queue_insert_tail(&qc->streams.free, &qs->queue);
  807.         ngx_reusable_connection(c, reusable);
  808.         return NULL;
  809.     }

  811.     cln->handler = ngx_quic_stream_cleanup_handler;
  812.     cln->data = sc;

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

  816.     return qs;
  817. }


  820. void
  821. ‌ngx_quic_cancelable_stream(ngx_connection_t *c)
  822. {
  823.     ngx_connection_t       *pc;
  824.     ngx_quic_stream_t      *qs;
  825.     ngx_quic_connection_t  *qc;

  827.     qs = c->quic;
  828.     pc = qs->parent;
  829.     qc = ngx_quic_get_connection(pc);

  831.     if (!qs->cancelable) {
  832.         qs->cancelable = 1;

  834.         if (ngx_quic_can_shutdown(pc) == NGX_OK) {
  835.             ngx_reusable_connection(pc, 1);

  837.             if (qc->shutdown) {
  838.                 ngx_quic_shutdown_quic(pc);
  839.             }
  840.         }
  841.     }
  842. }


  845. static void
  846. ‌ngx_quic_empty_handler(ngx_event_t *ev)
  847. {
  848. }


  851. static ssize_t
  852. ‌ngx_quic_stream_recv(ngx_connection_t *c, u_char *buf, size_t size)
  853. {
  854.     ssize_t             len;
  855.     ngx_buf_t          *b;
  856.     ngx_chain_t        *cl, *in;
  857.     ngx_event_t        *rev;
  858.     ngx_connection_t   *pc;
  859.     ngx_quic_stream_t  *qs;

  861.     qs = c->quic;
  862.     pc = qs->parent;
  863.     rev = c->read;

  865.     if (qs->recv_state == NGX_QUIC_STREAM_RECV_RESET_RECVD
  866.         || qs->recv_state == NGX_QUIC_STREAM_RECV_RESET_READ)
  867.     {
  868.         qs->recv_state = NGX_QUIC_STREAM_RECV_RESET_READ;
  869.         return NGX_ERROR;
  870.     }

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

  875.     if (size == 0) {
  876.         return 0;
  877.     }

  879.     in = ngx_quic_read_buffer(pc, &qs->recv, size);
  880.     if (in == NGX_CHAIN_ERROR) {
  881.         return NGX_ERROR;
  882.     }

  884.     len = 0;

  886.     for (cl = in; cl; cl = cl->next) {
  887.         b = cl->buf;
  888.         len += b->last - b->pos;
  889.         buf = ngx_cpymem(buf, b->pos, b->last - b->pos);
  890.     }

  892.     ngx_quic_free_chain(pc, in);

  894.     if (len == 0) {
  895.         rev->ready = 0;

  897.         if (qs->recv_state == NGX_QUIC_STREAM_RECV_DATA_RECVD
  898.             && qs->recv_offset == qs->recv_final_size)
  899.         {
  900.             qs->recv_state = NGX_QUIC_STREAM_RECV_DATA_READ;
  901.         }

  903.         if (qs->recv_state == NGX_QUIC_STREAM_RECV_DATA_READ) {
  904.             rev->eof = 1;
  905.             return 0;
  906.         }

  908.         ngx_log_debug1(NGX_LOG_DEBUG_EVENT, c->log, 0,
  909.                        "quic stream id:0x%xL recv() not ready", qs->id);
  910.         return NGX_AGAIN;
  911.     }

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

  916.     if (ngx_quic_update_flow(qs, qs->recv_offset + len) != NGX_OK) {
  917.         return NGX_ERROR;
  918.     }

  920.     return len;
  921. }


  924. static ssize_t
  925. ‌ngx_quic_stream_send(ngx_connection_t *c, u_char *buf, size_t size)
  926. {
  927.     ngx_buf_t    b;
  928.     ngx_chain_t  cl;

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

  932.     b.memory = 1;
  933.     b.pos = buf;
  934.     b.last = buf + size;

  936.     cl.buf = &b;
  937.     cl.next = NULL;

  939.     if (ngx_quic_stream_send_chain(c, &cl, 0) == NGX_CHAIN_ERROR) {
  940.         return NGX_ERROR;
  941.     }

  943.     if (b.pos == buf) {
  944.         return NGX_AGAIN;
  945.     }

  947.     return b.pos - buf;
  948. }


  951. static ngx_chain_t *
  952. ‌ngx_quic_stream_send_chain(ngx_connection_t *c, ngx_chain_t *in, off_t limit)
  953. {
  954.     uint64_t                n, flow;
  955.     ngx_event_t            *wev;
  956.     ngx_connection_t       *pc;
  957.     ngx_quic_stream_t      *qs;
  958.     ngx_quic_connection_t  *qc;

  960.     qs = c->quic;
  961.     pc = qs->parent;
  962.     qc = ngx_quic_get_connection(pc);
  963.     wev = c->write;

  965.     if (qs->send_state != NGX_QUIC_STREAM_SEND_READY
  966.         && qs->send_state != NGX_QUIC_STREAM_SEND_SEND)
  967.     {
  968.         wev->error = 1;
  969.         return NGX_CHAIN_ERROR;
  970.     }

  972.     qs->send_state = NGX_QUIC_STREAM_SEND_SEND;

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

  976.     if (flow == 0) {
  977.         wev->ready = 0;
  978.         return in;
  979.     }

  981.     if (limit == 0 || limit > (off_t) flow) {
  982.         limit = flow;
  983.     }

  985.     n = qs->send.size;

  987.     in = ngx_quic_write_buffer(pc, &qs->send, in, limit, qs->sent);
  988.     if (in == NGX_CHAIN_ERROR) {
  989.         return NGX_CHAIN_ERROR;
  990.     }

  992.     n = qs->send.size - n;
  993.     c->sent += n;
  994.     qs->sent += n;
  995.     qc->streams.sent += n;

  997.     if (flow == n) {
  998.         wev->ready = 0;
  999.     }

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

  1004.     if (ngx_quic_stream_flush(qs) != NGX_OK) {
  1005.         return NGX_CHAIN_ERROR;
  1006.     }

  1008.     return in;
  1009. }


  1012. static ngx_int_t
  1013. ‌ngx_quic_stream_flush(ngx_quic_stream_t *qs)
  1014. {
  1015.     off_t                   limit, len;
  1016.     ngx_uint_t              last;
  1017.     ngx_chain_t            *out;
  1018.     ngx_quic_frame_t       *frame;
  1019.     ngx_connection_t       *pc;
  1020.     ngx_quic_connection_t  *qc;

  1022.     if (qs->send_state != NGX_QUIC_STREAM_SEND_SEND) {
  1023.         return NGX_OK;
  1024.     }

  1026.     pc = qs->parent;
  1027.     qc = ngx_quic_get_connection(pc);

  1029.     if (qc->streams.send_max_data == 0) {
  1030.         qc->streams.send_max_data = qc->ctp.initial_max_data;
  1031.     }

  1033.     limit = ngx_min(qc->streams.send_max_data - qc->streams.send_offset,
  1034.                     qs->send_max_data - qs->send_offset);

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

  1039.     len = qs->send.offset;

  1041.     out = ngx_quic_read_buffer(pc, &qs->send, limit);
  1042.     if (out == NGX_CHAIN_ERROR) {
  1043.         return NGX_ERROR;
  1044.     }

  1046.     len = qs->send.offset - len;
  1047.     last = 0;

  1049.     if (qs->send_final_size != (uint64_t) -1
  1050.         && qs->send_final_size == qs->send.offset)
  1051.     {
  1052.         qs->send_state = NGX_QUIC_STREAM_SEND_DATA_SENT;
  1053.         last = 1;
  1054.     }

  1056.     if (len == 0 && !last) {
  1057.         return NGX_OK;
  1058.     }

  1060.     frame = ngx_quic_alloc_frame(pc);
  1061.     if (frame == NULL) {
  1062.         return NGX_ERROR;
  1063.     }

  1065.     frame->level = NGX_QUIC_ENCRYPTION_APPLICATION;
  1066.     frame->type = NGX_QUIC_FT_STREAM;
  1067.     frame->data = out;

  1069.     frame->u.stream.off = 1;
  1070.     frame->u.stream.len = 1;
  1071.     frame->u.stream.fin = last;

  1073.     frame->u.stream.stream_id = qs->id;
  1074.     frame->u.stream.offset = qs->send_offset;
  1075.     frame->u.stream.length = len;

  1077.     ngx_quic_queue_frame(qc, frame);

  1079.     qs->send_offset += len;
  1080.     qc->streams.send_offset += len;

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

  1086.     if (qs->connection == NULL) {
  1087.         return ngx_quic_close_stream(qs);
  1088.     }

  1090.     return NGX_OK;
  1091. }


  1094. static void
  1095. ‌ngx_quic_stream_cleanup_handler(void *data)
  1096. {
  1097.     ngx_connection_t *c = data;

  1099.     ngx_quic_stream_t      *qs;
  1100.     ngx_quic_connection_t  *qc;

  1102.     qs = c->quic;

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

  1107.     if (ngx_quic_shutdown_stream(c, NGX_RDWR_SHUTDOWN) != NGX_OK) {
  1108.         qs->connection = NULL;
  1109.         goto failed;
  1110.     }

  1112.     qs->connection = NULL;

  1114.     if (ngx_quic_close_stream(qs) != NGX_OK) {
  1115.         goto failed;
  1116.     }

  1118.     return;

  1120. failed:

  1122.     qc = ngx_quic_get_connection(qs->parent);
  1123.     qc->error = NGX_QUIC_ERR_INTERNAL_ERROR;

  1125.     ngx_post_event(&qc->close, &ngx_posted_events);
  1126. }


  1129. static ngx_int_t
  1130. ‌ngx_quic_close_stream(ngx_quic_stream_t *qs)
  1131. {
  1132.     ngx_connection_t       *pc;
  1133.     ngx_quic_frame_t       *frame;
  1134.     ngx_quic_connection_t  *qc;

  1136.     pc = qs->parent;
  1137.     qc = ngx_quic_get_connection(pc);

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

  1142.         if (qs->recv_state == NGX_QUIC_STREAM_RECV_RECV) {
  1143.             return NGX_OK;
  1144.         }

  1146.         if (qs->send_state != NGX_QUIC_STREAM_SEND_DATA_RECVD
  1147.             && qs->send_state != NGX_QUIC_STREAM_SEND_RESET_RECVD)
  1148.         {
  1149.             return NGX_OK;
  1150.         }
  1151.     }

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

  1156.     ngx_quic_free_buffer(pc, &qs->send);
  1157.     ngx_quic_free_buffer(pc, &qs->recv);

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

  1162.     if (qc->closing) {
  1163.         /* schedule handler call to continue ngx_quic_close_connection() */
  1164.         ngx_post_event(&qc->close, &ngx_posted_events);
  1165.         return NGX_OK;
  1166.     }

  1168.     if (!pc->reusable && ngx_quic_can_shutdown(pc) == NGX_OK) {
  1169.         ngx_reusable_connection(pc, 1);
  1170.     }

  1172.     if (qc->shutdown) {
  1173.         ngx_quic_shutdown_quic(pc);
  1174.         return NGX_OK;
  1175.     }

  1177.     if ((qs->id & NGX_QUIC_STREAM_SERVER_INITIATED) == 0) {
  1178.         frame = ngx_quic_alloc_frame(pc);
  1179.         if (frame == NULL) {
  1180.             return NGX_ERROR;
  1181.         }

  1183.         frame->level = NGX_QUIC_ENCRYPTION_APPLICATION;
  1184.         frame->type = NGX_QUIC_FT_MAX_STREAMS;

  1186.         if (qs->id & NGX_QUIC_STREAM_UNIDIRECTIONAL) {
  1187.             frame->u.max_streams.limit = ++qc->streams.client_max_streams_uni;
  1188.             frame->u.max_streams.bidi = 0;

  1190.         } else {
  1191.             frame->u.max_streams.limit = ++qc->streams.client_max_streams_bidi;
  1192.             frame->u.max_streams.bidi = 1;
  1193.         }

  1195.         ngx_quic_queue_frame(qc, frame);
  1196.     }

  1198.     return NGX_OK;
  1199. }


  1202. static ngx_int_t
  1203. ‌ngx_quic_can_shutdown(ngx_connection_t *c)
  1204. {
  1205.     ngx_rbtree_t           *tree;
  1206.     ngx_rbtree_node_t      *node;
  1207.     ngx_quic_stream_t      *qs;
  1208.     ngx_quic_connection_t  *qc;

  1210.     qc = ngx_quic_get_connection(c);

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

  1214.     if (tree->root != tree->sentinel) {
  1215.         for (node = ngx_rbtree_min(tree->root, tree->sentinel);
  1216.              node;
  1217.              node = ngx_rbtree_next(tree, node))
  1218.         {
  1219.             qs = (ngx_quic_stream_t *) node;

  1221.             if (!qs->cancelable) {
  1222.                 return NGX_DECLINED;
  1223.             }
  1224.         }
  1225.     }

  1227.     return NGX_OK;
  1228. }


  1231. ngx_int_t
  1232. ‌ngx_quic_handle_stream_frame(ngx_connection_t *c, ngx_quic_header_t *pkt,
  1233.     ngx_quic_frame_t *frame)
  1234. {
  1235.     uint64_t                  last;
  1236.     ngx_quic_stream_t        *qs;
  1237.     ngx_quic_connection_t    *qc;
  1238.     ngx_quic_stream_frame_t  *f;

  1240.     qc = ngx_quic_get_connection(c);
  1241.     f = &frame->u.stream;

  1243.     if ((f->stream_id & NGX_QUIC_STREAM_UNIDIRECTIONAL)
  1244.         && (f->stream_id & NGX_QUIC_STREAM_SERVER_INITIATED))
  1245.     {
  1246.         qc->error = NGX_QUIC_ERR_STREAM_STATE_ERROR;
  1247.         return NGX_ERROR;
  1248.     }

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

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

  1255.     if (qs == NULL) {
  1256.         return NGX_ERROR;
  1257.     }

  1259.     if (qs == NGX_QUIC_STREAM_GONE) {
  1260.         return NGX_OK;
  1261.     }

  1263.     if (qs->recv_state != NGX_QUIC_STREAM_RECV_RECV
  1264.         && qs->recv_state != NGX_QUIC_STREAM_RECV_SIZE_KNOWN)
  1265.     {
  1266.         return NGX_OK;
  1267.     }

  1269.     if (ngx_quic_control_flow(qs, last) != NGX_OK) {
  1270.         return NGX_ERROR;
  1271.     }

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

  1278.     if (last < qs->recv_offset) {
  1279.         return NGX_OK;
  1280.     }

  1282.     if (f->fin) {
  1283.         if (qs->recv_final_size != (uint64_t) -1 && qs->recv_final_size != last)
  1284.         {
  1285.             qc->error = NGX_QUIC_ERR_FINAL_SIZE_ERROR;
  1286.             return NGX_ERROR;
  1287.         }

  1289.         if (qs->recv_last > last) {
  1290.             qc->error = NGX_QUIC_ERR_FINAL_SIZE_ERROR;
  1291.             return NGX_ERROR;
  1292.         }

  1294.         qs->recv_final_size = last;
  1295.         qs->recv_state = NGX_QUIC_STREAM_RECV_SIZE_KNOWN;
  1296.     }

  1298.     if (ngx_quic_write_buffer(c, &qs->recv, frame->data, f->length, f->offset)
  1299.         == NGX_CHAIN_ERROR)
  1300.     {
  1301.         return NGX_ERROR;
  1302.     }

  1304.     if (qs->recv_state == NGX_QUIC_STREAM_RECV_SIZE_KNOWN
  1305.         && qs->recv.size == qs->recv_final_size)
  1306.     {
  1307.         qs->recv_state = NGX_QUIC_STREAM_RECV_DATA_RECVD;
  1308.     }

  1310.     if (qs->connection == NULL) {
  1311.         return ngx_quic_close_stream(qs);
  1312.     }

  1314.     if (f->offset <= qs->recv_offset) {
  1315.         ngx_quic_set_event(qs->connection->read);
  1316.     }

  1318.     return NGX_OK;
  1319. }


  1322. ngx_int_t
  1323. ‌ngx_quic_handle_max_data_frame(ngx_connection_t *c,
  1324.     ngx_quic_max_data_frame_t *f)
  1325. {
  1326.     ngx_rbtree_t           *tree;
  1327.     ngx_rbtree_node_t      *node;
  1328.     ngx_quic_stream_t      *qs;
  1329.     ngx_quic_connection_t  *qc;

  1331.     qc = ngx_quic_get_connection(c);
  1332.     tree = &qc->streams.tree;

  1334.     if (f->max_data <= qc->streams.send_max_data) {
  1335.         return NGX_OK;
  1336.     }

  1338.     if (tree->root == tree->sentinel
  1339.         || qc->streams.send_offset < qc->streams.send_max_data)
  1340.     {
  1341.         /* not blocked on MAX_DATA */
  1342.         qc->streams.send_max_data = f->max_data;
  1343.         return NGX_OK;
  1344.     }

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

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

  1351.         qs = (ngx_quic_stream_t *) node;
  1352.         node = ngx_rbtree_next(tree, node);

  1354.         if (ngx_quic_stream_flush(qs) != NGX_OK) {
  1355.             return NGX_ERROR;
  1356.         }
  1357.     }

  1359.     return NGX_OK;
  1360. }


  1363. ngx_int_t
  1364. ‌ngx_quic_handle_streams_blocked_frame(ngx_connection_t *c,
  1365.     ngx_quic_header_t *pkt, ngx_quic_streams_blocked_frame_t *f)
  1366. {
  1367.     return NGX_OK;
  1368. }


  1371. ngx_int_t
  1372. ‌ngx_quic_handle_data_blocked_frame(ngx_connection_t *c,
  1373.     ngx_quic_header_t *pkt, ngx_quic_data_blocked_frame_t *f)
  1374. {
  1375.     return ngx_quic_update_max_data(c);
  1376. }


  1379. ngx_int_t
  1380. ‌ngx_quic_handle_stream_data_blocked_frame(ngx_connection_t *c,
  1381.     ngx_quic_header_t *pkt, ngx_quic_stream_data_blocked_frame_t *f)
  1382. {
  1383.     ngx_quic_stream_t      *qs;
  1384.     ngx_quic_connection_t  *qc;

  1386.     qc = ngx_quic_get_connection(c);

  1388.     if ((f->id & NGX_QUIC_STREAM_UNIDIRECTIONAL)
  1389.         && (f->id & NGX_QUIC_STREAM_SERVER_INITIATED))
  1390.     {
  1391.         qc->error = NGX_QUIC_ERR_STREAM_STATE_ERROR;
  1392.         return NGX_ERROR;
  1393.     }

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

  1397.     if (qs == NULL) {
  1398.         return NGX_ERROR;
  1399.     }

  1401.     if (qs == NGX_QUIC_STREAM_GONE) {
  1402.         return NGX_OK;
  1403.     }

  1405.     return ngx_quic_update_max_stream_data(qs);
  1406. }


  1409. ngx_int_t
  1410. ‌ngx_quic_handle_max_stream_data_frame(ngx_connection_t *c,
  1411.     ngx_quic_header_t *pkt, ngx_quic_max_stream_data_frame_t *f)
  1412. {
  1413.     ngx_quic_stream_t      *qs;
  1414.     ngx_quic_connection_t  *qc;

  1416.     qc = ngx_quic_get_connection(c);

  1418.     if ((f->id & NGX_QUIC_STREAM_UNIDIRECTIONAL)
  1419.         && (f->id & NGX_QUIC_STREAM_SERVER_INITIATED) == 0)
  1420.     {
  1421.         qc->error = NGX_QUIC_ERR_STREAM_STATE_ERROR;
  1422.         return NGX_ERROR;
  1423.     }

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

  1427.     if (qs == NULL) {
  1428.         return NGX_ERROR;
  1429.     }

  1431.     if (qs == NGX_QUIC_STREAM_GONE) {
  1432.         return NGX_OK;
  1433.     }

  1435.     if (f->limit <= qs->send_max_data) {
  1436.         return NGX_OK;
  1437.     }

  1439.     if (qs->send_offset < qs->send_max_data) {
  1440.         /* not blocked on MAX_STREAM_DATA */
  1441.         qs->send_max_data = f->limit;
  1442.         return NGX_OK;
  1443.     }

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

  1447.     return ngx_quic_stream_flush(qs);
  1448. }


  1451. ngx_int_t
  1452. ‌ngx_quic_handle_reset_stream_frame(ngx_connection_t *c,
  1453.     ngx_quic_header_t *pkt, ngx_quic_reset_stream_frame_t *f)
  1454. {
  1455.     ngx_event_t            *rev;
  1456.     ngx_quic_stream_t      *qs;
  1457.     ngx_quic_connection_t  *qc;

  1459.     qc = ngx_quic_get_connection(c);

  1461.     if ((f->id & NGX_QUIC_STREAM_UNIDIRECTIONAL)
  1462.         && (f->id & NGX_QUIC_STREAM_SERVER_INITIATED))
  1463.     {
  1464.         qc->error = NGX_QUIC_ERR_STREAM_STATE_ERROR;
  1465.         return NGX_ERROR;
  1466.     }

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

  1470.     if (qs == NULL) {
  1471.         return NGX_ERROR;
  1472.     }

  1474.     if (qs == NGX_QUIC_STREAM_GONE) {
  1475.         return NGX_OK;
  1476.     }

  1478.     if (qs->recv_state == NGX_QUIC_STREAM_RECV_RESET_RECVD
  1479.         || qs->recv_state == NGX_QUIC_STREAM_RECV_RESET_READ)
  1480.     {
  1481.         return NGX_OK;
  1482.     }

  1484.     qs->recv_state = NGX_QUIC_STREAM_RECV_RESET_RECVD;

  1486.     if (ngx_quic_control_flow(qs, f->final_size) != NGX_OK) {
  1487.         return NGX_ERROR;
  1488.     }

  1490.     if (qs->recv_final_size != (uint64_t) -1
  1491.         && qs->recv_final_size != f->final_size)
  1492.     {
  1493.         qc->error = NGX_QUIC_ERR_FINAL_SIZE_ERROR;
  1494.         return NGX_ERROR;
  1495.     }

  1497.     if (qs->recv_last > f->final_size) {
  1498.         qc->error = NGX_QUIC_ERR_FINAL_SIZE_ERROR;
  1499.         return NGX_ERROR;
  1500.     }

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

  1504.     if (ngx_quic_update_flow(qs, qs->recv_final_size) != NGX_OK) {
  1505.         return NGX_ERROR;
  1506.     }

  1508.     if (qs->connection == NULL) {
  1509.         return ngx_quic_close_stream(qs);
  1510.     }

  1512.     rev = qs->connection->read;
  1513.     rev->error = 1;

  1515.     ngx_quic_set_event(rev);

  1517.     return NGX_OK;
  1518. }


  1521. ngx_int_t
  1522. ‌ngx_quic_handle_stop_sending_frame(ngx_connection_t *c,
  1523.     ngx_quic_header_t *pkt, ngx_quic_stop_sending_frame_t *f)
  1524. {
  1525.     ngx_quic_stream_t      *qs;
  1526.     ngx_quic_connection_t  *qc;

  1528.     qc = ngx_quic_get_connection(c);

  1530.     if ((f->id & NGX_QUIC_STREAM_UNIDIRECTIONAL)
  1531.         && (f->id & NGX_QUIC_STREAM_SERVER_INITIATED) == 0)
  1532.     {
  1533.         qc->error = NGX_QUIC_ERR_STREAM_STATE_ERROR;
  1534.         return NGX_ERROR;
  1535.     }

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

  1539.     if (qs == NULL) {
  1540.         return NGX_ERROR;
  1541.     }

  1543.     if (qs == NGX_QUIC_STREAM_GONE) {
  1544.         return NGX_OK;
  1545.     }

  1547.     if (ngx_quic_do_reset_stream(qs, f->error_code) != NGX_OK) {
  1548.         return NGX_ERROR;
  1549.     }

  1551.     if (qs->connection == NULL) {
  1552.         return ngx_quic_close_stream(qs);
  1553.     }

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

  1557.     return NGX_OK;
  1558. }


  1561. ngx_int_t
  1562. ‌ngx_quic_handle_max_streams_frame(ngx_connection_t *c,
  1563.     ngx_quic_header_t *pkt, ngx_quic_max_streams_frame_t *f)
  1564. {
  1565.     ngx_quic_connection_t  *qc;

  1567.     qc = ngx_quic_get_connection(c);

  1569.     if (f->bidi) {
  1570.         if (qc->streams.server_max_streams_bidi < f->limit) {
  1571.             qc->streams.server_max_streams_bidi = f->limit;

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

  1577.     } else {
  1578.         if (qc->streams.server_max_streams_uni < f->limit) {
  1579.             qc->streams.server_max_streams_uni = f->limit;

  1581.             ngx_log_debug1(NGX_LOG_DEBUG_EVENT, c->log, 0,
  1582.                            "quic max_streams_uni:%uL", f->limit);
  1583.         }
  1584.     }

  1586.     return NGX_OK;
  1587. }


  1590. void
  1591. ‌ngx_quic_handle_stream_ack(ngx_connection_t *c, ngx_quic_frame_t *f)
  1592. {
  1593.     uint64_t                acked;
  1594.     ngx_quic_stream_t      *qs;
  1595.     ngx_quic_connection_t  *qc;

  1597.     qc = ngx_quic_get_connection(c);

  1599.     switch (f->type) {

  1601.     case NGX_QUIC_FT_RESET_STREAM:

  1603.         qs = ngx_quic_find_stream(&qc->streams.tree, f->u.reset_stream.id);
  1604.         if (qs == NULL) {
  1605.             return;
  1606.         }

  1608.         qs->send_state = NGX_QUIC_STREAM_SEND_RESET_RECVD;

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

  1614.         break;

  1616.     case NGX_QUIC_FT_STREAM:

  1618.         qs = ngx_quic_find_stream(&qc->streams.tree, f->u.stream.stream_id);
  1619.         if (qs == NULL) {
  1620.             return;
  1621.         }

  1623.         acked = qs->acked;
  1624.         qs->acked += f->u.stream.length;

  1626.         if (f->u.stream.fin) {
  1627.             qs->fin_acked = 1;
  1628.         }

  1630.         if (qs->send_state == NGX_QUIC_STREAM_SEND_DATA_SENT
  1631.             && qs->acked == qs->sent && qs->fin_acked)
  1632.         {
  1633.             qs->send_state = NGX_QUIC_STREAM_SEND_DATA_RECVD;
  1634.         }

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

  1641.         if (qs->connection
  1642.             && qs->sent - acked == qc->conf->stream_buffer_size
  1643.             && f->u.stream.length > 0)
  1644.         {
  1645.             ngx_quic_set_event(qs->connection->write);
  1646.         }

  1648.         break;

  1650.     default:
  1651.         return;
  1652.     }

  1654.     if (qs->connection == NULL) {
  1655.         ngx_quic_close_stream(qs);
  1656.     }
  1657. }


  1660. static ngx_int_t
  1661. ‌ngx_quic_control_flow(ngx_quic_stream_t *qs, uint64_t last)
  1662. {
  1663.     uint64_t                len;
  1664.     ngx_connection_t       *pc;
  1665.     ngx_quic_connection_t  *qc;

  1667.     pc = qs->parent;
  1668.     qc = ngx_quic_get_connection(pc);

  1670.     if (last <= qs->recv_last) {
  1671.         return NGX_OK;
  1672.     }

  1674.     len = last - qs->recv_last;

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

  1681.     qs->recv_last += len;

  1683.     if (qs->recv_state == NGX_QUIC_STREAM_RECV_RECV
  1684.         && qs->recv_last > qs->recv_max_data)
  1685.     {
  1686.         qc->error = NGX_QUIC_ERR_FLOW_CONTROL_ERROR;
  1687.         return NGX_ERROR;
  1688.     }

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

  1692.     if (qc->streams.recv_last > qc->streams.recv_max_data) {
  1693.         qc->error = NGX_QUIC_ERR_FLOW_CONTROL_ERROR;
  1694.         return NGX_ERROR;
  1695.     }

  1697.     return NGX_OK;
  1698. }


  1701. static ngx_int_t
  1702. ‌ngx_quic_update_flow(ngx_quic_stream_t *qs, uint64_t last)
  1703. {
  1704.     uint64_t                len;
  1705.     ngx_connection_t       *pc;
  1706.     ngx_quic_connection_t  *qc;

  1708.     pc = qs->parent;
  1709.     qc = ngx_quic_get_connection(pc);

  1711.     if (last <= qs->recv_offset) {
  1712.         return NGX_OK;
  1713.     }

  1715.     len = last - qs->recv_offset;

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

  1720.     qs->recv_offset += len;

  1722.     if (qs->recv_max_data <= qs->recv_offset + qs->recv_window / 2) {
  1723.         if (ngx_quic_update_max_stream_data(qs) != NGX_OK) {
  1724.             return NGX_ERROR;
  1725.         }
  1726.     }

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

  1730.     if (qc->streams.recv_max_data
  1731.         <= qc->streams.recv_offset + qc->streams.recv_window / 2)
  1732.     {
  1733.         if (ngx_quic_update_max_data(pc) != NGX_OK) {
  1734.             return NGX_ERROR;
  1735.         }
  1736.     }

  1738.     return NGX_OK;
  1739. }


  1742. static ngx_int_t
  1743. ‌ngx_quic_update_max_stream_data(ngx_quic_stream_t *qs)
  1744. {
  1745.     uint64_t                recv_max_data;
  1746.     ngx_connection_t       *pc;
  1747.     ngx_quic_frame_t       *frame;
  1748.     ngx_quic_connection_t  *qc;

  1750.     pc = qs->parent;
  1751.     qc = ngx_quic_get_connection(pc);

  1753.     if (qs->recv_state != NGX_QUIC_STREAM_RECV_RECV) {
  1754.         return NGX_OK;
  1755.     }

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

  1759.     if (qs->recv_max_data == recv_max_data) {
  1760.         return NGX_OK;
  1761.     }

  1763.     qs->recv_max_data = recv_max_data;

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

  1769.     frame = ngx_quic_alloc_frame(pc);
  1770.     if (frame == NULL) {
  1771.         return NGX_ERROR;
  1772.     }

  1774.     frame->level = NGX_QUIC_ENCRYPTION_APPLICATION;
  1775.     frame->type = NGX_QUIC_FT_MAX_STREAM_DATA;
  1776.     frame->u.max_stream_data.id = qs->id;
  1777.     frame->u.max_stream_data.limit = qs->recv_max_data;

  1779.     ngx_quic_queue_frame(qc, frame);

  1781.     return NGX_OK;
  1782. }


  1785. static ngx_int_t
  1786. ‌ngx_quic_update_max_data(ngx_connection_t *c)
  1787. {
  1788.     uint64_t                recv_max_data;
  1789.     ngx_quic_frame_t       *frame;
  1790.     ngx_quic_connection_t  *qc;

  1792.     qc = ngx_quic_get_connection(c);

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

  1796.     if (qc->streams.recv_max_data == recv_max_data) {
  1797.         return NGX_OK;
  1798.     }

  1800.     qc->streams.recv_max_data = recv_max_data;

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

  1805.     frame = ngx_quic_alloc_frame(c);
  1806.     if (frame == NULL) {
  1807.         return NGX_ERROR;
  1808.     }

  1810.     frame->level = NGX_QUIC_ENCRYPTION_APPLICATION;
  1811.     frame->type = NGX_QUIC_FT_MAX_DATA;
  1812.     frame->u.max_data.max_data = qc->streams.recv_max_data;

  1814.     ngx_quic_queue_frame(qc, frame);

  1816.     return NGX_OK;
  1817. }


  1820. static void
  1821. ‌ngx_quic_set_event(ngx_event_t *ev)
  1822. {
  1823.     ev->ready = 1;

  1825.     if (ev->active) {
  1826.         ngx_post_event(ev, &ngx_posted_events);
  1827.     }
  1828. }

One Level Up Top Level