Fix a typo.

[cubemap] / stream.cpp

diff --git a/stream.cpp b/stream.cpp
index 6563b5b3d9e5f39f563aeecf2382ea72cb74f38c..594d77cba33afe387502e07fbf021b1d180104ea 100644 (file)
--- a/stream.cpp
+++ b/stream.cpp
@@ -2,14 +2,17 @@
 #include <errno.h>
 #include <limits.h>
 #include <netinet/in.h>
+#include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
 #include <sys/types.h>
+#include <algorithm>
 #include <string>
 #include <vector>
 
 #include "log.h"
-#include "metacube.h"
+#include "metacube2.h"
+#include "mutexlock.h"
 #include "state.pb.h"
 #include "stream.h"
 #include "util.h"
@@ -23,12 +26,14 @@ Stream::Stream(const string &url, size_t backlog_size, Encoding encoding)
           backlog_size(backlog_size),
          bytes_received(0),
          last_suitable_starting_point(-1),
-         mark_pool(NULL),
+         pacing_rate(~0U),
          queued_data_last_starting_point(-1)
 {
        if (data_fd == -1) {
                exit(1);
        }
+
+       pthread_mutex_init(&queued_data_mutex, NULL);
 }
 
 Stream::~Stream()
@@ -46,32 +51,17 @@ Stream::Stream(const StreamProto &serialized, int data_fd)
          data_fd(data_fd),
          backlog_size(serialized.backlog_size()),
          bytes_received(serialized.bytes_received()),
-         mark_pool(NULL),
+         pacing_rate(~0U),
          queued_data_last_starting_point(-1)
 {
        if (data_fd == -1) {
                exit(1);
        }
 
-       // Split old-style headers into HTTP and video headers.
-       if (!serialized.header().empty()) {
-               string header = serialized.header();
-               size_t split = header.find("\r\n\r\n");
-               if (split == string::npos) {
-                       http_header = header;
-                       stream_header = "";
-               } else {
-                       http_header = header.substr(0, split + 2);  // Split off the second \r\n.
-                       stream_header = header.substr(split, string::npos);
-               }
-       }
+       assert(serialized.has_last_suitable_starting_point());
+       last_suitable_starting_point = serialized.last_suitable_starting_point();
 
-       // Older versions did not set last_suitable_starting_point.
-       if (serialized.has_last_suitable_starting_point()) {
-               last_suitable_starting_point = serialized.last_suitable_starting_point();
-       } else {
-               last_suitable_starting_point = bytes_received;
-       }
+       pthread_mutex_init(&queued_data_mutex, NULL);
 }
 
 StreamProto Stream::serialize()
@@ -211,6 +201,7 @@ void Stream::add_data_raw(const vector<iovec> &orig_data)
 
 void Stream::add_data_deferred(const char *data, size_t bytes, StreamStartSuitability suitable_for_stream_start)
 {
+       MutexLock lock(&queued_data_mutex);
        assert(suitable_for_stream_start == SUITABLE_FOR_STREAM_START ||
               suitable_for_stream_start == NOT_SUITABLE_FOR_STREAM_START);
        if (suitable_for_stream_start == SUITABLE_FOR_STREAM_START) {
@@ -219,13 +210,14 @@ void Stream::add_data_deferred(const char *data, size_t bytes, StreamStartSuitab
 
        if (encoding == Stream::STREAM_ENCODING_METACUBE) {
                // Add a Metacube block header before the data.
-               metacube_block_header hdr;
-               memcpy(hdr.sync, METACUBE_SYNC, sizeof(hdr.sync));
+               metacube2_block_header hdr;
+               memcpy(hdr.sync, METACUBE2_SYNC, sizeof(hdr.sync));
                hdr.size = htonl(bytes);
-               hdr.flags = htonl(0);
+               hdr.flags = htons(0);
                if (suitable_for_stream_start == NOT_SUITABLE_FOR_STREAM_START) {
-                       hdr.flags |= htonl(METACUBE_FLAGS_NOT_SUITABLE_FOR_STREAM_START);
+                       hdr.flags |= htons(METACUBE_FLAGS_NOT_SUITABLE_FOR_STREAM_START);
                }
+               hdr.csum = htons(metacube2_compute_crc(&hdr));
 
                iovec iov;
                iov.iov_base = new char[bytes + sizeof(hdr)];
@@ -250,27 +242,36 @@ void Stream::add_data_deferred(const char *data, size_t bytes, StreamStartSuitab
 
 void Stream::process_queued_data()
 {
-       if (queued_data.empty()) {
-               return;
+       std::vector<iovec> queued_data_copy;
+       int queued_data_last_starting_point_copy = -1;
+
+       // Hold the lock for as short as possible, since add_data_raw() can possibly
+       // write to disk, which might disturb the input thread.
+       {
+               MutexLock lock(&queued_data_mutex);
+               if (queued_data.empty()) {
+                       return;
+               }
+
+               swap(queued_data, queued_data_copy);
+               swap(queued_data_last_starting_point, queued_data_last_starting_point_copy);
        }
 
        // Update the last suitable starting point for the stream,
        // if the queued data contains such a starting point.
-       assert(queued_data_last_starting_point < ssize_t(queued_data.size()));
-       if (queued_data_last_starting_point >= 0) {
+       assert(queued_data_last_starting_point_copy < ssize_t(queued_data_copy.size()));
+       if (queued_data_last_starting_point_copy >= 0) {
                last_suitable_starting_point = bytes_received;
-               for (int i = 0; i < queued_data_last_starting_point; ++i) {
-                       last_suitable_starting_point += queued_data[i].iov_len;
+               for (int i = 0; i < queued_data_last_starting_point_copy; ++i) {
+                       last_suitable_starting_point += queued_data_copy[i].iov_len;
                }
        }
 
-       add_data_raw(queued_data);
-       for (size_t i = 0; i < queued_data.size(); ++i) {
-               char *data = reinterpret_cast<char *>(queued_data[i].iov_base);
+       add_data_raw(queued_data_copy);
+       for (size_t i = 0; i < queued_data_copy.size(); ++i) {
+               char *data = reinterpret_cast<char *>(queued_data_copy[i].iov_base);
                delete[] data;
        }
-       queued_data.clear();
-       queued_data_last_starting_point = -1;
 
        // We have more data, so wake up all clients.
        if (to_process.empty()) {