Add support for setting fwmarks per-connection. Useful for shaping individual streams...
authorSteinar H. Gunderson <sgunderson@bigfoot.com>
Wed, 10 Apr 2013 19:10:00 +0000 (21:10 +0200)
committerSteinar H. Gunderson <sgunderson@bigfoot.com>
Wed, 10 Apr 2013 19:10:00 +0000 (21:10 +0200)
Makefile
main.cpp
markpool.cpp [new file with mode: 0644]
markpool.h [new file with mode: 0644]
server.cpp
server.h
serverpool.cpp
serverpool.h

index 8b47a9f..a9333f0 100644 (file)
--- 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
 
index 1773af3..1300508 100644 (file)
--- a/main.cpp
+++ b/main.cpp
@@ -18,6 +18,7 @@
 #include <map>
 #include <set>
 
+#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<pair<int, int>, MarkPool *> *mark_pools, int from, int to)
+{
+       pair<int, int> 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<pair<int, int>, 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<string> deserialized_stream_ids;
        map<string, Input *> 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<string> expecting_stream_ids = deserialized_stream_ids;
+       map<pair<int, int>, 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 (file)
index 0000000..196952e
--- /dev/null
@@ -0,0 +1,43 @@
+#include "markpool.h"
+#include "mutexlock.h"
+#include <stdio.h>
+#include <assert.h>
+#include <pthread.h>
+#include <queue>
+
+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 (file)
index 0000000..f90aeea
--- /dev/null
@@ -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 <pthread.h>
+#include <queue>
+
+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<int> free_marks;
+};
+
+#endif  // !defined(_MARKPOOL_H)
index 6e1005e..aed8779 100644 (file)
@@ -16,6 +16,7 @@
 #include <map>
 #include <algorithm>
 
+#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 {
index 300ec0d..7979493 100644 (file)
--- 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 {
        // <sleeping_clients>).
        std::vector<Client *> 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.
index dc7506d..09fe435 100644 (file)
@@ -68,3 +68,10 @@ vector<ClientStats> 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);
+       }       
+}
index 3083986..c22fb67 100644 (file)
@@ -5,6 +5,8 @@
 
 #include <vector>
 
+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();