Fix a crash when trying to get HLS fragments from a disconnected strema.

After the last patch, we would try to serve HLS fragments even if the
backend is down, but we had zeroed out the HTTP header, causing an
assertion failure with HTTP/1.1 clients (and an invalid HTTP response
for HTTP/1.0 clients). Fix by keeping the header and setting a special
“unavailable” flag instead (which allows us to keep sending HLS fragments).

diff --git a/httpinput.cpp b/httpinput.cpp
index cbffe3ae26292e11dd550156fce840120dd3031e..715e31c4a162f8d9ce168969901ae3ca4253a8c9 100644 (file)
--- a/httpinput.cpp
+++ b/httpinput.cpp
@@ -340,7 +340,8 @@ void HTTPInput::do_work()
                        pending_data.clear();
                        has_metacube_header = false;
                        for (int stream_index : stream_indices) {
-                               servers->set_header(stream_index, "", "");
+                               // Don't zero out the header; it might still be of use to HLS clients.
+                               servers->set_unavailable(stream_index);
                        }
 
                        {

diff --git a/server.cpp b/server.cpp
index 4773b479b92047b0be10f87095e14605799e4db1..35cb9d6bce869b354f73b96f1824e004cf1835ec 100644 (file)
--- a/server.cpp
+++ b/server.cpp
@@ -460,6 +460,13 @@ void Server::set_header(int stream_index, const string &http_header, const strin
        assert(stream_index >= 0 && stream_index < ssize_t(streams.size()));
        streams[stream_index]->set_header(http_header, stream_header);
 }
+
+void Server::set_unavailable(int stream_index)
+{
+       lock_guard<mutex> lock(mu);
+       assert(stream_index >= 0 && stream_index < ssize_t(streams.size()));
+       streams[stream_index]->set_unavailable();
+}
        
 void Server::set_pacing_rate(int stream_index, uint32_t pacing_rate)
 {
@@ -1108,7 +1115,7 @@ int Server::parse_request(Client *client)
        }
 
        if (client->stream_pos_end == Client::STREAM_POS_NO_END) {
-               if (stream->http_header.empty()) {
+               if (stream->unavailable) {
                        return 503;  // Service unavailable.
                }
 

diff --git a/server.h b/server.h
index d6e6e618c1c296fa10187c7fbd4fedb7712d4f02..0c172b5a306cc70f4fdd225d5708eaee4396b700 100644 (file)
--- a/server.h
+++ b/server.h
@@ -57,6 +57,9 @@ public:
                        const std::string &http_header,
                        const std::string &stream_header);
 
+       // Sets the given stream as unavailable (will always return 503 to new clients).
+       void set_unavailable(int stream_index);
+
        // Set that the given stream should use the given max pacing rate from now on.
        // NOTE: This should be set before any clients are connected!
        void set_pacing_rate(int stream_index, uint32_t pacing_rate);

diff --git a/serverpool.cpp b/serverpool.cpp
index c3fc69d04a3414fd00fee354966bc02a886c8d11..e9fea12944d13455ee7315685a32c7f2ed6fc9ae 100644 (file)
--- a/serverpool.cpp
+++ b/serverpool.cpp
@@ -178,6 +178,15 @@ void ServerPool::add_data(int stream_index, const char *data, size_t bytes, uint
        }
 }
 
+void ServerPool::set_unavailable(int stream_index)
+{
+       assert(stream_index >= 0 && stream_index < ssize_t(num_http_streams));
+
+       for (int i = 0; i < num_servers; ++i) {
+               servers[i].set_unavailable(stream_index);
+       }
+}
+
 void ServerPool::add_gen204(const std::string &url, const std::string &allow_origin)
 {
        for (int i = 0; i < num_servers; ++i) {

diff --git a/serverpool.h b/serverpool.h
index 9cac26e47133f270e91c040afa3091420433f387..253636a64c1de876b83fa051a97a1a12ff81cece 100644 (file)
--- a/serverpool.h
+++ b/serverpool.h
@@ -55,6 +55,9 @@ public:
                        const std::string &stream_header);
        void add_data(int stream_index, const char *data, size_t bytes, uint16_t metacube_flags, const RationalPTS &pts);
 
+       // Sets the given stream as unavailable on all the servers.
+       void set_unavailable(int stream_index);
+
        // Sets the max pacing rate for all the servers.
        void set_pacing_rate(int stream_index, uint32_t pacing_rate);
 

diff --git a/state.proto b/state.proto
index 28ec66a0f96d5827792c6536f16d80837cb648b9..c1e782fb5b71b6e64411bdc736b3a68d9b725b97 100644 (file)
--- a/state.proto
+++ b/state.proto
@@ -36,6 +36,7 @@ message FragmentStartProto {
 
 // Corresponds to struct Stream.
 message StreamProto {
+       optional bool unavailable = 14 [default=false];
        optional bytes http_header = 6;
        optional bytes stream_header = 7;
        repeated int32 data_fds = 8;

diff --git a/stream.cpp b/stream.cpp
index 6a5ef99f78fa80a8294b6bf9d3bf8ee52d0bb1c0..3ee1a434d2e6fe6c3cc73fcc84d89d860efd0f05 100644 (file)
--- a/stream.cpp
+++ b/stream.cpp
@@ -53,6 +53,7 @@ Stream::~Stream()
 
 Stream::Stream(const StreamProto &serialized, int data_fd)
        : url(serialized.url()),
+         unavailable(serialized.unavailable()),
          http_header(serialized.http_header()),
          stream_header(serialized.stream_header()),
          encoding(Stream::STREAM_ENCODING_RAW),  // Will be changed later.
@@ -84,6 +85,7 @@ Stream::Stream(const StreamProto &serialized, int data_fd)
 StreamProto Stream::serialize()
 {
        StreamProto serialized;
+       serialized.set_unavailable(unavailable);
        serialized.set_http_header(http_header);
        serialized.set_stream_header(stream_header);
        serialized.add_data_fds(data_fd);
@@ -154,6 +156,7 @@ void Stream::set_backlog_size(size_t new_size)
 
 void Stream::set_header(const std::string &new_http_header, const std::string &new_stream_header)
 {
+       unavailable = false;
        http_header = new_http_header;
        if (new_stream_header == stream_header) {
                return;

diff --git a/stream.h b/stream.h
index 931da2aa236e941ad72833a3af0d12cdbfac2ec2..9e384a9ffd141b63fee39f44c8140b29269d9020 100644 (file)
--- a/stream.h
+++ b/stream.h
@@ -47,8 +47,13 @@ struct Stream {
        void set_backlog_size(size_t new_size);
 
        // You should hold the owning Server's <mutex>, since it calls add_data_raw().
+       // Sets unavailable to false.
        void set_header(const std::string &new_http_header, const std::string &new_stream_header);
 
+       void set_unavailable() {
+               unavailable = true;
+       }
+
        // 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.
@@ -56,6 +61,10 @@ struct Stream {
 
        std::string url;
 
+       // If true, the backend is not completely connected, and thus, we cannot serve
+       // clients (except for historic HLS fragments).
+       bool unavailable = true;
+
        // The HTTP response header, without the trailing double newline.
        std::string http_header;