git.sesse.net Git - cubemap/blob - serverpool.cpp

   1 #include <assert.h>
   2 #include <stdlib.h>
   3
   4 #include "client.h"
   5 #include "server.h"
   6 #include "serverpool.h"
   7 #include "state.pb.h"
   8 #include "util.h"
   9
  10 using namespace std;
  11
  12 ServerPool::ServerPool(int size)
  13         : servers(new Server[size]),
  14           num_servers(size),
  15           clients_added(0)
  16 {
  17 }
  18
  19 ServerPool::~ServerPool()
  20 {
  21         delete[] servers;
  22 }
  23
  24 CubemapStateProto ServerPool::serialize()
  25 {
  26         CubemapStateProto state;
  27
  28         for (int i = 0; i < num_servers; ++i) {
  29                 CubemapStateProto local_state = servers[i].serialize();
  30
  31                 // The stream state should be identical between the servers, so we only store it once,
  32                 // save for the fds, which we keep around to distribute to the servers after re-exec.
  33                 if (i == 0) {
  34                         state.mutable_streams()->MergeFrom(local_state.streams());
  35                 } else {
  36                         assert(state.streams_size() == local_state.streams_size());
  37                         for (int j = 0; j < local_state.streams_size(); ++j) {
  38                                 assert(local_state.streams(j).data_fds_size() == 1);
  39                                 state.mutable_streams(j)->add_data_fds(local_state.streams(j).data_fds(0));
  40                         }
  41                 }
  42                 for (int j = 0; j < local_state.clients_size(); ++j) {
  43                         state.add_clients()->MergeFrom(local_state.clients(j));
  44                 }
  45         }
  46
  47         return state;
  48 }
  49
  50 void ServerPool::add_client(int sock)
  51 {
  52         servers[clients_added++ % num_servers].add_client_deferred(sock);
  53 }
  54
  55 void ServerPool::add_client_from_serialized(const ClientProto &client)
  56 {
  57         servers[clients_added++ % num_servers].add_client_from_serialized(client);
  58 }
  59
  60 int ServerPool::lookup_stream_by_url(const std::string &url) const
  61 {
  62         assert(servers != NULL);
  63         return servers[0].lookup_stream_by_url(url);
  64 }
  65
  66 int ServerPool::add_stream(const string &url, size_t backlog_size, Stream::Encoding encoding)
  67 {
  68         int stream_index = -1;
  69         for (int i = 0; i < num_servers; ++i) {
  70                 int stream_index2 = servers[i].add_stream(url, backlog_size, encoding);
  71                 if (i == 0) {
  72                         stream_index = stream_index2;
  73                 } else {
  74                         // Verify that all servers have this under the same stream index.
  75                         assert(stream_index == stream_index2);
  76                 }
  77         }
  78         return stream_index;
  79 }
  80
  81 int ServerPool::add_stream_from_serialized(const StreamProto &stream, const vector<int> &data_fds)
  82 {
  83         assert(!data_fds.empty());
  84         string contents;
  85         int stream_index = -1;
  86         for (int i = 0; i < num_servers; ++i) {
  87                 int data_fd;
  88                 if (i < int(data_fds.size())) {
  89                         // Reuse one of the existing file descriptors.
  90                         data_fd = data_fds[i];
  91                 } else {
  92                         // Clone the first one.
  93                         if (contents.empty()) {
  94                                 if (!read_tempfile(data_fds[0], &contents)) {
  95                                         exit(1);
  96                                 }
  97                         }
  98                         data_fd = make_tempfile(contents);
  99                 }
 100
 101                 int stream_index2 = servers[i].add_stream_from_serialized(stream, data_fd);
 102                 if (i == 0) {
 103                         stream_index = stream_index2;
 104                 } else {
 105                         // Verify that all servers have this under the same stream index.
 106                         assert(stream_index == stream_index2);
 107                 }
 108         }
 109
 110         // Close and delete any leftovers, if the number of servers was reduced.
 111         for (size_t i = num_servers; i < data_fds.size(); ++i) {
 112                 safe_close(data_fds[i]);  // Implicitly deletes the file.
 113         }
 114
 115         return stream_index;
 116 }
 117
 118 void ServerPool::set_header(int stream_index, const string &http_header, const string &stream_header)
 119 {
 120         for (int i = 0; i < num_servers; ++i) {
 121                 servers[i].set_header(stream_index, http_header, stream_header);
 122         }
 123 }
 124
 125 void ServerPool::add_data(int stream_index, const char *data, size_t bytes)
 126 {
 127         for (int i = 0; i < num_servers; ++i) {
 128                 servers[i].add_data_deferred(stream_index, data, bytes);
 129         }
 130 }
 131
 132 void ServerPool::run()
 133 {
 134         for (int i = 0; i < num_servers; ++i) {
 135                 servers[i].run();
 136         }
 137 }
 138
 139 void ServerPool::stop()
 140 {
 141         for (int i = 0; i < num_servers; ++i) {
 142                 servers[i].stop();
 143         }
 144 }
 145
 146 vector<ClientStats> ServerPool::get_client_stats() const
 147 {
 148         vector<ClientStats> ret;
 149         for (int i = 0; i < num_servers; ++i) {
 150                 vector<ClientStats> stats = servers[i].get_client_stats();
 151                 ret.insert(ret.end(), stats.begin(), stats.end());
 152         }
 153         return ret;
 154 }
 155
 156 void ServerPool::set_mark_pool(int stream_index, MarkPool *mark_pool)
 157 {
 158         for (int i = 0; i < num_servers; ++i) {
 159                 servers[i].set_mark_pool(stream_index, mark_pool);
 160         }
 161 }
 162
 163 void ServerPool::set_backlog_size(int stream_index, size_t new_size)
 164 {
 165         for (int i = 0; i < num_servers; ++i) {
 166                 servers[i].set_backlog_size(stream_index, new_size);
 167         }
 168 }
 169
 170 void ServerPool::set_encoding(int stream_index, Stream::Encoding encoding)
 171 {
 172         for (int i = 0; i < num_servers; ++i) {
 173                 servers[i].set_encoding(stream_index, encoding);
 174         }
 175 }