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

   1 #include <stdio.h>
   2 #include <string.h>
   3 #include <stdint.h>
   4 #include <unistd.h>
   5 #include <assert.h>
   6 #include <arpa/inet.h>
   7 #include <sys/socket.h>
   8 #include <pthread.h>
   9 #include <sys/types.h>
  10 #include <sys/ioctl.h>
  11 #include <sys/epoll.h>
  12 #include <sys/sendfile.h>
  13 #include <time.h>
  14 #include <signal.h>
  15 #include <errno.h>
  16 #include <vector>
  17 #include <string>
  18 #include <map>
  19 #include <algorithm>
  20
  21 #include "markpool.h"
  22 #include "metacube.h"
  23 #include "server.h"
  24 #include "mutexlock.h"
  25 #include "parse.h"
  26 #include "util.h"
  27 #include "state.pb.h"
  28
  29 using namespace std;
  30
  31 Client::Client(int sock)
  32         : sock(sock),
  33           fwmark(0),
  34           connect_time(time(NULL)),
  35           state(Client::READING_REQUEST),
  36           stream(NULL),
  37           header_or_error_bytes_sent(0),
  38           bytes_sent(0)
  39 {
  40         request.reserve(1024);
  41
  42         // Find the remote address, and convert it to ASCII.
  43         sockaddr_in6 addr;
  44         socklen_t addr_len = sizeof(addr);
  45
  46         if (getpeername(sock, reinterpret_cast<sockaddr *>(&addr), &addr_len) == -1) {
  47                 perror("getpeername");
  48                 remote_addr = "";
  49         } else {
  50                 char buf[INET6_ADDRSTRLEN];
  51                 if (inet_ntop(addr.sin6_family, &addr.sin6_addr, buf, sizeof(buf)) == NULL) {
  52                         perror("inet_ntop");
  53                         remote_addr = "";
  54                 } else {
  55                         remote_addr = buf;
  56                 }
  57         }
  58 }
  59
  60 Client::Client(const ClientProto &serialized, Stream *stream)
  61         : sock(serialized.sock()),
  62           remote_addr(serialized.remote_addr()),
  63           connect_time(serialized.connect_time()),
  64           state(State(serialized.state())),
  65           request(serialized.request()),
  66           stream_id(serialized.stream_id()),
  67           stream(stream),
  68           header_or_error(serialized.header_or_error()),
  69           header_or_error_bytes_sent(serialized.header_or_error_bytes_sent()),
  70           bytes_sent(serialized.bytes_sent())
  71 {
  72         if (stream->mark_pool != NULL) {
  73                 fwmark = stream->mark_pool->get_mark();
  74         } else {
  75                 fwmark = 0;  // No mark.
  76         }
  77         if (setsockopt(sock, SOL_SOCKET, SO_MARK, &fwmark, sizeof(fwmark)) == -1) {
  78                 if (fwmark != 0) {
  79                         perror("setsockopt(SO_MARK)");
  80                 }
  81         }
  82 }
  83
  84 ClientProto Client::serialize() const
  85 {
  86         ClientProto serialized;
  87         serialized.set_sock(sock);
  88         serialized.set_remote_addr(remote_addr);
  89         serialized.set_connect_time(connect_time);
  90         serialized.set_state(state);
  91         serialized.set_request(request);
  92         serialized.set_stream_id(stream_id);
  93         serialized.set_header_or_error(header_or_error);
  94         serialized.set_header_or_error_bytes_sent(serialized.header_or_error_bytes_sent());
  95         serialized.set_bytes_sent(bytes_sent);
  96         return serialized;
  97 }
  98
  99 ClientStats Client::get_stats() const
 100 {
 101         ClientStats stats;
 102         stats.stream_id = stream_id;
 103         stats.remote_addr = remote_addr;
 104         stats.connect_time = connect_time;
 105         stats.bytes_sent = bytes_sent;
 106         return stats;
 107 }
 108
 109 Stream::Stream(const string &stream_id)
 110         : stream_id(stream_id),
 111           data_fd(make_tempfile("")),
 112           data_size(0),
 113           mark_pool(NULL)
 114 {
 115         if (data_fd == -1) {
 116                 exit(1);
 117         }
 118 }
 119
 120 Stream::~Stream()
 121 {
 122         if (data_fd != -1) {
 123                 int ret;
 124                 do {
 125                         ret = close(data_fd);
 126                 } while (ret == -1 && errno == EINTR);
 127                 if (ret == -1) {
 128                         perror("close");
 129                 }
 130         }
 131 }
 132
 133 Stream::Stream(const StreamProto &serialized)
 134         : stream_id(serialized.stream_id()),
 135           header(serialized.header()),
 136           data_fd(make_tempfile(serialized.data())),
 137           data_size(serialized.data_size()),
 138           mark_pool(NULL)
 139 {
 140         if (data_fd == -1) {
 141                 exit(1);
 142         }
 143 }
 144
 145 StreamProto Stream::serialize()
 146 {
 147         StreamProto serialized;
 148         serialized.set_header(header);
 149         if (!read_tempfile(data_fd, serialized.mutable_data())) {  // Closes data_fd.
 150                 exit(1);
 151         }
 152         serialized.set_data_size(data_size);
 153         serialized.set_stream_id(stream_id);
 154         data_fd = -1;
 155         return serialized;
 156 }
 157
 158 void Stream::put_client_to_sleep(Client *client)
 159 {
 160         sleeping_clients.push_back(client);
 161 }
 162
 163 void Stream::wake_up_all_clients()
 164 {
 165         if (to_process.empty()) {
 166                 swap(sleeping_clients, to_process);
 167         } else {
 168                 to_process.insert(to_process.end(), sleeping_clients.begin(), sleeping_clients.end());
 169                 sleeping_clients.clear();
 170         }
 171 }
 172
 173 Server::Server()
 174 {
 175         pthread_mutex_init(&mutex, NULL);
 176         pthread_mutex_init(&queued_data_mutex, NULL);
 177
 178         epoll_fd = epoll_create(1024);  // Size argument is ignored.
 179         if (epoll_fd == -1) {
 180                 perror("epoll_fd");
 181                 exit(1);
 182         }
 183 }
 184
 185 Server::~Server()
 186 {
 187         int ret;
 188         do {
 189                 ret = close(epoll_fd);
 190         } while (ret == -1 && errno == EINTR);
 191
 192         if (ret == -1) {
 193                 perror("close(epoll_fd)");
 194         }
 195 }
 196
 197 vector<ClientStats> Server::get_client_stats() const
 198 {
 199         vector<ClientStats> ret;
 200
 201         MutexLock lock(&mutex);
 202         for (map<int, Client>::const_iterator client_it = clients.begin();
 203              client_it != clients.end();
 204              ++client_it) {
 205                 ret.push_back(client_it->second.get_stats());
 206         }
 207         return ret;
 208 }
 209
 210 void Server::do_work()
 211 {
 212         for ( ;; ) {
 213                 int nfds = epoll_wait(epoll_fd, events, EPOLL_MAX_EVENTS, EPOLL_TIMEOUT_MS);
 214                 if (nfds == -1 && errno == EINTR) {
 215                         if (should_stop) {
 216                                 return;
 217                         }
 218                         continue;
 219                 }
 220                 if (nfds == -1) {
 221                         perror("epoll_wait");
 222                         exit(1);
 223                 }
 224
 225                 MutexLock lock(&mutex);  // We release the mutex between iterations.
 226
 227                 process_queued_data();
 228
 229                 for (int i = 0; i < nfds; ++i) {
 230                         int fd = events[i].data.fd;
 231                         assert(clients.count(fd) != 0);
 232                         Client *client = &clients[fd];
 233
 234                         if (events[i].events & (EPOLLERR | EPOLLRDHUP | EPOLLHUP)) {
 235                                 close_client(client);
 236                                 continue;
 237                         }
 238
 239                         process_client(client);
 240                 }
 241
 242                 for (map<string, Stream *>::iterator stream_it = streams.begin();
 243                      stream_it != streams.end();
 244                      ++stream_it) {
 245                         vector<Client *> to_process;
 246                         swap(stream_it->second->to_process, to_process);
 247                         for (size_t i = 0; i < to_process.size(); ++i) {
 248                                 process_client(to_process[i]);
 249                         }
 250                 }
 251
 252                 if (should_stop) {
 253                         return;
 254                 }
 255         }
 256 }
 257
 258 CubemapStateProto Server::serialize()
 259 {
 260         // We don't serialize anything queued, so empty the queues.
 261         process_queued_data();
 262
 263         CubemapStateProto serialized;
 264         for (map<int, Client>::const_iterator client_it = clients.begin();
 265              client_it != clients.end();
 266              ++client_it) {
 267                 serialized.add_clients()->MergeFrom(client_it->second.serialize());
 268         }
 269         for (map<string, Stream *>::const_iterator stream_it = streams.begin();
 270              stream_it != streams.end();
 271              ++stream_it) {
 272                 serialized.add_streams()->MergeFrom(stream_it->second->serialize());
 273         }
 274         return serialized;
 275 }
 276
 277 void Server::add_client_deferred(int sock)
 278 {
 279         MutexLock lock(&queued_data_mutex);
 280         queued_add_clients.push_back(sock);
 281 }
 282
 283 void Server::add_client(int sock)
 284 {
 285         clients.insert(make_pair(sock, Client(sock)));
 286
 287         // Start listening on data from this socket.
 288         epoll_event ev;
 289         ev.events = EPOLLIN | EPOLLET | EPOLLRDHUP;
 290         ev.data.u64 = 0;  // Keep Valgrind happy.
 291         ev.data.fd = sock;
 292         if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, sock, &ev) == -1) {
 293                 perror("epoll_ctl(EPOLL_CTL_ADD)");
 294                 exit(1);
 295         }
 296
 297         process_client(&clients[sock]);
 298 }
 299
 300 void Server::add_client_from_serialized(const ClientProto &client)
 301 {
 302         MutexLock lock(&mutex);
 303         Stream *stream = find_stream(client.stream_id());
 304         clients.insert(make_pair(client.sock(), Client(client, stream)));
 305         Client *client_ptr = &clients[client.sock()];
 306
 307         // Start listening on data from this socket.
 308         epoll_event ev;
 309         if (client.state() == Client::READING_REQUEST) {
 310                 ev.events = EPOLLIN | EPOLLET | EPOLLRDHUP;
 311         } else {
 312                 // If we don't have more data for this client, we'll be putting it into
 313                 // the sleeping array again soon.
 314                 ev.events = EPOLLOUT | EPOLLET | EPOLLRDHUP;
 315         }
 316         ev.data.u64 = 0;  // Keep Valgrind happy.
 317         ev.data.fd = client.sock();
 318         if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, client.sock(), &ev) == -1) {
 319                 perror("epoll_ctl(EPOLL_CTL_ADD)");
 320                 exit(1);
 321         }
 322
 323         if (client_ptr->state == Client::SENDING_DATA &&
 324             client_ptr->bytes_sent == client_ptr->stream->data_size) {
 325                 client_ptr->stream->put_client_to_sleep(client_ptr);
 326         } else {
 327                 process_client(client_ptr);
 328         }
 329 }
 330
 331 void Server::add_stream(const string &stream_id)
 332 {
 333         MutexLock lock(&mutex);
 334         streams.insert(make_pair(stream_id, new Stream(stream_id)));
 335 }
 336
 337 void Server::add_stream_from_serialized(const StreamProto &stream)
 338 {
 339         MutexLock lock(&mutex);
 340         streams.insert(make_pair(stream.stream_id(), new Stream(stream)));
 341 }
 342
 343 void Server::set_header(const string &stream_id, const string &header)
 344 {
 345         MutexLock lock(&mutex);
 346         find_stream(stream_id)->header = header;
 347
 348         // If there are clients we haven't sent anything to yet, we should give
 349         // them the header, so push back into the SENDING_HEADER state.
 350         for (map<int, Client>::iterator client_it = clients.begin();
 351              client_it != clients.end();
 352              ++client_it) {
 353                 Client *client = &client_it->second;
 354                 if (client->state == Client::SENDING_DATA &&
 355                     client->bytes_sent == 0) {
 356                         construct_header(client);
 357                 }
 358         }
 359 }
 360
 361 void Server::set_mark_pool(const std::string &stream_id, MarkPool *mark_pool)
 362 {
 363         MutexLock lock(&mutex);
 364         assert(clients.empty());
 365         find_stream(stream_id)->mark_pool = mark_pool;
 366 }
 367
 368 void Server::add_data_deferred(const string &stream_id, const char *data, size_t bytes)
 369 {
 370         MutexLock lock(&queued_data_mutex);
 371         queued_data[stream_id].append(string(data, data + bytes));
 372 }
 373
 374 void Server::add_data(const string &stream_id, const char *data, ssize_t bytes)
 375 {
 376         Stream *stream = find_stream(stream_id);
 377         size_t pos = stream->data_size % BACKLOG_SIZE;
 378         stream->data_size += bytes;
 379
 380         if (pos + bytes > BACKLOG_SIZE) {
 381                 ssize_t to_copy = BACKLOG_SIZE - pos;
 382                 while (to_copy > 0) {
 383                         int ret = pwrite(stream->data_fd, data, to_copy, pos);
 384                         if (ret == -1 && errno == EINTR) {
 385                                 continue;
 386                         }
 387                         if (ret == -1) {
 388                                 perror("pwrite");
 389                                 // Dazed and confused, but trying to continue...
 390                                 break;
 391                         }
 392                         pos += ret;
 393                         data += ret;
 394                         to_copy -= ret;
 395                         bytes -= ret;
 396                 }
 397                 pos = 0;
 398         }
 399
 400         while (bytes > 0) {
 401                 int ret = pwrite(stream->data_fd, data, bytes, pos);
 402                 if (ret == -1 && errno == EINTR) {
 403                         continue;
 404                 }
 405                 if (ret == -1) {
 406                         perror("pwrite");
 407                         // Dazed and confused, but trying to continue...
 408                         break;
 409                 }
 410                 pos += ret;
 411                 data += ret;
 412                 bytes -= ret;
 413         }
 414
 415         stream->wake_up_all_clients();
 416 }
 417
 418 // See the .h file for postconditions after this function.
 419 void Server::process_client(Client *client)
 420 {
 421         switch (client->state) {
 422         case Client::READING_REQUEST: {
 423 read_request_again:
 424                 // Try to read more of the request.
 425                 char buf[1024];
 426                 int ret;
 427                 do {
 428                         ret = read(client->sock, buf, sizeof(buf));
 429                 } while (ret == -1 && errno == EINTR);
 430
 431                 if (ret == -1 && errno == EAGAIN) {
 432                         // No more data right now. Nothing to do.
 433                         // This is postcondition #2.
 434                         return;
 435                 }
 436                 if (ret == -1) {
 437                         perror("read");
 438                         close_client(client);
 439                         return;
 440                 }
 441                 if (ret == 0) {
 442                         // OK, the socket is closed.
 443                         close_client(client);
 444                         return;
 445                 }
 446
 447                 RequestParseStatus status = wait_for_double_newline(&client->request, buf, ret);
 448
 449                 switch (status) {
 450                 case RP_OUT_OF_SPACE:
 451                         fprintf(stderr, "WARNING: fd %d sent overlong request!\n", client->sock);
 452                         close_client(client);
 453                         return;
 454                 case RP_NOT_FINISHED_YET:
 455                         // OK, we don't have the entire header yet. Fine; we'll get it later.
 456                         // See if there's more data for us.
 457                         goto read_request_again;
 458                 case RP_EXTRA_DATA:
 459                         fprintf(stderr, "WARNING: fd %d had junk data after request!\n", client->sock);
 460                         close_client(client);
 461                         return;
 462                 case RP_FINISHED:
 463                         break;
 464                 }
 465
 466                 assert(status == RP_FINISHED);
 467
 468                 int error_code = parse_request(client);
 469                 if (error_code == 200) {
 470                         construct_header(client);
 471                 } else {
 472                         construct_error(client, error_code);
 473                 }
 474
 475                 // We've changed states, so fall through.
 476                 assert(client->state == Client::SENDING_ERROR ||
 477                        client->state == Client::SENDING_HEADER);
 478         }
 479         case Client::SENDING_ERROR:
 480         case Client::SENDING_HEADER: {
 481 sending_header_or_error_again:
 482                 int ret;
 483                 do {
 484                         ret = write(client->sock,
 485                                     client->header_or_error.data() + client->header_or_error_bytes_sent,
 486                                     client->header_or_error.size() - client->header_or_error_bytes_sent);
 487                 } while (ret == -1 && errno == EINTR);
 488
 489                 if (ret == -1 && errno == EAGAIN) {
 490                         // We're out of socket space, so now we're at the “low edge” of epoll's
 491                         // edge triggering. epoll will tell us when there is more room, so for now,
 492                         // just return.
 493                         // This is postcondition #4.
 494                         return;
 495                 }
 496
 497                 if (ret == -1) {
 498                         // Error! Postcondition #1.
 499                         perror("write");
 500                         close_client(client);
 501                         return;
 502                 }
 503
 504                 client->header_or_error_bytes_sent += ret;
 505                 assert(client->header_or_error_bytes_sent <= client->header_or_error.size());
 506
 507                 if (client->header_or_error_bytes_sent < client->header_or_error.size()) {
 508                         // We haven't sent all yet. Fine; go another round.
 509                         goto sending_header_or_error_again;
 510                 }
 511
 512                 // We're done sending the header or error! Clear it to release some memory.
 513                 client->header_or_error.clear();
 514
 515                 if (client->state == Client::SENDING_ERROR) {
 516                         // We're done sending the error, so now close.
 517                         // This is postcondition #1.
 518                         close_client(client);
 519                         return;
 520                 }
 521
 522                 // Start sending from the end. In other words, we won't send any of the backlog,
 523                 // but we'll start sending immediately as we get data.
 524                 // This is postcondition #3.
 525                 client->state = Client::SENDING_DATA;
 526                 client->bytes_sent = client->stream->data_size;
 527                 client->stream->put_client_to_sleep(client);
 528                 return;
 529         }
 530         case Client::SENDING_DATA: {
 531 sending_data_again:
 532                 // See if there's some data we've lost. Ideally, we should drop to a block boundary,
 533                 // but resync will be the mux's problem.
 534                 Stream *stream = client->stream;
 535                 size_t bytes_to_send = stream->data_size - client->bytes_sent;
 536                 if (bytes_to_send == 0) {
 537                         return;
 538                 }
 539                 if (bytes_to_send > BACKLOG_SIZE) {
 540                         fprintf(stderr, "WARNING: fd %d lost %lld bytes, maybe too slow connection\n",
 541                                 client->sock,
 542                                 (long long int)(bytes_to_send - BACKLOG_SIZE));
 543                         client->bytes_sent = stream->data_size - BACKLOG_SIZE;
 544                         bytes_to_send = BACKLOG_SIZE;
 545                 }
 546
 547                 // See if we need to split across the circular buffer.
 548                 bool more_data = false;
 549                 if ((client->bytes_sent % BACKLOG_SIZE) + bytes_to_send > BACKLOG_SIZE) {
 550                         bytes_to_send = BACKLOG_SIZE - (client->bytes_sent % BACKLOG_SIZE);
 551                         more_data = true;
 552                 }
 553
 554                 ssize_t ret;
 555                 do {
 556                         loff_t offset = client->bytes_sent % BACKLOG_SIZE;
 557                         ret = sendfile(client->sock, stream->data_fd, &offset, bytes_to_send);
 558                 } while (ret == -1 && errno == EINTR);
 559
 560                 if (ret == -1 && errno == EAGAIN) {
 561                         // We're out of socket space, so return; epoll will wake us up
 562                         // when there is more room.
 563                         // This is postcondition #4.
 564                         return;
 565                 }
 566                 if (ret == -1) {
 567                         // Error, close; postcondition #1.
 568                         perror("sendfile");
 569                         close_client(client);
 570                         return;
 571                 }
 572                 client->bytes_sent += ret;
 573
 574                 if (client->bytes_sent == stream->data_size) {
 575                         // We don't have any more data for this client, so put it to sleep.
 576                         // This is postcondition #3.
 577                         stream->put_client_to_sleep(client);
 578                 } else if (more_data) {
 579                         goto sending_data_again;
 580                 }
 581                 break;
 582         }
 583         default:
 584                 assert(false);
 585         }
 586 }
 587
 588 int Server::parse_request(Client *client)
 589 {
 590         vector<string> lines = split_lines(client->request);
 591         if (lines.empty()) {
 592                 return 400;  // Bad request (empty).
 593         }
 594
 595         vector<string> request_tokens = split_tokens(lines[0]);
 596         if (request_tokens.size() < 2) {
 597                 return 400;  // Bad request (empty).
 598         }
 599         if (request_tokens[0] != "GET") {
 600                 return 400;  // Should maybe be 405 instead?
 601         }
 602         if (streams.count(request_tokens[1]) == 0) {
 603                 return 404;  // Not found.
 604         }
 605
 606         client->stream_id = request_tokens[1];
 607         client->stream = find_stream(client->stream_id);
 608         if (client->stream->mark_pool != NULL) {
 609                 client->fwmark = client->stream->mark_pool->get_mark();
 610         } else {
 611                 client->fwmark = 0;  // No mark.
 612         }
 613         if (setsockopt(client->sock, SOL_SOCKET, SO_MARK, &client->fwmark, sizeof(client->fwmark)) == -1) {
 614                 if (client->fwmark != 0) {
 615                         perror("setsockopt(SO_MARK)");
 616                 }
 617         }
 618         client->request.clear();
 619
 620         return 200;  // OK!
 621 }
 622
 623 void Server::construct_header(Client *client)
 624 {
 625         client->header_or_error = find_stream(client->stream_id)->header;
 626
 627         // Switch states.
 628         client->state = Client::SENDING_HEADER;
 629
 630         epoll_event ev;
 631         ev.events = EPOLLOUT | EPOLLET | EPOLLRDHUP;
 632         ev.data.u64 = 0;  // Keep Valgrind happy.
 633         ev.data.fd = client->sock;
 634
 635         if (epoll_ctl(epoll_fd, EPOLL_CTL_MOD, client->sock, &ev) == -1) {
 636                 perror("epoll_ctl(EPOLL_CTL_MOD)");
 637                 exit(1);
 638         }
 639 }
 640
 641 void Server::construct_error(Client *client, int error_code)
 642 {
 643         char error[256];
 644         snprintf(error, 256, "HTTP/1.0 %d Error\r\nContent-type: text/plain\r\n\r\nSomething went wrong. Sorry.\r\n",
 645                 error_code);
 646         client->header_or_error = error;
 647
 648         // Switch states.
 649         client->state = Client::SENDING_ERROR;
 650
 651         epoll_event ev;
 652         ev.events = EPOLLOUT | EPOLLET | EPOLLRDHUP;
 653         ev.data.u64 = 0;  // Keep Valgrind happy.
 654         ev.data.fd = client->sock;
 655
 656         if (epoll_ctl(epoll_fd, EPOLL_CTL_MOD, client->sock, &ev) == -1) {
 657                 perror("epoll_ctl(EPOLL_CTL_MOD)");
 658                 exit(1);
 659         }
 660 }
 661
 662 template<class T>
 663 void delete_from(vector<T> *v, T elem)
 664 {
 665         typename vector<T>::iterator new_end = remove(v->begin(), v->end(), elem);
 666         v->erase(new_end, v->end());
 667 }
 668
 669 void Server::close_client(Client *client)
 670 {
 671         if (epoll_ctl(epoll_fd, EPOLL_CTL_DEL, client->sock, NULL) == -1) {
 672                 perror("epoll_ctl(EPOLL_CTL_DEL)");
 673                 exit(1);
 674         }
 675
 676         // This client could be sleeping, so we'll need to fix that. (Argh, O(n).)
 677         if (client->stream != NULL) {
 678                 delete_from(&client->stream->sleeping_clients, client);
 679                 delete_from(&client->stream->to_process, client);
 680                 if (client->stream->mark_pool != NULL) {
 681                         int fwmark = client->fwmark;
 682                         client->stream->mark_pool->release_mark(fwmark);
 683                 }
 684         }
 685
 686         // Bye-bye!
 687         int ret;
 688         do {
 689                 ret = close(client->sock);
 690         } while (ret == -1 && errno == EINTR);
 691
 692         if (ret == -1) {
 693                 perror("close");
 694         }
 695
 696         clients.erase(client->sock);
 697 }
 698
 699 Stream *Server::find_stream(const string &stream_id)
 700 {
 701         map<string, Stream *>::iterator it = streams.find(stream_id);
 702         assert(it != streams.end());
 703         return it->second;
 704 }
 705
 706 void Server::process_queued_data()
 707 {
 708         MutexLock lock(&queued_data_mutex);
 709
 710         for (size_t i = 0; i < queued_add_clients.size(); ++i) {
 711                 add_client(queued_add_clients[i]);
 712         }
 713         queued_add_clients.clear();
 714
 715         for (map<string, string>::iterator queued_it = queued_data.begin();
 716              queued_it != queued_data.end();
 717              ++queued_it) {
 718                 add_data(queued_it->first, queued_it->second.data(), queued_it->second.size());
 719         }
 720         queued_data.clear();
 721 }