X-Git-Url: https://git.sesse.net/?p=cubemap;a=blobdiff_plain;f=stream.cpp;h=64bf2e8053e7c0edd79b17f43ba12160a3dd8795;hp=50679ca5f5738c2bf0870d6816f24490284d3679;hb=b757a4a2ce9d24835b52a185134835762af2f50c;hpb=8095c4ac52431d9e593ece77bdc21566a2ed9240

diff --git a/stream.cpp b/stream.cpp
index 50679ca..64bf2e8 100644
--- a/stream.cpp
+++ b/stream.cpp
@@ -8,6 +8,7 @@
 #include <sys/types.h>
 #include <algorithm>
 #include <string>
+#include <queue>
 #include <vector>
 
 #include "log.h"
@@ -26,7 +27,6 @@ Stream::Stream(const string &url, size_t backlog_size, size_t prebuffering_bytes
           backlog_size(backlog_size),
 	  prebuffering_bytes(prebuffering_bytes),
 	  bytes_received(0),
-	  last_suitable_starting_point(-1),
 	  pacing_rate(~0U)
 {
 	if (data_fd == -1) {
@@ -58,8 +58,16 @@ Stream::Stream(const StreamProto &serialized, int data_fd)
 		exit(1);
 	}
 
-	assert(serialized.has_last_suitable_starting_point());
-	last_suitable_starting_point = serialized.last_suitable_starting_point();
+	for (int i = 0; i < serialized.suitable_starting_point_size(); ++i) {
+		ssize_t point = serialized.suitable_starting_point(i);
+		if (point == -1) {
+			// Can happen when upgrading from before 1.1.3,
+			// where this was an optional field with -1 signifying
+			// "no such point".
+			continue;
+		}
+		suitable_starting_points.push_back(point);
+	}
 
 	pthread_mutex_init(&queued_data_mutex, NULL);
 }
@@ -73,7 +81,9 @@ StreamProto Stream::serialize()
 	serialized.set_backlog_size(backlog_size);
 	serialized.set_prebuffering_bytes(prebuffering_bytes);
 	serialized.set_bytes_received(bytes_received);
-	serialized.set_last_suitable_starting_point(last_suitable_starting_point);
+	for (size_t i = 0; i < suitable_starting_points.size(); ++i) {
+		serialized.add_suitable_starting_point(suitable_starting_points[i]);
+	}
 	serialized.set_url(url);
 	data_fd = -1;
 	return serialized;
@@ -122,6 +132,7 @@ void Stream::set_backlog_size(size_t new_size)
 	vector<DataElement> data_elements;
 	data_elements.push_back(data_element);
 	add_data_raw(data_elements);
+	remove_obsolete_starting_points();
 }
 
 void Stream::put_client_to_sleep(Client *client)
@@ -133,7 +144,7 @@ void Stream::put_client_to_sleep(Client *client)
 vector<iovec> collect_iovecs(const vector<Stream::DataElement> &data, size_t bytes_wanted)
 {
 	vector<iovec> ret;
-	size_t max_iovecs = std::min<size_t>(data.size(), IOV_MAX);
+	size_t max_iovecs = min<size_t>(data.size(), IOV_MAX);
 	for (size_t i = 0; i < max_iovecs && bytes_wanted > 0; ++i) {
 		if (data[i].data.iov_len <= bytes_wanted) {
 			// Consume the entire iovec.
@@ -202,6 +213,16 @@ void Stream::add_data_raw(const vector<DataElement> &orig_data)
 	}
 }
 
+void Stream::remove_obsolete_starting_points()
+{
+	// We could do a binary search here (std::lower_bound), but it seems
+	// overkill for removing what's probably only a few points.
+	while (!suitable_starting_points.empty() &&
+	       bytes_received - suitable_starting_points[0] > backlog_size) {
+		suitable_starting_points.pop_front();
+	}
+}
+
 void Stream::add_data_deferred(const char *data, size_t bytes, StreamStartSuitability suitable_for_stream_start)
 {
 	MutexLock lock(&queued_data_mutex);
@@ -243,7 +264,7 @@ void Stream::add_data_deferred(const char *data, size_t bytes, StreamStartSuitab
 
 void Stream::process_queued_data()
 {
-	std::vector<DataElement> queued_data_copy;
+	vector<DataElement> queued_data_copy;
 
 	// Hold the lock for as short as possible, since add_data_raw() can possibly
 	// write to disk, which might disturb the input thread.
@@ -256,17 +277,30 @@ void Stream::process_queued_data()
 		swap(queued_data, queued_data_copy);
 	}
 
-	// Update the last suitable starting point for the stream,
-	// if the queued data contains such a starting point.
+	// Add suitable starting points for the stream, if the queued data
+	// contains such starting points. Note that we drop starting points
+	// if they're less than 10 kB apart, so that we don't get a huge
+	// amount of them for e.g. each and every MPEG-TS 188-byte cell.
+	// The 10 kB value is somewhat arbitrary, but at least it should make
+	// the RAM cost of saving the position ~0.1% (or less) of the actual
+	// data, and 10 kB is a very fine granularity in most streams.
+	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) {
-			last_suitable_starting_point = byte_position;
+			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) {
+				// p[n-1] - p[n-2] < 10 kB, so drop p[n-1].
+				suitable_starting_points.pop_back();
+			}
+			suitable_starting_points.push_back(byte_position);
 		}
 		byte_position += queued_data_copy[i].data.iov_len;
 	}
 
 	add_data_raw(queued_data_copy);
+	remove_obsolete_starting_points();
 	for (size_t i = 0; i < queued_data_copy.size(); ++i) {
 		char *data = reinterpret_cast<char *>(queued_data_copy[i].data.iov_base);
 		delete[] data;