From: Steinar H. Gunderson Date: Mon, 15 Apr 2019 07:33:36 +0000 (+0200) Subject: Keep the HLS backlog even if the stream header changes. X-Git-Tag: 1.4.3~4 X-Git-Url: https://git.sesse.net/?p=cubemap;a=commitdiff_plain;h=980ac162414c9fce62af4fdb9cfc282865b31572 Keep the HLS backlog even if the stream header changes. The typical case here is if we are streaming fMP4 and the encoder has to restart. The regular (non-HLS) stream has to cut off the backlog here, but HLS supports multiple headers, so we can add a discontinuity to keep the backlog seekable. I haven't actually found a client that will play correctly across the discontinuity, but hls.js can at least _seek_ across it, which is already useful. --- diff --git a/server.cpp b/server.cpp index 5c1f31b..4773b47 100644 --- a/server.cpp +++ b/server.cpp @@ -458,26 +458,7 @@ void Server::set_header(int stream_index, const string &http_header, const strin { lock_guard lock(mu); assert(stream_index >= 0 && stream_index < ssize_t(streams.size())); - Stream *stream = streams[stream_index].get(); - stream->http_header = http_header; - - if (stream_header != stream->stream_header) { - // We cannot start at any of the older starting points anymore, - // since they'd get the wrong header for the stream (not to mention - // that a changed header probably means the stream restarted, - // which means any client starting on the old one would probably - // stop playing properly at the change point). Next block - // should be a suitable starting point (if not, something is - // pretty strange), so it will fill up again soon enough. - stream->suitable_starting_points.clear(); - - if (!stream->fragments.empty()) { - stream->fragments.clear(); - ++stream->discontinuity_counter; - stream->clear_hls_playlist_cache(); - } - } - stream->stream_header = stream_header; + streams[stream_index]->set_header(http_header, stream_header); } void Server::set_pacing_rate(int stream_index, uint32_t pacing_rate) @@ -1116,9 +1097,6 @@ int Server::parse_request(Client *client) } Stream *stream = client->stream; - if (stream->http_header.empty()) { - return 503; // Service unavailable. - } if (client->serving_hls_playlist) { if (stream->encoding == Stream::STREAM_ENCODING_METACUBE) { @@ -1130,6 +1108,10 @@ int Server::parse_request(Client *client) } if (client->stream_pos_end == Client::STREAM_POS_NO_END) { + if (stream->http_header.empty()) { + return 503; // Service unavailable. + } + // 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). diff --git a/state.proto b/state.proto index 19b4b99..28ec66a 100644 --- a/state.proto +++ b/state.proto @@ -31,6 +31,7 @@ message ClientProto { message FragmentStartProto { optional int64 byte_position = 1; optional double pts = 2; + optional bool begins_header = 3 [default=false]; }; // Corresponds to struct Stream. diff --git a/stream.cpp b/stream.cpp index 2c7c42e..6a5ef99 100644 --- a/stream.cpp +++ b/stream.cpp @@ -77,7 +77,7 @@ Stream::Stream(const StreamProto &serialized, int data_fd) } for (const FragmentStartProto &fragment : serialized.fragment()) { - fragments.push_back(FragmentStart { size_t(fragment.byte_position()), fragment.pts() }); + fragments.push_back(FragmentStart { size_t(fragment.byte_position()), fragment.pts(), fragment.begins_header() }); } } @@ -96,6 +96,7 @@ StreamProto Stream::serialize() FragmentStartProto *proto = serialized.add_fragment(); proto->set_byte_position(fragment.byte_position); proto->set_pts(fragment.pts); + proto->set_begins_header(fragment.begins_header); } serialized.set_first_fragment_index(first_fragment_index); serialized.set_discontinuity_counter(discontinuity_counter); @@ -151,6 +152,45 @@ void Stream::set_backlog_size(size_t new_size) remove_obsolete_starting_points(); } +void Stream::set_header(const std::string &new_http_header, const std::string &new_stream_header) +{ + http_header = new_http_header; + if (new_stream_header == stream_header) { + return; + } + + // We cannot start at any of the older starting points anymore, + // since they'd get the wrong header for the stream (not to mention + // that a changed header probably means the stream restarted, + // which means any client starting on the old one would probably + // stop playing properly at the change point). Next block + // should be a suitable starting point (if not, something is + // pretty strange), so it will fill up again soon enough. + suitable_starting_points.clear(); + + // HLS, on the other hand, can deal with discontinuities and multiple + // headers. At least in theory (client support varies wildly). + if (!fragments.empty()) { + // Commit the old header to the backlog, so that we can serve it + // for all the old fragments for as long as they exist. + if (!stream_header.empty()) { + // End the current fragment and make a new one for the header. + fragments.push_back(Stream::FragmentStart { bytes_received, 0.0, true }); + process_queued_data(); + Stream::DataElement elem; + elem.data.iov_base = (char *)stream_header.data(); + elem.data.iov_len = stream_header.size(); + add_data_raw({ elem }); + remove_obsolete_starting_points(); + + // The discontinuity counter will be increased when + // this header goes out of the backlog. + } + clear_hls_playlist_cache(); + } + stream_header = new_stream_header; +} + void Stream::put_client_to_sleep(Client *client) { sleeping_clients.push_back(client); @@ -241,8 +281,12 @@ void Stream::remove_obsolete_starting_points() assert(backlog_size >= hls_backlog_margin); while (!fragments.empty() && bytes_received - fragments[0].byte_position > (backlog_size - hls_backlog_margin)) { + if (fragments[0].begins_header) { + ++discontinuity_counter; + } else { + ++first_fragment_index; + } fragments.pop_front(); - ++first_fragment_index; clear_hls_playlist_cache(); } } @@ -383,9 +427,11 @@ bool Stream::add_fragment_boundary(size_t byte_position, const RationalPTS &pts) { double pts_double = double(pts.pts) * pts.timebase_den / pts.timebase_num; - if (fragments.size() <= 1) { + if (fragments.size() <= 1 || + fragments[fragments.size() - 1].begins_header || + fragments[fragments.size() - 2].begins_header) { // Just starting up, so try to establish the first in-progress fragment. - fragments.push_back(FragmentStart{ byte_position, pts_double }); + fragments.push_back(FragmentStart{ byte_position, pts_double, false }); return false; } @@ -398,12 +444,12 @@ bool Stream::add_fragment_boundary(size_t byte_position, const RationalPTS &pts) double candidate_duration = pts_double - fragments[fragments.size() - 2].pts; if (lrintf(candidate_duration) <= hls_frag_duration && fabs(candidate_duration - hls_frag_duration) < fabs(current_duration - hls_frag_duration)) { - fragments.back() = FragmentStart{ byte_position, pts_double }; + fragments.back() = FragmentStart{ byte_position, pts_double, false }; return false; } else { // Extending the in-progress fragment would make it too long, // so finalize it and start a new in-progress fragment. - fragments.push_back(FragmentStart{ byte_position, pts_double }); + fragments.push_back(FragmentStart{ byte_position, pts_double, false }); return true; } } @@ -430,15 +476,57 @@ shared_ptr Stream::generate_hls_playlist(bool http_11, bool close_ string playlist = buf; - if (!stream_header.empty()) { - snprintf(buf, sizeof(buf), "#EXT-X-MAP:URI=\"%s?frag=header\"\r\n", url.c_str()); - playlist += buf; - } - - playlist += "\r\n"; if (fragments.size() >= 3) { + bool printed_header_for_this_group = false; + bool printed_first_header = false; for (size_t i = 0; i < fragments.size() - 2; ++i) { char buf[256]; + + if (fragments[i].begins_header) { + // End of this group. (We've already printed the header + // as part of the previous group.) + printed_header_for_this_group = false; + continue; + } + if (!printed_header_for_this_group) { + // Look forward until we find the header for this group (if any). + for (size_t j = i + 1; j < fragments.size() - 1; ++j) { + if (fragments[j].begins_header) { + if (printed_first_header) { + playlist += "#EXT-X-DISCONTINUITY\r\n"; + } + snprintf(buf, sizeof(buf), + "#EXT-X-MAP:URI=\"%s?frag=%" PRIu64 "-%" PRIu64 "\"\r\n", + url.c_str(), fragments[j].byte_position, + fragments[j + 1].byte_position); + playlist += buf; + printed_first_header = true; + printed_header_for_this_group = true; + break; + } + } + + if (!printed_header_for_this_group && !stream_header.empty()) { + if (printed_first_header) { + playlist += "#EXT-X-DISCONTINUITY\r\n"; + } + snprintf(buf, sizeof(buf), "#EXT-X-MAP:URI=\"%s?frag=header\"\r\n", url.c_str()); + playlist += buf; + } + + // Even if we didn't find anything, we don't want to search again for each fragment. + printed_first_header = true; + printed_header_for_this_group = true; + } + + if (fragments[i + 1].begins_header) { + // Since we only have start pts for each block and not duration, + // we have no idea how long this fragment is; the encoder restarted + // before it got to output the next pts. However, it's likely + // to be very short, so instead of trying to guess, we just skip it. + continue; + } + snprintf(buf, sizeof(buf), "#EXTINF:%f,\r\n%s?frag=%" PRIu64 "-%" PRIu64 "\r\n", fragments[i + 1].pts - fragments[i].pts, url.c_str(), diff --git a/stream.h b/stream.h index 5d570fd..931da2a 100644 --- a/stream.h +++ b/stream.h @@ -46,6 +46,9 @@ struct Stream { // Changes the backlog size, restructuring the data as needed. void set_backlog_size(size_t new_size); + // You should hold the owning Server's , since it calls add_data_raw(). + void set_header(const std::string &new_http_header, const std::string &new_stream_header); + // Mutex protecting and . // Note that if you want to hold both this and the owning server's // you will need to take before this one. @@ -112,7 +115,15 @@ struct Stream { // from points N-3..N-2. struct FragmentStart { uint64_t byte_position; - double pts; + double pts; // Unused if begins_header is true. + + // Whether the fragment started at this position is a stream header or not. + // Note that headers are stored _after_ the fragments they are headers for, + // so that they rotate out of the backlog last (and also because they are + // conveniently written then). The most current header is _not_ stored + // in the backlog; it is stored in stream_header. Only when replaced + // is it committed to the backlog and gets an entry here. + bool begins_header; }; std::deque fragments; uint64_t first_fragment_index = 0, discontinuity_counter = 0;