X-Git-Url: https://git.sesse.net/?p=cubemap;a=blobdiff_plain;f=server.cpp;h=99a3925e13f0bca65531f82902012408e0b5556c;hp=6fa4991b209f389da7f940aa73ff9462f472b8cd;hb=fb0ec9c7b3fb51c5da24af0183d35f3815e6ca0e;hpb=6889a665614e926437484a556124a5ff60363568 diff --git a/server.cpp b/server.cpp index 6fa4991..99a3925 100644 --- a/server.cpp +++ b/server.cpp @@ -35,6 +35,23 @@ using namespace std; extern AccessLogThread *access_log; +namespace { + +inline bool is_equal(timespec a, timespec b) +{ + return a.tv_sec == b.tv_sec && + a.tv_nsec == b.tv_nsec; +} + +inline bool is_earlier(timespec a, timespec b) +{ + if (a.tv_sec != b.tv_sec) + return a.tv_sec < b.tv_sec; + return a.tv_nsec < b.tv_nsec; +} + +} // namespace + Server::Server() { pthread_mutex_init(&mutex, NULL); @@ -90,6 +107,7 @@ void Server::do_work() process_queued_data(); + // Process each client where we have socket activity. for (int i = 0; i < nfds; ++i) { Client *client = reinterpret_cast(events[i].data.u64); @@ -101,6 +119,8 @@ void Server::do_work() process_client(client); } + // Process each client where its stream has new data, + // even if there was no socket activity. for (size_t i = 0; i < streams.size(); ++i) { vector to_process; swap(streams[i]->to_process, to_process); @@ -108,6 +128,49 @@ void Server::do_work() process_client(to_process[i]); } } + + // Finally, go through each client to see if it's timed out + // in the READING_REQUEST state. (Seemingly there are clients + // that can hold sockets up for days at a time without sending + // anything at all.) + timespec timeout_time; + if (clock_gettime(CLOCK_MONOTONIC_COARSE, &timeout_time) == -1) { + log_perror("clock_gettime(CLOCK_MONOTONIC_COARSE)"); + continue; + } + timeout_time.tv_sec -= REQUEST_READ_TIMEOUT_SEC; + while (!clients_ordered_by_connect_time.empty()) { + pair &connect_time_and_fd = clients_ordered_by_connect_time.front(); + + // See if we have reached the end of clients to process. + if (is_earlier(timeout_time, connect_time_and_fd.first)) { + break; + } + + // If this client doesn't exist anymore, just ignore it + // (it was deleted earlier). + std::map::iterator client_it = clients.find(connect_time_and_fd.second); + if (client_it == clients.end()) { + clients_ordered_by_connect_time.pop(); + continue; + } + Client *client = &client_it->second; + if (!is_equal(client->connect_time, connect_time_and_fd.first)) { + // Another client has taken this fd in the meantime. + clients_ordered_by_connect_time.pop(); + continue; + } + + if (client->state != Client::READING_REQUEST) { + // Only READING_REQUEST can time out. + clients_ordered_by_connect_time.pop(); + continue; + } + + // OK, it timed out. + close_client(client); + clients_ordered_by_connect_time.pop(); + } } } @@ -154,6 +217,17 @@ void Server::add_client(int sock) assert(ret.second == true); // Should not already exist. Client *client_ptr = &ret.first->second; + // Connection timestamps must be nondecreasing. I can't find any guarantee + // that even the monotonic clock can't go backwards by a small amount + // (think switching between CPUs with non-synchronized TSCs), so if + // this actually should happen, we hack around it by fudging + // connect_time. + if (!clients_ordered_by_connect_time.empty() && + is_earlier(client_ptr->connect_time, clients_ordered_by_connect_time.back().first)) { + client_ptr->connect_time = clients_ordered_by_connect_time.back().first; + } + clients_ordered_by_connect_time.push(make_pair(client_ptr->connect_time, sock)); + // Start listening on data from this socket. epoll_event ev; ev.events = EPOLLIN | EPOLLET | EPOLLRDHUP; @@ -182,6 +256,11 @@ void Server::add_client_from_serialized(const ClientProto &client) assert(ret.second == true); // Should not already exist. Client *client_ptr = &ret.first->second; + // Connection timestamps must be nondecreasing. + assert(clients_ordered_by_connect_time.empty() || + !is_earlier(client_ptr->connect_time, clients_ordered_by_connect_time.back().first)); + clients_ordered_by_connect_time.push(make_pair(client_ptr->connect_time, client.sock())); + // Start listening on data from this socket. epoll_event ev; if (client.state() == Client::READING_REQUEST) { @@ -238,6 +317,13 @@ void Server::set_backlog_size(int stream_index, size_t new_size) assert(stream_index >= 0 && stream_index < ssize_t(streams.size())); streams[stream_index]->set_backlog_size(new_size); } + +void Server::set_prebuffering_bytes(int stream_index, size_t new_amount) +{ + MutexLock lock(&mutex); + assert(stream_index >= 0 && stream_index < ssize_t(streams.size())); + streams[stream_index]->prebuffering_bytes = new_amount; +} void Server::set_encoding(int stream_index, Stream::Encoding encoding) { @@ -409,6 +495,7 @@ sending_header_or_error_again: 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. + // This is postcondition #3. stream->put_client_to_sleep(client); return; } @@ -492,6 +579,18 @@ int Server::parse_request(Client *client) return 400; // Bad request (empty). } + // Parse the headers, for logging purposes. + // TODO: Case-insensitivity. + multimap headers = extract_headers(lines, client->remote_addr); + multimap::const_iterator referer_it = headers.find("Referer"); + if (referer_it != headers.end()) { + client->referer = referer_it->second; + } + multimap::const_iterator user_agent_it = headers.find("User-Agent"); + if (user_agent_it != headers.end()) { + client->user_agent = user_agent_it->second; + } + vector request_tokens = split_tokens(lines[0]); if (request_tokens.size() < 2) { return 400; // Bad request (empty). @@ -519,6 +618,7 @@ int Server::parse_request(Client *client) } 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) {