Move the logic of load-balancing etc. into ServerPool, which frees external users...
authorSteinar H. Gunderson <sgunderson@bigfoot.com>
Sun, 7 Apr 2013 14:01:03 +0000 (16:01 +0200)
committerSteinar H. Gunderson <sgunderson@bigfoot.com>
Sun, 7 Apr 2013 14:01:03 +0000 (16:01 +0200)
Makefile
cubemap.cpp
input.cpp
server.h
serverpool.cpp [new file with mode: 0644]
serverpool.h [new file with mode: 0644]

index 257966a..2db0a75 100644 (file)
--- 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
 
index 8dff03b..4f03fea 100644 (file)
@@ -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<void *>(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);
index 541b5df..6a53c22 100644 (file)
--- a/input.cpp
+++ b/input.cpp
 #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);
        }
 }
 
index 7d94e17..2cb078e 100644 (file)
--- a/server.h
+++ b/server.h
@@ -3,10 +3,11 @@
 
 #include <stdint.h>
 #include <pthread.h>
+#include <sys/epoll.h>
 #include <string>
 #include <map>
+#include <vector>
 
-#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 (file)
index 0000000..f8fece6
--- /dev/null
@@ -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 (file)
index 0000000..74d6cb9
--- /dev/null
@@ -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)