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