Fix an issue where Metacube output would not include updated headers (e.g. on encoder...

diff --git a/httpinput.cpp b/httpinput.cpp
index 5fb55b378610e846f2d74f6074f82f1f95ee58b7..5a93969ff4801c425ca9dc418cc030147e12c808 100644 (file)
--- a/httpinput.cpp
+++ b/httpinput.cpp
@@ -490,7 +490,7 @@ void HTTPInput::process_data(char *ptr, size_t bytes)
 
        if (encoding == Input::INPUT_ENCODING_RAW) {
                for (size_t i = 0; i < stream_indices.size(); ++i) {
 
        if (encoding == Input::INPUT_ENCODING_RAW) {
                for (size_t i = 0; i < stream_indices.size(); ++i) {

-                       servers->add_data(stream_indices[i], ptr, bytes, SUITABLE_FOR_STREAM_START);
+                       servers->add_data(stream_indices[i], ptr, bytes, /*metacube_flags=*/0);

                }
                return;
        }
                }
                return;
        }


@@ -568,16 +568,9 @@ void HTTPInput::process_data(char *ptr, size_t bytes)
                        for (size_t i = 0; i < stream_indices.size(); ++i) {
                                servers->set_header(stream_indices[i], http_header, stream_header);
                        }
                        for (size_t i = 0; i < stream_indices.size(); ++i) {
                                servers->set_header(stream_indices[i], http_header, stream_header);
                        }

-               } else {
-                       StreamStartSuitability suitable_for_stream_start;
-                       if (flags & METACUBE_FLAGS_NOT_SUITABLE_FOR_STREAM_START) {
-                               suitable_for_stream_start = NOT_SUITABLE_FOR_STREAM_START;
-                       } else {
-                               suitable_for_stream_start = SUITABLE_FOR_STREAM_START;
-                       }
-                       for (size_t i = 0; i < stream_indices.size(); ++i) {
-                               servers->add_data(stream_indices[i], inner_data, size, suitable_for_stream_start);
-                       }
+               }
+               for (size_t i = 0; i < stream_indices.size(); ++i) {
+                       servers->add_data(stream_indices[i], inner_data, size, flags);

                }
 
                // Consume the block. This isn't the most efficient way of dealing with things
                }
 
                // Consume the block. This isn't the most efficient way of dealing with things

diff --git a/server.cpp b/server.cpp
index d3b936f0b9c1f869c34c5dbf05c1beb32ad90a8e..afaab9be6a3cfb0f464de29582554744e8faac70 100644 (file)
--- a/server.cpp
+++ b/server.cpp
@@ -373,10 +373,10 @@ void Server::add_gen204(const std::string &url, const std::string &allow_origin)
        ping_url_map[url] = allow_origin;
 }
 
        ping_url_map[url] = allow_origin;
 }
 

-void Server::add_data_deferred(int stream_index, const char *data, size_t bytes, StreamStartSuitability suitable_for_stream_start)
+void Server::add_data_deferred(int stream_index, const char *data, size_t bytes, uint16_t metacube_flags)

 {
        assert(stream_index >= 0 && stream_index < ssize_t(streams.size()));
 {
        assert(stream_index >= 0 && stream_index < ssize_t(streams.size()));

-       streams[stream_index]->add_data_deferred(data, bytes, suitable_for_stream_start);
+       streams[stream_index]->add_data_deferred(data, bytes, metacube_flags);

 }
 
 // See the .h file for postconditions after this function.     
 }
 
 // See the .h file for postconditions after this function.     

diff --git a/server.h b/server.h
index 320fcbcd8da8a274277a9633fbc0e0f1ca95a90c..b05f661aa6641d5aed95953b98b1d103fee8702f 100644 (file)
--- a/server.h
+++ b/server.h
@@ -48,7 +48,7 @@ public:
        // and the order between them are undefined.
        // XXX: header should ideally be ordered with respect to data.
        void add_client_deferred(int sock);
        // and the order between them are undefined.
        // XXX: header should ideally be ordered with respect to data.
        void add_client_deferred(int sock);

-       void add_data_deferred(int stream_index, const char *data, size_t bytes, StreamStartSuitability suitable_for_stream_start);
+       void add_data_deferred(int stream_index, const char *data, size_t bytes, uint16_t metacube_flags);

 
        // These should not be called while running, since that would violate
        // threading assumptions (ie., that epoll is only called from one thread
 
        // These should not be called while running, since that would violate
        // threading assumptions (ie., that epoll is only called from one thread

diff --git a/serverpool.cpp b/serverpool.cpp
index 2c663fbce37dd3d4e851a05de866bc7ceb9fb7e6..3e79b45ba41d065aa19021efe549eb1b65c08be6 100644 (file)
--- a/serverpool.cpp
+++ b/serverpool.cpp
@@ -147,7 +147,7 @@ void ServerPool::set_header(int stream_index, const string &http_header, const s
        }
 }
 
        }
 }
 

