From 019d476c729a6c1b6ac001e345d44339ba423221 Mon Sep 17 00:00:00 2001 From: "Steinar H. Gunderson" Date: Wed, 4 Apr 2018 00:40:33 +0200 Subject: [PATCH] Support delivering fragments from the backlog; intended for HLS. --- client.cpp | 2 + client.h | 4 ++ server.cpp | 110 +++++++++++++++++++++++++++++++++++++++++++++++----- state.proto | 1 + 4 files changed, 107 insertions(+), 10 deletions(-) diff --git a/client.cpp b/client.cpp index 4b7bcb7..c2cbcf5 100644 --- a/client.cpp +++ b/client.cpp @@ -66,6 +66,7 @@ Client::Client(const ClientProto &serialized, Stream *stream) header_or_short_response(serialized.header_or_short_response()), header_or_short_response_bytes_sent(serialized.header_or_short_response_bytes_sent()), stream_pos(serialized.stream_pos()), + stream_pos_end(serialized.stream_pos_end()), bytes_sent(serialized.bytes_sent()), bytes_lost(serialized.bytes_lost()), num_loss_events(serialized.num_loss_events()) @@ -120,6 +121,7 @@ ClientProto Client::serialize() const serialized.set_header_or_short_response(header_or_short_response); serialized.set_header_or_short_response_bytes_sent(serialized.header_or_short_response_bytes_sent()); serialized.set_stream_pos(stream_pos); + serialized.set_stream_pos_end(stream_pos_end); serialized.set_bytes_sent(bytes_sent); serialized.set_bytes_lost(bytes_lost); serialized.set_num_loss_events(num_loss_events); diff --git a/client.h b/client.h index ed04bc0..f1d14e6 100644 --- a/client.h +++ b/client.h @@ -87,6 +87,10 @@ struct Client { // these negative values will be translated to real numbers. size_t stream_pos = 0; + // Position at which to end the stream (one-past-the-end, used for fragments). + // -1 means never to end; this is the common case. + size_t stream_pos_end = 0; + // Number of bytes we've sent of data. Only relevant for SENDING_DATA. size_t bytes_sent = 0; diff --git a/server.cpp b/server.cpp index 1deb087..e01c9f4 100644 --- a/server.cpp +++ b/server.cpp @@ -545,12 +545,20 @@ sending_header_or_short_response_again: 0); client->state = Client::SENDING_DATA; goto sending_data; + } else if (client->stream_pos_end != Client::STREAM_POS_NO_END) { + // We're sending a fragment, and should have all of it, + // so start sending right away. + assert(client->stream_pos >= 0); + client->state = Client::SENDING_DATA; + goto sending_data; } else if (stream->prebuffering_bytes == 0) { // Start sending from the first keyframe we get. In other // words, we won't send any of the backlog, but we'll start // sending immediately as we get the next keyframe block. // Note that this is functionally identical to the next if branch, // except that we save a binary search. + assert(client->stream_pos == size_t(-1)); + assert(client->stream_pos_end == size_t(-1)); client->stream_pos = stream->bytes_received; client->state = Client::WAITING_FOR_KEYFRAME; } else { @@ -560,6 +568,8 @@ sending_header_or_short_response_again: // so that we are conservative and never add extra latency over just // waiting (assuming CBR or nearly so); otherwise, we could want e.g. // 100 kB prebuffer but end up sending a 10 MB GOP. + assert(client->stream_pos == size_t(-1)); + assert(client->stream_pos_end == size_t(-1)); deque::const_iterator starting_point_it = lower_bound(stream->suitable_starting_points.begin(), stream->suitable_starting_points.end(), @@ -611,9 +621,24 @@ sending_data: Stream *stream = client->stream; sending_data_again: - size_t bytes_to_send = stream->bytes_received - client->stream_pos; + size_t bytes_to_send; + if (client->stream_pos_end == size_t(-1)) { + bytes_to_send = stream->bytes_received - client->stream_pos; + } else { + bytes_to_send = client->stream_pos_end - client->stream_pos; + } assert(bytes_to_send <= stream->backlog_size); if (bytes_to_send == 0) { + if (client->stream_pos == client->stream_pos_end) { // We have a definite end, and we're at it. + if (more_requests(client)) { + // We're done sending the fragment, but should keep on reading new requests. + goto read_request_again; + } else { + // We're done sending the fragment, so now close. + // This is postcondition #1. + close_client(client); + } + } return; } @@ -645,13 +670,17 @@ sending_data_again: client->stream_pos += ret; client->bytes_sent += ret; - if (client->stream_pos == stream->bytes_received) { + assert(client->stream_pos_end == size_t(-1) || client->stream_pos <= client->stream_pos_end); + if (client->stream_pos == client->stream_pos_end) { + goto sending_data_again; // Will see that bytes_to_send == 0 and end. + } else if (client->stream_pos == stream->bytes_received) { // We don't have any more data for this client, so put it to sleep. // This is postcondition #3. stream->put_client_to_sleep(client); } else if (more_data && size_t(ret) == bytes_to_send) { goto sending_data_again; } + // We'll also get here for postcondition #4 (similar to the EAGAIN path above). break; } default: @@ -797,9 +826,19 @@ void Server::skip_lost_data(Client *client) size_t bytes_to_send = stream->bytes_received - client->stream_pos; if (bytes_to_send > stream->backlog_size) { size_t bytes_lost = bytes_to_send - stream->backlog_size; - client->stream_pos = stream->bytes_received - stream->backlog_size; client->bytes_lost += bytes_lost; ++client->num_loss_events; + if (!client->close_after_response) { + assert(client->stream_pos_end != size_t(-1)); + + // We've already sent a Content-length, so we can't just skip data. + // Close the connection immediately and hope the other side + // is able to figure out that there was an error and it needs to skip. + client->close_after_response = true; + client->stream_pos = client->stream_pos_end; + } else { + client->stream_pos = stream->bytes_received - stream->backlog_size; + } } } @@ -837,7 +876,36 @@ int Server::parse_request(Client *client) client->stream_pos = -2; url = url.substr(0, url.size() - 8); } else { - client->stream_pos = -1; + size_t pos = url.find("?frag="); + if (pos != string::npos) { + // Parse an endpoint of the type /stream.mp4?frag=1234-5678. + const char *ptr = url.c_str() + pos + 6; + char *endptr; + long long frag_start = strtol(ptr, &endptr, 10); + if (ptr == endptr || frag_start < 0 || frag_start == LLONG_MAX) { + return 400; // Bad request. + } + if (*endptr != '-') { + return 400; // Bad request. + } + ptr = endptr + 1; + + long long frag_end = strtol(ptr, &endptr, 10); + if (ptr == endptr || frag_end < frag_start || frag_end == LLONG_MAX) { + return 400; // Bad request. + } + + if (*endptr != '\0') { + return 400; // Bad request. + } + + url = url.substr(0, pos); + client->stream_pos = frag_start; + client->stream_pos_end = frag_end; + } else { + client->stream_pos = -1; + client->stream_pos_end = -1; + } } // Figure out if we're supposed to close the socket after we've delivered the response. @@ -873,11 +941,26 @@ int Server::parse_request(Client *client) return 503; // Service unavailable. } - // Streams currently never end, so we don't have a content-length, - // and can just as well tell the client it's Connection: close - // (otherwise, we'd have to implement chunking TE for no good reason). - // When we start to support fragments, this will change. - client->close_after_response = true; + if (client->stream_pos_end == size_t(-1)) { + // This stream won't end, so we don't have a content-length, + // and can just as well tell the client it's Connection: close + // (otherwise, we'd have to implement chunking TE for no good reason). + client->close_after_response = true; + } else { + if (stream->encoding == Stream::STREAM_ENCODING_METACUBE) { + // This doesn't make any sense, and is hard to implement, too. + return 416; // Range not satisfiable. + } + + // Check that we have the requested fragment in our backlog. + size_t buffer_end = stream->bytes_received; + size_t buffer_start = (buffer_end <= stream->backlog_size) ? 0 : buffer_end - stream->backlog_size; + + if (client->stream_pos_end > buffer_end || + client->stream_pos < buffer_start) { + return 416; // Range not satisfiable. + } + } client->stream = stream; if (setsockopt(client->sock, SOL_SOCKET, SO_MAX_PACING_RATE, &client->stream->pacing_rate, sizeof(client->stream->pacing_rate)) == -1) { @@ -894,6 +977,11 @@ void Server::construct_header(Client *client) { Stream *stream = client->stream; client->header_or_short_response = stream->http_header; + if (client->stream_pos_end != size_t(-1)) { + char buf[64]; + snprintf(buf, sizeof(buf), "Content-length: %zu\r\n", client->stream_pos_end - client->stream_pos); + client->header_or_short_response.append(buf); + } if (client->http_11) { assert(client->header_or_short_response.find("HTTP/1.0") == 0); client->header_or_short_response[7] = '1'; // Change to HTTP/1.1. @@ -921,7 +1009,9 @@ void Server::construct_header(Client *client) } else { assert(false); } - client->header_or_short_response.append(stream->stream_header); + if (client->stream_pos_end == size_t(-1)) { // Fragments don't contain stream headers. + client->header_or_short_response.append(stream->stream_header); + } // Switch states. client->state = Client::SENDING_HEADER; diff --git a/state.proto b/state.proto index e6aefb5..7643bbf 100644 --- a/state.proto +++ b/state.proto @@ -12,6 +12,7 @@ message ClientProto { optional bytes header_or_short_response = 5; optional int64 header_or_short_response_bytes_sent = 6; optional int64 stream_pos = 7; + optional int64 stream_pos_end = 20 [default=-1]; optional int64 bytes_sent = 10; optional int64 bytes_lost = 11; optional int64 num_loss_events = 12; -- 2.39.2