diff --git a/include/quicly.h b/include/quicly.h index e6f69c49..6d2aef23 100644 --- a/include/quicly.h +++ b/include/quicly.h @@ -1299,6 +1299,19 @@ int quicly_can_send_data(quicly_conn_t *conn, quicly_send_context_t *s); * the responsibility of the stream scheduler to maintain a list of such streams. */ quicly_error_t quicly_send_stream(quicly_stream_t *stream, quicly_send_context_t *s); +/** + * This function is same as `quicly_send_stream`, but can be used when the overhead of providing the stream payload can be minimized + * by using scattered I/O. + * @param emit_scattered the emit callback to be called instead of the standard `on_send_emit` callback; the callback returns + * the number of bytes that it has actually written + * @param available_bytes_hint bytes that might be available in the send buffer, starting from + * `stream->sendstate.acked.ranges[0].end`; `quicly_send_stream_scattered` prepares vectors only as + * much as required to send what has been hinted + */ +quicly_error_t quicly_send_stream_scattered(quicly_stream_t *stream, quicly_send_context_t *s, + size_t (*emit_scattered)(quicly_stream_t *stream, size_t off, const ptls_iovec_t *vecs, + size_t num_vecs, int *wrote_all), + size_t available_bytes_hint); /** * Builds a Version Negotiation packet. The generated packet might include a greasing version. * * @param versions zero-terminated list of versions to advertise; use `quicly_supported_versions` for sending the list of diff --git a/include/quicly/defaults.h b/include/quicly/defaults.h index 917c5ac1..93d84911 100644 --- a/include/quicly/defaults.h +++ b/include/quicly/defaults.h @@ -57,6 +57,14 @@ extern quicly_now_t quicly_default_now; */ extern quicly_crypto_engine_t quicly_default_crypto_engine; +int quicly_default_stream_scheduler_can_send(quicly_stream_scheduler_t *self, quicly_conn_t *conn, int conn_is_saturated); +quicly_error_t quicly_default_stream_scheduler_do_send_with(quicly_stream_scheduler_t *self, quicly_conn_t *conn, + quicly_send_context_t *s, + quicly_error_t (*do_send)(quicly_stream_t *, quicly_send_context_t *)); +quicly_error_t quicly_default_stream_scheduler_do_send(quicly_stream_scheduler_t *self, quicly_conn_t *conn, + quicly_send_context_t *s); +void quicly_default_stream_scheduler_update_state(quicly_stream_scheduler_t *self, quicly_stream_t *stream); + #define quicly_default_cc quicly_cc_type_reno #define quicly_default_init_cc quicly_cc_reno_init diff --git a/lib/defaults.c b/lib/defaults.c index 7406775f..fe9786eb 100644 --- a/lib/defaults.c +++ b/lib/defaults.c @@ -272,7 +272,7 @@ void quicly_free_default_cid_encryptor(quicly_cid_encryptor_t *_self) /** * See doc-comment of `st_quicly_default_scheduler_state_t` to understand the logic. */ -static int default_stream_scheduler_can_send(quicly_stream_scheduler_t *self, quicly_conn_t *conn, int conn_is_saturated) +int quicly_default_stream_scheduler_can_send(quicly_stream_scheduler_t *self, quicly_conn_t *conn, int conn_is_saturated) { struct st_quicly_default_scheduler_state_t *sched = &((struct _st_quicly_conn_public_t *)conn)->_default_scheduler; @@ -313,8 +313,9 @@ static void link_stream(struct st_quicly_default_scheduler_state_t *sched, quicl /** * See doc-comment of `st_quicly_default_scheduler_state_t` to understand the logic. */ -static quicly_error_t default_stream_scheduler_do_send(quicly_stream_scheduler_t *self, quicly_conn_t *conn, - quicly_send_context_t *s) +quicly_error_t quicly_default_stream_scheduler_do_send_with(quicly_stream_scheduler_t *self, quicly_conn_t *conn, + quicly_send_context_t *s, + quicly_error_t (*do_send)(quicly_stream_t *, quicly_send_context_t *)) { struct st_quicly_default_scheduler_state_t *sched = &((struct _st_quicly_conn_public_t *)conn)->_default_scheduler; int conn_is_blocked = quicly_is_blocked(conn); @@ -334,7 +335,7 @@ static quicly_error_t default_stream_scheduler_do_send(quicly_stream_scheduler_t continue; } /* send! */ - if ((ret = quicly_send_stream(stream, s)) != 0) { + if ((ret = do_send(stream, s)) != 0) { /* FIXME Stop quicly_send_stream emitting SENDBUF_FULL (happens when CWND is congested). Otherwise, we need to make * adjustments to the scheduler after popping a stream */ if (ret == QUICLY_ERROR_SENDBUF_FULL) { @@ -352,10 +353,16 @@ static quicly_error_t default_stream_scheduler_do_send(quicly_stream_scheduler_t return ret; } +quicly_error_t quicly_default_stream_scheduler_do_send(quicly_stream_scheduler_t *self, quicly_conn_t *conn, + quicly_send_context_t *s) +{ + return quicly_default_stream_scheduler_do_send_with(self, conn, s, quicly_send_stream); +} + /** * See doc-comment of `st_quicly_default_scheduler_state_t` to understand the logic. */ -static void default_stream_scheduler_update_state(quicly_stream_scheduler_t *self, quicly_stream_t *stream) +void quicly_default_stream_scheduler_update_state(quicly_stream_scheduler_t *self, quicly_stream_t *stream) { struct st_quicly_default_scheduler_state_t *sched = &((struct _st_quicly_conn_public_t *)stream->conn)->_default_scheduler; @@ -369,8 +376,11 @@ static void default_stream_scheduler_update_state(quicly_stream_scheduler_t *sel } } -quicly_stream_scheduler_t quicly_default_stream_scheduler = {default_stream_scheduler_can_send, default_stream_scheduler_do_send, - default_stream_scheduler_update_state}; +quicly_stream_scheduler_t quicly_default_stream_scheduler = { + .can_send = quicly_default_stream_scheduler_can_send, + .do_send = quicly_default_stream_scheduler_do_send, + .update_state = quicly_default_stream_scheduler_update_state, +}; quicly_stream_t *quicly_default_alloc_stream(quicly_context_t *ctx) { diff --git a/lib/quicly.c b/lib/quicly.c index e31fab4d..826e667a 100644 --- a/lib/quicly.c +++ b/lib/quicly.c @@ -4305,63 +4305,165 @@ int quicly_can_send_data(quicly_conn_t *conn, quicly_send_context_t *s) } /** - * If necessary, changes the frame representation from one without length field to one that has if necessary. Or, as an alternative, - * prepends PADDING frames. Upon return, `dst` points to the end of the frame being built. `*len`, `*wrote_all`, `*frame_type_at` - * are also updated reflecting their values post-adjustment. + * builds stream frame headers and setups the offset / scattered iovec arrays, and returns the size of the arrays */ -static inline void adjust_stream_frame_layout(uint8_t **dst, uint8_t *const dst_end, size_t *len, int *wrote_all, - uint8_t **frame_at) +static size_t prepare_scattered_emit(quicly_send_context_t *s, const uint16_t datagram_size, quicly_stream_id_t stream_id, + uint64_t off, size_t *len, ptls_iovec_t *vecs) { - size_t space_left = (dst_end - *dst) - *len, len_of_len = quicly_encodev_capacity(*len); + size_t num_vecs = 1, max_vecs = 10; - if (**frame_at == QUICLY_FRAME_TYPE_CRYPTO) { - /* CRYPTO frame: adjust payload length to make space for the length field, if necessary. */ - if (space_left < len_of_len) { - *len = dst_end - *dst - len_of_len; - *wrote_all = 0; - } - } else { - /* STREAM frame: insert length if space can be left for more frames. Otherwise, retain STREAM frame header omitting the - * length field, prepending PADDING if necessary. */ - if (space_left <= len_of_len) { - if (space_left != 0) { - memmove(*frame_at + space_left, *frame_at, *dst + *len - *frame_at); - memset(*frame_at, QUICLY_FRAME_TYPE_PADDING, space_left); - *dst += space_left; - *frame_at += space_left; - } - *dst += *len; - return; - } - **frame_at |= QUICLY_FRAME_TYPE_STREAM_BIT_LEN; + if (max_vecs > s->max_datagrams - s->num_datagrams) + max_vecs = s->max_datagrams - s->num_datagrams; + if (max_vecs > (s->send_window + datagram_size - 1) / datagram_size) + max_vecs = (s->send_window + datagram_size - 1) / datagram_size; + + const uint16_t packet_header_size = 1 + s->dcid->len + QUICLY_SEND_PN_SIZE; + const uint64_t max_off = off + *len; + const size_t tag_size = s->current.cipher->aead->algo->tag_size; + + off += vecs[0].len; + + /* build more frames (in the payload buffer in which additional datagrams will be allocated) */ + uint8_t *datagram_at = s->dst_end + tag_size; + assert(datagram_at == s->payload_buf.datagram + datagram_size); + for (; num_vecs < max_vecs && off < max_off && s->payload_buf.end - datagram_at >= datagram_size; + ++num_vecs, datagram_at += datagram_size) { + uint8_t *frame_at = datagram_at + packet_header_size, *p = frame_at; + *p++ = QUICLY_FRAME_TYPE_STREAM_BASE | QUICLY_FRAME_TYPE_STREAM_BIT_OFF; + p = quicly_encodev(p, stream_id); + p = quicly_encodev(p, off); + vecs[num_vecs] = ptls_iovec_init(p, datagram_at + datagram_size - tag_size - p); + off += vecs[num_vecs].len; + } + + /* adjust the size of the last vector, to avoid overflow of flow control */ + if (off > max_off) { + size_t overflow = off - max_off; + assert(vecs[num_vecs - 1].len > overflow); + vecs[num_vecs - 1].len -= overflow; + off -= overflow; + } + + *len = off - (max_off - *len); + return num_vecs; +} + +/** + * Assuming `dst` points to where the Length field of a CRYPTO frame should be inserted, inserts the field, changing `*len` and + * `*wrote_all` if necessary. Returns the end of the CRYPTO frame being adjusted. + */ +static uint8_t *adjust_crypto_frame_layout(uint8_t *dst, uint8_t *const dst_end, size_t *len, int *wrote_all) +{ + size_t space_left = (dst_end - dst) - *len, len_of_len = quicly_encodev_capacity(*len); + + if (space_left < len_of_len) { + *len = dst_end - dst - len_of_len; + *wrote_all = 0; } /* insert length before payload of `*len` bytes */ - memmove(*dst + len_of_len, *dst, *len); - *dst = quicly_encodev(*dst, *len); - *dst += *len; + memmove(dst + len_of_len, dst, *len); + dst = quicly_encodev(dst, *len); + dst += *len; + + return dst; } -quicly_error_t quicly_send_stream(quicly_stream_t *stream, quicly_send_context_t *s) +/** + * Assuming that `header` and `header_len` point to a STREAM frame without a Length field, either inserts a Length field or prepends + * a PADDING frame if necessary. Returns the size increase of the header. + */ +static size_t adjust_last_stream_frame(uint8_t *header, size_t header_len, uint16_t payload_size, size_t space_left) +{ + if (space_left == 0) + return 0; + + size_t len_len = quicly_encodev_capacity(payload_size); + + if (space_left <= len_len) { + /* prepend PADDING, as there is not enough space to insert the length field */ + memmove(header + header_len + space_left, header + header_len, payload_size); + memmove(header + space_left, header, header_len); + memset(header, QUICLY_FRAME_TYPE_PADDING, space_left); + return space_left; + } + + /* add the Length field */ + memmove(header + header_len + len_len, header + header_len, payload_size); + header[0] |= QUICLY_FRAME_TYPE_STREAM_BIT_LEN; + quicly_encodev(header + header_len, payload_size); + return len_len; +} + +static void commit_stream_frame(quicly_stream_t *stream, quicly_sent_t *sent, uint64_t off, const uint8_t *data, size_t len, + int wrote_all, int is_fin) +{ + /* update and log per-frame stats */ + if (stream->stream_id < 0) { + ++stream->conn->super.stats.num_frames_sent.crypto; + } else { + ++stream->conn->super.stats.num_frames_sent.stream; + } + QUICLY_PROBE(STREAM_SEND, stream->conn, stream->conn->stash.now, stream, off, data, len, is_fin, wrote_all); + QUICLY_LOG_CONN(stream_send, stream->conn, { + PTLS_LOG_ELEMENT_SIGNED(stream_id, stream->stream_id); + PTLS_LOG_ELEMENT_UNSIGNED(off, off); + PTLS_LOG_APPDATA_ELEMENT_HEXDUMP(data, data, len); + PTLS_LOG_ELEMENT_BOOL(is_fin, is_fin); + PTLS_LOG_ELEMENT_BOOL(wrote_all, wrote_all); + }); + QUICLY_PROBE(QUICTRACE_SEND_STREAM, stream->conn, stream->conn->stash.now, stream, off, len, is_fin); + + /* setup sentmap */ + sent->data.stream.stream_id = stream->stream_id; + sent->data.stream.args.start = off; + sent->data.stream.args.end = off + len + is_fin; +} + +static quicly_error_t update_stream_sendstate(quicly_stream_t *stream, uint64_t off, size_t len, int is_fin, int wrote_all) +{ + quicly_error_t ret; + + stream->conn->super.stats.num_bytes.stream_data_sent += len; + if (off < stream->sendstate.size_inflight) + stream->conn->super.stats.num_bytes.stream_data_resent += + (stream->sendstate.size_inflight < off + len ? stream->sendstate.size_inflight : off + len) - off; + + if (stream->sendstate.size_inflight < off + len) { + if (stream->stream_id >= 0) + stream->conn->egress.max_data.sent += off + len - stream->sendstate.size_inflight; + stream->sendstate.size_inflight = off + len; + } + if ((ret = quicly_ranges_subtract(&stream->sendstate.pending, off, off + len + is_fin)) != 0) + return ret; + if (wrote_all) { + if ((ret = quicly_ranges_subtract(&stream->sendstate.pending, stream->sendstate.size_inflight, UINT64_MAX)) != 0) + return ret; + } + + return ret; +} + +quicly_error_t quicly_send_stream_scattered(quicly_stream_t *stream, quicly_send_context_t *s, + size_t (*emit_scattered)(quicly_stream_t *, size_t, const ptls_iovec_t *, size_t, + int *), + size_t available_bytes_hint) { uint64_t off = stream->sendstate.pending.ranges[0].start; quicly_sent_t *sent; - uint8_t *dst; /* this pointer points to the current write position within the frame being built, while `s->dst` points to the - * beginning of the frame. */ - size_t len; + ptls_iovec_t payload_vecs[10]; + size_t num_vecs, len = available_bytes_hint; int wrote_all, is_fin; quicly_error_t ret; - /* write frame type, stream_id and offset, calculate capacity (and store that in `len`) */ + /* write frame type, stream_id and offset, cap `len` to the flow control limit */ if (stream->stream_id < 0) { if ((ret = allocate_ack_eliciting_frame(stream->conn, s, 1 + quicly_encodev_capacity(off) + 2 /* type + offset + len + 1-byte payload */, &sent, on_ack_stream)) != 0) return ret; - dst = s->dst; - *dst++ = QUICLY_FRAME_TYPE_CRYPTO; - dst = quicly_encodev(dst, off); - len = s->dst_end - dst; + s->dst[0] = QUICLY_FRAME_TYPE_CRYPTO; + payload_vecs[0].base = quicly_encodev(s->dst + 1, off); } else { uint8_t header[18], *hp = header + 1; hp = quicly_encodev(hp, stream->stream_id); @@ -4383,22 +4485,19 @@ quicly_error_t quicly_send_stream(quicly_stream_t *stream, quicly_send_context_t } memcpy(s->dst, header, hp - header); s->dst += hp - header; - len = 0; - wrote_all = 1; - is_fin = 1; - goto UpdateState; + commit_stream_frame(stream, sent, off, s->dst, 0, 1, 1); + update_stream_sendstate(stream, off, 0, 1, 1); + return 0; } if ((ret = allocate_ack_eliciting_frame(stream->conn, s, hp - header + 1, &sent, on_ack_stream)) != 0) return ret; - dst = s->dst; - memcpy(dst, header, hp - header); - dst += hp - header; - len = s->dst_end - dst; + memcpy(s->dst, header, hp - header); + payload_vecs[0].base = s->dst + (hp - header); /* cap by max_stream_data */ - if (off + len > stream->_send_aux.max_stream_data) + if (len > stream->_send_aux.max_stream_data - off) len = stream->_send_aux.max_stream_data - off; /* cap by max_data */ - if (off + len > stream->sendstate.size_inflight) { + if (len > stream->sendstate.size_inflight - off) { uint64_t new_bytes = off + len - stream->sendstate.size_inflight; if (new_bytes > stream->conn->egress.max_data.permitted - stream->conn->egress.max_data.sent) { size_t max_stream_data = @@ -4418,18 +4517,32 @@ quicly_error_t quicly_send_stream(quicly_stream_t *stream, quicly_send_context_t len = range_capacity; } - /* Write payload, adjusting len to actual size. Note that `on_send_emit` might fail (e.g., when underlying pread(2) fails), in - * which case the application will either close the connection immediately or reset the stream. If that happens, we return - * immediately without updating state. */ - assert(len != 0); + payload_vecs[0].len = s->dst_end - payload_vecs[0].base; + + if (emit_scattered != NULL && stream->stream_id >= 0 && get_epoch(s->current.first_byte) == QUICLY_EPOCH_1RTT) { + /* build (potentially) multiple STREAM frames spanning across multiple 1-RTT packets at once! */ + num_vecs = prepare_scattered_emit(s, stream->conn->egress.max_udp_payload_size, stream->stream_id, off, &len, payload_vecs); + } else { + /* build CYPTO / STREAM frame within the current datagram */ + if (len > payload_vecs[0].len) + len = payload_vecs[0].len; + emit_scattered = NULL; + } + + /* call the emit callback */ size_t emit_off = (size_t)(off - stream->sendstate.acked.ranges[0].end); + assert(len != 0); QUICLY_PROBE(STREAM_ON_SEND_EMIT, stream->conn, stream->conn->stash.now, stream, emit_off, len); QUICLY_LOG_CONN(stream_on_send_emit, stream->conn, { PTLS_LOG_ELEMENT_SIGNED(stream_id, stream->stream_id); PTLS_LOG_ELEMENT_UNSIGNED(off, off); PTLS_LOG_ELEMENT_UNSIGNED(capacity, len); }); - stream->callbacks->on_send_emit(stream, emit_off, dst, &len, &wrote_all); + if (emit_scattered != NULL) { + len = emit_scattered(stream, emit_off, payload_vecs, num_vecs, &wrote_all); + } else { + stream->callbacks->on_send_emit(stream, emit_off, payload_vecs[0].base, &len, &wrote_all); + } if (stream->conn->super.state >= QUICLY_STATE_CLOSING) { return QUICLY_ERROR_IS_CLOSING; } else if (stream->_send_aux.reset_stream.sender_state != QUICLY_SENDER_STATE_NONE) { @@ -4437,60 +4550,52 @@ quicly_error_t quicly_send_stream(quicly_stream_t *stream, quicly_send_context_t } assert(len != 0); - adjust_stream_frame_layout(&dst, s->dst_end, &len, &wrote_all, &s->dst); - - /* determine if the frame incorporates FIN */ - if (off + len == stream->sendstate.final_size) { - assert(!quicly_sendstate_is_open(&stream->sendstate)); - assert(s->dst != NULL); - is_fin = 1; - *s->dst |= QUICLY_FRAME_TYPE_STREAM_BIT_FIN; - } else { - is_fin = 0; - } - - /* update s->dst now that frame construction is complete */ - s->dst = dst; - -UpdateState: + /* adjust the frame layout and commit */ if (stream->stream_id < 0) { - ++stream->conn->super.stats.num_frames_sent.crypto; + /* CRYPTO frame */ + s->dst = adjust_crypto_frame_layout(payload_vecs[0].base, s->dst_end, &len, &wrote_all); + commit_stream_frame(stream, sent, off, s->dst - len, len, wrote_all, 0); + is_fin = 0; } else { - ++stream->conn->super.stats.num_frames_sent.stream; + /* STREAM frame */ + uint64_t frame_off = off; + size_t vec_index = 0; + for (; payload_vecs[vec_index].len < off + len - frame_off; ++vec_index) { + assert(payload_vecs[vec_index].base + payload_vecs[vec_index].len == s->dst_end); + commit_stream_frame(stream, sent, frame_off, payload_vecs[vec_index].base, payload_vecs[vec_index].len, 0, 0); + frame_off += payload_vecs[vec_index].len; + s->dst = s->dst_end; + if ((ret = allocate_ack_eliciting_frame(stream->conn, s, 1, &sent, on_ack_stream)) != 0) { + len = frame_off - off; + wrote_all = 0; + is_fin = 0; + goto UpdateStreamState; + } + } + if (off + len == stream->sendstate.final_size) { + assert(!quicly_sendstate_is_open(&stream->sendstate)); + is_fin = 1; + *s->dst |= QUICLY_FRAME_TYPE_STREAM_BIT_FIN; + } else { + is_fin = 0; + } + size_t last_frame_payload_size = off + len - frame_off; + size_t bytes_added = adjust_last_stream_frame(s->dst, payload_vecs[vec_index].base - s->dst, off + len - frame_off, + s->dst_end - (payload_vecs[vec_index].base + last_frame_payload_size)); + s->dst = payload_vecs[vec_index].base + last_frame_payload_size + bytes_added; + commit_stream_frame(stream, sent, frame_off, s->dst - last_frame_payload_size, last_frame_payload_size, wrote_all, is_fin); } - stream->conn->super.stats.num_bytes.stream_data_sent += len; - if (off < stream->sendstate.size_inflight) - stream->conn->super.stats.num_bytes.stream_data_resent += - (stream->sendstate.size_inflight < off + len ? stream->sendstate.size_inflight : off + len) - off; - QUICLY_PROBE(STREAM_SEND, stream->conn, stream->conn->stash.now, stream, off, s->dst - len, len, is_fin, wrote_all); - QUICLY_LOG_CONN(stream_send, stream->conn, { - PTLS_LOG_ELEMENT_SIGNED(stream_id, stream->stream_id); - PTLS_LOG_ELEMENT_UNSIGNED(off, off); - PTLS_LOG_APPDATA_ELEMENT_HEXDUMP(data, s->dst - len, len); - PTLS_LOG_ELEMENT_BOOL(is_fin, is_fin); - PTLS_LOG_ELEMENT_BOOL(wrote_all, wrote_all); - }); - QUICLY_PROBE(QUICTRACE_SEND_STREAM, stream->conn, stream->conn->stash.now, stream, off, len, is_fin); - /* update sendstate (and also MAX_DATA counter) */ - if (stream->sendstate.size_inflight < off + len) { - if (stream->stream_id >= 0) - stream->conn->egress.max_data.sent += off + len - stream->sendstate.size_inflight; - stream->sendstate.size_inflight = off + len; - } - if ((ret = quicly_ranges_subtract(&stream->sendstate.pending, off, off + len + is_fin)) != 0) - return ret; - if (wrote_all) { - if ((ret = quicly_ranges_subtract(&stream->sendstate.pending, stream->sendstate.size_inflight, UINT64_MAX)) != 0) - return ret; - } +UpdateStreamState: + /* update stream sendstate */ + update_stream_sendstate(stream, off, len, is_fin, wrote_all); - /* setup sentmap */ - sent->data.stream.stream_id = stream->stream_id; - sent->data.stream.args.start = off; - sent->data.stream.args.end = off + len + is_fin; + return ret; +} - return 0; +quicly_error_t quicly_send_stream(quicly_stream_t *stream, quicly_send_context_t *s) +{ + return quicly_send_stream_scattered(stream, s, NULL, SIZE_MAX); } static inline quicly_error_t init_acks_iter(quicly_conn_t *conn, quicly_sentmap_iter_t *iter) diff --git a/t/lossy.c b/t/lossy.c index 1b53a7ce..1cbff4d2 100644 --- a/t/lossy.c +++ b/t/lossy.c @@ -59,16 +59,14 @@ static void init_cond_even(struct loss_cond_t *cond) *cond = (struct loss_cond_t){cond_even_}; } +static ptls_cipher_context_t *cond_rand_cipher_ctx = NULL; + static int cond_rand_(struct loss_cond_t *cond) { - static ptls_cipher_context_t *c; - if (cond->data.rand_.bits_avail == 0) { - if (c == NULL) { - /* use different seed for each invocation */ + if (cond_rand_cipher_ctx == NULL) { static uint64_t key[2]; - c = ptls_cipher_new(&ptls_openssl_aes128ctr, 1, &key); - ++key[0]; + cond_rand_cipher_ctx = ptls_cipher_new(&ptls_openssl_aes128ctr, 1, &key); } /* initialize next `ntotal` bits, of which `nloss` bits are set */ cond->data.rand_.bits = 0; @@ -78,7 +76,7 @@ static int cond_rand_(struct loss_cond_t *cond) uint64_t mask; do { uint32_t v; - ptls_cipher_encrypt(c, &v, "01234567", 4); + ptls_cipher_encrypt(cond_rand_cipher_ctx, &v, "01234567", 4); mask = (uint64_t)1 << (v % cond->data.rand_.ratio.ntotal); } while ((cond->data.rand_.bits & mask) != 0); /* set the chosen bit */ @@ -536,6 +534,11 @@ static void test_bidirectional(void) void test_lossy(void) { + if (cond_rand_cipher_ctx != NULL) { + ptls_cipher_free(cond_rand_cipher_ctx); + cond_rand_cipher_ctx = NULL; + } + uint64_t handshake_timeout_backup = quic_ctx.handshake_timeout_rtt_multiplier; /* loss tests tend to incur gigantic (and artificial) latencies, which easily trigger handshake timeout. * for this test, we totally disable handshake timeout so we can focus on the loss test */ diff --git a/t/test.c b/t/test.c index ad2b508d..d67f5d60 100644 --- a/t/test.c +++ b/t/test.c @@ -99,6 +99,11 @@ int64_t quic_now = 1; quicly_context_t quic_ctx; quicly_stream_callbacks_t stream_callbacks = { on_destroy, quicly_streambuf_egress_shift, quicly_streambuf_egress_emit, on_egress_stop, on_ingress_receive, on_ingress_reset}; +quicly_stream_scheduler_t stream_scheduler = { + .can_send = quicly_default_stream_scheduler_can_send, + .do_send = quicly_default_stream_scheduler_do_send, + .update_state = quicly_default_stream_scheduler_update_state, +}; size_t on_destroy_callcnt; static void test_error_codes(void) @@ -155,6 +160,42 @@ static void test_error_codes(void) ok(!QUICLY_ERROR_IS_QUIC_APPLICATION(a)); } +static void test_adjust_crypto_frame_layout(void) +{ +#define TEST(_capacity, check) \ + do { \ + uint8_t buf[] = {0x06, 0x04, 'h', 'e', 'l', 'l', 'o', 0, 0, 0}; \ + uint8_t *dst = buf + 2, *const dst_end = buf + _capacity; \ + size_t len = 5; \ + int wrote_all = 1; \ + dst = adjust_crypto_frame_layout(dst, dst_end, &len, &wrote_all); \ + do { \ + check \ + } while (0); \ + } while (0); + + /* test CRYPTO frames that fit and don't when length is inserted */ + TEST(10, { + ok(dst == buf + 8); + ok(len == 5); + ok(wrote_all); + ok(memcmp(buf, "\x06\x04\x05hello", 8) == 0); + }); + TEST(18, { + ok(dst == buf + 8); + ok(len == 5); + ok(wrote_all); + ok(memcmp(buf, "\x06\x04\x05hello", 8) == 0); + }); + TEST(7, { + ok(dst == buf + 7); + ok(len == 4); + ok(!wrote_all); + ok(memcmp(buf, "\x06\x04\x04hell", 7) == 0); + }); +#undef TEST +} + static uint16_t test_enable_with_ratio255_random_value; static void test_enable_with_ratio255_get_random(void *p, size_t len) @@ -181,70 +222,164 @@ static void test_enable_with_ratio255(void) ok(num_enabled == 63 * (65535 / 255)); } -static void test_adjust_stream_frame_layout(void) +static void test_adjust_last_stream_frame(void) { -#define TEST(_is_crypto, _capacity, check) \ +#define TEST(space_left, check) \ do { \ - uint8_t buf[] = {0xff, 0x04, 'h', 'e', 'l', 'l', 'o', 0, 0, 0}; \ - uint8_t *dst = buf + 2, *const dst_end = buf + _capacity, *frame_at = buf; \ - size_t len = 5; \ - int wrote_all = 1; \ - buf[0] = _is_crypto ? 0x06 : 0x08; \ - adjust_stream_frame_layout(&dst, dst_end, &len, &wrote_all, &frame_at); \ + uint8_t buf[] = {0x08, 0x04, 'h', 'e', 'l', 'l', 'o', 0, 0, 0}; \ + size_t increase = adjust_last_stream_frame(buf, 2, 5, space_left); \ do { \ check \ } while (0); \ - } while (0); - - /* test CRYPTO frames that fit and don't when length is inserted */ - TEST(1, 10, { - ok(dst == buf + 8); - ok(len == 5); - ok(wrote_all); - ok(frame_at == buf); - ok(memcmp(buf, "\x06\x04\x05hello", 8) == 0); - }); - TEST(1, 8, { - ok(dst == buf + 8); - ok(len == 5); - ok(wrote_all); - ok(frame_at == buf); - ok(memcmp(buf, "\x06\x04\x05hello", 8) == 0); - }); - TEST(1, 7, { - ok(dst == buf + 7); - ok(len == 4); - ok(!wrote_all); - ok(frame_at == buf); - ok(memcmp(buf, "\x06\x04\x04hell", 7) == 0); - }); + } while (0) /* test STREAM frames */ - TEST(0, 9, { - ok(dst == buf + 8); - ok(len == 5); - ok(wrote_all); - ok(frame_at == buf); - ok(memcmp(buf, "\x0a\x04\x05hello", 8) == 0); + TEST(0, { + ok(increase == 0); + ok(memcmp(buf, "\x08\x04hello", 7) == 0); }); - TEST(0, 8, { - ok(dst == buf + 8); - ok(len == 5); - ok(wrote_all); - ok(frame_at == buf + 1); - ok(memcmp(buf, "\x00\x08\x04hello", 8) == 0); + TEST(1, { + ok(increase == 1); + ok(memcmp(buf, "\x00\x08\x04hello", 7) == 0); }); - TEST(0, 7, { - ok(dst == buf + 7); - ok(len == 5); - ok(wrote_all); - ok(frame_at == buf); - ok(memcmp(buf, "\x08\x04hello", 7) == 0); + TEST(2, { + ok(increase == 1); + ok(memcmp(buf, "\x0a\x04\x05hello", 7) == 0); }); #undef TEST } +static void test_prepare_scatter(void) +{ +#define DATAGRAM_SIZE 100 +#define STREAM_ID 4 +#define TAG_SIZE 16 + + quicly_cid_t dcid = {.cid = {0x41, 0x42, 0x43}, .len = 3}; + ptls_aead_context_t aead = {.algo = &ptls_openssl_aes128gcm}; + struct st_quicly_cipher_context_t cipher = {.aead = &aead}; + + assert(aead.algo->tag_size == TAG_SIZE); + +#define SETUP() \ + uint8_t buf[DATAGRAM_SIZE * 20]; \ + quicly_send_context_t s = { \ + .dcid = &dcid, \ + .current.cipher = &cipher, \ + .payload_buf.datagram = buf, \ + .payload_buf.end = buf + sizeof(buf), \ + .dst = buf + DATAGRAM_SIZE - 20 - TAG_SIZE, /* pretend as if only 20 bytes is left within the first datagram */ \ + .dst_end = buf + DATAGRAM_SIZE - TAG_SIZE, \ + .num_datagrams = 0, \ + .max_datagrams = 20, \ + .send_window = DATAGRAM_SIZE * 20, \ + }; \ + ptls_iovec_t vecs[10] = {{.base = s.dst + 2, .len = s.dst_end - (s.dst + 2)}}; /* 2 bytes for STREAM header(sid=0) */ \ + assert(vecs[0].len == 18) + +#define CHECK_VEC(index, expected_header, expected_len) \ + do { \ + ok(vecs[index].base == &buf[DATAGRAM_SIZE * (index) + 1 + dcid.len + QUICLY_SEND_PN_SIZE] + sizeof(expected_header) - 1); \ + ok(memcmp(vecs[index].base - (sizeof(expected_header) - 1), expected_header, sizeof(expected_header) - 1) == 0); \ + if ((expected_len) == SIZE_MAX) { \ + ok(vecs[index].base + vecs[index].len == buf + DATAGRAM_SIZE * ((index) + 1) - TAG_SIZE); \ + } else { \ + ok(vecs[index].len == (expected_len)); \ + } \ + } while (0) + + { /* basic check */ + SETUP(); + s.payload_buf.end = buf + DATAGRAM_SIZE * 5; /* limit output to 5 datagrams */ + size_t len = SIZE_MAX; + size_t num_vecs = prepare_scattered_emit(&s, DATAGRAM_SIZE, STREAM_ID, 0, &len, vecs); + ok(num_vecs == 5); + ok(vecs[0].len == 18); + CHECK_VEC(1, "\x0c\x04\x12", SIZE_MAX); + CHECK_VEC(2, "\x0c\x04\x40\x5d", SIZE_MAX); + CHECK_VEC(3, "\x0c\x04\x40\xa7", SIZE_MAX); + CHECK_VEC(4, "\x0c\x04\x40\xf1", SIZE_MAX); + } + + { /* no more than 10 vectors are expected to be returned */ + SETUP(); + size_t len = SIZE_MAX; + size_t num_vecs = prepare_scattered_emit(&s, DATAGRAM_SIZE, STREAM_ID, 0, &len, vecs); + ok(num_vecs == 10); + ok(vecs[0].len == 18); + CHECK_VEC(1, "\x0c\x04\x12", SIZE_MAX); + CHECK_VEC(2, "\x0c\x04\x40\x5d", SIZE_MAX); + CHECK_VEC(3, "\x0c\x04\x40\xa7", SIZE_MAX); + CHECK_VEC(4, "\x0c\x04\x40\xf1", SIZE_MAX); + } + + { /* if provided len is smaller than the space available in the first datagram, `vec[0].len` is reduced */ + SETUP(); + size_t len = 11; + size_t num_vecs = prepare_scattered_emit(&s, DATAGRAM_SIZE, STREAM_ID, 0, &len, vecs); + ok(num_vecs == 1); + ok(vecs[0].len == 11); + } + + { /* len might run out in the middle of the following datagrams */ + SETUP(); + size_t len = 150; + size_t num_vecs = prepare_scattered_emit(&s, DATAGRAM_SIZE, STREAM_ID, 0, &len, vecs); + ok(num_vecs == 3); + ok(vecs[0].len == 18); + CHECK_VEC(1, "\x0c\x04\x12", SIZE_MAX); + CHECK_VEC(2, "\x0c\x04\x40\x5d", 57); + } + + { /* prepare no more than the number of the datagrams that can be built */ + SETUP(); + s.num_datagrams = s.max_datagrams - 3; + size_t len = SIZE_MAX; + size_t num_vecs = prepare_scattered_emit(&s, DATAGRAM_SIZE, STREAM_ID, 0, &len, vecs); + ok(num_vecs == 3); + ok(vecs[0].len == 18); + CHECK_VEC(1, "\x0c\x04\x12", SIZE_MAX); + CHECK_VEC(2, "\x0c\x04\x40\x5d", SIZE_MAX); + } + + { /* prepare no more than send window */ + SETUP(); + s.send_window = DATAGRAM_SIZE; + size_t len = SIZE_MAX; + size_t num_vecs = prepare_scattered_emit(&s, DATAGRAM_SIZE, STREAM_ID, 0, &len, vecs); + ok(num_vecs == 1); + ok(vecs[0].len == 18); + } + + { /* prepare no more than send window (of 3 datagrams) */ + SETUP(); + s.send_window = DATAGRAM_SIZE * 3; + size_t len = SIZE_MAX; + size_t num_vecs = prepare_scattered_emit(&s, DATAGRAM_SIZE, STREAM_ID, 0, &len, vecs); + ok(num_vecs == 3); + ok(vecs[0].len == 18); + CHECK_VEC(1, "\x0c\x04\x12", SIZE_MAX); + CHECK_VEC(2, "\x0c\x04\x40\x5d", SIZE_MAX); + } + + { /* send window is rounded up to the datagram size */ + SETUP(); + s.send_window = DATAGRAM_SIZE + 1; + size_t len = SIZE_MAX; + size_t num_vecs = prepare_scattered_emit(&s, DATAGRAM_SIZE, STREAM_ID, 0, &len, vecs); + ok(num_vecs == 2); + ok(vecs[0].len == 18); + CHECK_VEC(1, "\x0c\x04\x12", SIZE_MAX); + } + +#undef CHECK_VEC +#undef SETUP +#undef DATAGRAM_SIZE +#undef STREAM_ID +#undef TAG_SIZE +} + static int64_t get_now_cb(quicly_now_t *self) { return quic_now; @@ -1102,6 +1237,50 @@ static void test_stats_foreach(void) #undef CHECK } +static size_t do_emit_scattered(quicly_stream_t *stream, size_t off, const ptls_iovec_t *vecs, size_t num_vecs, int *wrote_all) +{ + quicly_streambuf_t *sbuf = stream->data; + size_t bytes_written = 0; + + for (size_t i = 0; i < num_vecs; ++i) { + size_t vec_len = vecs[i].len; + quicly_sendbuf_emit(stream, &sbuf->egress, off, vecs[i].base, &vec_len, wrote_all); + bytes_written += vec_len; + if (*wrote_all) + break; + assert(vec_len == vecs[i].len); + off += vec_len; + } + + return bytes_written; +} + +static quicly_error_t do_send_scattered(quicly_stream_t *stream, quicly_send_context_t *s) +{ + assert(stream->callbacks == &stream_callbacks); + return quicly_send_stream_scattered(stream, s, do_emit_scattered, SIZE_MAX); +} + +static quicly_error_t scheduler_do_send_scattered(quicly_stream_scheduler_t *self, quicly_conn_t *conn, quicly_send_context_t *s) +{ + return quicly_default_stream_scheduler_do_send_with(self, conn, s, do_send_scattered); +} + +static void run_stream_tests(int scatter) +{ + if (scatter) { + assert(stream_scheduler.do_send == quicly_default_stream_scheduler_do_send); + stream_scheduler.do_send = scheduler_do_send_scattered; + } + + subtest("simple", test_simple); + subtest("stream-concurrency", test_stream_concurrency); + subtest("lossy", test_lossy); + + if (scatter) + stream_scheduler.do_send = quicly_default_stream_scheduler_do_send; +} + int main(int argc, char **argv) { static ptls_iovec_t cert; @@ -1118,6 +1297,7 @@ int main(int argc, char **argv) quic_ctx.transport_params.max_streams_bidi = 10; quic_ctx.stream_open = &stream_open; quic_ctx.now = &get_now; + quic_ctx.stream_scheduler = &stream_scheduler; fake_address.sa.sa_family = AF_INET; @@ -1161,14 +1341,15 @@ int main(int argc, char **argv) subtest("pacer", test_pacer); subtest("sentmap", test_sentmap); subtest("loss", test_loss); - subtest("adjust-stream-frame-layout", test_adjust_stream_frame_layout); + subtest("adjust-crypto-frame-layout", test_adjust_crypto_frame_layout); + subtest("adjust-last-stream-frame", test_adjust_last_stream_frame); + subtest("prepare-scatter", test_prepare_scatter); subtest("test-vector", test_vector); subtest("test-retry-aead", test_retry_aead); subtest("transport-parameters", test_transport_parameters); subtest("cid", test_cid); - subtest("simple", test_simple); - subtest("stream-concurrency", test_stream_concurrency); - subtest("lossy", test_lossy); + subtest("stream-subtests", run_stream_tests, 0); + subtest("stream-subtests-scattered", run_stream_tests, 1); subtest("test-nondecryptable-initial", test_nondecryptable_initial); subtest("set_cc", test_set_cc); subtest("ecn-index-from-bits", test_ecn_index_from_bits);