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

   1 #include <assert.h>
   2 #include <errno.h>
   3 #include <pthread.h>
   4 #include <stdio.h>
   5 #include <stdlib.h>
   6 #include <sys/epoll.h>
   7 #include <sys/sendfile.h>
   8 #include <sys/socket.h>
   9 #include <sys/types.h>
  10 #include <unistd.h>
  11 #include <algorithm>
  12 #include <map>
  13 #include <string>
  14 #include <utility>
  15 #include <vector>
  16
  17 #include "log.h"
  18 #include "markpool.h"
  19 #include "mutexlock.h"
  20 #include "parse.h"
  21 #include "server.h"
  22 #include "state.pb.h"
  23 #include "stream.h"
  24
  25 using namespace std;
  26
  27 Server::Server()
  28 {
  29         pthread_mutex_init(&mutex, NULL);
  30         pthread_mutex_init(&queued_data_mutex, NULL);
  31
  32         epoll_fd = epoll_create(1024);  // Size argument is ignored.
  33         if (epoll_fd == -1) {
  34                 log_perror("epoll_fd");
  35                 exit(1);
  36         }
  37 }
  38
  39 Server::~Server()
  40 {
  41         int ret;
  42         do {
  43                 ret = close(epoll_fd);
  44         } while (ret == -1 && errno == EINTR);
  45
  46         if (ret == -1) {
  47                 log_perror("close(epoll_fd)");
  48         }
  49 }
  50
  51 vector<ClientStats> Server::get_client_stats() const
  52 {
  53         vector<ClientStats> ret;
  54
  55         MutexLock lock(&mutex);
  56         for (map<int, Client>::const_iterator client_it = clients.begin();
  57              client_it != clients.end();
  58              ++client_it) {
  59                 ret.push_back(client_it->second.get_stats());
  60         }
  61         return ret;
  62 }
  63
  64 void Server::do_work()
  65 {
  66         for ( ;; ) {
  67                 int nfds = epoll_wait(epoll_fd, events, EPOLL_MAX_EVENTS, EPOLL_TIMEOUT_MS);
  68                 if (nfds == -1 && errno == EINTR) {
  69                         if (should_stop) {
  70                                 return;
  71                         }
  72                         continue;
  73                 }
  74                 if (nfds == -1) {
  75                         log_perror("epoll_wait");
  76                         exit(1);
  77                 }
  78
  79                 MutexLock lock(&mutex);  // We release the mutex between iterations.
  80
  81                 process_queued_data();
  82
  83                 for (int i = 0; i < nfds; ++i) {
  84                         Client *client = reinterpret_cast<Client *>(events[i].data.u64);
  85
  86                         if (events[i].events & (EPOLLERR | EPOLLRDHUP | EPOLLHUP)) {
  87                                 close_client(client);
  88                                 continue;
  89                         }
  90
  91                         process_client(client);
  92                 }
  93
  94                 for (map<string, Stream *>::iterator stream_it = streams.begin();
  95                      stream_it != streams.end();
  96                      ++stream_it) {
  97                         vector<Client *> to_process;
  98                         swap(stream_it->second->to_process, to_process);
  99                         for (size_t i = 0; i < to_process.size(); ++i) {
 100                                 process_client(to_process[i]);
 101                         }
 102                 }
 103
 104                 if (should_stop) {
 105                         return;
 106                 }
 107         }
 108 }
 109
 110 CubemapStateProto Server::serialize()
 111 {
 112         // We don't serialize anything queued, so empty the queues.
 113         process_queued_data();
 114
 115         CubemapStateProto serialized;
 116         for (map<int, Client>::const_iterator client_it = clients.begin();
 117              client_it != clients.end();
 118              ++client_it) {
 119                 serialized.add_clients()->MergeFrom(client_it->second.serialize());
 120         }
 121         for (map<string, Stream *>::const_iterator stream_it = streams.begin();
 122              stream_it != streams.end();
 123              ++stream_it) {
 124                 serialized.add_streams()->MergeFrom(stream_it->second->serialize());
 125         }
 126         return serialized;
 127 }
 128
 129 void Server::add_client_deferred(int sock)
 130 {
 131         MutexLock lock(&queued_data_mutex);
 132         queued_add_clients.push_back(sock);
 133 }
 134
 135 void Server::add_client(int sock)
 136 {
 137         clients.insert(make_pair(sock, Client(sock)));
 138
 139         // Start listening on data from this socket.
 140         epoll_event ev;
 141         ev.events = EPOLLIN | EPOLLET | EPOLLRDHUP;
 142         ev.data.u64 = reinterpret_cast<uint64_t>(&clients[sock]);
 143         if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, sock, &ev) == -1) {
 144                 log_perror("epoll_ctl(EPOLL_CTL_ADD)");
 145                 exit(1);
 146         }
 147
 148         process_client(&clients[sock]);
 149 }
 150
 151 void Server::add_client_from_serialized(const ClientProto &client)
 152 {
 153         MutexLock lock(&mutex);
 154         Stream *stream;
 155         map<string, Stream *>::iterator stream_it = streams.find(client.stream_id());
 156         if (stream_it == streams.end()) {
 157                 stream = NULL;
 158         } else {
 159                 stream = stream_it->second;
 160         }
 161         clients.insert(make_pair(client.sock(), Client(client, stream)));
 162         Client *client_ptr = &clients[client.sock()];
 163
 164         // Start listening on data from this socket.
 165         epoll_event ev;
 166         if (client.state() == Client::READING_REQUEST) {
 167                 ev.events = EPOLLIN | EPOLLET | EPOLLRDHUP;
 168         } else {
 169                 // If we don't have more data for this client, we'll be putting it into
 170                 // the sleeping array again soon.
 171                 ev.events = EPOLLOUT | EPOLLET | EPOLLRDHUP;
 172         }
 173         ev.data.u64 = 0;  // Keep Valgrind happy.
 174         ev.data.u64 = reinterpret_cast<uint64_t>(client_ptr);
 175         if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, client.sock(), &ev) == -1) {
 176                 log_perror("epoll_ctl(EPOLL_CTL_ADD)");
 177                 exit(1);
 178         }
 179
 180         if (client_ptr->state == Client::SENDING_DATA &&
 181             client_ptr->stream_pos == client_ptr->stream->bytes_received) {
 182                 client_ptr->stream->put_client_to_sleep(client_ptr);
 183         } else {
 184                 process_client(client_ptr);
 185         }
 186 }
 187
 188 void Server::add_stream(const string &stream_id, size_t backlog_size)
 189 {
 190         MutexLock lock(&mutex);
 191         streams.insert(make_pair(stream_id, new Stream(stream_id, backlog_size)));
 192 }
 193
 194 void Server::add_stream_from_serialized(const StreamProto &stream)
 195 {
 196         MutexLock lock(&mutex);
 197         streams.insert(make_pair(stream.stream_id(), new Stream(stream)));
 198 }
 199
 200 void Server::set_backlog_size(const std::string &stream_id, size_t new_size)
 201 {
 202         MutexLock lock(&mutex);
 203         assert(streams.count(stream_id) != 0);
 204         streams[stream_id]->set_backlog_size(new_size);
 205 }
 206
 207 void Server::set_header(const string &stream_id, const string &header)
 208 {
 209         MutexLock lock(&mutex);
 210         find_stream(stream_id)->header = header;
 211
 212         // If there are clients we haven't sent anything to yet, we should give
 213         // them the header, so push back into the SENDING_HEADER state.
 214         for (map<int, Client>::iterator client_it = clients.begin();
 215              client_it != clients.end();
 216              ++client_it) {
 217                 Client *client = &client_it->second;
 218                 if (client->state == Client::SENDING_DATA &&
 219                     client->stream_pos == 0) {
 220                         construct_header(client);
 221                 }
 222         }
 223 }
 224
 225 void Server::set_mark_pool(const std::string &stream_id, MarkPool *mark_pool)
 226 {
 227         MutexLock lock(&mutex);
 228         assert(clients.empty());
 229         find_stream(stream_id)->mark_pool = mark_pool;
 230 }
 231
 232 void Server::add_data_deferred(const string &stream_id, const char *data, size_t bytes)
 233 {
 234         MutexLock lock(&queued_data_mutex);
 235         queued_data[stream_id].append(string(data, data + bytes));
 236 }
 237
 238 // See the .h file for postconditions after this function.
 239 void Server::process_client(Client *client)
 240 {
 241         switch (client->state) {
 242         case Client::READING_REQUEST: {
 243 read_request_again:
 244                 // Try to read more of the request.
 245                 char buf[1024];
 246                 int ret;
 247                 do {
 248                         ret = read(client->sock, buf, sizeof(buf));
 249                 } while (ret == -1 && errno == EINTR);
 250
 251                 if (ret == -1 && errno == EAGAIN) {
 252                         // No more data right now. Nothing to do.
 253                         // This is postcondition #2.
 254                         return;
 255                 }
 256                 if (ret == -1) {
 257                         log_perror("read");
 258                         close_client(client);
 259                         return;
 260                 }
 261                 if (ret == 0) {
 262                         // OK, the socket is closed.
 263                         close_client(client);
 264                         return;
 265                 }
 266
 267                 RequestParseStatus status = wait_for_double_newline(&client->request, buf, ret);
 268
 269                 switch (status) {
 270                 case RP_OUT_OF_SPACE:
 271                         log(WARNING, "[%s] Client sent overlong request!", client->remote_addr.c_str());
 272                         close_client(client);
 273                         return;
 274                 case RP_NOT_FINISHED_YET:
 275                         // OK, we don't have the entire header yet. Fine; we'll get it later.
 276                         // See if there's more data for us.
 277                         goto read_request_again;
 278                 case RP_EXTRA_DATA:
 279                         log(WARNING, "[%s] Junk data after request!", client->remote_addr.c_str());
 280                         close_client(client);
 281                         return;
 282                 case RP_FINISHED:
 283                         break;
 284                 }
 285
 286                 assert(status == RP_FINISHED);
 287
 288                 int error_code = parse_request(client);
 289                 if (error_code == 200) {
 290                         construct_header(client);
 291                 } else {
 292                         construct_error(client, error_code);
 293                 }
 294
 295                 // We've changed states, so fall through.
 296                 assert(client->state == Client::SENDING_ERROR ||
 297                        client->state == Client::SENDING_HEADER);
 298         }
 299         case Client::SENDING_ERROR:
 300         case Client::SENDING_HEADER: {
 301 sending_header_or_error_again:
 302                 int ret;
 303                 do {
 304                         ret = write(client->sock,
 305                                     client->header_or_error.data() + client->header_or_error_bytes_sent,
 306                                     client->header_or_error.size() - client->header_or_error_bytes_sent);
 307                 } while (ret == -1 && errno == EINTR);
 308
 309                 if (ret == -1 && errno == EAGAIN) {
 310                         // We're out of socket space, so now we're at the “low edge” of epoll's
 311                         // edge triggering. epoll will tell us when there is more room, so for now,
 312                         // just return.
 313                         // This is postcondition #4.
 314                         return;
 315                 }
 316
 317                 if (ret == -1) {
 318                         // Error! Postcondition #1.
 319                         log_perror("write");
 320                         close_client(client);
 321                         return;
 322                 }
 323
 324                 client->header_or_error_bytes_sent += ret;
 325                 assert(client->header_or_error_bytes_sent <= client->header_or_error.size());
 326
 327                 if (client->header_or_error_bytes_sent < client->header_or_error.size()) {
 328                         // We haven't sent all yet. Fine; go another round.
 329                         goto sending_header_or_error_again;
 330                 }
 331
 332                 // We're done sending the header or error! Clear it to release some memory.
 333                 client->header_or_error.clear();
 334
 335                 if (client->state == Client::SENDING_ERROR) {
 336                         // We're done sending the error, so now close.
 337                         // This is postcondition #1.
 338                         close_client(client);
 339                         return;
 340                 }
 341
 342                 // Start sending from the end. In other words, we won't send any of the backlog,
 343                 // but we'll start sending immediately as we get data.
 344                 // This is postcondition #3.
 345                 client->state = Client::SENDING_DATA;
 346                 client->stream_pos = client->stream->bytes_received;
 347                 client->stream->put_client_to_sleep(client);
 348                 return;
 349         }
 350         case Client::SENDING_DATA: {
 351 sending_data_again:
 352                 // See if there's some data we've lost. Ideally, we should drop to a block boundary,
 353                 // but resync will be the mux's problem.
 354                 Stream *stream = client->stream;
 355                 size_t bytes_to_send = stream->bytes_received - client->stream_pos;
 356                 if (bytes_to_send == 0) {
 357                         return;
 358                 }
 359                 if (bytes_to_send > stream->backlog_size) {
 360                         log(WARNING, "[%s] Client lost %lld bytes, maybe too slow connection",
 361                                 client->remote_addr.c_str(),
 362                                 (long long int)(bytes_to_send - stream->backlog_size));
 363                         client->stream_pos = stream->bytes_received - stream->backlog_size;
 364                         client->bytes_lost += bytes_to_send - stream->backlog_size;
 365                         ++client->num_loss_events;
 366                         bytes_to_send = stream->backlog_size;
 367                 }
 368
 369                 // See if we need to split across the circular buffer.
 370                 bool more_data = false;
 371                 if ((client->stream_pos % stream->backlog_size) + bytes_to_send > stream->backlog_size) {
 372                         bytes_to_send = stream->backlog_size - (client->stream_pos % stream->backlog_size);
 373                         more_data = true;
 374                 }
 375
 376                 ssize_t ret;
 377                 do {
 378                         loff_t offset = client->stream_pos % stream->backlog_size;
 379                         ret = sendfile(client->sock, stream->data_fd, &offset, bytes_to_send);
 380                 } while (ret == -1 && errno == EINTR);
 381
 382                 if (ret == -1 && errno == EAGAIN) {
 383                         // We're out of socket space, so return; epoll will wake us up
 384                         // when there is more room.
 385                         // This is postcondition #4.
 386                         return;
 387                 }
 388                 if (ret == -1) {
 389                         // Error, close; postcondition #1.
 390                         log_perror("sendfile");
 391                         close_client(client);
 392                         return;
 393                 }
 394                 client->stream_pos += ret;
 395                 client->bytes_sent += ret;
 396
 397                 if (client->stream_pos == stream->bytes_received) {
 398                         // We don't have any more data for this client, so put it to sleep.
 399                         // This is postcondition #3.
 400                         stream->put_client_to_sleep(client);
 401                 } else if (more_data && size_t(ret) == bytes_to_send) {
 402                         goto sending_data_again;
 403                 }
 404                 break;
 405         }
 406         default:
 407                 assert(false);
 408         }
 409 }
 410
 411 int Server::parse_request(Client *client)
 412 {
 413         vector<string> lines = split_lines(client->request);
 414         if (lines.empty()) {
 415                 return 400;  // Bad request (empty).
 416         }
 417
 418         vector<string> request_tokens = split_tokens(lines[0]);
 419         if (request_tokens.size() < 2) {
 420                 return 400;  // Bad request (empty).
 421         }
 422         if (request_tokens[0] != "GET") {
 423                 return 400;  // Should maybe be 405 instead?
 424         }
 425         if (streams.count(request_tokens[1]) == 0) {
 426                 return 404;  // Not found.
 427         }
 428
 429         client->stream_id = request_tokens[1];
 430         client->stream = find_stream(client->stream_id);
 431         if (client->stream->mark_pool != NULL) {
 432                 client->fwmark = client->stream->mark_pool->get_mark();
 433         } else {
 434                 client->fwmark = 0;  // No mark.
 435         }
 436         if (setsockopt(client->sock, SOL_SOCKET, SO_MARK, &client->fwmark, sizeof(client->fwmark)) == -1) {
 437                 if (client->fwmark != 0) {
 438                         log_perror("setsockopt(SO_MARK)");
 439                 }
 440         }
 441         client->request.clear();
 442
 443         return 200;  // OK!
 444 }
 445
 446 void Server::construct_header(Client *client)
 447 {
 448         client->header_or_error = find_stream(client->stream_id)->header;
 449
 450         // Switch states.
 451         client->state = Client::SENDING_HEADER;
 452
 453         epoll_event ev;
 454         ev.events = EPOLLOUT | EPOLLET | EPOLLRDHUP;
 455         ev.data.u64 = reinterpret_cast<uint64_t>(client);
 456
 457         if (epoll_ctl(epoll_fd, EPOLL_CTL_MOD, client->sock, &ev) == -1) {
 458                 log_perror("epoll_ctl(EPOLL_CTL_MOD)");
 459                 exit(1);
 460         }
 461 }
 462
 463 void Server::construct_error(Client *client, int error_code)
 464 {
 465         char error[256];
 466         snprintf(error, 256, "HTTP/1.0 %d Error\r\nContent-type: text/plain\r\n\r\nSomething went wrong. Sorry.\r\n",
 467                 error_code);
 468         client->header_or_error = error;
 469
 470         // Switch states.
 471         client->state = Client::SENDING_ERROR;
 472
 473         epoll_event ev;
 474         ev.events = EPOLLOUT | EPOLLET | EPOLLRDHUP;
 475         ev.data.u64 = reinterpret_cast<uint64_t>(client);
 476
 477         if (epoll_ctl(epoll_fd, EPOLL_CTL_MOD, client->sock, &ev) == -1) {
 478                 log_perror("epoll_ctl(EPOLL_CTL_MOD)");
 479                 exit(1);
 480         }
 481 }
 482
 483 template<class T>
 484 void delete_from(vector<T> *v, T elem)
 485 {
 486         typename vector<T>::iterator new_end = remove(v->begin(), v->end(), elem);
 487         v->erase(new_end, v->end());
 488 }
 489
 490 void Server::close_client(Client *client)
 491 {
 492         if (epoll_ctl(epoll_fd, EPOLL_CTL_DEL, client->sock, NULL) == -1) {
 493                 log_perror("epoll_ctl(EPOLL_CTL_DEL)");
 494                 exit(1);
 495         }
 496
 497         // This client could be sleeping, so we'll need to fix that. (Argh, O(n).)
 498         if (client->stream != NULL) {
 499                 delete_from(&client->stream->sleeping_clients, client);
 500                 delete_from(&client->stream->to_process, client);
 501                 if (client->stream->mark_pool != NULL) {
 502                         int fwmark = client->fwmark;
 503                         client->stream->mark_pool->release_mark(fwmark);
 504                 }
 505         }
 506
 507         // Bye-bye!
 508         int ret;
 509         do {
 510                 ret = close(client->sock);
 511         } while (ret == -1 && errno == EINTR);
 512
 513         if (ret == -1) {
 514                 log_perror("close");
 515         }
 516
 517         clients.erase(client->sock);
 518 }
 519
 520 Stream *Server::find_stream(const string &stream_id)
 521 {
 522         map<string, Stream *>::iterator it = streams.find(stream_id);
 523         assert(it != streams.end());
 524         return it->second;
 525 }
 526
 527 void Server::process_queued_data()
 528 {
 529         MutexLock lock(&queued_data_mutex);
 530
 531         for (size_t i = 0; i < queued_add_clients.size(); ++i) {
 532                 add_client(queued_add_clients[i]);
 533         }
 534         queued_add_clients.clear();
 535
 536         for (map<string, string>::iterator queued_it = queued_data.begin();
 537              queued_it != queued_data.end();
 538              ++queued_it) {
 539                 Stream *stream = find_stream(queued_it->first);
 540                 stream->add_data(queued_it->second.data(), queued_it->second.size());
 541                 stream->wake_up_all_clients();
 542         }
 543         queued_data.clear();
 544 }