From: Steinar H. Gunderson Date: Wed, 10 Apr 2013 19:10:00 +0000 (+0200) Subject: Add support for setting fwmarks per-connection. Useful for shaping individual streams... X-Git-Tag: 1.0.0~162 X-Git-Url: https://git.sesse.net/?p=cubemap;a=commitdiff_plain;h=462cb16514a2b691ae4d4b76125b42fda83df712 Add support for setting fwmarks per-connection. Useful for shaping individual streams (for doing TCP pacing). --- diff --git a/Makefile b/Makefile index 8b47a9f..a9333f0 100644 --- a/Makefile +++ b/Makefile @@ -4,7 +4,7 @@ PROTOC=protoc CXXFLAGS=-Wall -O2 -g LDLIBS=-lpthread -lprotobuf -OBJS=main.o server.o serverpool.o mutexlock.o input.o parse.o state.pb.o +OBJS=main.o server.o serverpool.o mutexlock.o input.o parse.o markpool.o state.pb.o all: cubemap diff --git a/main.cpp b/main.cpp index 1773af3..1300508 100644 --- a/main.cpp +++ b/main.cpp @@ -18,6 +18,7 @@ #include #include +#include "markpool.h" #include "metacube.h" #include "parse.h" #include "server.h" @@ -254,6 +255,33 @@ CubemapStateProto read_tempfile(int state_fd) return state; } + +// Reuse mark pools if one already exists. +MarkPool *get_mark_pool(map, MarkPool *> *mark_pools, int from, int to) +{ + pair mark_range(from, to); + if (mark_pools->count(mark_range) != 0) { + return (*mark_pools)[mark_range]; + } + + // Check if we're overlapping some other mark pool. + for (map, MarkPool *>::const_iterator mp_it = mark_pools->begin(); + mp_it != mark_pools->end(); + ++mp_it) { + int other_from = mp_it->first.first; + int other_to = mp_it->first.second; + if ((from >= other_from && from < other_to) || + (to >= other_from && to < other_to)) { + fprintf(stderr, "WARNING: Mark pool %d-%d partially overlaps with %d-%d, you may get duplicate marks.\n", + from, to, other_from, other_to); + fprintf(stderr, " Mark pools must either be completely disjunct, or completely overlapping.\n"); + } + } + + MarkPool *mark_pool = new MarkPool(from, to); + mark_pools->insert(make_pair(mark_range, mark_pool)); + return mark_pool; +} int main(int argc, char **argv) { @@ -270,6 +298,7 @@ int main(int argc, char **argv) servers = new ServerPool(num_servers); + CubemapStateProto loaded_state; int server_sock = -1, old_port = -1; set deserialized_stream_ids; map deserialized_inputs; @@ -278,7 +307,7 @@ int main(int argc, char **argv) fprintf(stderr, "Deserializing state from previous process... "); int state_fd = atoi(argv[3]); - CubemapStateProto loaded_state = read_tempfile(state_fd); + loaded_state = read_tempfile(state_fd); serialize_start.tv_sec = loaded_state.serialize_start_sec(); serialize_start.tv_usec = loaded_state.serialize_start_usec(); @@ -289,12 +318,6 @@ int main(int argc, char **argv) deserialized_stream_ids.insert(loaded_state.streams(i).stream_id()); } - // Put back the existing clients. It doesn't matter which server we - // allocate them to, so just do round-robin. - for (int i = 0; i < loaded_state.clients_size(); ++i) { - servers->add_client_from_serialized(loaded_state.clients(i)); - } - // Deserialize the inputs. Note that we don't actually add them to any state yet. for (int i = 0; i < loaded_state.inputs_size(); ++i) { deserialized_inputs.insert(make_pair( @@ -311,6 +334,7 @@ int main(int argc, char **argv) // Find all streams in the configuration file, and create them. set expecting_stream_ids = deserialized_stream_ids; + map, MarkPool *> mark_pools; for (unsigned i = 0; i < config.size(); ++i) { if (config[i].keyword != "stream") { continue; @@ -324,6 +348,31 @@ int main(int argc, char **argv) servers->add_stream(stream_id); } expecting_stream_ids.erase(stream_id); + + // Set up marks, if so desired. + if (config[i].parameters.count("mark")) { + string mark_str = config[i].parameters["mark"]; + size_t split = mark_str.find_first_of('-'); + if (split == string::npos) { + fprintf(stderr, "WARNING: Invalid mark specification '%s' (expected 'X-Y'), ignoring.\n", + mark_str.c_str()); + continue; + } + + string from_str(mark_str.begin(), mark_str.begin() + split); + string to_str(mark_str.begin() + split + 1, mark_str.end()); + int from = atoi(from_str.c_str()); + int to = atoi(to_str.c_str()); + + if (from <= 0 || from >= 65536 || to <= 0 || to >= 65536) { + fprintf(stderr, "WARNING: Mark pool range %d-%d is outside legal range [1,65536>, ignoring.\n", + from, to); + continue; + } + + MarkPool *mark_pool = get_mark_pool(&mark_pools, from, to); + servers->set_mark_pool(stream_id, mark_pool); + } } // Warn about any servers we've lost. @@ -396,6 +445,15 @@ int main(int argc, char **argv) input->run(); inputs.push_back(input); } + + if (is_reexec) { + // Put back the existing clients. It doesn't matter which server we + // allocate them to, so just do round-robin. However, we need to add + // them after the mark pools have been set up. + for (int i = 0; i < loaded_state.clients_size(); ++i) { + servers->add_client_from_serialized(loaded_state.clients(i)); + } + } // All deserialized inputs should now have been taken care of, one way or the other. assert(deserialized_inputs.empty()); diff --git a/markpool.cpp b/markpool.cpp new file mode 100644 index 0000000..196952e --- /dev/null +++ b/markpool.cpp @@ -0,0 +1,43 @@ +#include "markpool.h" +#include "mutexlock.h" +#include +#include +#include +#include + +MarkPool::MarkPool(int start, int end) + : start(start), end(end) +{ + assert(start > 0 && start < 65536); + assert(end > 0 && end < 65536); + + for (int i = start; i < end; ++i) { + free_marks.push(i); + } + + pthread_mutex_init(&mutex, NULL); +} + +int MarkPool::get_mark() +{ + MutexLock lock(&mutex); + if (free_marks.empty()) { + fprintf(stderr, "WARNING: Out of free marks in mark pool %d-%d, session will not be marked.\n", + start, end); + fprintf(stderr, " To fix, increase the pool size and HUP the server.\n"); + return 0; + } + int mark = free_marks.front(); + free_marks.pop(); + return mark; +} + +void MarkPool::release_mark(int mark) +{ + if (mark == 0) { + return; + } + + MutexLock lock(&mutex); + free_marks.push(mark); +} diff --git a/markpool.h b/markpool.h new file mode 100644 index 0000000..f90aeea --- /dev/null +++ b/markpool.h @@ -0,0 +1,25 @@ +#ifndef _MARKPOOL_H +#define _MARKPOOL_H + +// A class that hands out fwmarks from a given range in a thread-safe fashion. +// If the range is empty, it returns 0. + +#include +#include + +class MarkPool { +public: + // Limits are [start, end>. Numbers are 16-bit, so above 65535 do not make sense. + MarkPool(int start, int end); + + int get_mark(); + void release_mark(int mark); + +private: + int start, end; + + pthread_mutex_t mutex; + std::queue free_marks; +}; + +#endif // !defined(_MARKPOOL_H) diff --git a/server.cpp b/server.cpp index 6e1005e..aed8779 100644 --- a/server.cpp +++ b/server.cpp @@ -16,6 +16,7 @@ #include #include +#include "markpool.h" #include "metacube.h" #include "server.h" #include "mutexlock.h" @@ -26,6 +27,7 @@ using namespace std; Client::Client(int sock) : sock(sock), + fwmark(0), connect_time(time(NULL)), state(Client::READING_REQUEST), stream(NULL), @@ -64,6 +66,16 @@ Client::Client(const ClientProto &serialized, Stream *stream) header_or_error_bytes_sent(serialized.header_or_error_bytes_sent()), bytes_sent(serialized.bytes_sent()) { + if (stream->mark_pool != NULL) { + fwmark = stream->mark_pool->get_mark(); + } else { + fwmark = 0; // No mark. + } + if (setsockopt(sock, SOL_SOCKET, SO_MARK, &fwmark, sizeof(fwmark)) == -1) { + if (fwmark != 0) { + perror("setsockopt(SO_MARK)"); + } + } } ClientProto Client::serialize() const @@ -94,7 +106,8 @@ ClientStats Client::get_stats() const Stream::Stream(const string &stream_id) : stream_id(stream_id), data(new char[BACKLOG_SIZE]), - data_size(0) + data_size(0), + mark_pool(NULL) { memset(data, 0, BACKLOG_SIZE); } @@ -108,7 +121,8 @@ Stream::Stream(const StreamProto &serialized) : stream_id(serialized.stream_id()), header(serialized.header()), data(new char[BACKLOG_SIZE]), - data_size(serialized.data_size()) + data_size(serialized.data_size()), + mark_pool(NULL) { assert(serialized.data().size() == BACKLOG_SIZE); memcpy(data, serialized.data().data(), BACKLOG_SIZE); @@ -358,6 +372,13 @@ void Server::set_header(const string &stream_id, const string &header) } } } + +void Server::set_mark_pool(const std::string &stream_id, MarkPool *mark_pool) +{ + MutexLock lock(&mutex); + assert(clients.empty()); + find_stream(stream_id)->mark_pool = mark_pool; +} void Server::add_data_deferred(const string &stream_id, const char *data, size_t bytes) { @@ -582,6 +603,16 @@ int Server::parse_request(Client *client) 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(); + } else { + client->fwmark = 0; // No mark. + } + if (setsockopt(client->sock, SOL_SOCKET, SO_MARK, &client->fwmark, sizeof(client->fwmark)) == -1) { + if (client->fwmark != 0) { + perror("setsockopt(SO_MARK)"); + } + } client->request.clear(); return 200; // OK! @@ -644,8 +675,12 @@ void Server::close_client(Client *client) if (client->stream != NULL) { 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); + } } - + // Bye-bye! int ret; do { diff --git a/server.h b/server.h index 300ec0d..7979493 100644 --- a/server.h +++ b/server.h @@ -16,6 +16,7 @@ class ClientProto; class CubemapStateProto; +class MarkPool; class Stream; class StreamProto; @@ -40,6 +41,9 @@ struct Client { // The file descriptor associated with this socket. int sock; + // The fwmark associated with this socket (or 0). + int fwmark; + // Some information only used for logging. std::string remote_addr; time_t connect_time; @@ -98,6 +102,9 @@ struct Stream { // ). std::vector to_process; + // What pool to fetch marks from, or NULL. + MarkPool *mark_pool; + // Put client to sleep, since there is no more data for it; we will on // longer listen on POLLOUT until we get more data. Also, it will be put // in the list of clients to wake up when we do. @@ -127,6 +134,10 @@ public: // Set header (both HTTP header and any stream headers) for the given stream. void set_header(const std::string &stream_id, const std::string &header); + // Set that the given stream should use the given mark pool from now on. + // NOTE: This should be set before any clients are connected! + void set_mark_pool(const std::string &stream_id, MarkPool *mark_pool); + // These will be deferred until the next time an iteration in do_work() happens, // and the order between them are undefined. // XXX: header should ideally be ordered with respect to data. diff --git a/serverpool.cpp b/serverpool.cpp index dc7506d..09fe435 100644 --- a/serverpool.cpp +++ b/serverpool.cpp @@ -68,3 +68,10 @@ vector ServerPool::get_client_stats() const } return ret; } + +void ServerPool::set_mark_pool(const std::string &stream_id, MarkPool *mark_pool) +{ + for (int i = 0; i < num_servers; ++i) { + servers[i].set_mark_pool(stream_id, mark_pool); + } +} diff --git a/serverpool.h b/serverpool.h index 3083986..c22fb67 100644 --- a/serverpool.h +++ b/serverpool.h @@ -5,6 +5,8 @@ #include +class MarkPool; + // Provides services such as load-balancing between a number of Server instances. class ServerPool { public: @@ -27,6 +29,9 @@ public: void set_header(const std::string &stream_id, const std::string &header); void add_data(const std::string &stream_id, const char *data, size_t bytes); + // Connects the given stream to the given mark pool for all the servers. + void set_mark_pool(const std::string &stream_id, MarkPool *mark_pool); + // Starts all the servers. void run();