X-Git-Url: https://git.sesse.net/?p=cubemap;a=blobdiff_plain;f=stream.cpp;h=b7a92c2293832e7d77c3b0c8b4c20e604b195a44;hp=e88ad779648e7639fd35a2926476eb446075cb1c;hb=d19e6c3ed8e85f5a39793903114d4de38b5a1609;hpb=979a284b4039b0ea74525b700b9f1089b8c4248d

diff --git a/stream.cpp b/stream.cpp
index e88ad77..b7a92c2 100644
--- a/stream.cpp
+++ b/stream.cpp
@@ -1,15 +1,17 @@
 #include <assert.h>
 #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 "metacube2.h"
+#include "mutexlock.h"
 #include "state.pb.h"
 #include "stream.h"
 #include "util.h"
@@ -29,6 +31,8 @@ Stream::Stream(const string &url, size_t backlog_size, Encoding encoding)
 	if (data_fd == -1) {
 		exit(1);
 	}
+
+	pthread_mutex_init(&queued_data_mutex, NULL);
 }
 
 Stream::~Stream()
@@ -72,6 +76,8 @@ Stream::Stream(const StreamProto &serialized, int data_fd)
 	} else {
 		last_suitable_starting_point = bytes_received;
 	}
+
+	pthread_mutex_init(&queued_data_mutex, NULL);
 }
 
 StreamProto Stream::serialize()
@@ -211,6 +217,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) {
@@ -226,6 +233,7 @@ void Stream::add_data_deferred(const char *data, size_t bytes, StreamStartSuitab
 		if (suitable_for_stream_start == 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 +258,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()) {