-void ServerPool::add_data(int stream_index, const char *data, size_t bytes, StreamStartSuitability suitable_for_stream_start)
+void ServerPool::add_data(int stream_index, const char *data, size_t bytes, uint16_t metacube_flags)

 {
        assert(stream_index >= 0 && stream_index < ssize_t(num_http_streams + udp_streams.size()));
 
 {
        assert(stream_index >= 0 && stream_index < ssize_t(num_http_streams + udp_streams.size()));
 


@@ -159,7 +159,7 @@ void ServerPool::add_data(int stream_index, const char *data, size_t bytes, Stre
 
        // HTTP stream.
        for (int i = 0; i < num_servers; ++i) {
 
        // HTTP stream.
        for (int i = 0; i < num_servers; ++i) {

-               servers[i].add_data_deferred(stream_index, data, bytes, suitable_for_stream_start);
+               servers[i].add_data_deferred(stream_index, data, bytes, metacube_flags);

        }
 }
 
        }
 }
 

diff --git a/serverpool.h b/serverpool.h
index e52aee8acf6ac92db9001dfbd72e12aa473c61af..c72623da13664fca6d54809f42bfb0393ac22835 100644 (file)
--- a/serverpool.h
+++ b/serverpool.h
@@ -41,7 +41,7 @@ public:
        void set_header(int stream_index,
                        const std::string &http_header,
                        const std::string &stream_header);
        void set_header(int stream_index,
                        const std::string &http_header,
                        const std::string &stream_header);

-       void add_data(int stream_index, const char *data, size_t bytes, StreamStartSuitability suitable_for_stream_start);
+       void add_data(int stream_index, const char *data, size_t bytes, uint16_t metacube_flags);

 
        // Sets the max pacing rate for all the servers.
        void set_pacing_rate(int stream_index, uint32_t pacing_rate);
 
        // Sets the max pacing rate for all the servers.
        void set_pacing_rate(int stream_index, uint32_t pacing_rate);

diff --git a/stream.cpp b/stream.cpp
index b498b2e6f7108ced5fab2f0f4d87109521671ca8..7a2c0998f60f02e945045194e04415f7ad1dafc2 100644 (file)
--- a/stream.cpp
+++ b/stream.cpp
@@ -128,7 +128,7 @@ void Stream::set_backlog_size(size_t new_size)
        DataElement data_element;
        data_element.data.iov_base = const_cast<char *>(existing_data.data());
        data_element.data.iov_len = existing_data.size();
        DataElement data_element;
        data_element.data.iov_base = const_cast<char *>(existing_data.data());
        data_element.data.iov_len = existing_data.size();

-       data_element.suitable_for_stream_start = NOT_SUITABLE_FOR_STREAM_START;  // Ignored by add_data_raw().
+       data_element.metacube_flags = 0;  // Ignored by add_data_raw().

 
        vector<DataElement> data_elements;
        data_elements.push_back(data_element);
 
        vector<DataElement> data_elements;
        data_elements.push_back(data_element);


@@ -177,7 +177,7 @@ vector<Stream::DataElement> remove_iovecs(const vector<Stream::DataElement> &dat
                        Stream::DataElement data_element;
                        data_element.data.iov_base = reinterpret_cast<char *>(data[i].data.iov_base) + bytes_wanted;
                        data_element.data.iov_len = data[i].data.iov_len - bytes_wanted;
                        Stream::DataElement data_element;
                        data_element.data.iov_base = reinterpret_cast<char *>(data[i].data.iov_base) + bytes_wanted;
                        data_element.data.iov_len = data[i].data.iov_len - bytes_wanted;

-                       data_element.suitable_for_stream_start = NOT_SUITABLE_FOR_STREAM_START;
+                       data_element.metacube_flags = METACUBE_FLAGS_NOT_SUITABLE_FOR_STREAM_START;

                        ret.push_back(data_element);
                        bytes_wanted = 0;
                }
                        ret.push_back(data_element);
                        bytes_wanted = 0;
                }


@@ -224,24 +224,29 @@ void Stream::remove_obsolete_starting_points()
        }
 }
 
        }
 }
 

