Mercurial > urweb

annotate src/c/http.c @ 2289:78820fa8f5a7

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Fix bugs for lock calculation and SQL parsing and add support for tasks.
author Ziv Scully <ziv@mit.edu>
date Sun, 15 Nov 2015 14:18:35 -0500
parents 3b4a5604ed97
children
rev   line source
adamc@1268 1 #include "config.h"
adamc@1268 2
adamc@116 3 #include <stdio.h>
adamc@116 4 #include <string.h>
adamc@502 5 #include <stdlib.h>
adamc@116 6 #include <sys/types.h>
adamc@116 7 #include <sys/socket.h>
adamc@116 8 #include <netinet/in.h>
adam@1917 9 #include <netinet/tcp.h>
adam@1779 10 #include <arpa/inet.h>
adamc@472 11 #include <unistd.h>
adamc@502 12 #include <signal.h>
adamc@856 13 #include <stdarg.h>
adamc@116 14
adamc@138 15 #include <pthread.h>
adamc@138 16
adamc@244 17 #include "urweb.h"
adamc@853 18 #include "request.h"
adamc@859 19 #include "queue.h"
adamc@117 20
adamc@1094 21 extern uw_app uw_application;
adamc@1094 22
adam@1914 23 int uw_backlog = SOMAXCONN;
adam@1930 24 static int keepalive = 0, quiet = 0;
adamc@116 25
grrwlf@2069 26 #define qfprintf(f, fmt, args...) do { if(!quiet) fprintf(f, fmt, ##args); } while(0)
grrwlf@2069 27 #define qprintf(fmt, args...) do { if(!quiet) printf(fmt, ##args); } while(0)
grrwlf@2069 28
adamc@854 29 static char *get_header(void *data, const char *h) {
adamc@854 30 char *s = data;
adamc@854 31 int len = strlen(h);
adamc@854 32 char *p;
adamc@854 33
adamc@1134 34 while ((p = strchr(s, ':'))) {
adamc@854 35 if (p - s == len && !strncasecmp(s, h, len)) {
adamc@854 36 return p + 2;
adamc@854 37 } else {
adamc@854 38 if ((s = strchr(p, 0)) && s[1] != 0)
adamc@854 39 s += 2;
adamc@854 40 else
adamc@854 41 return NULL;
adamc@854 42 }
adamc@854 43 }
adamc@854 44
adamc@854 45 return NULL;
adamc@854 46 }
adamc@854 47
adam@1799 48 static char *get_env(void *data, const char *name) {
adam@1799 49 return getenv(name);
adam@1799 50 }
adam@1799 51
adamc@856 52 static void on_success(uw_context ctx) {
adamc@856 53 uw_write_header(ctx, "HTTP/1.1 200 OK\r\n");
adamc@856 54 }
adamc@856 55
adamc@856 56 static void on_failure(uw_context ctx) {
adamc@856 57 uw_write_header(ctx, "HTTP/1.1 500 Internal Server Error\r\n");
adamc@856 58 }
adamc@856 59
adamc@856 60 static void log_error(void *data, const char *fmt, ...) {
adamc@856 61 va_list ap;
adamc@856 62 va_start(ap, fmt);
adamc@856 63
adamc@856 64 vfprintf(stderr, fmt, ap);
adamc@856 65 }
adamc@856 66
adamc@856 67 static void log_debug(void *data, const char *fmt, ...) {
adam@1930 68 if (!quiet) {
adam@1930 69 va_list ap;
adam@1930 70 va_start(ap, fmt);
adamc@856 71
adam@1930 72 vprintf(fmt, ap);
adam@1930 73 }
adamc@856 74 }
adamc@856 75
grrwlf@1997 76 static uw_loggers ls = {NULL, log_error, log_debug};
grrwlf@1997 77
adamc@853 78 static void *worker(void *data) {
adamc@853 79 int me = *(int *)data;
grrwlf@1997 80 uw_context ctx = uw_request_new_context(me, &uw_application, &ls);
adam@1937 81 size_t buf_size = 1024;
adam@1917 82 char *buf = malloc(buf_size), *back = buf;
adamc@853 83 uw_request_context rc = uw_new_request_context();
adam@1917 84 int sock = 0;
adamc@698 85
adamc@272 86 while (1) {
adam@1917 87 if (sock == 0) {
adam@1917 88 back = buf;
adam@1917 89 sock = uw_dequeue();
adam@1917 90 }
adamc@138 91
adam@2073 92 uw_set_remoteSock(ctx, sock);
adam@2073 93
grrwlf@2069 94 qprintf("Handling connection with thread #%d.\n", me);
adamc@138 95
adamc@138 96 while (1) {
adamc@737 97 int r;
adam@1917 98 char *method, *path, *query_string, *headers, *body, *after, *s, *s2;
adamc@737 99
adamc@742 100 if (back - buf == buf_size - 1) {
adamc@737 101 char *new_buf;
grrwlf@2070 102 size_t new_buf_size = buf_size*2;
grrwlf@2070 103 new_buf = realloc(buf, new_buf_size);
grrwlf@2070 104 if(!new_buf) {
grrwlf@2070 105 qfprintf(stderr, "Realloc failed while receiving header\n");
grrwlf@2070 106 close(sock);
grrwlf@2070 107 sock = 0;
grrwlf@2070 108 break;
grrwlf@2070 109 }
grrwlf@2070 110 buf_size = new_buf_size;
adamc@737 111 back = new_buf + (back - buf);
adamc@737 112 buf = new_buf;
adamc@737 113 }
adamc@737 114
adam@1937 115 *back = 0;
adam@1937 116 body = strstr(buf, "\r\n\r\n");
adam@1937 117 if (body == NULL) {
adam@1937 118 r = recv(sock, back, buf_size - 1 - (back - buf), 0);
adamc@138 119
adam@1937 120 if (r < 0) {
grrwlf@2071 121 qfprintf(stderr, "Recv failed while receiving header, retcode %d errno %m\n", r);
adam@1937 122 close(sock);
adam@1937 123 sock = 0;
adam@1937 124 break;
adam@1937 125 }
adam@1937 126
adam@1937 127 if (r == 0) {
grrwlf@2069 128 qprintf("Connection closed.\n");
adam@1937 129 close(sock);
adam@1937 130 sock = 0;
adam@1937 131 break;
adam@1937 132 }
adam@1937 133
adam@1937 134 back += r;
adam@1937 135 *back = 0;
adamc@138 136 }
adamc@138 137
adam@1937 138 if (body != NULL || (body = strstr(buf, "\r\n\r\n"))) {
adamc@853 139 request_result rr;
adam@1917 140 int should_keepalive = 0;
adamc@138 141
adamc@854 142 body[0] = body[1] = 0;
adamc@854 143 body += 4;
adamc@854 144
adamc@854 145 if ((s = strcasestr(buf, "\r\nContent-Length: ")) && s < body) {
adamc@853 146 int clen;
adamc@737 147
adamc@854 148 if (sscanf(s + 18, "%d\r\n", &clen) != 1) {
adamc@853 149 fprintf(stderr, "Malformed Content-Length header\n");
adam@1917 150 close(sock);
adam@1917 151 sock = 0;
adamc@167 152 break;
adamc@167 153 }
adamc@167 154
adamc@854 155 while (back - body < clen) {
adamc@854 156 if (back - buf == buf_size - 1) {
adamc@854 157 char *new_buf;
grrwlf@2070 158 size_t new_buf_size = buf_size * 2;
grrwlf@2070 159 new_buf = realloc(buf, new_buf_size);
grrwlf@2070 160 if(!new_buf) {
grrwlf@2070 161 qfprintf(stderr, "Realloc failed while receiving content\n");
grrwlf@2070 162 close(sock);
grrwlf@2070 163 sock = 0;
grrwlf@2070 164 goto done;
grrwlf@2070 165 }
adamc@854 166
grrwlf@2070 167 buf_size = new_buf_size;
adamc@854 168 back = new_buf + (back - buf);
adamc@854 169 body = new_buf + (body - buf);
adamc@854 170 s = new_buf + (s - buf);
adamc@854 171
adamc@854 172 buf = new_buf;
adamc@854 173 }
adamc@854 174
adamc@854 175 r = recv(sock, back, buf_size - 1 - (back - buf), 0);
adamc@854 176
adamc@854 177 if (r < 0) {
grrwlf@2071 178 qfprintf(stderr, "Recv failed while receiving content, retcode %d errno %m\n", r);
adamc@854 179 close(sock);
adam@1917 180 sock = 0;
adamc@854 181 goto done;
adamc@854 182 }
adamc@854 183
adamc@854 184 if (r == 0) {
grrwlf@2069 185 qfprintf(stderr, "Connection closed.\n");
adamc@854 186 close(sock);
adam@1917 187 sock = 0;
adamc@854 188 goto done;
adamc@854 189 }
adamc@854 190
adamc@854 191 back += r;
adamc@854 192 *back = 0;
adamc@854 193 }
adam@1917 194
adam@1917 195 after = body + clen;
adam@1917 196 } else
adam@1917 197 after = body;
adamc@138 198
adamc@1119 199 body[-4] = '\r';
adamc@1119 200 body[-3] = '\n';
adamc@1119 201
adamc@854 202 if (!(s = strstr(buf, "\r\n"))) {
adamc@854 203 fprintf(stderr, "No newline in request\n");
adamc@854 204 close(sock);
adam@1917 205 sock = 0;
adamc@854 206 goto done;
adamc@854 207 }
adamc@854 208
adamc@1119 209 body[-4] = body[-3] = 0;
adamc@1119 210
adamc@854 211 *s = 0;
adamc@854 212 headers = s + 2;
adamc@854 213 method = s = buf;
adamc@854 214
adam@1426 215 strsep(&s, " ");
adam@1426 216 if (!s) {
adamc@854 217 fprintf(stderr, "No first space in HTTP command\n");
adamc@854 218 close(sock);
adam@1917 219 sock = 0;
adamc@854 220 goto done;
adamc@854 221 }
adamc@854 222 path = s;
adamc@854 223
adamc@1134 224 if ((s = strchr(path, ' ')))
adamc@854 225 *s = 0;
adamc@854 226
adamc@1134 227 if ((s = strchr(path, '?'))) {
adamc@854 228 *s = 0;
adamc@854 229 query_string = s+1;
adamc@854 230 }
adamc@854 231 else
adamc@854 232 query_string = NULL;
adamc@854 233
adamc@854 234 s = headers;
adamc@1134 235 while ((s2 = strchr(s, '\r'))) {
adam@1937 236 if (s2 == s) {
adam@1937 237 *s = 0;
adam@1937 238 break;
adam@1937 239 }
adam@1937 240
adamc@854 241 s = s2;
adamc@854 242
adamc@854 243 if (s[1] == 0)
adamc@854 244 break;
adamc@854 245
adamc@854 246 *s = 0;
adamc@854 247 s += 2;
adamc@854 248 }
adamc@854 249
adamc@854 250 uw_set_headers(ctx, get_header, headers);
adam@1799 251 uw_set_env(ctx, get_env, NULL);
adamc@854 252
grrwlf@2069 253 qprintf("Serving URI %s....\n", path);
adamc@856 254 rr = uw_request(rc, ctx, method, path, query_string, body, back - body,
adamc@856 255 on_success, on_failure,
adamc@856 256 NULL, log_error, log_debug,
adamc@863 257 sock, uw_really_send, close);
adam@1917 258
adam@1915 259 if (rr != KEEP_OPEN) {
adam@1917 260 if (keepalive) {
adam@1917 261 char *connection = uw_Basis_requestHeader(ctx, "Connection");
adam@1917 262
adam@1917 263 should_keepalive = !(connection && !strcmp(connection, "close"));
adam@1917 264 }
adam@1917 265
adam@1917 266 if (!should_keepalive)
adam@1917 267 uw_write_header(ctx, "Connection: close\r\n");
adam@1917 268
adam@1934 269 if (!uw_has_contentLength(ctx)) {
adam@1934 270 char clen[100];
adam@1934 271
adam@1934 272 sprintf(clen, "Content-length: %d\r\n", uw_pagelen(ctx));
adam@1934 273 uw_write_header(ctx, clen);
adam@1934 274 }
adam@1934 275
adam@1915 276 uw_send(ctx, sock);
adam@1915 277 }
adamc@138 278
adam@1917 279 if (rr == SERVED || rr == FAILED) {
adam@1917 280 if (should_keepalive) {
adam@1917 281 // In case any other requests are queued up, shift
adam@1917 282 // unprocessed part of buffer to front.
adam@1917 283 int kept = back - after;
adam@1944 284
adam@1945 285 if (kept == 0) {
adam@1944 286 // No pipelining going on here.
adam@1944 287 // We'd might as well try to switch to a different connection,
adam@1944 288 // while we wait for more input on this one.
adam@1944 289 uw_enqueue(sock);
adam@1944 290 sock = 0;
adam@1944 291 } else {
adam@1944 292 // More input! Move it to the front and continue in this loop.
adam@1944 293 memmove(buf, after, kept);
adam@1944 294 back = buf + kept;
adam@1944 295 }
adam@1917 296 } else {
adam@1917 297 close(sock);
adam@1917 298 sock = 0;
adam@1917 299 }
adam@1931 300 } else if (rr == KEEP_OPEN)
adam@1931 301 sock = 0;
adam@1931 302 else
adamc@853 303 fprintf(stderr, "Illegal uw_request return code: %d\n", rr);
adamc@853 304
adamc@138 305 break;
adamc@138 306 }
adamc@116 307 }
adamc@116 308
adamc@854 309 done:
adamc@311 310 uw_reset(ctx);
adamc@116 311 }
adamc@1138 312
adamc@1138 313 return NULL;
adamc@116 314 }
adamc@102 315
adamc@477 316 static void help(char *cmd) {
adam@2172 317 printf("Usage: %s [-p <port>] [-a <IP address>] [-t <thread count>] [-k] [-q] [-T SEC]\nThe '-k' option turns on HTTP keepalive.\nThe '-q' option turns off some chatter on stdout.\nThe '-T' option sets socket recv timeout (0 disables timeout, default is 5 sec).\n", cmd);
adamc@477 318 }
adamc@477 319
adamc@502 320 static void sigint(int signum) {
adamc@502 321 printf("Exiting....\n");
adamc@502 322 exit(0);
adamc@502 323 }
adamc@502 324
adamc@138 325 int main(int argc, char *argv[]) {
adamc@116 326 // The skeleton for this function comes from Beej's sockets tutorial.
adamc@138 327 int sockfd; // listen on sock_fd
adamc@116 328 struct sockaddr_in my_addr;
adamc@116 329 struct sockaddr_in their_addr; // connector's address information
adamc@1134 330 socklen_t sin_size;
adamc@1134 331 int yes = 1, uw_port = 8080, nthreads = 1, i, *names, opt;
grrwlf@2071 332 int recv_timeout_sec = 5;
adamc@502 333
adamc@502 334 signal(SIGINT, sigint);
adamc@505 335 signal(SIGPIPE, SIG_IGN);
adamc@505 336
adam@1779 337 my_addr.sin_addr.s_addr = INADDR_ANY; // auto-fill with my IP
adam@1779 338 memset(my_addr.sin_zero, '\0', sizeof my_addr.sin_zero);
adam@1779 339
grrwlf@2071 340 while ((opt = getopt(argc, argv, "hp:a:t:kqT:")) != -1) {
adamc@472 341 switch (opt) {
adamc@472 342 case '?':
adam@1930 343 fprintf(stderr, "Unknown command-line option\n");
adamc@477 344 help(argv[0]);
adamc@472 345 return 1;
adamc@138 346
adamc@477 347 case 'h':
adamc@477 348 help(argv[0]);
adamc@477 349 return 0;
adamc@477 350
adamc@472 351 case 'p':
adamc@472 352 uw_port = atoi(optarg);
adamc@472 353 if (uw_port <= 0) {
adamc@472 354 fprintf(stderr, "Invalid port number\n");
adamc@477 355 help(argv[0]);
adamc@472 356 return 1;
adamc@472 357 }
adamc@472 358 break;
adamc@472 359
adam@1779 360 case 'a':
adam@1779 361 if (!inet_pton(AF_INET, optarg, &my_addr.sin_addr)) {
adam@1779 362 fprintf(stderr, "Invalid IP address\n");
adam@1779 363 help(argv[0]);
adam@1779 364 return 1;
adam@1779 365 }
adam@1779 366 break;
adam@1779 367
adamc@472 368 case 't':
adamc@472 369 nthreads = atoi(optarg);
adamc@472 370 if (nthreads <= 0) {
adamc@472 371 fprintf(stderr, "Invalid thread count\n");
adamc@477 372 help(argv[0]);
adamc@472 373 return 1;
adamc@472 374 }
adamc@472 375 break;
adamc@472 376
adam@1917 377 case 'k':
adam@1917 378 keepalive = 1;
adam@1917 379 break;
adam@1917 380
grrwlf@2071 381 case 'T':
grrwlf@2071 382 recv_timeout_sec = atoi(optarg);
grrwlf@2071 383 if (recv_timeout_sec < 0) {
grrwlf@2071 384 fprintf(stderr, "Invalid recv timeout\n");
grrwlf@2071 385 help(argv[0]);
grrwlf@2071 386 return 1;
grrwlf@2071 387 }
grrwlf@2071 388 break;
grrwlf@2071 389
adam@1930 390 case 'q':
adam@1930 391 quiet = 1;
adam@1930 392 break;
adam@1930 393
adamc@472 394 default:
adamc@472 395 fprintf(stderr, "Unexpected getopt() behavior\n");
adamc@472 396 return 1;
adamc@472 397 }
adamc@138 398 }
adamc@138 399
grrwlf@1997 400 uw_request_init(&uw_application, &ls);
adamc@687 401
adamc@138 402 names = calloc(nthreads, sizeof(int));
adamc@116 403
adamc@116 404 sockfd = socket(PF_INET, SOCK_STREAM, 0); // do some error checking!
adamc@116 405
adamc@116 406 if (sockfd < 0) {
adamc@116 407 fprintf(stderr, "Listener socket creation failed\n");
adamc@116 408 return 1;
adamc@116 409 }
adamc@116 410
adamc@116 411 if (setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(int)) < 0) {
adamc@116 412 fprintf(stderr, "Listener socket option setting failed\n");
adamc@116 413 return 1;
adamc@116 414 }
adamc@116 415
adamc@116 416 my_addr.sin_family = AF_INET; // host byte order
adamc@311 417 my_addr.sin_port = htons(uw_port); // short, network byte order
adamc@116 418
adamc@116 419 if (bind(sockfd, (struct sockaddr *)&my_addr, sizeof my_addr) < 0) {
adamc@116 420 fprintf(stderr, "Listener socket bind failed\n");
adamc@116 421 return 1;
adamc@116 422 }
adamc@116 423
adamc@311 424 if (listen(sockfd, uw_backlog) < 0) {
adamc@116 425 fprintf(stderr, "Socket listen failed\n");
adamc@116 426 return 1;
adamc@116 427 }
adamc@116 428
adamc@116 429 sin_size = sizeof their_addr;
adamc@116 430
grrwlf@2069 431 qprintf("Listening on port %d....\n", uw_port);
adamc@116 432
adamc@667 433 {
adamc@667 434 pthread_t thread;
adamc@667 435
grrwlf@1997 436 pruner_data *pd = (pruner_data *)malloc(sizeof(pruner_data));
grrwlf@1997 437 pd->app = &uw_application;
grrwlf@1997 438 pd->loggers = &ls;
grrwlf@1997 439
grrwlf@1997 440 if (pthread_create_big(&thread, NULL, client_pruner, pd)) {
adamc@667 441 fprintf(stderr, "Error creating pruner thread\n");
adamc@667 442 return 1;
adamc@667 443 }
adamc@667 444 }
adamc@667 445
adamc@138 446 for (i = 0; i < nthreads; ++i) {
adamc@138 447 pthread_t thread;
adamc@138 448 names[i] = i;
adam@1522 449 if (pthread_create_big(&thread, NULL, worker, &names[i])) {
adamc@138 450 fprintf(stderr, "Error creating worker thread #%d\n", i);
adamc@138 451 return 1;
adamc@138 452 }
adamc@138 453 }
adamc@138 454
adamc@116 455 while (1) {
adamc@138 456 int new_fd = accept(sockfd, (struct sockaddr *)&their_addr, &sin_size);
adamc@116 457
adamc@116 458 if (new_fd < 0) {
grrwlf@2069 459 qfprintf(stderr, "Socket accept failed\n");
adam@1948 460 } else {
grrwlf@2069 461 qprintf("Accepted connection.\n");
adam@1948 462
adam@1948 463 if (keepalive) {
adam@1948 464 int flag = 1;
adam@1948 465 setsockopt(new_fd, IPPROTO_TCP, TCP_NODELAY, (char *) &flag, sizeof(int));
adam@1948 466 }
adam@1948 467
grrwlf@2071 468 if(recv_timeout_sec>0) {
grrwlf@2071 469 int ret;
grrwlf@2071 470 struct timeval tv;
grrwlf@2071 471 memset(&tv, 0, sizeof(struct timeval));
grrwlf@2071 472 tv.tv_sec = recv_timeout_sec;
grrwlf@2071 473 ret = setsockopt(new_fd, SOL_SOCKET, SO_RCVTIMEO, (char *)&tv, sizeof(struct timeval));
grrwlf@2071 474 if(ret != 0) {
grrwlf@2071 475 qfprintf(stderr, "Timeout setting failed, errcode %d errno '%m'\n", ret);
grrwlf@2071 476 }
grrwlf@2071 477 }
grrwlf@2071 478
adam@1948 479 uw_enqueue(new_fd);
adamc@116 480 }
adamc@116 481 }
adamc@102 482 }
adamc@1121 483
adamc@1121 484 void *uw_init_client_data() {
adamc@1121 485 return NULL;
adamc@1121 486 }
adamc@1121 487
adamc@1121 488 void uw_free_client_data(void *data) {
adamc@1121 489 }
adamc@1121 490
adamc@1121 491 void uw_copy_client_data(void *dst, void *src) {
adamc@1121 492 }
adamc@1121 493
adamc@1121 494 void uw_do_expunge(uw_context ctx, uw_Basis_client cli, void *data) {
adam@1941 495 uw_ensure_transaction(ctx);
adam@1941 496 uw_get_app(ctx)->expunger(ctx, cli);
adam@1941 497
adam@1941 498 if (uw_commit(ctx))
adam@1941 499 uw_error(ctx, UNLIMITED_RETRY, "Rerunning expunge transaction");
adamc@1121 500 }
adamc@1121 501
adamc@1121 502 void uw_post_expunge(uw_context ctx, void *data) {
adamc@1121 503 }
adam@1320 504
adam@1320 505 int uw_supports_direct_status = 1;