Store multiple suitable starting points for each stream.

We still only use the last one. This will change soon.

diff --git a/server.cpp b/server.cpp
index 7116a73f7e8ab9a3bf3fec9d6973275aa0f23307..946dc5d1df12d0c5b9c3d88058c0b5dad68c5a8d 100644 (file)
--- a/server.cpp
+++ b/server.cpp
@@ -479,14 +479,15 @@ sending_header_or_error_again:
        }
        case Client::WAITING_FOR_KEYFRAME: {
                Stream *stream = client->stream;
-               if (ssize_t(client->stream_pos) > stream->last_suitable_starting_point) {
+               if (stream->suitable_starting_points.empty() ||
+                   client->stream_pos > stream->suitable_starting_points.back()) {
                        // We haven't received a keyframe since this stream started waiting,
                        // so keep on waiting for one.
                        // This is postcondition #3.
                        stream->put_client_to_sleep(client);
                        return;
                }
-               client->stream_pos = stream->last_suitable_starting_point;
+               client->stream_pos = stream->suitable_starting_points.back();
                client->state = Client::PREBUFFERING;
                // Fall through.
        }

diff --git a/state.proto b/state.proto
index 3a611086966d17ec85da3f99d77fd706dea5b201..942254012ffae5a309ed6bf6f95c1a965e2fe8dc 100644 (file)
--- a/state.proto
+++ b/state.proto
@@ -26,7 +26,7 @@ message StreamProto {
        optional int64 backlog_size = 5 [default=10485760];
        optional int64 prebuffering_bytes = 10 [default=0];
        optional int64 bytes_received = 3;
-       optional int64 last_suitable_starting_point = 9;
+       repeated int64 suitable_starting_point = 9;
        optional string url = 4;
 };
 

diff --git a/stream.cpp b/stream.cpp
index 50679ca5f5738c2bf0870d6816f24490284d3679..d416803fafb03dd23b4e591cf46bf061179a102b 100644 (file)
--- 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)
@@ -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);
@@ -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;

diff --git a/stream.h b/stream.h
index 9a9982a3541d489f72b63a18228f1e1d1910acca..c98feb46e3bb2e09f5913c1ea31a018e02680075 100644 (file)
--- a/stream.h
+++ b/stream.h
@@ -8,6 +8,7 @@
 #include <stdint.h>
 #include <sys/types.h>
 #include <sys/uio.h>
+#include <deque>
 #include <string>
 #include <vector>
 
@@ -75,9 +76,9 @@ struct Stream {
        // than <backlog_size>, since the buffer wraps.
        size_t bytes_received;
 
-       // The last point in the stream that is suitable to start new clients at
-       // (after having sent the header). -1 if no such point exists yet.
-       ssize_t last_suitable_starting_point;
+       // A list of points in the stream that is suitable to start new clients at
+       // (after having sent the header). Empty if no such point exists yet.
+       std::deque<size_t> suitable_starting_points;
        
        // Clients that are in SENDING_DATA, but that we don't listen on,
        // because we currently don't have any data for them.
@@ -117,8 +118,14 @@ private:
 
        // Adds data directly to the stream file descriptor, without adding headers or
        // going through <queued_data>.
-       // You should hold the owning Server's <mutex>.
+       // You should hold the owning Server's <mutex>, and probably call
+       // remove_obsolete_starting_points() afterwards.
        void add_data_raw(const std::vector<DataElement> &data);
+
+       // Remove points from <suitable_starting_points> that are no longer
+       // in the backlog.
+       // You should hold the owning Server's <mutex>.
+       void remove_obsolete_starting_points();
 };
 
 #endif  // !defined(_STREAM_H)