Mercurial > urweb

annotate src/c/http.c @ 2033:ea0ecd5fa9df

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