quic_sstream.c
/*
* Copyright 2022-2023 The OpenSSL Project Authors. All Rights Reserved.
*
* Licensed under the Apache License 2.0 (the "License"). You may not use
* this file except in compliance with the License. You can obtain a copy
* in the file LICENSE in the source distribution or at
* https://www.openssl.org/source/license.html
*/
#include "internal/quic_stream.h"
#include "internal/uint_set.h"
#include "internal/common.h"
#include "internal/ring_buf.h"
/*
* ==================================================================
* QUIC Send Stream
*/
struct quic_sstream_st {
struct ring_buf ring_buf;
/*
* Any logical byte in the stream is in one of these states:
*
* - NEW: The byte has not yet been transmitted, or has been lost and is
* in need of retransmission.
*
* - IN_FLIGHT: The byte has been transmitted but is awaiting
* acknowledgement. We continue to store the data in case we return
* to the NEW state.
*
* - ACKED: The byte has been acknowledged and we can cease storing it.
* We do not necessarily cull it immediately, so there may be a delay
* between reaching the ACKED state and the buffer space actually being
* recycled.
*
* A logical byte in the stream is
*
* - in the NEW state if it is in new_set;
* - is in the ACKED state if it is in acked_set
* (and may or may not have been culled);
* - is in the IN_FLIGHT state otherwise.
*
* Invariant: No logical byte is ever in both new_set and acked_set.
*/
UINT_SET new_set, acked_set;
/*
* The current size of the stream is ring_buf.head_offset. If
* have_final_size is true, this is also the final size of the stream.
*/
unsigned int have_final_size : 1;
unsigned int sent_final_size : 1;
unsigned int acked_final_size : 1;
unsigned int cleanse : 1;
};
static void qss_cull(QUIC_SSTREAM *qss);
QUIC_SSTREAM *ossl_quic_sstream_new(size_t init_buf_size)
{
QUIC_SSTREAM *qss;
qss = OPENSSL_zalloc(sizeof(QUIC_SSTREAM));
if (qss == NULL)
return NULL;
ring_buf_init(&qss->ring_buf);
if (!ring_buf_resize(&qss->ring_buf, init_buf_size, 0)) {
ring_buf_destroy(&qss->ring_buf, 0);
OPENSSL_free(qss);
return NULL;
}
ossl_uint_set_init(&qss->new_set);
ossl_uint_set_init(&qss->acked_set);
return qss;
}
void ossl_quic_sstream_free(QUIC_SSTREAM *qss)
{
if (qss == NULL)
return;
ossl_uint_set_destroy(&qss->new_set);
ossl_uint_set_destroy(&qss->acked_set);
ring_buf_destroy(&qss->ring_buf, qss->cleanse);
OPENSSL_free(qss);
}
int ossl_quic_sstream_get_stream_frame(QUIC_SSTREAM *qss,
size_t skip,
OSSL_QUIC_FRAME_STREAM *hdr,
OSSL_QTX_IOVEC *iov,
size_t *num_iov)
{
size_t num_iov_ = 0, src_len = 0, total_len = 0, i;
uint64_t max_len;
const unsigned char *src = NULL;
UINT_SET_ITEM *range = ossl_list_uint_set_head(&qss->new_set);
if (*num_iov < 2)
return 0;
for (i = 0; i < skip && range != NULL; ++i)
range = ossl_list_uint_set_next(range);
if (range == NULL) {
if (i < skip)
/* Don't return FIN for infinitely increasing skip */
return 0;
/* No new bytes to send, but we might have a FIN */
if (!qss->have_final_size || qss->sent_final_size)
return 0;
hdr->offset = qss->ring_buf.head_offset;
hdr->len = 0;
hdr->is_fin = 1;
*num_iov = 0;
return 1;
}
/*
* We can only send a contiguous range of logical bytes in a single
* stream frame, so limit ourselves to the range of the first set entry.
*
* Set entries never have 'adjacent' entries so we don't have to worry
* about them here.
*/
max_len = range->range.end - range->range.start + 1;
for (i = 0;; ++i) {
if (total_len >= max_len)
break;
if (!ring_buf_get_buf_at(&qss->ring_buf,
range->range.start + total_len,
&src, &src_len))
return 0;
if (src_len == 0)
break;
assert(i < 2);
if (total_len + src_len > max_len)
src_len = (size_t)(max_len - total_len);
iov[num_iov_].buf = src;
iov[num_iov_].buf_len = src_len;
total_len += src_len;
++num_iov_;
}
hdr->offset = range->range.start;
hdr->len = total_len;
hdr->is_fin = qss->have_final_size
&& hdr->offset + hdr->len == qss->ring_buf.head_offset;
*num_iov = num_iov_;
return 1;
}
int ossl_quic_sstream_has_pending(QUIC_SSTREAM *qss)
{
OSSL_QUIC_FRAME_STREAM shdr;
OSSL_QTX_IOVEC iov[2];
size_t num_iov = OSSL_NELEM(iov);
return ossl_quic_sstream_get_stream_frame(qss, 0, &shdr, iov, &num_iov);
}
uint64_t ossl_quic_sstream_get_cur_size(QUIC_SSTREAM *qss)
{
return qss->ring_buf.head_offset;
}
int ossl_quic_sstream_mark_transmitted(QUIC_SSTREAM *qss,
uint64_t start,
uint64_t end)
{
UINT_RANGE r;
r.start = start;
r.end = end;
if (!ossl_uint_set_remove(&qss->new_set, &r))
return 0;
return 1;
}
int ossl_quic_sstream_mark_transmitted_fin(QUIC_SSTREAM *qss,
uint64_t final_size)
{
/*
* We do not really need final_size since we already know the size of the
* stream, but this serves as a sanity check.
*/
if (!qss->have_final_size || final_size != qss->ring_buf.head_offset)
return 0;
qss->sent_final_size = 1;
return 1;
}
int ossl_quic_sstream_mark_lost(QUIC_SSTREAM *qss,
uint64_t start,
uint64_t end)
{
UINT_RANGE r;
r.start = start;
r.end = end;
/*
* We lost a range of stream data bytes, so reinsert them into the new set,
* so that they are returned once more by ossl_quic_sstream_get_stream_frame.
*/
if (!ossl_uint_set_insert(&qss->new_set, &r))
return 0;
return 1;
}
int ossl_quic_sstream_mark_lost_fin(QUIC_SSTREAM *qss)
{
if (qss->acked_final_size)
/* Does not make sense to lose a FIN after it has been ACKed */
return 0;
/* FIN was lost, so we need to transmit it again. */
qss->sent_final_size = 0;
return 1;
}
int ossl_quic_sstream_mark_acked(QUIC_SSTREAM *qss,
uint64_t start,
uint64_t end)
{
UINT_RANGE r;
r.start = start;
r.end = end;
if (!ossl_uint_set_insert(&qss->acked_set, &r))
return 0;
qss_cull(qss);
return 1;
}
int ossl_quic_sstream_mark_acked_fin(QUIC_SSTREAM *qss)
{
if (!qss->have_final_size)
/* Cannot ack final size before we have a final size */
return 0;
qss->acked_final_size = 1;
return 1;
}
void ossl_quic_sstream_fin(QUIC_SSTREAM *qss)
{
if (qss->have_final_size)
return;
qss->have_final_size = 1;
}
int ossl_quic_sstream_get_final_size(QUIC_SSTREAM *qss, uint64_t *final_size)
{
if (!qss->have_final_size)
return 0;
if (final_size != NULL)
*final_size = qss->ring_buf.head_offset;
return 1;
}
int ossl_quic_sstream_append(QUIC_SSTREAM *qss,
const unsigned char *buf,
size_t buf_len,
size_t *consumed)
{
size_t l, consumed_ = 0;
UINT_RANGE r;
struct ring_buf old_ring_buf = qss->ring_buf;
if (qss->have_final_size) {
*consumed = 0;
return 0;
}
/*
* Note: It is assumed that ossl_quic_sstream_append will be called during a
* call to e.g. SSL_write and this function is therefore designed to support
* such semantics. In particular, the buffer pointed to by buf is only
* assumed to be valid for the duration of this call, therefore we must copy
* the data here. We will later copy-and-encrypt the data during packet
* encryption, so this is a two-copy design. Supporting a one-copy design in
* the future will require applications to use a different kind of API.
* Supporting such changes in future will require corresponding enhancements
* to this code.
*/
while (buf_len > 0) {
l = ring_buf_push(&qss->ring_buf, buf, buf_len);
if (l == 0)
break;
buf += l;
buf_len -= l;
consumed_ += l;
}
if (consumed_ > 0) {
r.start = old_ring_buf.head_offset;
r.end = r.start + consumed_ - 1;
assert(r.end + 1 == qss->ring_buf.head_offset);
if (!ossl_uint_set_insert(&qss->new_set, &r)) {
qss->ring_buf = old_ring_buf;
*consumed = 0;
return 0;
}
}
*consumed = consumed_;
return 1;
}
static void qss_cull(QUIC_SSTREAM *qss)
{
UINT_SET_ITEM *h = ossl_list_uint_set_head(&qss->acked_set);
/*
* Potentially cull data from our ring buffer. This can happen once data has
* been ACKed and we know we are never going to have to transmit it again.
*
* Since we use a ring buffer design for simplicity, we cannot cull byte n +
* k (for k > 0) from the ring buffer until byte n has also been culled.
* This means if parts of the stream get acknowledged out of order we might
* keep around some data we technically don't need to for a while. The
* impact of this is likely to be small and limited to quite a short
* duration, and doesn't justify the use of a more complex design.
*/
/*
* We only need to check the first range entry in the integer set because we
* can only cull contiguous areas at the start of the ring buffer anyway.
*/
if (h != NULL)
ring_buf_cpop_range(&qss->ring_buf, h->range.start, h->range.end,
qss->cleanse);
}
int ossl_quic_sstream_set_buffer_size(QUIC_SSTREAM *qss, size_t num_bytes)
{
return ring_buf_resize(&qss->ring_buf, num_bytes, qss->cleanse);
}
size_t ossl_quic_sstream_get_buffer_size(QUIC_SSTREAM *qss)
{
return qss->ring_buf.alloc;
}
size_t ossl_quic_sstream_get_buffer_used(QUIC_SSTREAM *qss)
{
return ring_buf_used(&qss->ring_buf);
}
size_t ossl_quic_sstream_get_buffer_avail(QUIC_SSTREAM *qss)
{
return ring_buf_avail(&qss->ring_buf);
}
int ossl_quic_sstream_is_totally_acked(QUIC_SSTREAM *qss)
{
UINT_RANGE r;
uint64_t cur_size;
if (qss->have_final_size && !qss->acked_final_size)
return 0;
if (ossl_quic_sstream_get_cur_size(qss) == 0)
return 1;
if (ossl_list_uint_set_num(&qss->acked_set) != 1)
return 0;
r = ossl_list_uint_set_head(&qss->acked_set)->range;
cur_size = qss->ring_buf.head_offset;
/*
* The invariants of UINT_SET guarantee a single list element if we have a
* single contiguous range, which is what we should have if everything has
* been acked.
*/
assert(r.end + 1 <= cur_size);
return r.start == 0 && r.end + 1 == cur_size;
}
void ossl_quic_sstream_adjust_iov(size_t len,
OSSL_QTX_IOVEC *iov,
size_t num_iov)
{
size_t running = 0, i, iovlen;
for (i = 0, running = 0; i < num_iov; ++i) {
iovlen = iov[i].buf_len;
if (running >= len)
iov[i].buf_len = 0;
else if (running + iovlen > len)
iov[i].buf_len = len - running;
running += iovlen;
}
}
void ossl_quic_sstream_set_cleanse(QUIC_SSTREAM *qss, int cleanse)
{
qss->cleanse = cleanse;
}
