X-Git-Url: https://git.sesse.net/?p=cubemap;a=blobdiff_plain;f=server.cpp;h=6570b5b73d084531d0c7099fd1979422f56616db;hp=6bac2c63feefffeaf677fd636e39886f8dc92ebd;hb=061988af511f42da3cd584b4d983177504ddc177;hpb=24c4b8ed26779f3316a97191c037342561932ed9 diff --git a/server.cpp b/server.cpp index 6bac2c6..6570b5b 100644 --- a/server.cpp +++ b/server.cpp @@ -23,7 +23,6 @@ #include "accesslog.h" #include "log.h" #include "metacube2.h" -#include "mutexlock.h" #include "parse.h" #include "server.h" #include "state.pb.h" @@ -57,9 +56,6 @@ inline bool is_earlier(timespec a, timespec b) Server::Server() { - pthread_mutex_init(&mutex, NULL); - pthread_mutex_init(&queued_clients_mutex, NULL); - epoll_fd = epoll_create(1024); // Size argument is ignored. if (epoll_fd == -1) { log_perror("epoll_fd"); @@ -69,10 +65,6 @@ Server::Server() Server::~Server() { - for (size_t i = 0; i < streams.size(); ++i) { - delete streams[i]; - } - safe_close(epoll_fd); } @@ -80,11 +72,9 @@ vector Server::get_client_stats() const { vector ret; - MutexLock lock(&mutex); - for (map::const_iterator client_it = clients.begin(); - client_it != clients.end(); - ++client_it) { - ret.push_back(client_it->second.get_stats()); + lock_guard lock(mu); + for (const auto &fd_and_client : clients) { + ret.push_back(fd_and_client.second.get_stats()); } return ret; } @@ -106,7 +96,7 @@ void Server::do_work() exit(1); } - MutexLock lock(&mutex); // We release the mutex between iterations. + lock_guard lock(mu); // We release the mutex between iterations. process_queued_data(); @@ -124,11 +114,11 @@ void Server::do_work() // 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) { + for (unique_ptr &stream : streams) { vector to_process; - swap(streams[i]->to_process, to_process); - for (size_t i = 0; i < to_process.size(); ++i) { - process_client(to_process[i]); + swap(stream->to_process, to_process); + for (Client *client : to_process) { + process_client(client); } } @@ -152,7 +142,7 @@ void Server::do_work() // If this client doesn't exist anymore, just ignore it // (it was deleted earlier). - map::iterator client_it = clients.find(connect_time_and_fd.second); + auto client_it = clients.find(connect_time_and_fd.second); if (client_it == clients.end()) { clients_ordered_by_connect_time.pop(); continue; @@ -177,7 +167,7 @@ void Server::do_work() } } -CubemapStateProto Server::serialize() +CubemapStateProto Server::serialize(unordered_map *short_response_pool) { // We don't serialize anything queued, so empty the queues. process_queued_data(); @@ -189,37 +179,32 @@ CubemapStateProto Server::serialize() // // 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); + for (auto &fd_and_client : clients) { + skip_lost_data(&fd_and_client.second); } CubemapStateProto serialized; - for (map::const_iterator client_it = clients.begin(); - client_it != clients.end(); - ++client_it) { - serialized.add_clients()->MergeFrom(client_it->second.serialize()); + for (const auto &fd_and_client : clients) { + serialized.add_clients()->MergeFrom(fd_and_client.second.serialize(short_response_pool)); } - for (size_t i = 0; i < streams.size(); ++i) { - serialized.add_streams()->MergeFrom(streams[i]->serialize()); + for (unique_ptr &stream : streams) { + serialized.add_streams()->MergeFrom(stream->serialize()); } return serialized; } void Server::add_client_deferred(int sock, Acceptor *acceptor) { - MutexLock lock(&queued_clients_mutex); + lock_guard lock(queued_clients_mutex); queued_add_clients.push_back(std::make_pair(sock, acceptor)); } void Server::add_client(int sock, Acceptor *acceptor) { const bool is_tls = acceptor->is_tls(); - pair::iterator, bool> ret = - clients.insert(make_pair(sock, Client(sock))); - assert(ret.second == true); // Should not already exist. - Client *client_ptr = &ret.first->second; + auto inserted = clients.insert(make_pair(sock, Client(sock))); + assert(inserted.second == true); // Should not already exist. + Client *client_ptr = &inserted.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 @@ -252,7 +237,7 @@ void Server::add_client(int sock, Acceptor *acceptor) if (is_tls) { assert(tls_server_contexts.count(acceptor)); client_ptr->tls_context = tls_accept(tls_server_contexts[acceptor]); - if (client_ptr->tls_context == NULL) { + if (client_ptr->tls_context == nullptr) { log(ERROR, "tls_accept() failed"); close_client(client_ptr); return; @@ -263,21 +248,20 @@ void Server::add_client(int sock, Acceptor *acceptor) process_client(client_ptr); } -void Server::add_client_from_serialized(const ClientProto &client) +void Server::add_client_from_serialized(const ClientProto &client, const vector> &short_responses) { - MutexLock lock(&mutex); + lock_guard lock(mu); Stream *stream; int stream_index = lookup_stream_by_url(client.url()); if (stream_index == -1) { assert(client.state() != Client::SENDING_DATA); - stream = NULL; + stream = nullptr; } else { - stream = streams[stream_index]; + stream = streams[stream_index].get(); } - pair::iterator, bool> ret = - clients.insert(make_pair(client.sock(), Client(client, stream))); - assert(ret.second == true); // Should not already exist. - Client *client_ptr = &ret.first->second; + auto inserted = clients.insert(make_pair(client.sock(), Client(client, short_responses, stream))); + assert(inserted.second == true); // Should not already exist. + Client *client_ptr = &inserted.first->second; // Connection timestamps must be nondecreasing. assert(clients_ordered_by_connect_time.empty() || @@ -325,51 +309,51 @@ int Server::lookup_stream_by_url(const string &url) const int Server::add_stream(const string &url, size_t backlog_size, size_t prebuffering_bytes, Stream::Encoding encoding, Stream::Encoding src_encoding) { - MutexLock lock(&mutex); + lock_guard lock(mu); stream_url_map.insert(make_pair(url, streams.size())); - streams.push_back(new Stream(url, backlog_size, prebuffering_bytes, encoding, src_encoding)); + streams.emplace_back(new Stream(url, backlog_size, prebuffering_bytes, encoding, src_encoding)); return streams.size() - 1; } int Server::add_stream_from_serialized(const StreamProto &stream, int data_fd) { - MutexLock lock(&mutex); + lock_guard lock(mu); stream_url_map.insert(make_pair(stream.url(), streams.size())); - streams.push_back(new Stream(stream, data_fd)); + streams.emplace_back(new Stream(stream, data_fd)); return streams.size() - 1; } void Server::set_backlog_size(int stream_index, size_t new_size) { - MutexLock lock(&mutex); + lock_guard lock(mu); 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); + lock_guard lock(mu); 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) { - MutexLock lock(&mutex); + lock_guard lock(mu); assert(stream_index >= 0 && stream_index < ssize_t(streams.size())); streams[stream_index]->encoding = encoding; } void Server::set_src_encoding(int stream_index, Stream::Encoding encoding) { - MutexLock lock(&mutex); + lock_guard lock(mu); assert(stream_index >= 0 && stream_index < ssize_t(streams.size())); streams[stream_index]->src_encoding = encoding; } void Server::set_header(int stream_index, const string &http_header, const string &stream_header) { - MutexLock lock(&mutex); + lock_guard lock(mu); assert(stream_index >= 0 && stream_index < ssize_t(streams.size())); streams[stream_index]->http_header = http_header; @@ -388,7 +372,7 @@ void Server::set_header(int stream_index, const string &http_header, const strin void Server::set_pacing_rate(int stream_index, uint32_t pacing_rate) { - MutexLock lock(&mutex); + lock_guard lock(mu); assert(clients.empty()); assert(stream_index >= 0 && stream_index < ssize_t(streams.size())); streams[stream_index]->pacing_rate = pacing_rate; @@ -396,7 +380,7 @@ void Server::set_pacing_rate(int stream_index, uint32_t pacing_rate) void Server::add_gen204(const std::string &url, const std::string &allow_origin) { - MutexLock lock(&mutex); + lock_guard lock(mu); assert(clients.empty()); ping_url_map[url] = allow_origin; } @@ -430,7 +414,7 @@ void Server::process_client(Client *client) { switch (client->state) { case Client::READING_REQUEST: { - if (client->tls_context != NULL) { + if (client->tls_context != nullptr) { if (send_pending_tls_data(client)) { // send_pending_tls_data() hit postconditions #1 or #4. return; @@ -441,7 +425,7 @@ read_request_again: // Try to read more of the request. char buf[1024]; int ret; - if (client->tls_context == NULL) { + if (client->tls_context == nullptr) { ret = read_nontls_data(client, buf, sizeof(buf)); if (ret == -1) { // read_nontls_data() hit postconditions #1 or #2. @@ -511,8 +495,8 @@ sending_header_or_short_response_again: int ret; do { ret = write(client->sock, - client->header_or_short_response.data() + client->header_or_short_response_bytes_sent, - client->header_or_short_response.size() - client->header_or_short_response_bytes_sent); + client->header_or_short_response->data() + client->header_or_short_response_bytes_sent, + client->header_or_short_response->size() - client->header_or_short_response_bytes_sent); } while (ret == -1 && errno == EINTR); if (ret == -1 && errno == EAGAIN) { @@ -531,37 +515,52 @@ sending_header_or_short_response_again: } client->header_or_short_response_bytes_sent += ret; - assert(client->header_or_short_response_bytes_sent <= client->header_or_short_response.size()); + assert(client->header_or_short_response_bytes_sent <= client->header_or_short_response->size()); - if (client->header_or_short_response_bytes_sent < client->header_or_short_response.size()) { + if (client->header_or_short_response_bytes_sent < client->header_or_short_response->size()) { // We haven't sent all yet. Fine; go another round. goto sending_header_or_short_response_again; } // We're done sending the header or error! Clear it to release some memory. - client->header_or_short_response.clear(); + client->header_or_short_response = nullptr; + client->header_or_short_response_holder.clear(); + client->header_or_short_response_ref.reset(); if (client->state == Client::SENDING_SHORT_RESPONSE) { - // We're done sending the error, so now close. - // This is postcondition #1. - close_client(client); + if (more_requests(client)) { + // We're done sending the error, but should keep on reading new requests. + goto read_request_again; + } else { + // We're done sending the error, so now close. + // This is postcondition #1. + close_client(client); + } return; } Stream *stream = client->stream; - if (client->stream_pos == size_t(-2)) { + if (client->stream_pos == Client::STREAM_POS_AT_START) { // Start sending from the beginning of the backlog. client->stream_pos = min( stream->bytes_received - stream->backlog_size, 0); client->state = Client::SENDING_DATA; goto sending_data; + } else if (client->stream_pos_end != Client::STREAM_POS_NO_END) { + // We're sending a fragment, and should have all of it, + // so start sending right away. + assert(client->stream_pos >= 0); + client->state = Client::SENDING_DATA; + goto sending_data; } else if (stream->prebuffering_bytes == 0) { // 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. // Note that this is functionally identical to the next if branch, // except that we save a binary search. + assert(client->stream_pos == Client::STREAM_POS_AT_END); + assert(client->stream_pos_end == Client::STREAM_POS_NO_END); client->stream_pos = stream->bytes_received; client->state = Client::WAITING_FOR_KEYFRAME; } else { @@ -571,6 +570,8 @@ sending_header_or_short_response_again: // so that we are conservative and never add extra latency over just // waiting (assuming CBR or nearly so); otherwise, we could want e.g. // 100 kB prebuffer but end up sending a 10 MB GOP. + assert(client->stream_pos == Client::STREAM_POS_AT_END); + assert(client->stream_pos_end == Client::STREAM_POS_NO_END); deque::const_iterator starting_point_it = lower_bound(stream->suitable_starting_points.begin(), stream->suitable_starting_points.end(), @@ -622,9 +623,24 @@ sending_data: Stream *stream = client->stream; sending_data_again: - size_t bytes_to_send = stream->bytes_received - client->stream_pos; + size_t bytes_to_send; + if (client->stream_pos_end == Client::STREAM_POS_NO_END) { + bytes_to_send = stream->bytes_received - client->stream_pos; + } else { + bytes_to_send = client->stream_pos_end - client->stream_pos; + } assert(bytes_to_send <= stream->backlog_size); if (bytes_to_send == 0) { + if (client->stream_pos == client->stream_pos_end) { // We have a definite end, and we're at it. + if (more_requests(client)) { + // We're done sending the fragment, but should keep on reading new requests. + goto read_request_again; + } else { + // We're done sending the fragment, so now close. + // This is postcondition #1. + close_client(client); + } + } return; } @@ -656,13 +672,17 @@ sending_data_again: client->stream_pos += ret; client->bytes_sent += ret; - if (client->stream_pos == stream->bytes_received) { + assert(client->stream_pos_end == Client::STREAM_POS_NO_END || client->stream_pos <= client->stream_pos_end); + if (client->stream_pos == client->stream_pos_end) { + goto sending_data_again; // Will see that bytes_to_send == 0 and end. + } else if (client->stream_pos == stream->bytes_received) { // We don't have any more data for this client, so put it to sleep. // This is postcondition #3. stream->put_client_to_sleep(client); } else if (more_data && size_t(ret) == bytes_to_send) { goto sending_data_again; } + // We'll also get here for postcondition #4 (similar to the EAGAIN path above). break; } default: @@ -673,9 +693,9 @@ sending_data_again: bool Server::send_pending_tls_data(Client *client) { // See if there's data from the TLS library to write. - if (client->tls_data_to_send == NULL) { + if (client->tls_data_to_send == nullptr) { client->tls_data_to_send = tls_get_write_buffer(client->tls_context, &client->tls_data_left_to_send); - if (client->tls_data_to_send == NULL) { + if (client->tls_data_to_send == nullptr) { // Really no data to send. return false; } @@ -704,7 +724,7 @@ send_data_again: if (ret > 0 && size_t(ret) == client->tls_data_left_to_send) { // All data has been sent, so we don't need to go to sleep. tls_buffer_clear(client->tls_context); - client->tls_data_to_send = NULL; + client->tls_data_to_send = nullptr; return false; } @@ -802,21 +822,32 @@ read_again: void Server::skip_lost_data(Client *client) { Stream *stream = client->stream; - if (stream == NULL) { + if (stream == nullptr) { 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; + if (!client->close_after_response) { + assert(client->stream_pos_end != Client::STREAM_POS_NO_END); + + // We've already sent a Content-length, so we can't just skip data. + // Close the connection immediately and hope the other side + // is able to figure out that there was an error and it needs to skip. + client->close_after_response = true; + client->stream_pos = client->stream_pos_end; + } else { + client->stream_pos = stream->bytes_received - stream->backlog_size; + } } } int Server::parse_request(Client *client) { vector lines = split_lines(client->request); + client->request.clear(); if (lines.empty()) { return 400; // Bad request (empty). } @@ -834,7 +865,7 @@ int Server::parse_request(Client *client) } vector request_tokens = split_tokens(lines[0]); - if (request_tokens.size() < 2) { + if (request_tokens.size() < 3) { return 400; // Bad request (empty). } if (request_tokens[0] != "GET") { @@ -844,10 +875,64 @@ int Server::parse_request(Client *client) string url = request_tokens[1]; client->url = url; if (url.size() > 8 && url.find("?backlog") == url.size() - 8) { - client->stream_pos = -2; + client->stream_pos = Client::STREAM_POS_AT_START; url = url.substr(0, url.size() - 8); } else { - client->stream_pos = -1; + size_t pos = url.find("?frag="); + if (pos != string::npos) { + // Parse an endpoint of the type /stream.mp4?frag=1234-5678. + const char *ptr = url.c_str() + pos + 6; + + // "?frag=header" is special. + if (strcmp(ptr, "header") == 0) { + client->stream_pos = Client::STREAM_POS_HEADER_ONLY; + client->stream_pos_end = -1; + } else { + char *endptr; + long long frag_start = strtol(ptr, &endptr, 10); + if (ptr == endptr || frag_start < 0 || frag_start == LLONG_MAX) { + return 400; // Bad request. + } + if (*endptr != '-') { + return 400; // Bad request. + } + ptr = endptr + 1; + + long long frag_end = strtol(ptr, &endptr, 10); + if (ptr == endptr || frag_end < frag_start || frag_end == LLONG_MAX) { + return 400; // Bad request. + } + + if (*endptr != '\0') { + return 400; // Bad request. + } + + client->stream_pos = frag_start; + client->stream_pos_end = frag_end; + } + url = url.substr(0, pos); + } else { + client->stream_pos = -1; + client->stream_pos_end = -1; + } + } + + // Figure out if we're supposed to close the socket after we've delivered the response. + string protocol = request_tokens[2]; + if (protocol.find("HTTP/") != 0) { + return 400; // Bad request. + } + client->close_after_response = false; + client->http_11 = true; + if (protocol == "HTTP/1.0") { + // No persistent connections. + client->close_after_response = true; + client->http_11 = false; + } else { + multimap::const_iterator connection_it = headers.find("Connection"); + if (connection_it != headers.end() && connection_it->second == "close") { + client->close_after_response = true; + } } map::const_iterator stream_url_map_it = stream_url_map.find(url); @@ -860,11 +945,32 @@ int Server::parse_request(Client *client) } } - Stream *stream = streams[stream_url_map_it->second]; + Stream *stream = streams[stream_url_map_it->second].get(); if (stream->http_header.empty()) { return 503; // Service unavailable. } + if (client->stream_pos_end == Client::STREAM_POS_NO_END) { + // This stream won't end, so we don't have a content-length, + // and can just as well tell the client it's Connection: close + // (otherwise, we'd have to implement chunking TE for no good reason). + client->close_after_response = true; + } else { + if (stream->encoding == Stream::STREAM_ENCODING_METACUBE) { + // This doesn't make any sense, and is hard to implement, too. + return 416; // Range not satisfiable. + } + + // Check that we have the requested fragment in our backlog. + size_t buffer_end = stream->bytes_received; + size_t buffer_start = (buffer_end <= stream->backlog_size) ? 0 : buffer_end - stream->backlog_size; + + if (client->stream_pos_end > buffer_end || + client->stream_pos < buffer_start) { + return 416; // Range not satisfiable. + } + } + 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) { @@ -879,39 +985,71 @@ int Server::parse_request(Client *client) void Server::construct_header(Client *client) { Stream *stream = client->stream; + string response = stream->http_header; + if (client->stream_pos == Client::STREAM_POS_HEADER_ONLY) { + char buf[64]; + snprintf(buf, sizeof(buf), "Content-length: %zu\r\n", stream->stream_header.size()); + response.append(buf); + } else if (client->stream_pos_end != Client::STREAM_POS_NO_END) { + char buf[64]; + snprintf(buf, sizeof(buf), "Content-length: %zu\r\n", client->stream_pos_end - client->stream_pos); + response.append(buf); + } + if (client->http_11) { + assert(response.find("HTTP/1.0") == 0); + response[7] = '1'; // Change to HTTP/1.1. + if (client->close_after_response) { + response.append("Connection: close\r\n"); + } + } else { + assert(client->close_after_response); + } if (stream->encoding == Stream::STREAM_ENCODING_RAW) { - client->header_or_short_response = stream->http_header + - "\r\n" + - stream->stream_header; + response.append("\r\n"); } else if (stream->encoding == Stream::STREAM_ENCODING_METACUBE) { - client->header_or_short_response = stream->http_header + - "Content-encoding: metacube\r\n" + - "\r\n"; + response.append("Content-encoding: metacube\r\n\r\n"); if (!stream->stream_header.empty()) { metacube2_block_header hdr; memcpy(hdr.sync, METACUBE2_SYNC, sizeof(hdr.sync)); hdr.size = htonl(stream->stream_header.size()); hdr.flags = htons(METACUBE_FLAGS_HEADER); hdr.csum = htons(metacube2_compute_crc(&hdr)); - client->header_or_short_response.append( - string(reinterpret_cast(&hdr), sizeof(hdr))); + response.append(string(reinterpret_cast(&hdr), sizeof(hdr))); } - client->header_or_short_response.append(stream->stream_header); } else { assert(false); } + if (client->stream_pos == Client::STREAM_POS_HEADER_ONLY) { + client->state = Client::SENDING_SHORT_RESPONSE; + response.append(stream->stream_header); + } else { + client->state = Client::SENDING_HEADER; + if (client->stream_pos_end == Client::STREAM_POS_NO_END) { // Fragments don't contain stream headers. + response.append(stream->stream_header); + } + } + + client->header_or_short_response_holder = move(response); + client->header_or_short_response = &client->header_or_short_response_holder; // Switch states. - client->state = Client::SENDING_HEADER; change_epoll_events(client, EPOLLOUT | EPOLLET | EPOLLRDHUP); } void Server::construct_error(Client *client, int error_code) { char error[256]; - snprintf(error, 256, "HTTP/1.0 %d Error\r\nContent-type: text/plain\r\n\r\nSomething went wrong. Sorry.\r\n", - error_code); - client->header_or_short_response = error; + if (client->http_11 && client->close_after_response) { + snprintf(error, sizeof(error), + "HTTP/1.1 %d Error\r\nContent-type: text/plain\r\nConnection: close\r\n\r\nSomething went wrong. Sorry.\r\n", + error_code); + } else { + snprintf(error, sizeof(error), + "HTTP/1.%d %d Error\r\nContent-type: text/plain\r\nContent-length: 30\r\n\r\nSomething went wrong. Sorry.\r\n", + client->http_11, error_code); + } + client->header_or_short_response_holder = error; + client->header_or_short_response = &client->header_or_short_response_holder; // Switch states. client->state = Client::SENDING_SHORT_RESPONSE; @@ -923,19 +1061,25 @@ void Server::construct_204(Client *client) map::const_iterator ping_url_map_it = ping_url_map.find(client->url); assert(ping_url_map_it != ping_url_map.end()); - if (ping_url_map_it->second.empty()) { - client->header_or_short_response = - "HTTP/1.0 204 No Content\r\n" - "\r\n"; + string response; + if (client->http_11) { + response = "HTTP/1.1 204 No Content\r\n"; + if (client->close_after_response) { + response.append("Connection: close\r\n"); + } } else { - char response[256]; - snprintf(response, 256, - "HTTP/1.0 204 No Content\r\n" - "Access-Control-Allow-Origin: %s\r\n" - "\r\n", - ping_url_map_it->second.c_str()); - client->header_or_short_response = response; + response = "HTTP/1.0 204 No Content\r\n"; + assert(client->close_after_response); + } + if (!ping_url_map_it->second.empty()) { + response.append("Access-Control-Allow-Origin: "); + response.append(ping_url_map_it->second); + response.append("\r\n"); } + response.append("\r\n"); + + client->header_or_short_response_holder = move(response); + client->header_or_short_response = &client->header_or_short_response_holder; // Switch states. client->state = Client::SENDING_SHORT_RESPONSE; @@ -951,13 +1095,13 @@ void delete_from(vector *v, T elem) void Server::close_client(Client *client) { - if (epoll_ctl(epoll_fd, EPOLL_CTL_DEL, client->sock, NULL) == -1) { + if (epoll_ctl(epoll_fd, EPOLL_CTL_DEL, client->sock, nullptr) == -1) { log_perror("epoll_ctl(EPOLL_CTL_DEL)"); exit(1); } // This client could be sleeping, so we'll need to fix that. (Argh, O(n).) - if (client->stream != NULL) { + if (client->stream != nullptr) { delete_from(&client->stream->sleeping_clients, client); delete_from(&client->stream->to_process, client); } @@ -987,18 +1131,38 @@ void Server::change_epoll_events(Client *client, uint32_t events) } } +bool Server::more_requests(Client *client) +{ + if (client->close_after_response) { + return false; + } + + // Switch states and reset the parsers. We don't reset statistics. + client->state = Client::READING_REQUEST; + client->url.clear(); + client->stream = NULL; + client->header_or_short_response = nullptr; + client->header_or_short_response_holder.clear(); + client->header_or_short_response_ref.reset(); + client->header_or_short_response_bytes_sent = 0; + + change_epoll_events(client, EPOLLIN | EPOLLET | EPOLLRDHUP); // No TLS handshake, so no EPOLLOUT needed. + + return true; +} + void Server::process_queued_data() { { - MutexLock lock(&queued_clients_mutex); + lock_guard lock(queued_clients_mutex); - for (size_t i = 0; i < queued_add_clients.size(); ++i) { - add_client(queued_add_clients[i].first, queued_add_clients[i].second); + for (const pair &id_and_acceptor : queued_add_clients) { + add_client(id_and_acceptor.first, id_and_acceptor.second); } queued_add_clients.clear(); } - for (size_t i = 0; i < streams.size(); ++i) { - streams[i]->process_queued_data(); + for (unique_ptr &stream : streams) { + stream->process_queued_data(); } }