X-Git-Url: https://git.sesse.net/?p=cubemap;a=blobdiff_plain;f=server.cpp;h=5c1f31bef451300f5cdf335261a7c9fd2aaedbae;hp=266d41809765075caaa750be0fd06809d707481d;hb=5ee2e95320160f0d01a9af5cea5c799c3ce1eaa9;hpb=ae994771c0747d43bd1ed422224f4caacb95ca9f diff --git a/server.cpp b/server.cpp index 266d418..5c1f31b 100644 --- a/server.cpp +++ b/server.cpp @@ -1,13 +1,18 @@ #include #include +#include +#include +#include +#include #include +#include #include #include +#include #include #include #include #include -#include #include #include #include @@ -15,75 +20,115 @@ #include #include -#include "markpool.h" -#include "mutexlock.h" +#include "ktls.h" +#include "tlse.h" + +#include "acceptor.h" +#include "accesslog.h" +#include "log.h" +#include "metacube2.h" #include "parse.h" #include "server.h" #include "state.pb.h" #include "stream.h" +#include "util.h" + +#ifndef SO_MAX_PACING_RATE +#define SO_MAX_PACING_RATE 47 +#endif using namespace std; -Server::Server() +extern AccessLogThread *access_log; + +namespace { + +inline bool is_equal(timespec a, timespec b) { - pthread_mutex_init(&mutex, NULL); - pthread_mutex_init(&queued_data_mutex, NULL); + 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() +{ epoll_fd = epoll_create(1024); // Size argument is ignored. if (epoll_fd == -1) { - perror("epoll_fd"); + log_perror("epoll_fd"); exit(1); } } Server::~Server() { - int ret; - do { - ret = close(epoll_fd); - } while (ret == -1 && errno == EINTR); - - if (ret == -1) { - perror("close(epoll_fd)"); - } + safe_close(epoll_fd); } 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; } -void Server::do_work() +vector Server::get_hls_zombies() { - for ( ;; ) { - int nfds = epoll_wait(epoll_fd, events, EPOLL_MAX_EVENTS, EPOLL_TIMEOUT_MS); - if (nfds == -1 && errno == EINTR) { - if (should_stop) { - return; - } - continue; + vector ret; + + timespec now; + if (clock_gettime(CLOCK_MONOTONIC_COARSE, &now) == -1) { + log_perror("clock_gettime(CLOCK_MONOTONIC_COARSE)"); + return ret; + } + + lock_guard lock(mu); + for (auto it = hls_zombies.begin(); it != hls_zombies.end(); ) { + if (is_earlier(it->second.expires, now)) { + hls_zombies.erase(it++); + } else { + ret.push_back(it->second); + ++it; } - if (nfds == -1) { - perror("epoll_wait"); + } + return ret; +} + +void Server::do_work() +{ + while (!should_stop()) { + // Wait until there's activity on at least one of the fds, + // or 20 ms (about one frame at 50 fps) has elapsed. + // + // We could in theory wait forever and rely on wakeup() + // from add_client_deferred() and add_data_deferred(), + // but wakeup is a pretty expensive operation, and the + // two threads might end up fighting over a lock, so it's + // seemingly (much) more efficient to just have a timeout here. + int nfds = epoll_pwait(epoll_fd, events, EPOLL_MAX_EVENTS, EPOLL_TIMEOUT_MS, &sigset_without_usr1_block); + if (nfds == -1 && errno != EINTR) { + log_perror("epoll_wait"); exit(1); } - MutexLock lock(&mutex); // We release the mutex between iterations. + lock_guard lock(mu); // We release the mutex between iterations. process_queued_data(); + // Process each client where we have socket activity. for (int i = 0; i < nfds; ++i) { - int fd = events[i].data.fd; - assert(clients.count(fd) != 0); - Client *client = &clients[fd]; + Client *client = reinterpret_cast(events[i].data.ptr); if (events[i].events & (EPOLLERR | EPOLLRDHUP | EPOLLHUP)) { close_client(client); @@ -93,136 +138,385 @@ void Server::do_work() process_client(client); } - for (map::iterator stream_it = streams.begin(); - stream_it != streams.end(); - ++stream_it) { + // Process each client where its stream has new data, + // even if there was no socket activity. + for (unique_ptr &stream : streams) { vector to_process; - swap(stream_it->second->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); } } - if (should_stop) { - return; + // 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()) { + const 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). + auto 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(); } } } -CubemapStateProto Server::serialize() +CubemapStateProto Server::serialize(unordered_map *short_response_pool) { // 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 (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 (map::const_iterator stream_it = streams.begin(); - stream_it != streams.end(); - ++stream_it) { - serialized.add_streams()->MergeFrom(stream_it->second->serialize()); + for (unique_ptr &stream : streams) { + serialized.add_streams()->MergeFrom(stream->serialize()); + } + for (const auto &key_and_zombie : hls_zombies) { + HLSZombieProto *proto = serialized.add_hls_zombies(); + proto->set_key(key_and_zombie.first); + + const HLSZombie &zombie = key_and_zombie.second; + proto->set_remote_addr(zombie.remote_addr); + proto->set_url(zombie.url); + proto->set_referer(zombie.referer); + proto->set_user_agent(zombie.user_agent); + proto->set_expires_sec(zombie.expires.tv_sec); + proto->set_expires_nsec(zombie.expires.tv_nsec); } return serialized; } -void Server::add_client_deferred(int sock) +void Server::add_client_deferred(int sock, Acceptor *acceptor) { - MutexLock lock(&queued_data_mutex); - queued_add_clients.push_back(sock); + lock_guard lock(queued_clients_mutex); + queued_add_clients.push_back(std::make_pair(sock, acceptor)); } -void Server::add_client(int sock) +void Server::add_client(int sock, Acceptor *acceptor) { - clients.insert(make_pair(sock, Client(sock))); + const bool is_tls = acceptor->is_tls(); + auto inserted = clients.insert(make_pair(sock, Client(sock))); + assert(inserted.second == true); // Should not already exist. + Client *client_ptr = &inserted.first->second; + + start_client_timeout_timer(client_ptr); // Start listening on data from this socket. epoll_event ev; - ev.events = EPOLLIN | EPOLLET | EPOLLRDHUP; - ev.data.u64 = 0; // Keep Valgrind happy. - ev.data.fd = sock; + if (is_tls) { + // Even in the initial state (READING_REQUEST), TLS needs to + // send data for the handshake, and thus might end up needing + // to know about EPOLLOUT. + ev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLRDHUP; + } else { + // EPOLLOUT will be added once we go out of READING_REQUEST. + ev.events = EPOLLIN | EPOLLET | EPOLLRDHUP; + } + ev.data.ptr = client_ptr; if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, sock, &ev) == -1) { - perror("epoll_ctl(EPOLL_CTL_ADD)"); + log_perror("epoll_ctl(EPOLL_CTL_ADD)"); exit(1); } - process_client(&clients[sock]); + if (is_tls) { + assert(tls_server_contexts.count(acceptor)); + client_ptr->tls_context = tls_accept(tls_server_contexts[acceptor]); + if (client_ptr->tls_context == nullptr) { + log(ERROR, "tls_accept() failed"); + close_client(client_ptr); + return; + } + tls_make_exportable(client_ptr->tls_context, 1); + } + + 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); - Stream *stream = find_stream(client.stream_id()); - clients.insert(make_pair(client.sock(), Client(client, stream))); - Client *client_ptr = &clients[client.sock()]; + 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 = nullptr; + } else { + stream = streams[stream_index].get(); + } + 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() || + !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) { - ev.events = EPOLLIN | EPOLLET | EPOLLRDHUP; + // See the corresponding comment in Server::add_client(). + if (client.has_tls_context()) { + ev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLRDHUP; + } else { + ev.events = EPOLLIN | EPOLLET | EPOLLRDHUP; + } } else { // If we don't have more data for this client, we'll be putting it into // the sleeping array again soon. ev.events = EPOLLOUT | EPOLLET | EPOLLRDHUP; } - ev.data.u64 = 0; // Keep Valgrind happy. - ev.data.fd = client.sock(); + ev.data.ptr = client_ptr; if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, client.sock(), &ev) == -1) { - perror("epoll_ctl(EPOLL_CTL_ADD)"); + log_perror("epoll_ctl(EPOLL_CTL_ADD)"); exit(1); } - if (client_ptr->state == Client::SENDING_DATA && - client_ptr->bytes_sent == 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); } } -void Server::add_stream(const string &stream_id, size_t backlog_size) +void Server::start_client_timeout_timer(Client *client) +{ + // 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 (clock_gettime(CLOCK_MONOTONIC_COARSE, &client->connect_time) == -1) { + log_perror("clock_gettime(CLOCK_MONOTONIC_COARSE)"); + } else { + if (!clients_ordered_by_connect_time.empty() && + is_earlier(client->connect_time, clients_ordered_by_connect_time.back().first)) { + client->connect_time = clients_ordered_by_connect_time.back().first; + } + clients_ordered_by_connect_time.push(make_pair(client->connect_time, client->sock)); + } +} + +int Server::lookup_stream_by_url(const string &url) const +{ + const auto stream_url_it = stream_url_map.find(url); + if (stream_url_it == stream_url_map.end()) { + return -1; + } + return stream_url_it->second; +} + +int Server::add_stream(const string &url, + const string &hls_url, + size_t backlog_size, + size_t prebuffering_bytes, + Stream::Encoding encoding, + Stream::Encoding src_encoding, + unsigned hls_frag_duration, + size_t hls_backlog_margin, + const string &allow_origin) +{ + lock_guard lock(mu); + stream_url_map.insert(make_pair(url, streams.size())); + if (!hls_url.empty()) { + stream_hls_url_map.insert(make_pair(hls_url, streams.size())); + } + streams.emplace_back(new Stream(url, backlog_size, prebuffering_bytes, encoding, src_encoding, hls_frag_duration, hls_backlog_margin, allow_origin)); + return streams.size() - 1; +} + +int Server::add_stream_from_serialized(const StreamProto &stream, int data_fd) +{ + lock_guard lock(mu); + stream_url_map.insert(make_pair(stream.url(), streams.size())); + // stream_hls_url_map will be updated in register_hls_url(), since it is not part + // of the serialized state (it will always be picked out from the configuration). + streams.emplace_back(new Stream(stream, data_fd)); + return streams.size() - 1; +} + +void Server::add_hls_zombie_from_serialized(const HLSZombieProto &zombie_proto) +{ + lock_guard lock(mu); + HLSZombie zombie; + zombie.remote_addr = zombie_proto.remote_addr(); + zombie.url = zombie_proto.url(); + zombie.referer = zombie_proto.referer(); + zombie.user_agent = zombie_proto.user_agent(); + zombie.expires.tv_sec = zombie_proto.expires_sec(); + zombie.expires.tv_nsec = zombie_proto.expires_nsec(); + hls_zombies[zombie_proto.key()] = move(zombie); +} + +void Server::set_backlog_size(int stream_index, size_t new_size) +{ + 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); - streams.insert(make_pair(stream_id, new Stream(stream_id, backlog_size))); + 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) +{ + 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) +{ + lock_guard lock(mu); + assert(stream_index >= 0 && stream_index < ssize_t(streams.size())); + streams[stream_index]->src_encoding = encoding; +} + +void Server::set_hls_frag_duration(int stream_index, unsigned hls_frag_duration) +{ + lock_guard lock(mu); + assert(stream_index >= 0 && stream_index < ssize_t(streams.size())); + streams[stream_index]->hls_frag_duration = hls_frag_duration; +} + +void Server::set_hls_backlog_margin(int stream_index, size_t hls_backlog_margin) +{ + lock_guard lock(mu); + assert(stream_index >= 0 && stream_index < ssize_t(streams.size())); + assert(hls_backlog_margin < streams[stream_index]->backlog_size); + streams[stream_index]->hls_backlog_margin = hls_backlog_margin; +} + +void Server::set_allow_origin(int stream_index, const string &allow_origin) +{ + lock_guard lock(mu); + assert(stream_index >= 0 && stream_index < ssize_t(streams.size())); + streams[stream_index]->allow_origin = allow_origin; } -void Server::add_stream_from_serialized(const StreamProto &stream) +void Server::register_hls_url(int stream_index, const string &hls_url) { - MutexLock lock(&mutex); - streams.insert(make_pair(stream.stream_id(), new Stream(stream))); + lock_guard lock(mu); + assert(stream_index >= 0 && stream_index < ssize_t(streams.size())); + assert(!hls_url.empty()); + stream_hls_url_map.insert(make_pair(hls_url, stream_index)); } -void Server::set_header(const string &stream_id, const string &header) +void Server::set_header(int stream_index, const string &http_header, const string &stream_header) { - MutexLock lock(&mutex); - find_stream(stream_id)->header = header; + lock_guard lock(mu); + assert(stream_index >= 0 && stream_index < ssize_t(streams.size())); + Stream *stream = streams[stream_index].get(); + stream->http_header = http_header; + + if (stream_header != stream->stream_header) { + // We cannot start at any of the older starting points anymore, + // since they'd get the wrong header for the stream (not to mention + // that a changed header probably means the stream restarted, + // which means any client starting on the old one would probably + // stop playing properly at the change point). Next block + // should be a suitable starting point (if not, something is + // pretty strange), so it will fill up again soon enough. + stream->suitable_starting_points.clear(); - // 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->bytes_sent == 0) { - construct_header(client); + if (!stream->fragments.empty()) { + stream->fragments.clear(); + ++stream->discontinuity_counter; + stream->clear_hls_playlist_cache(); } } + stream->stream_header = stream_header; } -void Server::set_mark_pool(const std::string &stream_id, MarkPool *mark_pool) +void Server::set_pacing_rate(int stream_index, uint32_t pacing_rate) { - MutexLock lock(&mutex); + lock_guard lock(mu); assert(clients.empty()); - find_stream(stream_id)->mark_pool = mark_pool; + assert(stream_index >= 0 && stream_index < ssize_t(streams.size())); + streams[stream_index]->pacing_rate = pacing_rate; +} + +void Server::add_gen204(const std::string &url, const std::string &allow_origin) +{ + lock_guard lock(mu); + assert(clients.empty()); + ping_url_map[url] = allow_origin; +} + +void Server::create_tls_context_for_acceptor(const Acceptor *acceptor) +{ + assert(acceptor->is_tls()); + + bool is_server = true; + TLSContext *server_context = tls_create_context(is_server, TLS_V12); + + const string &cert = acceptor->get_certificate_chain(); + int num_cert = tls_load_certificates(server_context, reinterpret_cast(cert.data()), cert.size()); + assert(num_cert > 0); // Should have been checked by config earlier. + + const string &key = acceptor->get_private_key(); + int num_key = tls_load_private_key(server_context, reinterpret_cast(key.data()), key.size()); + assert(num_key > 0); // Should have been checked by config earlier. + + tls_server_contexts.insert(make_pair(acceptor, server_context)); } -void Server::add_data_deferred(const string &stream_id, const char *data, size_t bytes) +void Server::add_data_deferred(int stream_index, const char *data, size_t bytes, uint16_t metacube_flags, const RationalPTS &pts) { - MutexLock lock(&queued_data_mutex); - queued_data[stream_id].append(string(data, data + bytes)); + assert(stream_index >= 0 && stream_index < ssize_t(streams.size())); + streams[stream_index]->add_data_deferred(data, bytes, metacube_flags, pts); } // See the .h file for postconditions after this function. @@ -230,35 +524,36 @@ void Server::process_client(Client *client) { switch (client->state) { case Client::READING_REQUEST: { + if (client->tls_context != nullptr && !client->in_ktls_mode) { + if (send_pending_tls_data(client)) { + // send_pending_tls_data() hit postconditions #1 or #4. + return; + } + } + read_request_again: // Try to read more of the request. char buf[1024]; int ret; - do { - ret = read(client->sock, buf, sizeof(buf)); - } while (ret == -1 && errno == EINTR); - - if (ret == -1 && errno == EAGAIN) { - // No more data right now. Nothing to do. - // This is postcondition #2. - return; - } - if (ret == -1) { - perror("read"); - close_client(client); - return; - } - if (ret == 0) { - // OK, the socket is closed. - close_client(client); - return; + if (client->tls_context == nullptr || client->in_ktls_mode) { + ret = read_plain_data(client, buf, sizeof(buf)); + if (ret == -1) { + // read_plain_data() hit postconditions #1 or #2. + return; + } + } else { + ret = read_tls_data(client, buf, sizeof(buf)); + if (ret == -1) { + // read_tls_data() hit postconditions #1, #2 or #4. + return; + } } RequestParseStatus status = wait_for_double_newline(&client->request, buf, ret); switch (status) { case RP_OUT_OF_SPACE: - fprintf(stderr, "WARNING: fd %d sent overlong request!\n", client->sock); + log(WARNING, "[%s] Client sent overlong request!", client->remote_addr.c_str()); close_client(client); return; case RP_NOT_FINISHED_YET: @@ -266,7 +561,7 @@ read_request_again: // See if there's more data for us. goto read_request_again; case RP_EXTRA_DATA: - fprintf(stderr, "WARNING: fd %d had junk data after request!\n", client->sock); + log(WARNING, "[%s] Junk data after request!", client->remote_addr.c_str()); close_client(client); return; case RP_FINISHED: @@ -277,23 +572,29 @@ read_request_again: int error_code = parse_request(client); if (error_code == 200) { - construct_header(client); + if (client->serving_hls_playlist) { + construct_hls_playlist(client); + } else { + construct_stream_header(client); + } + } else if (error_code == 204) { + construct_204(client); } else { construct_error(client, error_code); } // We've changed states, so fall through. - assert(client->state == Client::SENDING_ERROR || + assert(client->state == Client::SENDING_SHORT_RESPONSE || client->state == Client::SENDING_HEADER); } - case Client::SENDING_ERROR: + case Client::SENDING_SHORT_RESPONSE: case Client::SENDING_HEADER: { -sending_header_or_error_again: +sending_header_or_short_response_again: int ret; do { ret = write(client->sock, - client->header_or_error.data() + client->header_or_error_bytes_sent, - client->header_or_error.size() - client->header_or_error_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) { @@ -306,64 +607,166 @@ sending_header_or_error_again: if (ret == -1) { // Error! Postcondition #1. - perror("write"); + log_perror("write"); close_client(client); return; } - client->header_or_error_bytes_sent += ret; - assert(client->header_or_error_bytes_sent <= client->header_or_error.size()); + client->header_or_short_response_bytes_sent += ret; + assert(client->header_or_short_response_bytes_sent <= client->header_or_short_response->size()); - if (client->header_or_error_bytes_sent < client->header_or_error.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_error_again; + goto sending_header_or_short_response_again; } // We're done sending the header or error! Clear it to release some memory. - client->header_or_error.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_ERROR) { - // We're done sending the error, so now close. - // This is postcondition #1. - close_client(client); + if (client->state == Client::SENDING_SHORT_RESPONSE) { + 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; } - // 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. - // This is postcondition #3. + Stream *stream = client->stream; + hls_zombies.erase(client->get_hls_zombie_key()); + 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(ssize_t(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 { + // We're not going to send anything to the client before we have + // N bytes. However, this wait might be boring; we can just as well + // use it to send older data if we have it. We use lower_bound() + // 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(), + stream->bytes_received - stream->prebuffering_bytes); + if (starting_point_it == stream->suitable_starting_points.end()) { + // None found. Just put us at the end, and then wait for the + // first keyframe to appear. + client->stream_pos = stream->bytes_received; + client->state = Client::WAITING_FOR_KEYFRAME; + } else { + client->stream_pos = *starting_point_it; + client->state = Client::PREBUFFERING; + goto prebuffering; + } + } + // Fall through. + } + case Client::WAITING_FOR_KEYFRAME: { + Stream *stream = client->stream; + if (stream->suitable_starting_points.empty() || + client->stream_pos > stream->suitable_starting_points.back()) { + // 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->suitable_starting_points.back(); + client->state = Client::PREBUFFERING; + // Fall through. + } + case Client::PREBUFFERING: { +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. + // This is postcondition #3. + stream->put_client_to_sleep(client); + return; + } client->state = Client::SENDING_DATA; - client->bytes_sent = client->stream->bytes_received; - client->stream->put_client_to_sleep(client); - return; + // 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. +sending_data: + skip_lost_data(client); Stream *stream = client->stream; - size_t bytes_to_send = stream->bytes_received - client->bytes_sent; + +sending_data_again: + 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. + // Add (or overwrite) a HLS zombie. + timespec now; + if (clock_gettime(CLOCK_MONOTONIC_COARSE, &now) == -1) { + log_perror("clock_gettime(CLOCK_MONOTONIC_COARSE)"); + } else { + HLSZombie zombie; + zombie.remote_addr = client->remote_addr; + zombie.referer = client->referer; + zombie.user_agent = client->user_agent; + zombie.url = client->stream->url + "?frag="; + zombie.expires = now; + zombie.expires.tv_sec += client->stream->hls_frag_duration * 3; + hls_zombies[client->get_hls_zombie_key()] = move(zombie); + } + 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; } - if (bytes_to_send > stream->backlog_size) { - fprintf(stderr, "WARNING: fd %d lost %lld bytes, maybe too slow connection\n", - client->sock, - (long long int)(bytes_to_send - stream->backlog_size)); - client->bytes_sent = stream->bytes_received - stream->backlog_size; - bytes_to_send = stream->backlog_size; - } // See if we need to split across the circular buffer. bool more_data = false; - if ((client->bytes_sent % stream->backlog_size) + bytes_to_send > stream->backlog_size) { - bytes_to_send = stream->backlog_size - (client->bytes_sent % stream->backlog_size); + if ((client->stream_pos % stream->backlog_size) + bytes_to_send > stream->backlog_size) { + bytes_to_send = stream->backlog_size - (client->stream_pos % stream->backlog_size); more_data = true; } ssize_t ret; do { - loff_t offset = client->bytes_sent % 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); @@ -375,19 +778,24 @@ sending_data_again: } if (ret == -1) { // Error, close; postcondition #1. - perror("sendfile"); + log_perror("sendfile"); close_client(client); return; } + client->stream_pos += ret; client->bytes_sent += ret; - if (client->bytes_sent == 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) { + } 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: @@ -395,34 +803,357 @@ sending_data_again: } } +namespace { + +void flush_pending_data(int sock) +{ + // Flush pending data, which would otherwise wait for the 200ms TCP_CORK timer + // to elapsed; does not cancel out TCP_CORK (since that still takes priority), + // but does a one-off flush. + int one = 1; + if (setsockopt(sock, SOL_TCP, TCP_NODELAY, &one, sizeof(one)) == -1) { + log_perror("setsockopt(TCP_NODELAY)"); + // Can still continue. + } +} + +} // namespace + +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 == 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 == nullptr) { + // Really no data to send. + return false; + } + } + +send_data_again: + int ret; + do { + ret = write(client->sock, client->tls_data_to_send, client->tls_data_left_to_send); + } while (ret == -1 && errno == EINTR); + assert(ret < 0 || size_t(ret) <= client->tls_data_left_to_send); + + if (ret == -1 && errno == EAGAIN) { + // We're out of socket space, so now we're at the “low edge” of epoll's + // edge triggering. epoll will tell us when there is more room, so for now, + // just return. + // This is postcondition #4. + return true; + } + if (ret == -1) { + // Error! Postcondition #1. + log_perror("write"); + close_client(client); + return true; + } + 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 + // (although we are likely to do so immediately afterwards, + // due to lack of client data). + tls_buffer_clear(client->tls_context); + client->tls_data_to_send = nullptr; + + // Flush the data we just wrote, since the client probably + // is waiting for it. + flush_pending_data(client->sock); + return false; + } + + // More data to send, so try again. + client->tls_data_to_send += ret; + client->tls_data_left_to_send -= ret; + goto send_data_again; +} + +int Server::read_plain_data(Client *client, char *buf, size_t max_size) +{ + int ret; + do { + ret = read(client->sock, buf, max_size); + } while (ret == -1 && errno == EINTR); + + if (ret == -1 && errno == EAGAIN) { + // No more data right now. Nothing to do. + // This is postcondition #2. + return -1; + } + if (ret == -1) { + log_perror("read"); + close_client(client); + return -1; + } + if (ret == 0) { + // OK, the socket is closed. + close_client(client); + return -1; + } + + return ret; +} + +int Server::read_tls_data(Client *client, char *buf, size_t max_size) +{ +read_again: + assert(!client->in_ktls_mode); + + int ret; + do { + ret = read(client->sock, buf, max_size); + } while (ret == -1 && errno == EINTR); + + if (ret == -1 && errno == EAGAIN) { + // No more data right now. Nothing to do. + // This is postcondition #2. + return -1; + } + if (ret == -1) { + log_perror("read"); + close_client(client); + return -1; + } + if (ret == 0) { + // OK, the socket is closed. + close_client(client); + return -1; + } + + // Give it to the TLS library. + int err = tls_consume_stream(client->tls_context, reinterpret_cast(buf), ret, nullptr); + if (err < 0) { + log_tls_error("tls_consume_stream", err); + close_client(client); + return -1; + } + if (err == 0) { + // Not consumed any data. See if we can read more. + goto read_again; + } + + // Read any decrypted data available for us. (We can reuse buf, since it's free now.) + ret = tls_read(client->tls_context, reinterpret_cast(buf), max_size); + if (ret == 0) { + // No decrypted data for us yet, but there might be some more handshaking + // to send. Do that if needed, then look for more data. + if (send_pending_tls_data(client)) { + // send_pending_tls_data() hit postconditions #1 or #4. + return -1; + } + goto read_again; + } + if (ret < 0) { + log_tls_error("tls_read", ret); + close_client(client); + return -1; + } + + if (tls_established(client->tls_context)) { + // We're ready to enter kTLS mode, unless we still have some + // handshake data to send (which then must be sent as non-kTLS). + if (send_pending_tls_data(client)) { + // send_pending_tls_data() hit postconditions #1 or #4. + return -1; + } + int err = tls_make_ktls(client->tls_context, client->sock); // Don't overwrite ret. + if (err < 0) { + log_tls_error("tls_make_ktls", ret); + close_client(client); + return -1; + } + client->in_ktls_mode = true; + } + + assert(ret > 0); + return ret; +} + +// 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; + 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->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). } + // Parse the headers, for logging purposes. + HTTPHeaderMultimap headers = extract_headers(lines, client->remote_addr); + const auto referer_it = headers.find("Referer"); + if (referer_it != headers.end()) { + client->referer = referer_it->second; + } + const auto user_agent_it = headers.find("User-Agent"); + if (user_agent_it != headers.end()) { + client->user_agent = user_agent_it->second; + } + const auto x_playback_session_id_it = headers.find("X-Playback-Session-Id"); + if (x_playback_session_id_it != headers.end()) { + client->x_playback_session_id = x_playback_session_id_it->second; + } else { + client->x_playback_session_id.clear(); + } + 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") { return 400; // Should maybe be 405 instead? } - if (streams.count(request_tokens[1]) == 0) { - return 404; // Not found. + + string url = request_tokens[1]; + client->url = url; + if (url.size() > 8 && url.find("?backlog") == url.size() - 8) { + client->stream_pos = Client::STREAM_POS_AT_START; + url = url.substr(0, url.size() - 8); + } else { + 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 { + const auto connection_it = headers.find("Connection"); + if (connection_it != headers.end() && connection_it->second == "close") { + client->close_after_response = true; + } + } + + const auto stream_url_map_it = stream_url_map.find(url); + if (stream_url_map_it != stream_url_map.end()) { + // Serve a regular stream.. + client->stream = streams[stream_url_map_it->second].get(); + client->serving_hls_playlist = false; + } else { + const auto stream_hls_url_map_it = stream_hls_url_map.find(url); + if (stream_hls_url_map_it != stream_hls_url_map.end()) { + // Serve HLS playlist. + client->stream = streams[stream_hls_url_map_it->second].get(); + client->serving_hls_playlist = true; + } else { + const auto ping_url_map_it = ping_url_map.find(url); + if (ping_url_map_it == ping_url_map.end()) { + return 404; // Not found. + } else { + // Serve a ping (204 no error). + return 204; + } + } + } + + Stream *stream = client->stream; + if (stream->http_header.empty()) { + return 503; // Service unavailable. + } + + if (client->serving_hls_playlist) { + if (stream->encoding == Stream::STREAM_ENCODING_METACUBE) { + // This doesn't make any sense, and is hard to implement, too. + return 404; + } else { + return 200; + } } - client->stream_id = request_tokens[1]; - client->stream = find_stream(client->stream_id); - if (client->stream->mark_pool != NULL) { - client->fwmark = client->stream->mark_pool->get_mark(); + 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 { - client->fwmark = 0; // No mark. + 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. + } } - if (setsockopt(client->sock, SOL_SOCKET, SO_MARK, &client->fwmark, sizeof(client->fwmark)) == -1) { - if (client->fwmark != 0) { - perror("setsockopt(SO_MARK)"); + + 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(); @@ -430,103 +1161,266 @@ int Server::parse_request(Client *client) return 200; // OK! } -void Server::construct_header(Client *client) +void Server::construct_stream_header(Client *client) { - client->header_or_error = find_stream(client->stream_id)->header; - - // Switch states. - client->state = Client::SENDING_HEADER; + 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: %" PRIu64 "\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->allow_origin.empty()) { + response.append("Access-Control-Allow-Origin: "); + response.append(stream->allow_origin); + response.append("\r\n"); + } + if (stream->encoding == Stream::STREAM_ENCODING_RAW) { + response.append("\r\n"); + } else if (stream->encoding == Stream::STREAM_ENCODING_METACUBE) { + 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)); + response.append(string(reinterpret_cast(&hdr), sizeof(hdr))); + } + } 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); + } + } - epoll_event ev; - ev.events = EPOLLOUT | EPOLLET | EPOLLRDHUP; - ev.data.u64 = 0; // Keep Valgrind happy. - ev.data.fd = client->sock; + client->header_or_short_response_holder = move(response); + client->header_or_short_response = &client->header_or_short_response_holder; - if (epoll_ctl(epoll_fd, EPOLL_CTL_MOD, client->sock, &ev) == -1) { - perror("epoll_ctl(EPOLL_CTL_MOD)"); - exit(1); - } + // Switch states. + 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_error = 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_ERROR; + client->state = Client::SENDING_SHORT_RESPONSE; + change_epoll_events(client, EPOLLOUT | EPOLLET | EPOLLRDHUP); +} - epoll_event ev; - ev.events = EPOLLOUT | EPOLLET | EPOLLRDHUP; - ev.data.u64 = 0; // Keep Valgrind happy. - ev.data.fd = client->sock; +void Server::construct_hls_playlist(Client *client) +{ + Stream *stream = client->stream; + shared_ptr *cache; + if (client->http_11) { + if (client->close_after_response) { + cache = &stream->hls_playlist_http11_close; + } else { + cache = &stream->hls_playlist_http11_persistent; + } + } else { + assert(client->close_after_response); + cache = &stream->hls_playlist_http10; + } - if (epoll_ctl(epoll_fd, EPOLL_CTL_MOD, client->sock, &ev) == -1) { - perror("epoll_ctl(EPOLL_CTL_MOD)"); - exit(1); + if (*cache == nullptr) { + *cache = stream->generate_hls_playlist(client->http_11, client->close_after_response); + } + client->header_or_short_response_ref = *cache; + client->header_or_short_response = cache->get(); + + // Switch states. + client->state = Client::SENDING_SHORT_RESPONSE; + change_epoll_events(client, EPOLLOUT | EPOLLET | EPOLLRDHUP); +} + +void Server::construct_204(Client *client) +{ + const auto ping_url_map_it = ping_url_map.find(client->url); + assert(ping_url_map_it != ping_url_map.end()); + + 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 { + 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; + change_epoll_events(client, EPOLLOUT | EPOLLET | EPOLLRDHUP); } +namespace { + template void delete_from(vector *v, T elem) { typename vector::iterator new_end = remove(v->begin(), v->end(), elem); v->erase(new_end, v->end()); } + +void send_ktls_close(int sock) +{ + uint8_t record_type = 21; // Alert. + uint8_t body[] = { + 1, // Warning level (but still fatal!). + 0, // close_notify. + }; + + int cmsg_len = sizeof(record_type); + char buf[CMSG_SPACE(cmsg_len)]; + + msghdr msg = {0}; + msg.msg_control = buf; + msg.msg_controllen = sizeof(buf); + cmsghdr *cmsg = CMSG_FIRSTHDR(&msg); + cmsg->cmsg_level = SOL_TLS; + cmsg->cmsg_type = TLS_SET_RECORD_TYPE; + cmsg->cmsg_len = CMSG_LEN(cmsg_len); + *CMSG_DATA(cmsg) = record_type; + msg.msg_controllen = cmsg->cmsg_len; + + iovec msg_iov; + msg_iov.iov_base = body; + msg_iov.iov_len = sizeof(body); + msg.msg_iov = &msg_iov; + msg.msg_iovlen = 1; + + int err; + do { + err = sendmsg(sock, &msg, 0); + } while (err == -1 && errno == EINTR); // Ignore all other errors. +} + +} // namespace void Server::close_client(Client *client) { - if (epoll_ctl(epoll_fd, EPOLL_CTL_DEL, client->sock, NULL) == -1) { - perror("epoll_ctl(EPOLL_CTL_DEL)"); + 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); - if (client->stream->mark_pool != NULL) { - int fwmark = client->fwmark; - client->stream->mark_pool->release_mark(fwmark); + } + + if (client->tls_context) { + if (client->in_ktls_mode) { + // Keep GnuTLS happy. + send_ktls_close(client->sock); } + tls_destroy_context(client->tls_context); } - // Bye-bye! - int ret; - do { - ret = close(client->sock); - } while (ret == -1 && errno == EINTR); + // Log to access_log. + access_log->write(client->get_stats()); - if (ret == -1) { - perror("close"); - } + // Bye-bye! + safe_close(client->sock); clients.erase(client->sock); } - -Stream *Server::find_stream(const string &stream_id) + +void Server::change_epoll_events(Client *client, uint32_t events) +{ + epoll_event ev; + ev.events = events; + ev.data.ptr = client; + + if (epoll_ctl(epoll_fd, EPOLL_CTL_MOD, client->sock, &ev) == -1) { + log_perror("epoll_ctl(EPOLL_CTL_MOD)"); + exit(1); + } +} + +bool Server::more_requests(Client *client) { - map::iterator it = streams.find(stream_id); - assert(it != streams.end()); - return it->second; + if (client->close_after_response) { + return false; + } + + // Log to access_log. + access_log->write(client->get_stats()); + + flush_pending_data(client->sock); + + // 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; + client->bytes_sent = 0; + start_client_timeout_timer(client); + + change_epoll_events(client, EPOLLIN | EPOLLET | EPOLLRDHUP); // No TLS handshake, so no EPOLLOUT needed. + + return true; } void Server::process_queued_data() { - MutexLock lock(&queued_data_mutex); + { + lock_guard lock(queued_clients_mutex); - for (size_t i = 0; i < queued_add_clients.size(); ++i) { - add_client(queued_add_clients[i]); + for (const pair &id_and_acceptor : queued_add_clients) { + add_client(id_and_acceptor.first, id_and_acceptor.second); + } + queued_add_clients.clear(); } - queued_add_clients.clear(); - - for (map::iterator queued_it = queued_data.begin(); - queued_it != queued_data.end(); - ++queued_it) { - Stream *stream = find_stream(queued_it->first); - stream->add_data(queued_it->second.data(), queued_it->second.size()); + + for (unique_ptr &stream : streams) { + stream->process_queued_data(); } - queued_data.clear(); }