X-Git-Url: https://git.sesse.net/?p=cubemap;a=blobdiff_plain;f=stream.cpp;h=42edee95aae01176e94c3a63ab47704a95b19a36;hp=17b5949172a531cae5a37eb4f8948dbf177bb4a9;hb=e05f9f880fb97c3510546a60d57b5da0b0752591;hpb=1e15bf9054e65adfce268578f4e474c980ef6443 diff --git a/stream.cpp b/stream.cpp index 17b5949..42edee9 100644 --- a/stream.cpp +++ b/stream.cpp @@ -1,27 +1,30 @@ #include #include +#include #include #include #include -#include +#include #include #include #include "log.h" -#include "metacube.h" +#include "metacube2.h" #include "state.pb.h" #include "stream.h" #include "util.h" using namespace std; -Stream::Stream(const string &stream_id, size_t backlog_size, Encoding encoding) - : stream_id(stream_id), +Stream::Stream(const string &url, size_t backlog_size, Encoding encoding) + : url(url), encoding(encoding), data_fd(make_tempfile("")), backlog_size(backlog_size), bytes_received(0), - mark_pool(NULL) + last_suitable_starting_point(-1), + mark_pool(NULL), + queued_data_last_starting_point(-1) { if (data_fd == -1) { exit(1); @@ -36,14 +39,15 @@ Stream::~Stream() } Stream::Stream(const StreamProto &serialized, int data_fd) - : stream_id(serialized.stream_id()), + : url(serialized.url()), http_header(serialized.http_header()), stream_header(serialized.stream_header()), encoding(Stream::STREAM_ENCODING_RAW), // Will be changed later. data_fd(data_fd), backlog_size(serialized.backlog_size()), bytes_received(serialized.bytes_received()), - mark_pool(NULL) + mark_pool(NULL), + queued_data_last_starting_point(-1) { if (data_fd == -1) { exit(1); @@ -61,6 +65,13 @@ Stream::Stream(const StreamProto &serialized, int data_fd) stream_header = header.substr(split, string::npos); } } + + // 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; + } } StreamProto Stream::serialize() @@ -71,7 +82,8 @@ StreamProto Stream::serialize() serialized.add_data_fds(data_fd); serialized.set_backlog_size(backlog_size); serialized.set_bytes_received(bytes_received); - serialized.set_stream_id(stream_id); + serialized.set_last_suitable_starting_point(last_suitable_starting_point); + serialized.set_url(url); data_fd = -1; return serialized; } @@ -129,7 +141,8 @@ void Stream::put_client_to_sleep(Client *client) vector collect_iovecs(const vector &data, size_t bytes_wanted) { vector ret; - for (size_t i = 0; i < data.size() && bytes_wanted > 0; ++i) { + size_t max_iovecs = std::min(data.size(), IOV_MAX); + for (size_t i = 0; i < max_iovecs && bytes_wanted > 0; ++i) { if (data[i].iov_len <= bytes_wanted) { // Consume the entire iovec. ret.push_back(data[i]); @@ -185,7 +198,7 @@ void Stream::add_data_raw(const vector &orig_data) } while (ret == -1 && errno == EINTR); if (ret == -1) { - log_perror("pwrite"); + log_perror("pwritev"); // Dazed and confused, but trying to continue... return; } @@ -196,30 +209,44 @@ void Stream::add_data_raw(const vector &orig_data) } } -void Stream::add_data_deferred(const char *data, size_t bytes) +void Stream::add_data_deferred(const char *data, size_t bytes, StreamStartSuitability suitable_for_stream_start) { + 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) { + queued_data_last_starting_point = queued_data.size(); + } + 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 |= htons(METACUBE_FLAGS_NOT_SUITABLE_FOR_STREAM_START); + } + hdr.csum = htons(metacube2_compute_crc(&hdr)); iovec iov; - iov.iov_base = new char[sizeof(hdr)]; + iov.iov_base = new char[bytes + sizeof(hdr)]; + iov.iov_len = bytes + sizeof(hdr); + memcpy(iov.iov_base, &hdr, sizeof(hdr)); - iov.iov_len = sizeof(hdr); + memcpy(reinterpret_cast(iov.iov_base) + sizeof(hdr), data, bytes); + + queued_data.push_back(iov); + } else if (encoding == Stream::STREAM_ENCODING_RAW) { + // Just add the data itself. + iovec iov; + iov.iov_base = new char[bytes]; + memcpy(iov.iov_base, data, bytes); + iov.iov_len = bytes; + queued_data.push_back(iov); } else { - assert(encoding == Stream::STREAM_ENCODING_RAW); + assert(false); } - - // Add the data itself. - iovec iov; - iov.iov_base = new char[bytes]; - memcpy(iov.iov_base, data, bytes); - iov.iov_len = bytes; - queued_data.push_back(iov); } void Stream::process_queued_data() @@ -228,12 +255,23 @@ void Stream::process_queued_data() return; } + // 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) { + 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; + } + } + add_data_raw(queued_data); for (size_t i = 0; i < queued_data.size(); ++i) { char *data = reinterpret_cast(queued_data[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()) {