From: Steinar H. Gunderson Date: Sun, 7 Apr 2013 14:01:03 +0000 (+0200) Subject: Move the logic of load-balancing etc. into ServerPool, which frees external users... X-Git-Tag: 1.0.0~193 X-Git-Url: https://git.sesse.net/?p=cubemap;a=commitdiff_plain;h=b59fa7ce2d47f135ea027548cc89f937a5fa875b Move the logic of load-balancing etc. into ServerPool, which frees external users from knowing anything about NUM_SERVERS. --- diff --git a/Makefile b/Makefile index 257966a..2db0a75 100644 --- a/Makefile +++ b/Makefile @@ -4,7 +4,7 @@ PROTOC=protoc CXXFLAGS=-Wall -O2 -g LDLIBS=-lcurl -lpthread -lprotobuf -OBJS=cubemap.o server.o mutexlock.o input.o state.pb.o +OBJS=cubemap.o server.o serverpool.o mutexlock.o input.o state.pb.o all: cubemap diff --git a/cubemap.cpp b/cubemap.cpp index 8dff03b..4f03fea 100644 --- a/cubemap.cpp +++ b/cubemap.cpp @@ -18,6 +18,7 @@ #include "metacube.h" #include "server.h" +#include "serverpool.h" #include "input.h" #include "state.pb.h" @@ -27,7 +28,7 @@ using namespace std; -Server *servers = NULL; +ServerPool *servers = NULL; volatile bool hupped = false; void hup(int ignored) @@ -100,7 +101,7 @@ void *acceptor_thread_run(void *arg) } // Pick a server, round-robin, and hand over the socket to it. - servers[num_accepted % NUM_SERVERS].add_client(sock); + servers->add_client(sock); ++num_accepted; } } @@ -279,8 +280,7 @@ int main(int argc, char **argv) exit(1); } - // Create the servers. - servers = new Server[NUM_SERVERS]; + servers = new ServerPool(NUM_SERVERS); int server_sock = -1, old_port = -1; if (argc == 4 && strcmp(argv[2], "-state") == 0) { @@ -290,15 +290,13 @@ int main(int argc, char **argv) // Deserialize the streams. for (int i = 0; i < loaded_state.streams_size(); ++i) { - for (int j = 0; j < NUM_SERVERS; ++j) { - servers[j].add_stream_from_serialized(loaded_state.streams(i)); - } + servers->add_stream_from_serialized(loaded_state.streams(i)); } // 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[i % NUM_SERVERS].add_client_from_serialized(loaded_state.clients(i)); + servers->add_client_from_serialized(loaded_state.clients(i)); } // Deserialize the server socket. @@ -308,9 +306,7 @@ int main(int argc, char **argv) fprintf(stderr, "done.\n"); } else{ // TODO: This should come from the config file. - for (int i = 0; i < NUM_SERVERS; ++i) { - servers[i].add_stream(STREAM_ID); - } + servers->add_stream(STREAM_ID); } // Open a new server socket if we do not already have one, or if we changed ports. @@ -323,10 +319,7 @@ int main(int argc, char **argv) server_sock = create_server_socket(port); } - // Start up all the servers! - for (int i = 0; i < NUM_SERVERS; ++i) { - servers[i].run(); - } + servers->run(); pthread_t acceptor_thread; pthread_create(&acceptor_thread, NULL, acceptor_thread_run, reinterpret_cast(server_sock)); @@ -346,10 +339,10 @@ int main(int argc, char **argv) CubemapStateProto state; state.set_server_sock(server_sock); state.set_port(port); - for (int i = 0; i < NUM_SERVERS; ++i) { - servers[i].stop(); + for (int i = 0; i < NUM_SERVERS; ++i) { + servers->get_server(i)->stop(); - CubemapStateProto local_state = servers[i].serialize(); + CubemapStateProto local_state = servers->get_server(i)->serialize(); // The stream state should be identical between the servers, so we only store it once. if (i == 0) { @@ -359,7 +352,7 @@ int main(int argc, char **argv) state.add_clients()->MergeFrom(local_state.clients(j)); } } - delete[] servers; + delete servers; fprintf(stderr, "Serializing state and re-execing...\n"); int state_fd = make_tempfile(state); diff --git a/input.cpp b/input.cpp index 541b5df..6a53c22 100644 --- a/input.cpp +++ b/input.cpp @@ -18,10 +18,11 @@ #include "mutexlock.h" #include "input.h" #include "server.h" +#include "serverpool.h" using namespace std; -extern Server *servers; +extern ServerPool *servers; Input::Input(const string &stream_id, const string &url) : stream_id(stream_id), @@ -146,13 +147,9 @@ void Input::process_block(const char *data, uint32_t size, uint32_t flags) { if (flags & METACUBE_FLAGS_HEADER) { string header(data, data + size); - for (int i = 0; i < NUM_SERVERS; ++i) { - servers[i].set_header(stream_id, header); - } + servers->set_header(stream_id, header); } else { - for (int i = 0; i < NUM_SERVERS; ++i) { - servers[i].add_data(stream_id, data, size); - } + servers->add_data(stream_id, data, size); } } diff --git a/server.h b/server.h index 7d94e17..2cb078e 100644 --- a/server.h +++ b/server.h @@ -3,10 +3,11 @@ #include #include +#include #include #include +#include -#define NUM_SERVERS 4 #define BACKLOG_SIZE 1048576 #define EPOLL_MAX_EVENTS 8192 #define EPOLL_TIMEOUT_MS 20 diff --git a/serverpool.cpp b/serverpool.cpp new file mode 100644 index 0000000..f8fece6 --- /dev/null +++ b/serverpool.cpp @@ -0,0 +1,60 @@ +#include "serverpool.h" + +using namespace std; + +ServerPool::ServerPool(int size) + : servers(new Server[size]), + num_servers(size), + clients_added(0) +{ +} + +ServerPool::~ServerPool() +{ + delete[] servers; +} + +void ServerPool::add_client(int sock) +{ + servers[clients_added++ % num_servers].add_client(sock); +} + +void ServerPool::add_client_from_serialized(const ClientProto &client) +{ + servers[clients_added++ % num_servers].add_client_from_serialized(client); +} + +void ServerPool::add_stream(const std::string &stream_id) +{ + for (int i = 0; i < num_servers; ++i) { + servers[i].add_stream(stream_id); + } +} + +void ServerPool::add_stream_from_serialized(const StreamProto &stream) +{ + for (int i = 0; i < num_servers; ++i) { + servers[i].add_stream_from_serialized(stream); + } +} + +void ServerPool::set_header(const std::string &stream_id, const std::string &header) +{ + for (int i = 0; i < num_servers; ++i) { + servers[i].set_header(stream_id, header); + } +} + +void ServerPool::add_data(const std::string &stream_id, const char *data, size_t bytes) +{ + for (int i = 0; i < num_servers; ++i) { + servers[i].add_data(stream_id, data, bytes); + } +} + +void ServerPool::run() +{ + for (int i = 0; i < num_servers; ++i) { + servers[i].run(); + } +} diff --git a/serverpool.h b/serverpool.h new file mode 100644 index 0000000..74d6cb9 --- /dev/null +++ b/serverpool.h @@ -0,0 +1,38 @@ +#ifndef _SERVERPOOL_H +#define _SERVERPOOL_H 1 + +#include "server.h" + +// Provides services such as load-balancing between a number of Server instances. +class ServerPool { +public: + ServerPool(int num_servers); + ~ServerPool(); + + // Accessor. Only to be used in rare situations, really. + // The ServerPool retains ownership. + Server *get_server(int num) { return &servers[num]; } + + // Picks a server (round-robin) and allocates the given client to it. + void add_client(int sock); + void add_client_from_serialized(const ClientProto &client); + + // Adds the given stream to all the servers. + void add_stream(const std::string &stream_id); + void add_stream_from_serialized(const StreamProto &stream); + + // Adds the given data to all the servers. + 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); + + // Starts all the servers. + void run(); + +private: + Server *servers; + int num_servers, clients_added; + + ServerPool(const ServerPool &); +}; + +#endif // !defined(_SERVERPOOL_H)