X-Git-Url: https://git.sesse.net/?p=cubemap;a=blobdiff_plain;f=server.cpp;h=6fa4991b209f389da7f940aa73ff9462f472b8cd;hp=7acf501236e1ee2db9cac22db0b72a5ddf8ecf69;hb=6889a665614e926437484a556124a5ff60363568;hpb=44f74dbce7ed19d8218318ce5ecc0da2fb4ae9f0 diff --git a/server.cpp b/server.cpp index 7acf501..6fa4991 100644 --- a/server.cpp +++ b/server.cpp @@ -19,8 +19,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" @@ -28,6 +27,10 @@ #include "stream.h" #include "util.h" +#ifndef SO_MAX_PACING_RATE +#define SO_MAX_PACING_RATE 47 +#endif + using namespace std; extern AccessLogThread *access_log; @@ -35,7 +38,7 @@ extern AccessLogThread *access_log; Server::Server() { pthread_mutex_init(&mutex, NULL); - pthread_mutex_init(&queued_data_mutex, NULL); + pthread_mutex_init(&queued_clients_mutex, NULL); epoll_fd = epoll_create(1024); // Size argument is ignored. if (epoll_fd == -1) { @@ -140,7 +143,7 @@ CubemapStateProto Server::serialize() void Server::add_client_deferred(int sock) { - MutexLock lock(&queued_data_mutex); + MutexLock lock(&queued_clients_mutex); queued_add_clients.push_back(sock); } @@ -194,8 +197,10 @@ 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::PREBUFFERING || + (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); @@ -211,11 +216,11 @@ int Server::lookup_stream_by_url(const std::string &url) const return url_it->second; } -int Server::add_stream(const string &url, size_t backlog_size, Stream::Encoding encoding) +int Server::add_stream(const string &url, size_t backlog_size, size_t prebuffering_bytes, Stream::Encoding encoding) { MutexLock lock(&mutex); url_map.insert(make_pair(url, streams.size())); - streams.push_back(new Stream(url, backlog_size, encoding)); + streams.push_back(new Stream(url, backlog_size, prebuffering_bytes, encoding)); return streams.size() - 1; } @@ -247,33 +252,20 @@ void Server::set_header(int stream_index, const string &http_header, const strin assert(stream_index >= 0 && stream_index < ssize_t(streams.size())); streams[stream_index]->http_header = http_header; streams[stream_index]->stream_header = stream_header; - - // If there are clients we haven't sent anything to yet, we should give - // them the header, so push back into the SENDING_HEADER state. - for (map::iterator client_it = clients.begin(); - client_it != clients.end(); - ++client_it) { - Client *client = &client_it->second; - if (client->state == Client::SENDING_DATA && - client->stream_pos == 0) { - construct_header(client); - } - } } -void Server::set_mark_pool(int stream_index, MarkPool *mark_pool) +void Server::set_pacing_rate(int stream_index, uint32_t pacing_rate) { MutexLock lock(&mutex); assert(clients.empty()); assert(stream_index >= 0 && stream_index < ssize_t(streams.size())); - streams[stream_index]->mark_pool = mark_pool; + streams[stream_index]->pacing_rate = pacing_rate; } -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. @@ -380,22 +372,49 @@ 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; 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::PREBUFFERING; + // Fall through. + } + case Client::PREBUFFERING: { + Stream *stream = client->stream; + size_t bytes_to_send = stream->bytes_received - client->stream_pos; + assert(bytes_to_send <= stream->backlog_size); + if (bytes_to_send < stream->prebuffering_bytes) { + // We don't have enough bytes buffered to start this client yet. + stream->put_client_to_sleep(client); + return; + } + client->state = Client::SENDING_DATA; + // Fall through. + } case Client::SENDING_DATA: { skip_lost_data(client); Stream *stream = client->stream; @@ -416,7 +435,7 @@ sending_data_again: ssize_t ret; do { - loff_t offset = client->stream_pos % stream->backlog_size; + off_t offset = client->stream_pos % stream->backlog_size; ret = sendfile(client->sock, stream->data_fd, &offset, bytes_to_send); } while (ret == -1 && errno == EINTR); @@ -454,18 +473,15 @@ sending_data_again: void Server::skip_lost_data(Client *client) { Stream *stream = client->stream; + if (stream == NULL) { + return; + } 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); } } @@ -497,16 +513,16 @@ int Server::parse_request(Client *client) return 404; // Not found. } - client->url = request_tokens[1]; - client->stream = streams[url_map_it->second]; - if (client->stream->mark_pool != NULL) { - client->fwmark = client->stream->mark_pool->get_mark(); - } else { - client->fwmark = 0; // No mark. + Stream *stream = streams[url_map_it->second]; + if (stream->http_header.empty()) { + return 503; // Service unavailable. } - if (setsockopt(client->sock, SOL_SOCKET, SO_MARK, &client->fwmark, sizeof(client->fwmark)) == -1) { - if (client->fwmark != 0) { - log_perror("setsockopt(SO_MARK)"); + + client->url = request_tokens[1]; + client->stream = stream; + if (setsockopt(client->sock, SOL_SOCKET, SO_MAX_PACING_RATE, &client->stream->pacing_rate, sizeof(client->stream->pacing_rate)) == -1) { + if (client->stream->pacing_rate != ~0U) { + log_perror("setsockopt(SO_MAX_PACING_RATE)"); } } client->request.clear(); @@ -526,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))); } @@ -589,10 +606,6 @@ void Server::close_client(Client *client) if (client->stream != NULL) { delete_from(&client->stream->sleeping_clients, client); delete_from(&client->stream->to_process, client); - if (client->stream->mark_pool != NULL) { - int fwmark = client->fwmark; - client->stream->mark_pool->release_mark(fwmark); - } } // Log to access_log. @@ -606,12 +619,14 @@ void Server::close_client(Client *client) void Server::process_queued_data() { - MutexLock lock(&queued_data_mutex); + { + MutexLock lock(&queued_clients_mutex); - for (size_t i = 0; i < queued_add_clients.size(); ++i) { - add_client(queued_add_clients[i]); + for (size_t i = 0; i < queued_add_clients.size(); ++i) { + add_client(queued_add_clients[i]); + } + queued_add_clients.clear(); } - queued_add_clients.clear(); for (size_t i = 0; i < streams.size(); ++i) { streams[i]->process_queued_data();