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 5c1f31bef451300f5cdf335261a7c9fd2aaedbae..4773b479b92047b0be10f87095e14605799e4db1 100644 (file)
--- 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<mutex> 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 19b4b9961191b39674fa1ed25ccc61cbbcf2111b..28ec66a0f96d5827792c6536f16d80837cb648b9 100644 (file)
--- 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 2c7c42ed75154fbff911fcdcf53421c7dd7500e3..6a5ef99f78fa80a8294b6bf9d3bf8ee52d0bb1c0 100644 (file)
--- 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<const string> 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 5d570fded6f6bec0a2387d12b845466d61eee1cf..931da2aa236e941ad72833a3af0d12cdbfac2ec2 100644 (file)
--- 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 <mutex>, since it calls add_data_raw().
+       void set_header(const std::string &new_http_header, const std::string &new_stream_header);
+
        // Mutex protecting <queued_data> and <queued_data_last_starting_point>.
        // Note that if you want to hold both this and the owning server's
        // <mutex> you will need to take <mutex> 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<FragmentStart> fragments;
        uint64_t first_fragment_index = 0, discontinuity_counter = 0;