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

   1 #include <stdio.h>
   2 #include <string.h>
   3 #include <stdint.h>
   4 #include <assert.h>
   5 #include <arpa/inet.h>
   6 #include <sys/socket.h>
   7 #include <pthread.h>
   8 #include <sys/types.h>
   9 #include <sys/ioctl.h>
  10 #include <sys/types.h>
  11 #include <sys/socket.h>
  12 #include <netdb.h>
  13 #include <poll.h>
  14 #include <errno.h>
  15 #include <vector>
  16 #include <string>
  17 #include <map>
  18
  19 #include "metacube.h"
  20 #include "mutexlock.h"
  21 #include "input.h"
  22 #include "server.h"
  23 #include "serverpool.h"
  24 #include "parse.h"
  25
  26 using namespace std;
  27
  28 extern ServerPool *servers;
  29
  30 // Extremely rudimentary URL parsing.
  31 bool parse_url(const string &url, string *host, string *port, string *path)
  32 {
  33         if (url.find("http://") != 0) {
  34                 return false;
  35         }
  36
  37         string rest = url.substr(strlen("http://"));
  38         size_t split = rest.find_first_of(":/");
  39         if (split == string::npos) {
  40                 // http://foo
  41                 *host = rest;
  42                 *port = "http";
  43                 *path = "/";
  44                 return true;
  45         }
  46
  47         *host = string(rest.begin(), rest.begin() + split);
  48         char ch = rest[split];  // Colon or slash.
  49         rest = string(rest.begin() + split + 1, rest.end());
  50
  51         if (ch == ':') {
  52                 // Parse the port.
  53                 split = rest.find_first_of('/');
  54                 if (split == string::npos) {
  55                         // http://foo:1234
  56                         *port = rest;
  57                         *path = "/";
  58                         return true;
  59                 } else {
  60                         // http://foo:1234/bar
  61                         *port = string(rest.begin(), rest.begin() + split);
  62                         *path = string(rest.begin() + split, rest.end());
  63                         return true;
  64                 }
  65         }
  66
  67         // http://foo/bar
  68         *port = "http";
  69         *path = rest;
  70         return true;
  71 }
  72
  73 Input::Input(const string &stream_id, const string &url)
  74         : state(NOT_CONNECTED),
  75           stream_id(stream_id),
  76           url(url),
  77           has_metacube_header(false),
  78           sock(-1)
  79 {
  80 }
  81
  82 void Input::run()
  83 {
  84         should_stop = false;
  85
  86         // Joinable is already the default, but it's good to be certain.
  87         pthread_attr_t attr;
  88         pthread_attr_init(&attr);
  89         pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
  90         pthread_create(&worker_thread, &attr, Input::do_work_thunk, this);
  91 }
  92
  93 void Input::stop()
  94 {
  95         should_stop = true;
  96
  97         if (pthread_join(worker_thread, NULL) == -1) {
  98                 perror("pthread_join");
  99                 exit(1);
 100         }
 101 }
 102
 103 void *Input::do_work_thunk(void *arg)
 104 {
 105         Input *input = static_cast<Input *>(arg);
 106         input->do_work();
 107         return NULL;
 108 }
 109
 110 int Input::lookup_and_connect(const string &host, const string &port)
 111 {
 112         addrinfo *ai;
 113         int err = getaddrinfo(host.c_str(), port.c_str(), NULL, &ai);
 114         if (err == -1) {
 115                 fprintf(stderr, "WARNING: Lookup of '%s' failed (%s).\n",
 116                         host.c_str(), gai_strerror(err));
 117                 freeaddrinfo(ai);
 118                 return -1;
 119         }
 120
 121         // Connect to everything in turn until we have a socket.
 122         while (ai && !should_stop) {
 123                 int sock = socket(ai->ai_family, SOCK_STREAM, IPPROTO_TCP);
 124                 if (sock == -1) {
 125                         // Could be e.g. EPROTONOSUPPORT. The show must go on.
 126                         continue;
 127                 }
 128
 129                 do {
 130                         err = connect(sock, ai->ai_addr, ai->ai_addrlen);
 131                 } while (err == -1 && errno == EINTR);
 132
 133                 if (err != -1) {
 134                         freeaddrinfo(ai);
 135                         return sock;
 136                 }
 137
 138                 ai = ai->ai_next;
 139         }
 140
 141         // Give the last one as error.
 142         fprintf(stderr, "WARNING: Connect to '%s' failed (%s)\n",
 143                 host.c_str(), strerror(errno));
 144         freeaddrinfo(ai);
 145         return -1;
 146 }
 147
 148 void Input::do_work()
 149 {
 150         while (!should_stop) {
 151                 if (state == SENDING_REQUEST || state == RECEIVING_HEADER || state == RECEIVING_DATA) {
 152                         // Since we are non-blocking, we need to wait for the right state first.
 153                         // Wait up to 50 ms, then check should_stop.
 154                         pollfd pfd;
 155                         pfd.fd = sock;
 156                         pfd.events = (state == SENDING_REQUEST) ? POLLOUT : POLLIN;
 157                         pfd.events |= POLLRDHUP;
 158
 159                         int nfds = poll(&pfd, 1, 50);
 160                         if (nfds == 0 || (nfds == -1 && errno == EAGAIN)) {
 161                                 continue;
 162                         }
 163                         if (nfds == -1) {
 164                                 perror("poll");
 165                                 state = CLOSING_SOCKET;
 166                         }
 167                 }
 168
 169                 switch (state) {
 170                 case NOT_CONNECTED:
 171                         request.clear();
 172                         request_bytes_sent = 0;
 173                         response.clear();
 174
 175                         if (!parse_url(url, &host, &port, &path)) {
 176                                 fprintf(stderr, "Failed to parse URL '%s'\n", url.c_str());
 177                                 break;
 178                         }
 179
 180                         sock = lookup_and_connect(host, port);
 181                         if (sock != -1) {
 182                                 // Yay, successful connect. Try to set it as nonblocking.
 183                                 int one = 1;
 184                                 if (ioctl(sock, FIONBIO, &one) == -1) {
 185                                         perror("ioctl(FIONBIO)");
 186                                         state = CLOSING_SOCKET;
 187                                 } else {
 188                                         state = SENDING_REQUEST;
 189                                         request = "GET " + path + " HTTP/1.0\r\nUser-Agent: cubemap\r\n\r\n";
 190                                         request_bytes_sent = 0;
 191                                 }
 192                         }
 193                         break;
 194                 case SENDING_REQUEST: {
 195                         size_t to_send = request.size() - request_bytes_sent;
 196                         int ret;
 197
 198                         do {
 199                                 ret = write(sock, request.data() + request_bytes_sent, to_send);
 200                         } while (ret == -1 && errno == EINTR);
 201
 202                         if (ret == -1) {
 203                                 perror("write");
 204                                 state = CLOSING_SOCKET;
 205                                 continue;
 206                         }
 207
 208                         assert(ret >= 0);
 209                         request_bytes_sent += ret;
 210
 211                         if (request_bytes_sent == request.size()) {
 212                                 state = RECEIVING_HEADER;
 213                         }
 214                         break;
 215                 }
 216                 case RECEIVING_HEADER: {
 217                         char buf[4096];
 218                         int ret;
 219
 220                         do {
 221                                 ret = read(sock, buf, sizeof(buf));
 222                         } while (ret == -1 && errno == EINTR);
 223
 224                         if (ret == -1) {
 225                                 perror("read");
 226                                 state = CLOSING_SOCKET;
 227                                 continue;
 228                         }
 229
 230                         if (ret == 0) {
 231                                 // This really shouldn't happen...
 232                                 fprintf(stderr, "Socket unexpectedly closed while reading header\n");
 233                                 state = CLOSING_SOCKET;
 234                                 continue;
 235                         }
 236
 237                         RequestParseStatus status = wait_for_double_newline(&response, buf, ret);
 238
 239                         if (status == RP_OUT_OF_SPACE) {
 240                                 fprintf(stderr, "WARNING: fd %d sent overlong response!\n", sock);
 241                                 state = CLOSING_SOCKET;
 242                                 continue;
 243                         } else if (status == RP_NOT_FINISHED_YET) {
 244                                 continue;
 245                         }
 246
 247                         // OK, so we're fine, but there might be some of the actual data after the response.
 248                         // We'll need to deal with that separately.
 249                         string extra_data;
 250                         if (status == RP_EXTRA_DATA) {
 251                                 char *ptr = static_cast<char *>(
 252                                         memmem(response.data(), response.size(), "\r\n\r\n", 4));
 253                                 assert(ptr != NULL);
 254                                 extra_data = string(ptr, &response[0] + response.size());
 255                                 response.resize(ptr - response.data());
 256                         }
 257
 258                         // TODO: Check that the response is 200, save the headers, etc.
 259
 260                         if (!extra_data.empty()) {
 261                                 process_data(&extra_data[0], extra_data.size());
 262                         }
 263
 264                         state = RECEIVING_DATA;
 265                         break;
 266                 }
 267                 case RECEIVING_DATA: {
 268                         char buf[4096];
 269                         int ret;
 270
 271                         do {
 272                                 ret = read(sock, buf, sizeof(buf));
 273                         } while (ret == -1 && errno == EINTR);
 274
 275                         if (ret == -1) {
 276                                 perror("read");
 277                                 state = CLOSING_SOCKET;
 278                                 continue;
 279                         }
 280
 281                         if (ret == 0) {
 282                                 // This really shouldn't happen...
 283                                 fprintf(stderr, "Socket unexpectedly closed while reading header\n");
 284                                 state = CLOSING_SOCKET;
 285                                 continue;
 286                         }
 287
 288                         process_data(buf, ret);
 289                         break;
 290                 }
 291                 case CLOSING_SOCKET: {
 292                         int err;
 293                         do {
 294                                 err = close(sock);
 295                         } while (err == -1 && errno == EINTR);
 296
 297                         if (err == -1) {
 298                                 perror("close");
 299                         }
 300
 301                         state = NOT_CONNECTED;
 302                         break;
 303                 }
 304                 default:
 305                         assert(false);
 306                 }
 307
 308                 // If we are still in NOT_CONNECTED, either something went wrong,
 309                 // or the connection just got closed.
 310                 // The earlier steps have already given the error message, if any.
 311                 if (state == NOT_CONNECTED && !should_stop) {
 312                         fprintf(stderr, "Waiting 0.2 second and restarting...\n");
 313                         usleep(200000);
 314                 }
 315         }
 316 }
 317
 318 void Input::process_data(char *ptr, size_t bytes)
 319 {
 320         pending_data.insert(pending_data.end(), ptr, ptr + bytes);
 321
 322         for ( ;; ) {
 323                 // If we don't have enough data (yet) for even the Metacube header, just return.
 324                 if (pending_data.size() < sizeof(metacube_block_header)) {
 325                         return;
 326                 }
 327
 328                 // Make sure we have the Metacube sync header at the start.
 329                 // We may need to skip over junk data (it _should_ not happen, though).
 330                 if (!has_metacube_header) {
 331                         char *ptr = static_cast<char *>(
 332                                 memmem(pending_data.data(), pending_data.size(),
 333                                        METACUBE_SYNC, strlen(METACUBE_SYNC)));
 334                         if (ptr == NULL) {
 335                                 // OK, so we didn't find the sync marker. We know then that
 336                                 // we do not have the _full_ marker in the buffer, but we
 337                                 // could have N-1 bytes. Drop everything before that,
 338                                 // and then give up.
 339                                 drop_pending_data(pending_data.size() - (strlen(METACUBE_SYNC) - 1));
 340                                 return;
 341                         } else {
 342                                 // Yay, we found the header. Drop everything (if anything) before it.
 343                                 drop_pending_data(ptr - pending_data.data());
 344                                 has_metacube_header = true;
 345
 346                                 // Re-check that we have the entire header; we could have dropped data.
 347                                 if (pending_data.size() < sizeof(metacube_block_header)) {
 348                                         return;
 349                                 }
 350                         }
 351                 }
 352
 353                 // Now it's safe to read the header.
 354                 metacube_block_header *hdr = reinterpret_cast<metacube_block_header *>(pending_data.data());
 355                 assert(memcmp(hdr->sync, METACUBE_SYNC, sizeof(hdr->sync)) == 0);
 356                 uint32_t size = ntohl(hdr->size);
 357                 uint32_t flags = ntohl(hdr->flags);
 358
 359                 // See if we have the entire block. If not, wait for more data.
 360                 if (pending_data.size() < sizeof(metacube_block_header) + size) {
 361                         return;
 362                 }
 363
 364                 // Send this block on to the data.
 365                 char *inner_data = pending_data.data() + sizeof(metacube_block_header);
 366                 if (flags & METACUBE_FLAGS_HEADER) {
 367                         string header(inner_data, inner_data + size);
 368                         servers->set_header(stream_id, header);
 369                 } else {
 370                         servers->add_data(stream_id, inner_data, size);
 371                 }
 372
 373                 // Consume the block. This isn't the most efficient way of dealing with things
 374                 // should we have many blocks, but these routines don't need to be too efficient
 375                 // anyway.
 376                 pending_data.erase(pending_data.begin(), pending_data.begin() + sizeof(metacube_block_header) + size);
 377                 has_metacube_header = false;
 378         }
 379 }
 380
 381 void Input::drop_pending_data(size_t num_bytes)
 382 {
 383         if (num_bytes == 0) {
 384                 return;
 385         }
 386         fprintf(stderr, "Warning: Dropping %lld junk bytes from stream, maybe it is not a Metacube stream?\n",
 387                 (long long)num_bytes);
 388         pending_data.erase(pending_data.begin(), pending_data.begin() + num_bytes);
 389 }
 390