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
#include <map>
#include <set>
+#include "markpool.h"
#include "metacube.h"
#include "parse.h"
#include "server.h"
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)
{
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;
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();
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(
// 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;
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.
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());
--- /dev/null
+#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);
+}
--- /dev/null
+#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)
#include <map>
#include <algorithm>
+#include "markpool.h"
#include "metacube.h"
#include "server.h"
#include "mutexlock.h"
Client::Client(int sock)
: sock(sock),
+ fwmark(0),
connect_time(time(NULL)),
state(Client::READING_REQUEST),
stream(NULL),
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
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);
}
: 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);
}
}
}
+
+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)
{
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!
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 {
class ClientProto;
class CubemapStateProto;
+class MarkPool;
class Stream;
class StreamProto;
// 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;
// <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.
// 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.
}
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);
+ }
+}
#include <vector>
+class MarkPool;
+
// Provides services such as load-balancing between a number of Server instances.
class ServerPool {
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();
projects / cubemap / commitdiff