-void Stream::add_data_deferred(const char *data, size_t bytes, StreamStartSuitability suitable_for_stream_start)
+void Stream::add_data_deferred(const char *data, size_t bytes, uint16_t metacube_flags)

 {
 {

+       // For regular output, we don't want to send the client twice
+       // (it's already sent out together with the HTTP header).
+       // However, for Metacube output, we need to send it so that
+       // the Cubemap instance in the other end has a chance to update it.
+       // It may come twice in its stream, but Cubemap doesn't care.
+       if (encoding == Stream::STREAM_ENCODING_RAW &&
+           (metacube_flags & METACUBE_FLAGS_HEADER) != 0) {
+               return;
+       }
+

        MutexLock lock(&queued_data_mutex);
        MutexLock lock(&queued_data_mutex);

-       assert(suitable_for_stream_start == SUITABLE_FOR_STREAM_START ||
-              suitable_for_stream_start == NOT_SUITABLE_FOR_STREAM_START);

 
        DataElement data_element;
 
        DataElement data_element;

-       data_element.suitable_for_stream_start = suitable_for_stream_start;
+       data_element.metacube_flags = metacube_flags;

 
        if (encoding == Stream::STREAM_ENCODING_METACUBE) {
                // Add a Metacube block header before the data.
                metacube2_block_header hdr;
                memcpy(hdr.sync, METACUBE2_SYNC, sizeof(hdr.sync));
                hdr.size = htonl(bytes);
 
        if (encoding == Stream::STREAM_ENCODING_METACUBE) {
                // Add a Metacube block header before the data.
                metacube2_block_header hdr;
                memcpy(hdr.sync, METACUBE2_SYNC, sizeof(hdr.sync));
                hdr.size = htonl(bytes);

-               hdr.flags = htons(0);
-               if (suitable_for_stream_start == NOT_SUITABLE_FOR_STREAM_START) {
-                       hdr.flags |= htons(METACUBE_FLAGS_NOT_SUITABLE_FOR_STREAM_START);
-               }
+               hdr.flags = htons(metacube_flags);

                hdr.csum = htons(metacube2_compute_crc(&hdr));
 
                data_element.data.iov_base = new char[bytes + sizeof(hdr)];
                hdr.csum = htons(metacube2_compute_crc(&hdr));
 
                data_element.data.iov_base = new char[bytes + sizeof(hdr)];


@@ -288,7 +293,7 @@ void Stream::process_queued_data()
        static const int minimum_start_point_distance = 10240;
        size_t byte_position = bytes_received;
        for (size_t i = 0; i < queued_data_copy.size(); ++i) {
        static const int minimum_start_point_distance = 10240;
        size_t byte_position = bytes_received;
        for (size_t i = 0; i < queued_data_copy.size(); ++i) {

-               if (queued_data_copy[i].suitable_for_stream_start == SUITABLE_FOR_STREAM_START) {
+               if ((queued_data_copy[i].metacube_flags & METACUBE_FLAGS_NOT_SUITABLE_FOR_STREAM_START) == 0) {

                        size_t num_points = suitable_starting_points.size();
                        if (num_points >= 2 &&
                            suitable_starting_points[num_points - 1] - suitable_starting_points[num_points - 2] < minimum_start_point_distance) {
                        size_t num_points = suitable_starting_points.size();
                        if (num_points >= 2 &&
                            suitable_starting_points[num_points - 1] - suitable_starting_points[num_points - 2] < minimum_start_point_distance) {

diff --git a/stream.h b/stream.h
index 057a48cbe08e516be02f756ce4aeac3701672ee2..9995765b2db95042fbe1c6369e9d3c2114896400 100644 (file)
--- a/stream.h
+++ b/stream.h
@@ -15,11 +15,6 @@
 class StreamProto;
 struct Client;
 
 class StreamProto;
 struct Client;
 

-enum StreamStartSuitability {
-       NOT_SUITABLE_FOR_STREAM_START,
-       SUITABLE_FOR_STREAM_START,
-};
-

 struct Stream {
        // Must be in sync with StreamConfig::Encoding.
        enum Encoding { STREAM_ENCODING_RAW = 0, STREAM_ENCODING_METACUBE };
 struct Stream {
        // Must be in sync with StreamConfig::Encoding.
        enum Encoding { STREAM_ENCODING_RAW = 0, STREAM_ENCODING_METACUBE };


@@ -99,7 +94,7 @@ struct Stream {
        // The data pointers in the iovec are owned by us.
        struct DataElement {
                iovec data;
        // The data pointers in the iovec are owned by us.
        struct DataElement {
                iovec data;

-               StreamStartSuitability suitable_for_stream_start;
+               uint16_t metacube_flags;

        };
        std::vector<DataElement> queued_data;
 
        };
        std::vector<DataElement> queued_data;
 


@@ -110,7 +105,7 @@ struct Stream {
 
        // Add more data to <queued_data>, adding Metacube headers if needed.
        // Does not take ownership of <data>.
 
        // Add more data to <queued_data>, adding Metacube headers if needed.
        // Does not take ownership of <data>.

-       void add_data_deferred(const char *data, size_t bytes, StreamStartSuitability suitable_for_stream_start);
+       void add_data_deferred(const char *data, size_t bytes, uint16_t metacube_flags);

 
        // Add queued data to the stream, if any.
        // You should hold the owning Server's <mutex>.
 
        // Add queued data to the stream, if any.
        // You should hold the owning Server's <mutex>.

diff --git a/udpinput.cpp b/udpinput.cpp
index 559e9cd29c0c773c7519f814eea5cc3b265e96db..71364f1e79d2e98084b4c0b358fbd11c06c59f1c 100644 (file)
--- a/udpinput.cpp
+++ b/udpinput.cpp
@@ -226,7 +226,7 @@ void UDPInput::do_work()
                }
                
                for (size_t i = 0; i < stream_indices.size(); ++i) {
                }
                
                for (size_t i = 0; i < stream_indices.size(); ++i) {

-                       servers->add_data(stream_indices[i], packet_buf, ret, SUITABLE_FOR_STREAM_START);
+                       servers->add_data(stream_indices[i], packet_buf, ret, /*metacube_flags=*/0);

                }
        }
 }
                }
        }
 }