X-Git-Url: https://git.sesse.net/?p=cubemap;a=blobdiff_plain;f=server.cpp;h=4442b34cee7899661bae68e15e1ed223018abdc2;hp=5bd0d555c18194ca112a04b21cb3d9344272722e;hb=74cd48ffef90d7d0752e37a4515e4ecfb68f7c9d;hpb=9abb89bcf7940e2ada9d708f86a218a56334f68d diff --git a/server.cpp b/server.cpp index 5bd0d55..4442b34 100644 --- a/server.cpp +++ b/server.cpp @@ -20,7 +20,7 @@ #include "accesslog.h" #include "log.h" #include "markpool.h" -#include "metacube.h" +#include "metacube2.h" #include "mutexlock.h" #include "parse.h" #include "server.h" @@ -113,6 +113,19 @@ CubemapStateProto Server::serialize() // We don't serialize anything queued, so empty the queues. process_queued_data(); + // Set all clients in a consistent state before serializing + // (ie., they have no remaining lost data). Otherwise, increasing + // the backlog could take clients into a newly valid area of the backlog, + // sending a stream of zeros instead of skipping the data as it should. + // + // TODO: Do this when clients are added back from serialized state instead; + // it would probably be less wasteful. + for (map::iterator client_it = clients.begin(); + client_it != clients.end(); + ++client_it) { + skip_lost_data(&client_it->second); + } + CubemapStateProto serialized; for (map::const_iterator client_it = clients.begin(); client_it != clients.end(); @@ -181,8 +194,9 @@ void Server::add_client_from_serialized(const ClientProto &client) exit(1); } - if (client_ptr->state == Client::SENDING_DATA && - client_ptr->stream_pos == client_ptr->stream->bytes_received) { + if (client_ptr->state == Client::WAITING_FOR_KEYFRAME || + (client_ptr->state == Client::SENDING_DATA && + client_ptr->stream_pos == client_ptr->stream->bytes_received)) { client_ptr->stream->put_client_to_sleep(client_ptr); } else { process_client(client_ptr); @@ -256,11 +270,11 @@ void Server::set_mark_pool(int stream_index, MarkPool *mark_pool) streams[stream_index]->mark_pool = mark_pool; } -void Server::add_data_deferred(int stream_index, const char *data, size_t bytes) +void Server::add_data_deferred(int stream_index, const char *data, size_t bytes, StreamStartSuitability suitable_for_stream_start) { MutexLock lock(&queued_data_mutex); assert(stream_index >= 0 && stream_index < ssize_t(streams.size())); - streams[stream_index]->add_data_deferred(data, bytes); + streams[stream_index]->add_data_deferred(data, bytes, suitable_for_stream_start); } // See the .h file for postconditions after this function. @@ -367,36 +381,47 @@ sending_header_or_error_again: return; } - // Start sending from the end. In other words, we won't send any of the backlog, - // but we'll start sending immediately as we get data. + // Start sending from the first keyframe we get. In other + // words, we won't send any of the backlog, but we'll start + // sending immediately as we get the next keyframe block. // This is postcondition #3. - client->state = Client::SENDING_DATA; - client->stream_pos = client->stream->bytes_received; + if (client->stream_pos == size_t(-2)) { + client->stream_pos = std::min( + client->stream->bytes_received - client->stream->backlog_size, + 0); + client->state = Client::SENDING_DATA; + } else { + // client->stream_pos should be -1, but it might not be, + // if we have clients from an older version. + client->stream_pos = client->stream->bytes_received; + client->state = Client::WAITING_FOR_KEYFRAME; + } client->stream->put_client_to_sleep(client); return; } + case Client::WAITING_FOR_KEYFRAME: { + Stream *stream = client->stream; + if (ssize_t(client->stream_pos) > stream->last_suitable_starting_point) { + // 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->state = Client::SENDING_DATA; + // Fall through. + } case Client::SENDING_DATA: { -sending_data_again: - // See if there's some data we've lost. Ideally, we should drop to a block boundary, - // but resync will be the mux's problem. + skip_lost_data(client); Stream *stream = client->stream; + +sending_data_again: size_t bytes_to_send = stream->bytes_received - client->stream_pos; + assert(bytes_to_send <= stream->backlog_size); if (bytes_to_send == 0) { return; } - if (bytes_to_send > stream->backlog_size) { - size_t bytes_lost = bytes_to_send - stream->backlog_size; - client->stream_pos = stream->bytes_received - stream->backlog_size; - client->bytes_lost += bytes_lost; - ++client->num_loss_events; - bytes_to_send = stream->backlog_size; - - double loss_fraction = double(client->bytes_lost) / double(client->bytes_lost + client->bytes_sent); - log(WARNING, "[%s] Client lost %lld bytes (total loss: %.2f%%), maybe too slow connection", - client->remote_addr.c_str(), - (long long int)(bytes_lost), - 100.0 * loss_fraction); - } // See if we need to split across the circular buffer. bool more_data = false; @@ -440,6 +465,26 @@ sending_data_again: } } +// See if there's some data we've lost. Ideally, we should drop to a block boundary, +// but resync will be the mux's problem. +void Server::skip_lost_data(Client *client) +{ + Stream *stream = client->stream; + size_t bytes_to_send = stream->bytes_received - client->stream_pos; + if (bytes_to_send > stream->backlog_size) { + size_t bytes_lost = bytes_to_send - stream->backlog_size; + client->stream_pos = stream->bytes_received - stream->backlog_size; + client->bytes_lost += bytes_lost; + ++client->num_loss_events; + + double loss_fraction = double(client->bytes_lost) / double(client->bytes_lost + client->bytes_sent); + log(WARNING, "[%s] Client lost %lld bytes (total loss: %.2f%%), maybe too slow connection", + client->remote_addr.c_str(), + (long long int)(bytes_lost), + 100.0 * loss_fraction); + } +} + int Server::parse_request(Client *client) { vector lines = split_lines(client->request); @@ -455,7 +500,15 @@ int Server::parse_request(Client *client) return 400; // Should maybe be 405 instead? } - map::const_iterator url_map_it = url_map.find(request_tokens[1]); + string url = request_tokens[1]; + if (url.find("?backlog") == url.size() - 8) { + client->stream_pos = -2; + url = url.substr(0, url.size() - 8); + } else { + client->stream_pos = -1; + } + + map::const_iterator url_map_it = url_map.find(url); if (url_map_it == url_map.end()) { return 404; // Not found. } @@ -489,10 +542,11 @@ void Server::construct_header(Client *client) "Content-encoding: metacube\r\n" + "\r\n"; if (!stream->stream_header.empty()) { - 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(stream->stream_header.size()); - hdr.flags = htonl(METACUBE_FLAGS_HEADER); + hdr.flags = htons(METACUBE_FLAGS_HEADER); + hdr.csum = htons(metacube2_compute_crc(&hdr)); client->header_or_error.append( string(reinterpret_cast(&hdr), sizeof(hdr))); }