nginx-quic: src/event/ngx_event

comparison src/event/ngx_event_quic.c @ 7811:72d20158c814 quic

Added reordering support for STREAM frames. Each stream node now includes incoming frames queue and sent/received counters for tracking offset. The sent counter is not used, c->sent is used, not like in crypto buffers, which have no connections.

author	Vladimir Homutov <vl@nginx.com>
date	Wed, 15 Apr 2020 11:11:54 +0300
parents	167d32476737
children	76839f55bc48

comparison

equal deleted inserted replaced

-:167d32476737
+:72d20158c814
 uint64_t                          largest_ack; /* number received from peer */
 ngx_queue_t                       frames;
 ngx_queue_t                       sent;
 } ngx_quic_send_ctx_t;
-/* ordered frames stream context */
-typedef struct {
-uint64_t                          sent;
-uint64_t                          received;
-ngx_queue_t                       frames;
-size_t                            total; /* size of buffered data */
-} ngx_quic_frames_stream_t;
 struct ngx_quic_connection_s {
 ngx_str_t                         scid;
 ngx_str_t                         dcid;
 ngx_quic_frame_handler_pt handler);
 static ngx_int_t ngx_quic_crypto_input(ngx_connection_t *c,
 ngx_quic_frame_t *frame);
 static ngx_int_t ngx_quic_handle_stream_frame(ngx_connection_t *c,
-ngx_quic_header_t *pkt, ngx_quic_stream_frame_t *frame);
+ngx_quic_header_t *pkt, ngx_quic_frame_t *frame);
+static ngx_int_t ngx_quic_stream_input(ngx_connection_t *c,
+ngx_quic_frame_t *frame);
+static ngx_quic_stream_t *ngx_quic_add_stream(ngx_connection_t *c,
+ngx_quic_stream_frame_t *f);
 static ngx_int_t ngx_quic_handle_max_streams(ngx_connection_t *c);
 static ngx_int_t ngx_quic_handle_streams_blocked_frame(ngx_connection_t *c,
 ngx_quic_header_t *pkt, ngx_quic_streams_blocked_frame_t *f);
 static ngx_int_t ngx_quic_handle_stream_data_blocked_frame(ngx_connection_t *c,
 ngx_quic_header_t *pkt, ngx_quic_stream_data_blocked_frame_t *f);
 ngx_quic_close_connection(ngx_connection_t *c)
 {
 #if (NGX_DEBUG)
 ngx_uint_t              ns;
 #endif
+ngx_uint_t              i;
 ngx_pool_t             *pool;
 ngx_event_t            *rev;
 ngx_rbtree_t           *tree;
 ngx_rbtree_node_t      *node;
 ngx_quic_stream_t      *qs;
 ngx_quic_connection_t  *qc;
 ngx_log_debug0(NGX_LOG_DEBUG_EVENT, c->log, 0, "close quic connection");
-// TODO: free frames from reorder queue if any
 qc = c->quic;
 if (qc) {
+for (i = 0; i < NGX_QUIC_ENCRYPTION_LAST; i++) {
+ngx_quic_free_frames(c, &qc->crypto[i].frames);
+}
 qc->closing = 1;
 tree = &qc->streams.tree;
 if (tree->root != tree->sentinel) {
 if (c->read->timer_set) {
 case NGX_QUIC_FT_STREAM4:
 case NGX_QUIC_FT_STREAM5:
 case NGX_QUIC_FT_STREAM6:
 case NGX_QUIC_FT_STREAM7:
-if (ngx_quic_handle_stream_frame(c, pkt, &frame.u.stream)
+if (ngx_quic_handle_stream_frame(c, pkt, &frame) != NGX_OK) {
-!= NGX_OK)
-{
 return NGX_ERROR;
 }
 ack_this = 1;
 break;
 static ngx_int_t
 ngx_quic_handle_ordered_frame(ngx_connection_t *c, ngx_quic_frames_stream_t *fs,
 ngx_quic_frame_t *frame, ngx_quic_frame_handler_pt handler)
 {
 size_t                     full_len;
+ngx_int_t                  rc;
 ngx_queue_t               *q;
 ngx_quic_ordered_frame_t  *f;
 f = &frame->u.ord;
 /* intersecting data range, frame modified */
 }
 /* f->offset == fs->received */
-if (handler(c, frame) != NGX_OK) {
+rc = handler(c, frame);
-return NGX_ERROR;
+if (rc == NGX_ERROR) {
-}
+return NGX_ERROR;
+} else if (rc == NGX_DONE) {
+/* handler destroyed stream, queue no longer exists */
+return NGX_OK;
+}
+/* rc == NGX_OK */
 fs->received += f->length;
 /* now check the queue if we can continue with buffered frames */
 /* frame was adjusted, proceed to input */
 }
 /* f->offset == fs->received */
-if (handler(c, frame) != NGX_OK) {
+rc = handler(c, frame);
+if (rc == NGX_ERROR) {
 return NGX_ERROR;
+} else if (rc == NGX_DONE) {
+/* handler destroyed stream, queue no longer exists */
+return NGX_OK;
 }
 fs->received += f->length;
 fs->total -= full_len;
 return NGX_OK;
 }
 static ngx_int_t
-ngx_quic_handle_stream_frame(ngx_connection_t *c,
+ngx_quic_handle_stream_frame(ngx_connection_t *c, ngx_quic_header_t *pkt,
-ngx_quic_header_t *pkt, ngx_quic_stream_frame_t *f)
+ngx_quic_frame_t *frame)
 {
-size_t                  n;
+ngx_quic_stream_t         *sn;
-ngx_buf_t              *b;
+ngx_quic_connection_t     *qc;
-ngx_event_t            *rev;
+ngx_quic_stream_frame_t   *f;
-ngx_quic_stream_t      *sn;
+ngx_quic_frames_stream_t  *fs;
-ngx_quic_connection_t  *qc;
 qc = c->quic;
+f = &frame->u.stream;
 sn = ngx_quic_find_stream(&qc->streams.tree, f->stream_id);
-if (sn) {
+if (sn == NULL) {
+sn = ngx_quic_add_stream(c, f);
+if (sn == NULL) {
+return NGX_ERROR;
+}
+}
+fs = &sn->fs;
+return ngx_quic_handle_ordered_frame(c, fs, frame, ngx_quic_stream_input);
+}
+static ngx_int_t
+ngx_quic_stream_input(ngx_connection_t *c, ngx_quic_frame_t *frame)
+{
+ngx_buf_t                *b;
+ngx_event_t              *rev;
+ngx_quic_stream_t        *sn;
+ngx_quic_connection_t    *qc;
+ngx_quic_stream_frame_t  *f;
+qc = c->quic;
+f = &frame->u.stream;
+sn = ngx_quic_find_stream(&qc->streams.tree, f->stream_id);
+if (sn == NULL) {
+// TODO: possible?
+// deleted while stream is in reordering queue?
+return NGX_ERROR;
+}
+if (sn->fs.received != 0) {
 ngx_log_debug0(NGX_LOG_DEBUG_EVENT, c->log, 0, "existing stream");
 b = sn->b;
 if ((size_t) ((b->pos - b->start) + (b->end - b->last)) < f->length) {
 ngx_log_error(NGX_LOG_INFO, c->log, 0, "no space in stream buffer");
 if (rev->active) {
 rev->handler(rev);
 }
+/* check if stream was destroyed */
+if (ngx_quic_find_stream(&qc->streams.tree, f->stream_id) == NULL) {
+return NGX_DONE;
+}
 return NGX_OK;
 }
+b = sn->b;
+b->last = ngx_cpymem(b->last, f->data, f->length);
+rev = sn->c->read;
+rev->ready = 1;
+if (f->fin) {
+rev->pending_eof = 1;
+}
+if ((f->stream_id & NGX_QUIC_STREAM_UNIDIRECTIONAL) == 0) {
+ngx_quic_handle_max_streams(c);
+}
+qc->streams.handler(sn->c);
+/* check if stream was destroyed */
+if (ngx_quic_find_stream(&qc->streams.tree, f->stream_id) == NULL) {
+return NGX_DONE;
+}
+return NGX_OK;
+}
+static ngx_quic_stream_t *
+ngx_quic_add_stream(ngx_connection_t *c, ngx_quic_stream_frame_t *f)
+{
+size_t                  n;
+ngx_quic_stream_t      *sn;
+ngx_quic_connection_t  *qc;
+qc = c->quic;
+// TODO: check increasing IDs
 ngx_log_debug0(NGX_LOG_DEBUG_EVENT, c->log, 0, "stream is new");
 n = (f->stream_id & NGX_QUIC_STREAM_UNIDIRECTIONAL)
 ? qc->tp.initial_max_stream_data_uni
 n = NGX_QUIC_STREAM_BUFSIZE;
 }
 if (n < f->length) {
 ngx_log_error(NGX_LOG_INFO, c->log, 0, "no space in stream buffer");
-return NGX_ERROR;
+return NULL;
 }
 sn = ngx_quic_create_stream(c, f->stream_id, n);
 if (sn == NULL) {
-return NGX_ERROR;
+return NULL;
 }
-b = sn->b;
+return sn;
-b->last = ngx_cpymem(b->last, f->data, f->length);
-rev = sn->c->read;
-rev->ready = 1;
-if (f->fin) {
-rev->pending_eof = 1;
-}
-if ((f->stream_id & NGX_QUIC_STREAM_UNIDIRECTIONAL) == 0) {
-ngx_quic_handle_max_streams(c);
-}
-qc->streams.handler(sn->c);
-return NGX_OK;
 }
 static ngx_int_t
 ngx_quic_handle_max_streams(ngx_connection_t *c)
 } else {
 return NGX_ERROR;
 }
 } while (q != ngx_queue_sentinel(&ctx->frames));
 return NGX_OK;
 }
 ngx_quic_free_frames(ngx_connection_t *c, ngx_queue_t *frames)
 {
 ngx_queue_t       *q;
 ngx_quic_frame_t  *f;
-q = ngx_queue_head(frames);
 do {
+q = ngx_queue_head(frames);
+if (q == ngx_queue_sentinel(frames)) {
+break;
+}
+ngx_queue_remove(q);
 f = ngx_queue_data(q, ngx_quic_frame_t, queue);
-q = ngx_queue_next(q);
 ngx_quic_free_frame(c, f);
+} while (1);
-} while (q != ngx_queue_sentinel(frames));
 }
 /* pack a group of frames [start; end) into memory p and send as single packet */
 static ngx_int_t
 static void
 ngx_quic_push_handler(ngx_event_t *ev)
 {
-ngx_connection_t       *c;
+ngx_connection_t  *c;
 ngx_log_debug0(NGX_LOG_DEBUG_EVENT, ev->log, 0, "push timer");
 c = ev->data;
 sn->b = ngx_create_temp_buf(pool, rcvbuf_size);
 if (sn->b == NULL) {
 ngx_destroy_pool(pool);
 return NULL;
 }
+ngx_queue_init(&sn->fs.frames);
 log = ngx_palloc(pool, sizeof(ngx_log_t));
 if (log == NULL) {
 ngx_destroy_pool(pool);
 return NULL;
 if (qc->closing) {
 ngx_post_event(pc->read, &ngx_posted_events);
 return;
 }
+ngx_quic_free_frames(pc, &qs->fs.frames);
 if ((qs->id & 0x03) == NGX_QUIC_STREAM_UNIDIRECTIONAL) {
 /* do not send fin for client unidirectional streams */
 return;
 }

Mercurial > hg > nginx-quic

comparison src/event/ngx_event_quic.c @ 7811:72d20158c814 quic