X-Git-Url: https://git.sesse.net/?p=cubemap;a=blobdiff_plain;f=server.cpp;h=6b92fc6af5eb6eec6fa9983a587681dfc2e14623;hp=31e5c0a31da327bc80499c543bbb4dee932b74bb;hb=fbfb955ee7233030357ec32c0d613f9279700d70;hpb=97bdb597d4847308ce9d6982505b56a3a09e930b diff --git a/server.cpp b/server.cpp index 31e5c0a..6b92fc6 100644 --- a/server.cpp +++ b/server.cpp @@ -3,7 +3,6 @@ #include #include #include -#include #include #include #include @@ -13,16 +12,103 @@ #include #include #include +#include #include "metacube.h" #include "server.h" #include "mutexlock.h" +#include "parse.h" +#include "state.pb.h" using namespace std; +Client::Client(int sock) + : sock(sock), + state(Client::READING_REQUEST), + stream(NULL), + header_or_error_bytes_sent(0), + bytes_sent(0) +{ + request.reserve(1024); +} + +Client::Client(const ClientProto &serialized, Stream *stream) + : sock(serialized.sock()), + state(State(serialized.state())), + request(serialized.request()), + stream_id(serialized.stream_id()), + stream(stream), + header_or_error(serialized.header_or_error()), + header_or_error_bytes_sent(serialized.header_or_error_bytes_sent()), + bytes_sent(serialized.bytes_sent()) +{ +} + +ClientProto Client::serialize() const +{ + ClientProto serialized; + serialized.set_sock(sock); + serialized.set_state(state); + serialized.set_request(request); + serialized.set_stream_id(stream_id); + serialized.set_header_or_error(header_or_error); + serialized.set_header_or_error_bytes_sent(serialized.header_or_error_bytes_sent()); + serialized.set_bytes_sent(bytes_sent); + return serialized; +} + +Stream::Stream(const string &stream_id) + : stream_id(stream_id), + data(new char[BACKLOG_SIZE]), + data_size(0) +{ + memset(data, 0, BACKLOG_SIZE); +} + +Stream::~Stream() +{ + delete[] data; +} + +Stream::Stream(const StreamProto &serialized) + : stream_id(serialized.stream_id()), + header(serialized.header()), + data(new char[BACKLOG_SIZE]), + data_size(serialized.data_size()) +{ + assert(serialized.data().size() == BACKLOG_SIZE); + memcpy(data, serialized.data().data(), BACKLOG_SIZE); +} + +StreamProto Stream::serialize() const +{ + StreamProto serialized; + serialized.set_header(header); + serialized.set_data(string(data, data + BACKLOG_SIZE)); + serialized.set_data_size(data_size); + serialized.set_stream_id(stream_id); + return serialized; +} + +void Stream::put_client_to_sleep(Client *client) +{ + sleeping_clients.push_back(client); +} + +void Stream::wake_up_all_clients() +{ + if (to_process.empty()) { + swap(sleeping_clients, to_process); + } else { + to_process.insert(to_process.end(), sleeping_clients.begin(), sleeping_clients.end()); + sleeping_clients.clear(); + } +} + Server::Server() { pthread_mutex_init(&mutex, NULL); + pthread_mutex_init(&queued_data_mutex, NULL); epoll_fd = epoll_create(1024); // Size argument is ignored. if (epoll_fd == -1) { @@ -31,10 +117,40 @@ Server::Server() } } +Server::~Server() +{ + int ret; + do { + ret = close(epoll_fd); + } while (ret == -1 && errno == EINTR); + + if (ret == -1) { + perror("close(epoll_fd)"); + } +} + void Server::run() { - pthread_t thread; - pthread_create(&thread, NULL, Server::do_work_thunk, this); + should_stop = false; + + // Joinable is already the default, but it's good to be certain. + pthread_attr_t attr; + pthread_attr_init(&attr); + pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE); + pthread_create(&worker_thread, &attr, Server::do_work_thunk, this); +} + +void Server::stop() +{ + { + MutexLock lock(&mutex); + should_stop = true; + } + + if (pthread_join(worker_thread, NULL) == -1) { + perror("pthread_join"); + exit(1); + } } void *Server::do_work_thunk(void *arg) @@ -47,44 +163,472 @@ void *Server::do_work_thunk(void *arg) void Server::do_work() { for ( ;; ) { - MutexLock lock(&mutex); - printf("server thread running\n"); - sleep(1); + int nfds = epoll_wait(epoll_fd, events, EPOLL_MAX_EVENTS, EPOLL_TIMEOUT_MS); + if (nfds == -1 && errno == EINTR) { + continue; + } + if (nfds == -1) { + perror("epoll_wait"); + exit(1); + } + + MutexLock lock(&mutex); // We release the mutex between iterations. + + if (should_stop) { + return; + } + + process_queued_data(); + + for (int i = 0; i < nfds; ++i) { + int fd = events[i].data.fd; + assert(clients.count(fd) != 0); + Client *client = &clients[fd]; + + if (events[i].events & (EPOLLERR | EPOLLRDHUP | EPOLLHUP)) { + close_client(client); + continue; + } + + process_client(client); + } + + for (map::iterator stream_it = streams.begin(); + stream_it != streams.end(); + ++stream_it) { + 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]); + } + } } } - -void Server::add_client(int sock) + +CubemapStateProto Server::serialize() { - MutexLock lock(&mutex); - Client new_client; - new_client.state = Client::READING_REQUEST; - new_client.header_bytes_sent = 0; - new_client.bytes_sent = 0; + // We don't serialize anything queued, so empty the queues. + process_queued_data(); - clients.insert(make_pair(sock, new_client)); + 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 (map::const_iterator stream_it = streams.begin(); + stream_it != streams.end(); + ++stream_it) { + serialized.add_streams()->MergeFrom(stream_it->second->serialize()); + } + return serialized; +} + +void Server::add_client_deferred(int sock) +{ + MutexLock lock(&queued_data_mutex); + queued_add_clients.push_back(sock); +} + +void Server::add_client(int sock) +{ + clients.insert(make_pair(sock, Client(sock))); // Start listening on data from this socket. epoll_event ev; - ev.events = EPOLLIN; + ev.events = EPOLLIN | EPOLLET | EPOLLRDHUP; + ev.data.u64 = 0; // Keep Valgrind happy. ev.data.fd = sock; if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, sock, &ev) == -1) { perror("epoll_ctl(EPOLL_CTL_ADD)"); exit(1); } + + process_client(&clients[sock]); } - + +void Server::add_client_from_serialized(const ClientProto &client) +{ + 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()]; + + // Start listening on data from this socket. + epoll_event ev; + if (client.state() == Client::READING_REQUEST) { + 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(); + if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, client.sock(), &ev) == -1) { + perror("epoll_ctl(EPOLL_CTL_ADD)"); + exit(1); + } + + if (client_ptr->state == Client::SENDING_DATA && + client_ptr->bytes_sent == client_ptr->stream->data_size) { + client_ptr->stream->put_client_to_sleep(client_ptr); + } else { + process_client(client_ptr); + } +} + void Server::add_stream(const string &stream_id) { - // TODO + MutexLock lock(&mutex); + streams.insert(make_pair(stream_id, new Stream(stream_id))); +} + +void Server::add_stream_from_serialized(const StreamProto &stream) +{ + MutexLock lock(&mutex); + streams.insert(make_pair(stream.stream_id(), new Stream(stream))); } void Server::set_header(const string &stream_id, const string &header) { - // TODO - printf("got header! %lu bytes\n", header.size()); + MutexLock lock(&mutex); + find_stream(stream_id)->header = 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->bytes_sent == 0) { + construct_header(client); + } + } } - + +void Server::add_data_deferred(const string &stream_id, const char *data, size_t bytes) +{ + MutexLock lock(&queued_data_mutex); + queued_data[stream_id].append(string(data, data + bytes)); +} + void Server::add_data(const string &stream_id, const char *data, size_t bytes) { - // TODO + Stream *stream = find_stream(stream_id); + size_t pos = stream->data_size % BACKLOG_SIZE; + stream->data_size += bytes; + + if (pos + bytes > BACKLOG_SIZE) { + size_t to_copy = BACKLOG_SIZE - pos; + memcpy(stream->data + pos, data, to_copy); + data += to_copy; + bytes -= to_copy; + pos = 0; + } + + memcpy(stream->data + pos, data, bytes); + stream->wake_up_all_clients(); +} + +// See the .h file for postconditions after this function. +void Server::process_client(Client *client) +{ + switch (client->state) { + case Client::READING_REQUEST: { +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; + } + + 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); + close_client(client); + return; + case RP_NOT_FINISHED_YET: + // OK, we don't have the entire header yet. Fine; we'll get it later. + // 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); + close_client(client); + return; + case RP_FINISHED: + break; + } + + assert(status == RP_FINISHED); + + int error_code = parse_request(client); + if (error_code == 200) { + construct_header(client); + } else { + construct_error(client, error_code); + } + + // We've changed states, so fall through. + assert(client->state == Client::SENDING_ERROR || + client->state == Client::SENDING_HEADER); + } + case Client::SENDING_ERROR: + case Client::SENDING_HEADER: { +sending_header_or_error_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); + } while (ret == -1 && errno == EINTR); + + 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; + } + + if (ret == -1) { + // Error! Postcondition #1. + 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()); + + if (client->header_or_error_bytes_sent < client->header_or_error.size()) { + // We haven't sent all yet. Fine; go another round. + goto sending_header_or_error_again; + } + + // We're done sending the header or error! Clear it to release some memory. + client->header_or_error.clear(); + + if (client->state == Client::SENDING_ERROR) { + // 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. + client->state = Client::SENDING_DATA; + client->bytes_sent = client->stream->data_size; + client->stream->put_client_to_sleep(client); + return; + } + case Client::SENDING_DATA: { + // 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. + Stream *stream = client->stream; + size_t bytes_to_send = stream->data_size - client->bytes_sent; + if (bytes_to_send == 0) { + return; + } + if (bytes_to_send > BACKLOG_SIZE) { + fprintf(stderr, "WARNING: fd %d lost %lld bytes, maybe too slow connection\n", + client->sock, + (long long int)(bytes_to_send - BACKLOG_SIZE)); + client->bytes_sent = stream->data_size - BACKLOG_SIZE; + bytes_to_send = BACKLOG_SIZE; + } + + // See if we need to split across the circular buffer. + ssize_t ret; + if ((client->bytes_sent % BACKLOG_SIZE) + bytes_to_send > BACKLOG_SIZE) { + size_t bytes_first_part = BACKLOG_SIZE - (client->bytes_sent % BACKLOG_SIZE); + + iovec iov[2]; + iov[0].iov_base = const_cast(stream->data + (client->bytes_sent % BACKLOG_SIZE)); + iov[0].iov_len = bytes_first_part; + + iov[1].iov_base = const_cast(stream->data); + iov[1].iov_len = bytes_to_send - bytes_first_part; + + do { + ret = writev(client->sock, iov, 2); + } while (ret == -1 && errno == EINTR); + } else { + do { + ret = write(client->sock, + stream->data + (client->bytes_sent % BACKLOG_SIZE), + bytes_to_send); + } while (ret == -1 && errno == EINTR); + } + if (ret == -1 && errno == EAGAIN) { + // We're out of socket space, so return; epoll will wake us up + // when there is more room. + // This is postcondition #4. + return; + } + if (ret == -1) { + // Error, close; postcondition #1. + perror("write/writev"); + close_client(client); + return; + } + client->bytes_sent += ret; + + if (client->bytes_sent == stream->data_size) { + // 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 { + // XXX: Do we need to go another round here to explicitly + // get the EAGAIN? + } + break; + } + default: + assert(false); + } +} + +int Server::parse_request(Client *client) +{ + vector lines = split_lines(client->request); + if (lines.empty()) { + return 400; // Bad request (empty). + } + + vector request_tokens = split_tokens(lines[0]); + if (request_tokens.size() < 2) { + 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. + } + + client->stream_id = request_tokens[1]; + client->stream = find_stream(client->stream_id); + client->request.clear(); + + return 200; // OK! +} + +void Server::construct_header(Client *client) +{ + client->header_or_error = "HTTP/1.0 200 OK\r\nContent-type: video/x-flv\r\nCache-Control: no-cache\r\n\r\n" + + find_stream(client->stream_id)->header; + + // Switch states. + client->state = Client::SENDING_HEADER; + + epoll_event ev; + ev.events = EPOLLOUT | EPOLLET | EPOLLRDHUP; + ev.data.u64 = 0; // Keep Valgrind happy. + ev.data.fd = client->sock; + + if (epoll_ctl(epoll_fd, EPOLL_CTL_MOD, client->sock, &ev) == -1) { + perror("epoll_ctl(EPOLL_CTL_MOD)"); + exit(1); + } +} + +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; + + // Switch states. + client->state = Client::SENDING_ERROR; + + epoll_event ev; + ev.events = EPOLLOUT | EPOLLET | EPOLLRDHUP; + ev.data.u64 = 0; // Keep Valgrind happy. + ev.data.fd = client->sock; + + if (epoll_ctl(epoll_fd, EPOLL_CTL_MOD, client->sock, &ev) == -1) { + perror("epoll_ctl(EPOLL_CTL_MOD)"); + exit(1); + } +} + +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 Server::close_client(Client *client) +{ + if (epoll_ctl(epoll_fd, EPOLL_CTL_DEL, client->sock, NULL) == -1) { + 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) { + delete_from(&client->stream->sleeping_clients, client); + delete_from(&client->stream->to_process, client); + } + + // Bye-bye! + int ret; + do { + ret = close(client->sock); + } while (ret == -1 && errno == EINTR); + + if (ret == -1) { + perror("close"); + } + + clients.erase(client->sock); +} + +Stream *Server::find_stream(const string &stream_id) +{ + map::iterator it = streams.find(stream_id); + assert(it != streams.end()); + return it->second; +} + +void Server::process_queued_data() +{ + MutexLock lock(&queued_data_mutex); + + for (size_t i = 0; i < queued_add_clients.size(); ++i) { + add_client(queued_add_clients[i]); + } + queued_add_clients.clear(); + + for (map::iterator queued_it = queued_data.begin(); + queued_it != queued_data.end(); + ++queued_it) { + add_data(queued_it->first, queued_it->second.data(), queued_it->second.size()); + } + queued_data.clear(); }