From c5cdeee3a4c99abbf7303b47634b0a748e50bcb4 Mon Sep 17 00:00:00 2001 From: "Steinar H. Gunderson" Date: Sat, 13 Apr 2013 19:04:27 +0200 Subject: [PATCH] Send data using sendfile() instead of write(). Seems to be something like 30% lower CPU usage. --- server.cpp | 105 +++++++++++++++++++++++++++++++++++------------------ server.h | 15 ++++++-- 2 files changed, 80 insertions(+), 40 deletions(-) diff --git a/server.cpp b/server.cpp index 8881912..c9b942f 100644 --- a/server.cpp +++ b/server.cpp @@ -9,6 +9,7 @@ #include #include #include +#include #include #include #include @@ -22,6 +23,7 @@ #include "server.h" #include "mutexlock.h" #include "parse.h" +#include "util.h" #include "state.pb.h" using namespace std; @@ -106,36 +108,50 @@ ClientStats Client::get_stats() const Stream::Stream(const string &stream_id) : stream_id(stream_id), - data(new char[BACKLOG_SIZE]), + data_fd(make_tempfile("")), data_size(0), mark_pool(NULL) { - memset(data, 0, BACKLOG_SIZE); + if (data_fd == -1) { + exit(1); + } } Stream::~Stream() { - delete[] data; + if (data_fd != -1) { + int ret; + do { + ret = close(data_fd); + } while (ret == -1 && errno == EINTR); + if (ret == -1) { + perror("close"); + } + } } Stream::Stream(const StreamProto &serialized) : stream_id(serialized.stream_id()), header(serialized.header()), - data(new char[BACKLOG_SIZE]), + data_fd(make_tempfile(serialized.data())), data_size(serialized.data_size()), mark_pool(NULL) { - assert(serialized.data().size() == BACKLOG_SIZE); - memcpy(data, serialized.data().data(), BACKLOG_SIZE); + if (data_fd == -1) { + exit(1); + } } -StreamProto Stream::serialize() const +StreamProto Stream::serialize() { StreamProto serialized; serialized.set_header(header); - serialized.set_data(string(data, data + BACKLOG_SIZE)); + if (!read_tempfile(data_fd, serialized.mutable_data())) { // Closes data_fd. + exit(1); + } serialized.set_data_size(data_size); serialized.set_stream_id(stream_id); + data_fd = -1; return serialized; } @@ -355,21 +371,47 @@ void Server::add_data_deferred(const string &stream_id, const char *data, size_t queued_data[stream_id].append(string(data, data + bytes)); } -void Server::add_data(const string &stream_id, const char *data, size_t bytes) +void Server::add_data(const string &stream_id, const char *data, ssize_t bytes) { 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; + ssize_t to_copy = BACKLOG_SIZE - pos; + while (to_copy > 0) { + int ret = pwrite(stream->data_fd, data, to_copy, pos); + if (ret == -1 && errno == EINTR) { + continue; + } + if (ret == -1) { + perror("pwrite"); + // Dazed and confused, but trying to continue... + break; + } + pos += ret; + data += ret; + to_copy -= ret; + bytes -= ret; + } pos = 0; } - memcpy(stream->data + pos, data, bytes); + while (bytes > 0) { + int ret = pwrite(stream->data_fd, data, bytes, pos); + if (ret == -1 && errno == EINTR) { + continue; + } + if (ret == -1) { + perror("pwrite"); + // Dazed and confused, but trying to continue... + break; + } + pos += ret; + data += ret; + bytes -= ret; + } + stream->wake_up_all_clients(); } @@ -486,6 +528,7 @@ sending_header_or_error_again: return; } 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. Stream *stream = client->stream; @@ -502,27 +545,18 @@ sending_header_or_error_again: } // See if we need to split across the circular buffer. - ssize_t ret; + bool more_data = false; 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; + bytes_to_send = BACKLOG_SIZE - (client->bytes_sent % BACKLOG_SIZE); + more_data = true; + } - iov[1].iov_base = const_cast(stream->data); - iov[1].iov_len = bytes_to_send - bytes_first_part; + ssize_t ret; + do { + loff_t offset = client->bytes_sent % BACKLOG_SIZE; + ret = sendfile(client->sock, stream->data_fd, &offset, bytes_to_send); + } while (ret == -1 && errno == EINTR); - 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. @@ -531,7 +565,7 @@ sending_header_or_error_again: } if (ret == -1) { // Error, close; postcondition #1. - perror("write/writev"); + perror("sendfile"); close_client(client); return; } @@ -541,9 +575,8 @@ sending_header_or_error_again: // 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? + } else if (more_data) { + goto sending_data_again; } break; } diff --git a/server.h b/server.h index 8d5261f..e9b7431 100644 --- a/server.h +++ b/server.h @@ -81,15 +81,22 @@ struct Stream { // Serialization/deserialization. Stream(const StreamProto &serialized); - StreamProto serialize() const; + StreamProto serialize(); std::string stream_id; // The HTTP response header, plus the video stream header (if any). std::string header; - // The stream data itself, stored in a circular buffer. - char *data; + // The stream data itself, stored in a circular buffer.q + // + // We store our data in a file, so that we can send the data to the + // kernel only once (with write()). We then use sendfile() for each + // client, which effectively zero-copies it out of the kernel's buffer + // cache. This is significantly more efficient than doing write() from + // a userspace memory buffer, since the latter makes the kernel copy + // the same data from userspace many times. + int data_fd; // How many bytes contains. Can very well be larger than BACKLOG_SIZE, // since the buffer wraps. @@ -217,7 +224,7 @@ private: void process_queued_data(); void add_client(int sock); - void add_data(const std::string &stream_id, const char *data, size_t bytes); + void add_data(const std::string &stream_id, const char *data, ssize_t bytes); }; #endif // !defined(_SERVER_H) -- 2.39.2