asyncresponse.c (33699B)
1 /* Feel free to use this example code in any way 2 you see fit (Public Domain) */ 3 4 /** 5 * @file asyncresponse.c 6 * @brief Example for answering a request from another thread while 7 * the daemon is driven by an external select() loop. The 8 * main thread owns libmicrohttpd and does nothing but poll; 9 * every slow request gets a worker thread of its own and the 10 * connection is suspended for as long as that worker has 11 * nothing to say. 12 * 13 * "GET /" serves a small page whose JavaScript renders the 14 * data as it trickles in, "GET /events" is the stream behind 15 * it (a response is queued immediately and the content reader 16 * suspends between chunks), "GET /slow" suspends in the access 17 * handler until a complete answer is ready, and "GET /fast" is 18 * answered on the spot so that it can be used to show that the 19 * event loop never blocks. 20 * 21 * This example needs POSIX threads. 22 * @author Christian Grothoff 23 */ 24 25 #include <sys/types.h> 26 #include <sys/select.h> 27 #include <sys/socket.h> 28 #include <microhttpd.h> 29 #include <errno.h> 30 #include <pthread.h> 31 #include <signal.h> 32 #include <stdio.h> 33 #include <stdlib.h> 34 #include <string.h> 35 #include <time.h> 36 #include <unistd.h> 37 38 /** 39 * Port to listen on if none was given on the command line. Pass 0 to 40 * let the operating system pick a free port; the port that was 41 * actually bound is then printed on standard output. 42 */ 43 #define DEFAULT_PORT 8888 44 45 /** 46 * Time the worker thread pretends to need for one step, in 47 * milliseconds. 48 */ 49 #define DEFAULT_STEP_MS 250 50 51 /** 52 * Number of steps that make up one job. 53 */ 54 #define DEFAULT_STEPS 20 55 56 /** 57 * Size of the buffer in which a job collects the bytes that the 58 * content reader has not picked up yet. 59 */ 60 #define MAX_PAYLOAD 4096 61 62 /** 63 * Block size we ask MHD to use when it queries our content reader. 64 */ 65 #define IO_BLOCK_SIZE 1024 66 67 /** 68 * How long we are willing to wait for the last connections to go away 69 * when the server is shutting down, in milliseconds. 70 */ 71 #define DRAIN_TIMEOUT_MS 2000 72 73 74 /** 75 * The front page. Its JavaScript opens the "/events" stream and adds 76 * a row for every server-sent event as it arrives, so that the delay 77 * between the steps is plainly visible. The "ping" button fires a 78 * request against "/fast" and reports the round trip time; it stays in 79 * the low milliseconds even while several jobs are running, which is 80 * the whole point of suspending instead of blocking. 81 */ 82 static const char *PAGE = 83 "<!DOCTYPE html>\n" 84 "<html lang='en'>\n" 85 "<head>\n" 86 "<meta charset='utf-8'>\n" 87 "<title>libmicrohttpd: answering from another thread</title>\n" 88 "<style>\n" 89 "body { font-family: sans-serif; max-width: 40em; margin: 2em auto; }\n" 90 "#bar { height: 1em; border: 1px solid #888; margin: 1em 0; }\n" 91 "#fill { height: 100%; width: 0; background: #4a90d9; transition: width .2s; }\n" 92 "#log { font-family: monospace; font-size: 90%; list-style: none; padding: 0; }\n" 93 "#log li { border-bottom: 1px solid #eee; padding: 2px 0; }\n" 94 "button { margin-right: .5em; }\n" 95 "</style>\n" 96 "</head>\n" 97 "<body>\n" 98 "<h1>Answering from another thread</h1>\n" 99 "<p>The server runs a single external <code>select()</code> loop. The job\n" 100 "below is computed by a thread of its own, and the connection carrying it is\n" 101 "suspended whenever that thread has nothing to say. Nothing is buffered: each\n" 102 "row appears the moment the worker produced it.</p>\n" 103 "<button id='go'>Start a job</button>\n" 104 "<button id='ping'>Ping /fast</button>\n" 105 "<div id='bar'><div id='fill'></div></div>\n" 106 "<p id='status'>idle</p>\n" 107 "<ul id='log'></ul>\n" 108 "<script>\n" 109 "const $ = (id) => document.getElementById(id);\n" 110 "let es = null;\n" 111 "let t0 = 0;\n" 112 "const stop = () => { if (es) { es.close(); es = null; } };\n" 113 "$('go').onclick = () => {\n" 114 " stop ();\n" 115 " $('log').innerHTML = '';\n" 116 " $('fill').style.width = '0';\n" 117 " $('status').textContent = 'running';\n" 118 " t0 = performance.now ();\n" 119 " es = new EventSource ('/events');\n" 120 " es.addEventListener ('step', (e) => {\n" 121 " const d = JSON.parse (e.data);\n" 122 " const li = document.createElement ('li');\n" 123 " li.textContent = '+' + Math.round (performance.now () - t0) +\n" 124 " ' ms ' + d.label;\n" 125 " $('log').appendChild (li);\n" 126 " $('fill').style.width = (100 * d.n / d.total) + '%';\n" 127 " });\n" 128 " es.addEventListener ('done', () => {\n" 129 " stop ();\n" 130 " $('status').textContent = 'done after ' +\n" 131 " Math.round (performance.now () - t0) + ' ms';\n" 132 " });\n" 133 " es.onerror = () => { stop (); $('status').textContent = 'connection lost'; };\n" 134 "};\n" 135 "$('ping').onclick = async () => {\n" 136 " const t = performance.now ();\n" 137 " await fetch ('/fast', { cache: 'no-store' });\n" 138 " $('status').textContent = '/fast answered in ' +\n" 139 " Math.round (performance.now () - t) + ' ms';\n" 140 "};\n" 141 "</script>\n" 142 "</body>\n" 143 "</html>\n"; 144 145 146 /** 147 * State shared between the thread that runs libmicrohttpd and the 148 * worker thread that produces the answer. Everything below @e lock is 149 * protected by it. 150 * 151 * The structure is reference counted, with one reference held by MHD 152 * (dropped in #stream_done() or #request_completed()) and one held by 153 * the worker (dropped when the worker function returns). Reaching zero 154 * does not free the job: only the main thread does that, in 155 * #reap_jobs(), because it is also the thread that has to join the 156 * worker. 157 */ 158 struct Job 159 { 160 /** 161 * Kept in a doubly linked list of all jobs, so that the shutdown 162 * code can reach every worker. The list is protected by 163 * #jobs_lock, never by @e lock. 164 */ 165 struct Job *next; 166 167 /** 168 * See @e next. 169 */ 170 struct Job *prev; 171 172 /** 173 * Protects every field below, and---just as importantly---keeps the 174 * connection alive across #MHD_resume_connection(). 175 */ 176 pthread_mutex_t lock; 177 178 /** 179 * Used to wake the worker out of its sleep when the client goes away 180 * or the server is shutting down. 181 */ 182 pthread_cond_t cond; 183 184 /** 185 * The worker thread. Only touched by the main thread, under 186 * #jobs_lock. 187 */ 188 pthread_t tid; 189 190 /** 191 * The connection this job answers. Set to NULL by the MHD side as 192 * soon as MHD is done with the connection; the worker must not touch 193 * it after that, which is why @e abandoned is checked under @e lock 194 * before every #MHD_resume_connection(). 195 */ 196 struct MHD_Connection *connection; 197 198 /** 199 * Bytes produced by the worker that the content reader has not 200 * picked up yet. 201 */ 202 char payload[MAX_PAYLOAD]; 203 204 /** 205 * Number of valid bytes in @e payload. 206 */ 207 size_t fill; 208 209 /** 210 * Number of bytes of @e payload already handed to MHD. 211 */ 212 size_t off; 213 214 /** 215 * Number of references, see the comment on the structure. 216 */ 217 unsigned int rc; 218 219 /** 220 * Non-zero if the answer is streamed as it is produced ("/events"), 221 * zero if the client only ever sees the finished result ("/slow"). 222 * A streaming job resumes the connection after every step, the other 223 * kind only once, at the very end. 224 */ 225 int stream; 226 227 /** 228 * Non-zero once the worker has produced everything it is going to 229 * produce. 230 */ 231 int finished; 232 233 /** 234 * Non-zero while the connection is suspended (or is just about to 235 * be, which under @e lock is the same thing). 236 */ 237 int suspended; 238 239 /** 240 * Non-zero once MHD is done with the connection. The worker must 241 * not call any MHD function on @e connection any more. 242 */ 243 int abandoned; 244 245 /** 246 * Non-zero if the worker should give up early. 247 */ 248 int stop; 249 250 /** 251 * Non-zero once the worker thread was started. Main thread only. 252 */ 253 int started; 254 255 /** 256 * Non-zero once the worker thread was joined. Main thread only. 257 */ 258 int joined; 259 }; 260 261 262 /** 263 * Head of the list of all jobs. 264 */ 265 static struct Job *jobs_head; 266 267 /** 268 * Protects #jobs_head and the list links of every job. Lock order is 269 * #jobs_lock before any `struct Job`'s @e lock, never the other way 270 * round. 271 */ 272 static pthread_mutex_t jobs_lock = PTHREAD_MUTEX_INITIALIZER; 273 274 /** 275 * Set by the signal handler, read by the main loop. 276 */ 277 static volatile sig_atomic_t shutdown_requested; 278 279 /** 280 * Written to by the signal handler so that a blocking select() returns 281 * at once. MHD's own wakeup goes through its inter-thread 282 * communication channel, but our signal is our own business. 283 */ 284 static int wake_pipe[2] = { -1, -1 }; 285 286 /** 287 * Milliseconds the worker sleeps per step. 288 */ 289 static unsigned int step_ms = DEFAULT_STEP_MS; 290 291 /** 292 * Number of steps that make up one job. 293 */ 294 static unsigned int total_steps = DEFAULT_STEPS; 295 296 297 /** 298 * Signal handler for SIGINT and SIGTERM. Does the two things that are 299 * safe to do here: set a flag and poke a pipe. 300 * 301 * @param sig the signal that was received 302 */ 303 static void 304 signal_handler (int sig) 305 { 306 static const char c = 'x'; 307 308 (void) sig; /* Unused. Silent compiler warning. */ 309 shutdown_requested = 1; 310 if (0 > write (wake_pipe[1], 311 &c, 312 1)) 313 { 314 /* Nothing useful can be done about this here. */ 315 } 316 } 317 318 319 /** 320 * Allocate a job and put it on the global list. 321 * 322 * @param connection the connection the job answers 323 * @param stream non-zero to deliver the steps as they are produced 324 * @return the new job with a reference count of two (one for MHD, one 325 * for the worker that the caller is about to start), NULL on 326 * error 327 */ 328 static struct Job * 329 job_create (struct MHD_Connection *connection, 330 int stream) 331 { 332 struct Job *job; 333 334 job = malloc (sizeof (struct Job)); 335 if (NULL == job) 336 return NULL; 337 memset (job, 338 0, 339 sizeof (struct Job)); 340 if (0 != pthread_mutex_init (&job->lock, 341 NULL)) 342 { 343 free (job); 344 return NULL; 345 } 346 if (0 != pthread_cond_init (&job->cond, 347 NULL)) 348 { 349 pthread_mutex_destroy (&job->lock); 350 free (job); 351 return NULL; 352 } 353 job->connection = connection; 354 job->stream = stream; 355 job->rc = 2; 356 pthread_mutex_lock (&jobs_lock); 357 job->next = jobs_head; 358 if (NULL != jobs_head) 359 jobs_head->prev = job; 360 jobs_head = job; 361 pthread_mutex_unlock (&jobs_lock); 362 return job; 363 } 364 365 366 /** 367 * Drop a reference. Note that this never frees the job: the main 368 * thread has to join the worker first, and only it may do that. See 369 * #reap_jobs(). 370 * 371 * @param job the job to release 372 */ 373 static void 374 job_unref (struct Job *job) 375 { 376 pthread_mutex_lock (&job->lock); 377 job->rc--; 378 pthread_mutex_unlock (&job->lock); 379 } 380 381 382 /** 383 * Called by the MHD side once MHD is done with the connection. After 384 * this returns, the worker will not touch the connection again. 385 * 386 * @param job the job to detach from its connection 387 */ 388 static void 389 job_detach (struct Job *job) 390 { 391 pthread_mutex_lock (&job->lock); 392 job->abandoned = 1; 393 job->connection = NULL; 394 /* Wake the worker out of its sleep: if the client hung up in the 395 middle of the transfer there is no point in producing the rest. */ 396 pthread_cond_signal (&job->cond); 397 pthread_mutex_unlock (&job->lock); 398 job_unref (job); 399 } 400 401 402 /** 403 * Resume the connection of @a job if it is suspended. 404 * 405 * Must be called with @a job's lock held, and it deliberately calls 406 * into MHD while holding it: the MHD thread has to take the very same 407 * lock in #job_detach() before it can finish the connection, so as 408 * long as we hold the lock the connection cannot go away underneath 409 * us. The reverse order never occurs---MHD invokes our callbacks 410 * without holding any of its own locks that #MHD_resume_connection() 411 * would need---so this cannot deadlock. 412 * 413 * @param job the job whose connection to resume 414 */ 415 static void 416 job_resume_locked (struct Job *job) 417 { 418 if ( (0 == job->suspended) || 419 (0 != job->abandoned) ) 420 return; 421 job->suspended = 0; 422 MHD_resume_connection (job->connection); 423 } 424 425 426 /** 427 * Sleep for one step, or until the job is woken up. 428 * 429 * @param job the job to sleep on 430 * @return non-zero if the worker should stop early 431 */ 432 static int 433 job_sleep (struct Job *job) 434 { 435 struct timespec ts; 436 int stop; 437 438 if (0 != clock_gettime (CLOCK_REALTIME, 439 &ts)) 440 return 1; 441 ts.tv_sec += (time_t) (step_ms / 1000); 442 ts.tv_nsec += (long) (step_ms % 1000) * 1000000L; 443 if (1000000000L <= ts.tv_nsec) 444 { 445 ts.tv_nsec -= 1000000000L; 446 ts.tv_sec++; 447 } 448 pthread_mutex_lock (&job->lock); 449 while ( (0 == job->stop) && 450 (0 == job->abandoned) ) 451 { 452 if (ETIMEDOUT == pthread_cond_timedwait (&job->cond, 453 &job->lock, 454 &ts)) 455 break; 456 } 457 stop = (0 != job->stop) || (0 != job->abandoned); 458 pthread_mutex_unlock (&job->lock); 459 return stop; 460 } 461 462 463 /** 464 * Append @a len bytes from @a data to the job's payload buffer, 465 * compacting the buffer first. Must be called with @a job's lock 466 * held. A real application would need a policy for the case that the 467 * client cannot keep up; here we simply drop the step, which cannot 468 * happen with the sizes used in this example. 469 * 470 * @param job the job to append to 471 * @param data the bytes to append 472 * @param len the number of bytes to append 473 */ 474 static void 475 job_append_locked (struct Job *job, 476 const char *data, 477 size_t len) 478 { 479 if (0 != job->off) 480 { 481 memmove (job->payload, 482 &job->payload[job->off], 483 job->fill - job->off); 484 job->fill -= job->off; 485 job->off = 0; 486 } 487 if (len > MAX_PAYLOAD - job->fill) 488 return; 489 memcpy (&job->payload[job->fill], 490 data, 491 len); 492 job->fill += len; 493 } 494 495 496 /** 497 * The worker. This stands in for whatever expensive computation, 498 * database cursor or remote query a real application would run: it 499 * produces one server-sent event per step and resumes the connection 500 * whenever it has something new to offer. 501 * 502 * Note that the only MHD function this thread ever calls is 503 * #MHD_resume_connection(). Building and queueing the response is the 504 * business of the thread that runs the daemon. 505 * 506 * @param cls the `struct Job` to work on 507 * @return always NULL 508 */ 509 static void * 510 worker (void *cls) 511 { 512 struct Job *job = cls; 513 char event[256]; 514 unsigned int i; 515 int len; 516 517 for (i = 1; i <= total_steps; i++) 518 { 519 if (0 != job_sleep (job)) 520 break; 521 if (0 != job->stream) 522 len = snprintf (event, 523 sizeof (event), 524 "event: step\ndata: {\"n\":%u,\"total\":%u," 525 "\"label\":\"step %u of %u done\"}\n\n", 526 i, 527 total_steps, 528 i, 529 total_steps); 530 else 531 len = snprintf (event, 532 sizeof (event), 533 "step %u of %u done\n", 534 i, 535 total_steps); 536 if ( (0 >= len) || 537 (sizeof (event) <= (size_t) len) ) 538 break; 539 pthread_mutex_lock (&job->lock); 540 job_append_locked (job, 541 event, 542 (size_t) len); 543 /* Only a streaming job has anything to show yet. The other kind 544 stays suspended until the very last step. */ 545 if (0 != job->stream) 546 job_resume_locked (job); 547 pthread_mutex_unlock (&job->lock); 548 } 549 if (0 != job->stream) 550 len = snprintf (event, 551 sizeof (event), 552 "event: done\ndata: {\"n\":%u,\"total\":%u}\n\n", 553 total_steps, 554 total_steps); 555 else 556 len = snprintf (event, 557 sizeof (event), 558 "all %u steps done\n", 559 total_steps); 560 pthread_mutex_lock (&job->lock); 561 if ( (0 < len) && 562 (sizeof (event) > (size_t) len) ) 563 job_append_locked (job, 564 event, 565 (size_t) len); 566 job->finished = 1; 567 /* The last resume is not optional: if the connection were left 568 suspended, nothing would ever wake it up again, and the daemon 569 could not be stopped. */ 570 job_resume_locked (job); 571 pthread_mutex_unlock (&job->lock); 572 job_unref (job); 573 return NULL; 574 } 575 576 577 /** 578 * Start the worker thread of @a job. 579 * 580 * @param job the job whose worker to start 581 * @return #MHD_YES on success 582 */ 583 static enum MHD_Result 584 job_start (struct Job *job) 585 { 586 pthread_mutex_lock (&jobs_lock); 587 if (0 != pthread_create (&job->tid, 588 NULL, 589 &worker, 590 job)) 591 { 592 pthread_mutex_unlock (&jobs_lock); 593 return MHD_NO; 594 } 595 job->started = 1; 596 pthread_mutex_unlock (&jobs_lock); 597 return MHD_YES; 598 } 599 600 601 /** 602 * Content reader for the "/events" stream. This is the interesting 603 * one: when the worker has produced nothing since the last call we 604 * suspend the connection and return zero, instead of returning zero on 605 * its own---which would make MHD ask again immediately and burn a CPU 606 * core for the whole duration of the transfer. 607 * 608 * @param cls our `struct Job` 609 * @param pos number of bytes already returned for this response 610 * @param buf where to copy the data 611 * @param max maximum number of bytes to copy to @a buf 612 * @return number of bytes written to @a buf, 0 if the connection was 613 * suspended, MHD_CONTENT_READER_END_OF_STREAM at the end 614 */ 615 static ssize_t 616 stream_reader (void *cls, 617 uint64_t pos, 618 char *buf, 619 size_t max) 620 { 621 struct Job *job = cls; 622 size_t ready; 623 624 (void) pos; /* Unused. Silent compiler warning. */ 625 pthread_mutex_lock (&job->lock); 626 if (job->off == job->fill) 627 { 628 job->off = 0; 629 job->fill = 0; 630 if (0 != job->finished) 631 { 632 pthread_mutex_unlock (&job->lock); 633 return MHD_CONTENT_READER_END_OF_STREAM; 634 } 635 /* Nothing to send yet. Take the connection out of the event loop; 636 the worker will put it back in. */ 637 job->suspended = 1; 638 MHD_suspend_connection (job->connection); 639 pthread_mutex_unlock (&job->lock); 640 return 0; 641 } 642 ready = job->fill - job->off; 643 if (ready > max) 644 ready = max; 645 memcpy (buf, 646 &job->payload[job->off], 647 ready); 648 job->off += ready; 649 pthread_mutex_unlock (&job->lock); 650 return (ssize_t) ready; 651 } 652 653 654 /** 655 * Called by MHD when the response object of an "/events" request is 656 * destroyed, which happens for a completed transfer just as well as 657 * for a client that went away in the middle of one. This is the MHD 658 * side of the job's reference count. 659 * 660 * @param cls our `struct Job` 661 */ 662 static void 663 stream_done (void *cls) 664 { 665 job_detach (cls); 666 } 667 668 669 /** 670 * Handle "GET /events": queue a streaming response right away and let 671 * the content reader deal with the fact that the data does not exist 672 * yet. 673 * 674 * @param connection the connection to answer 675 * @return #MHD_YES on success 676 */ 677 static enum MHD_Result 678 handle_events (struct MHD_Connection *connection) 679 { 680 struct MHD_Response *response; 681 struct Job *job; 682 enum MHD_Result ret; 683 684 job = job_create (connection, 685 1); 686 if (NULL == job) 687 return MHD_NO; 688 response = MHD_create_response_from_callback (MHD_SIZE_UNKNOWN, 689 IO_BLOCK_SIZE, 690 &stream_reader, 691 job, 692 &stream_done); 693 if (NULL == response) 694 { 695 /* No response object exists, so nobody will call stream_done() for 696 us; drop both references by hand. The worker was never 697 started. */ 698 job_unref (job); 699 job_unref (job); 700 return MHD_NO; 701 } 702 if (MHD_NO == job_start (job)) 703 { 704 MHD_destroy_response (response); 705 job_unref (job); 706 return MHD_NO; 707 } 708 (void) MHD_add_response_header (response, 709 MHD_HTTP_HEADER_CONTENT_TYPE, 710 "text/event-stream"); 711 (void) MHD_add_response_header (response, 712 MHD_HTTP_HEADER_CACHE_CONTROL, 713 "no-cache"); 714 ret = MHD_queue_response (connection, 715 MHD_HTTP_OK, 716 response); 717 MHD_destroy_response (response); 718 return ret; 719 } 720 721 722 /** 723 * Handle "GET /slow": the answer is only useful as a whole, so instead 724 * of streaming we suspend the connection until the worker is done. 725 * MHD then calls this function again---on its own thread---and that is 726 * where the response is queued. #MHD_queue_response() is never called 727 * from the worker. 728 * 729 * @param connection the connection to answer 730 * @param req_cls the per-request pointer, holding our `struct Job` 731 * @return #MHD_YES on success 732 */ 733 static enum MHD_Result 734 handle_slow (struct MHD_Connection *connection, 735 void **req_cls) 736 { 737 struct MHD_Response *response; 738 struct Job *job = *req_cls; 739 char answer[MAX_PAYLOAD]; 740 size_t len; 741 enum MHD_Result ret; 742 743 if (NULL == job) 744 { 745 /* First call for this request: start the work and ask MHD to come 746 back to us. */ 747 job = job_create (connection, 748 0); 749 if (NULL == job) 750 return MHD_NO; 751 if (MHD_NO == job_start (job)) 752 { 753 job_unref (job); 754 job_unref (job); 755 return MHD_NO; 756 } 757 *req_cls = job; 758 return MHD_YES; 759 } 760 pthread_mutex_lock (&job->lock); 761 if (0 == job->finished) 762 { 763 /* Still working. Park the connection; the worker resumes it when 764 it is done, and MHD will then enter this function once more. */ 765 job->suspended = 1; 766 MHD_suspend_connection (connection); 767 pthread_mutex_unlock (&job->lock); 768 return MHD_YES; 769 } 770 len = job->fill - job->off; 771 if (len > sizeof (answer)) 772 len = sizeof (answer); 773 memcpy (answer, 774 &job->payload[job->off], 775 len); 776 pthread_mutex_unlock (&job->lock); 777 778 response = MHD_create_response_from_buffer_copy (len, 779 answer); 780 if (NULL == response) 781 return MHD_NO; 782 (void) MHD_add_response_header (response, 783 MHD_HTTP_HEADER_CONTENT_TYPE, 784 "text/plain"); 785 ret = MHD_queue_response (connection, 786 MHD_HTTP_OK, 787 response); 788 MHD_destroy_response (response); 789 return ret; 790 } 791 792 793 /** 794 * Queue a complete response that is already sitting in memory. 795 * 796 * @param connection the connection to answer 797 * @param status the HTTP status code to use 798 * @param mime the value for the "Content-Type" header 799 * @param body the body to send 800 * @return #MHD_YES on success 801 */ 802 static enum MHD_Result 803 queue_static (struct MHD_Connection *connection, 804 unsigned int status, 805 const char *mime, 806 const char *body) 807 { 808 struct MHD_Response *response; 809 enum MHD_Result ret; 810 811 response = MHD_create_response_from_buffer_copy (strlen (body), 812 body); 813 if (NULL == response) 814 return MHD_NO; 815 (void) MHD_add_response_header (response, 816 MHD_HTTP_HEADER_CONTENT_TYPE, 817 mime); 818 (void) MHD_add_response_header (response, 819 MHD_HTTP_HEADER_CACHE_CONTROL, 820 "no-store"); 821 ret = MHD_queue_response (connection, 822 status, 823 response); 824 MHD_destroy_response (response); 825 return ret; 826 } 827 828 829 static enum MHD_Result 830 answer_to_connection (void *cls, 831 struct MHD_Connection *connection, 832 const char *url, 833 const char *method, 834 const char *version, 835 const char *upload_data, 836 size_t *upload_data_size, 837 void **req_cls) 838 { 839 (void) cls; /* Unused. Silent compiler warning. */ 840 (void) version; /* Unused. Silent compiler warning. */ 841 (void) upload_data; /* Unused. Silent compiler warning. */ 842 (void) upload_data_size; /* Unused. Silent compiler warning. */ 843 844 if (0 != strcmp (method, 845 MHD_HTTP_METHOD_GET)) 846 return MHD_NO; 847 if (0 == strcmp (url, 848 "/")) 849 return queue_static (connection, 850 MHD_HTTP_OK, 851 "text/html", 852 PAGE); 853 if (0 == strcmp (url, 854 "/fast")) 855 return queue_static (connection, 856 MHD_HTTP_OK, 857 "text/plain", 858 "pong\n"); 859 if ( (0 == strcmp (url, 860 "/events")) || 861 (0 == strcmp (url, 862 "/slow")) ) 863 { 864 if (0 != shutdown_requested) 865 return queue_static (connection, 866 MHD_HTTP_SERVICE_UNAVAILABLE, 867 "text/plain", 868 "shutting down\n"); 869 if (0 == strcmp (url, 870 "/events")) 871 return handle_events (connection); 872 return handle_slow (connection, 873 req_cls); 874 } 875 return queue_static (connection, 876 MHD_HTTP_NOT_FOUND, 877 "text/plain", 878 "not found\n"); 879 } 880 881 882 /** 883 * Called by MHD when a request is done. For "/slow" this is where the 884 * MHD side of the reference count is dropped; the other handlers leave 885 * @a req_cls alone, so there is nothing to do for them. 886 * 887 * @param cls closure, unused 888 * @param connection the connection that finished 889 * @param req_cls the per-request pointer 890 * @param toe why the request ended 891 */ 892 static void 893 request_completed (void *cls, 894 struct MHD_Connection *connection, 895 void **req_cls, 896 enum MHD_RequestTerminationCode toe) 897 { 898 struct Job *job = *req_cls; 899 900 (void) cls; /* Unused. Silent compiler warning. */ 901 (void) connection; /* Unused. Silent compiler warning. */ 902 (void) toe; /* Unused. Silent compiler warning. */ 903 if (NULL == job) 904 return; 905 *req_cls = NULL; 906 job_detach (job); 907 } 908 909 910 /** 911 * Free every job that nobody references any more, joining its worker 912 * on the way. Called from the main loop, and thus always from the 913 * thread that also runs the daemon. 914 */ 915 static void 916 reap_jobs (void) 917 { 918 struct Job *job; 919 struct Job *next; 920 unsigned int rc; 921 922 pthread_mutex_lock (&jobs_lock); 923 for (job = jobs_head; NULL != job; job = next) 924 { 925 next = job->next; 926 pthread_mutex_lock (&job->lock); 927 rc = job->rc; 928 pthread_mutex_unlock (&job->lock); 929 if (0 != rc) 930 continue; 931 if ( (0 != job->started) && 932 (0 == job->joined) ) 933 { 934 pthread_join (job->tid, 935 NULL); 936 job->joined = 1; 937 } 938 if (NULL != job->prev) 939 job->prev->next = job->next; 940 else 941 jobs_head = job->next; 942 if (NULL != job->next) 943 job->next->prev = job->prev; 944 pthread_cond_destroy (&job->cond); 945 pthread_mutex_destroy (&job->lock); 946 free (job); 947 } 948 pthread_mutex_unlock (&jobs_lock); 949 } 950 951 952 /** 953 * Ask every worker to stop and wait until they all did. 954 * 955 * This has to happen before the daemon is stopped: a worker that is 956 * still asleep may be holding the only promise to resume a suspended 957 * connection, and stopping a daemon that has suspended connections is 958 * an API violation. Once every worker has returned, no connection can 959 * be suspended any more, because the workers resume before they exit 960 * and the content reader only ever suspends while a worker is running. 961 */ 962 static void 963 stop_all_jobs (void) 964 { 965 struct Job *job; 966 967 pthread_mutex_lock (&jobs_lock); 968 for (job = jobs_head; NULL != job; job = job->next) 969 { 970 pthread_mutex_lock (&job->lock); 971 job->stop = 1; 972 pthread_cond_signal (&job->cond); 973 pthread_mutex_unlock (&job->lock); 974 } 975 for (job = jobs_head; NULL != job; job = job->next) 976 { 977 if ( (0 != job->started) && 978 (0 == job->joined) ) 979 { 980 pthread_join (job->tid, 981 NULL); 982 job->joined = 1; 983 } 984 } 985 pthread_mutex_unlock (&jobs_lock); 986 } 987 988 989 /** 990 * Number of connections the daemon is currently handling. 991 * 992 * @param daemon the daemon to query 993 * @return the number of connections, 0 if it cannot be determined 994 */ 995 static unsigned int 996 connection_count (struct MHD_Daemon *daemon) 997 { 998 const union MHD_DaemonInfo *info; 999 1000 info = MHD_get_daemon_info (daemon, 1001 MHD_DAEMON_INFO_CURRENT_CONNECTIONS); 1002 if (NULL == info) 1003 return 0; 1004 return info->num_connections; 1005 } 1006 1007 1008 int 1009 main (int argc, 1010 char *const *argv) 1011 { 1012 struct MHD_Daemon *daemon; 1013 const union MHD_DaemonInfo *info; 1014 struct sigaction sa; 1015 fd_set rs; 1016 fd_set ws; 1017 fd_set es; 1018 struct timeval tv; 1019 struct timeval *tvp; 1020 MHD_socket max; 1021 uint64_t mhd_timeout; 1022 unsigned long port = DEFAULT_PORT; 1023 unsigned int drained_ms = 0; 1024 char drain_buf[64]; 1025 int draining = 0; 1026 int ret = 0; 1027 1028 if (1 < argc) 1029 port = strtoul (argv[1], 1030 NULL, 1031 10); 1032 if (2 < argc) 1033 step_ms = (unsigned int) strtoul (argv[2], 1034 NULL, 1035 10); 1036 if (3 < argc) 1037 total_steps = (unsigned int) strtoul (argv[3], 1038 NULL, 1039 10); 1040 if ( (65535 < port) || 1041 (0 == total_steps) ) 1042 { 1043 fprintf (stderr, 1044 "Usage: %s [PORT [STEP_MS [STEPS]]]\n" 1045 "Pass 0 as PORT to let the system pick a free one.\n", 1046 argv[0]); 1047 return 1; 1048 } 1049 1050 if (0 > pipe (wake_pipe)) 1051 { 1052 fprintf (stderr, 1053 "Failed to create wakeup pipe: %s\n", 1054 strerror (errno)); 1055 return 1; 1056 } 1057 memset (&sa, 1058 0, 1059 sizeof (sa)); 1060 sa.sa_handler = &signal_handler; 1061 sigaction (SIGINT, 1062 &sa, 1063 NULL); 1064 sigaction (SIGTERM, 1065 &sa, 1066 NULL); 1067 sa.sa_handler = SIG_IGN; 1068 sigaction (SIGPIPE, 1069 &sa, 1070 NULL); 1071 1072 /* No MHD_USE_INTERNAL_POLLING_THREAD: the loop below is ours. 1073 MHD_ALLOW_SUSPEND_RESUME implies MHD_USE_ITC, and that channel is 1074 what lets a worker thread break us out of the select() call. */ 1075 daemon = MHD_start_daemon (MHD_ALLOW_SUSPEND_RESUME | MHD_USE_ERROR_LOG, 1076 (uint16_t) port, 1077 NULL, NULL, 1078 &answer_to_connection, NULL, 1079 MHD_OPTION_NOTIFY_COMPLETED, 1080 &request_completed, NULL, 1081 MHD_OPTION_END); 1082 if (NULL == daemon) 1083 { 1084 fprintf (stderr, 1085 "Failed to start daemon.\n"); 1086 return 1; 1087 } 1088 info = MHD_get_daemon_info (daemon, 1089 MHD_DAEMON_INFO_BIND_PORT); 1090 if ( (NULL == info) || 1091 (0 == info->port) ) 1092 { 1093 fprintf (stderr, 1094 "Failed to determine bound port.\n"); 1095 MHD_stop_daemon (daemon); 1096 return 1; 1097 } 1098 printf ("Listening on port %u\n", 1099 (unsigned int) info->port); 1100 fflush (stdout); 1101 1102 while (1) 1103 { 1104 if ( (0 != shutdown_requested) && 1105 (0 == draining) ) 1106 { 1107 /* Every worker has to be gone before the daemon may be stopped. */ 1108 stop_all_jobs (); 1109 draining = 1; 1110 } 1111 if ( (0 != draining) && 1112 ( (0 == connection_count (daemon)) || 1113 (DRAIN_TIMEOUT_MS <= drained_ms) ) ) 1114 break; 1115 1116 FD_ZERO (&rs); 1117 FD_ZERO (&ws); 1118 FD_ZERO (&es); 1119 max = 0; 1120 if (MHD_YES != MHD_get_fdset (daemon, 1121 &rs, 1122 &ws, 1123 &es, 1124 &max)) 1125 { 1126 fprintf (stderr, 1127 "Failed to obtain the file descriptor set.\n"); 1128 ret = 1; 1129 break; 1130 } 1131 /* Our own descriptors go into the very same sets. */ 1132 FD_SET (wake_pipe[0], 1133 &rs); 1134 if (max < wake_pipe[0]) 1135 max = wake_pipe[0]; 1136 1137 if (0 != draining) 1138 { 1139 /* Look at the drain deadline regularly. */ 1140 tv.tv_sec = 0; 1141 tv.tv_usec = 50 * 1000; 1142 tvp = &tv; 1143 drained_ms += 50; 1144 } 1145 else if (MHD_YES == MHD_get_timeout64 (daemon, 1146 &mhd_timeout)) 1147 { 1148 tv.tv_sec = (time_t) (mhd_timeout / 1000); 1149 tv.tv_usec = ((long) (mhd_timeout % 1000)) * 1000; 1150 tvp = &tv; 1151 } 1152 else 1153 { 1154 /* MHD has nothing to wait for. With every connection suspended 1155 this is the normal case, and the loop simply sleeps until a 1156 worker resumes one---which reaches us through MHD's ITC, as it 1157 is part of the read set above. */ 1158 tvp = NULL; 1159 } 1160 1161 if (-1 == select ((int) max + 1, 1162 &rs, 1163 &ws, 1164 &es, 1165 tvp)) 1166 { 1167 if (EINTR == errno) 1168 continue; 1169 fprintf (stderr, 1170 "Aborting due to error during select: %s\n", 1171 strerror (errno)); 1172 ret = 1; 1173 break; 1174 } 1175 if (FD_ISSET (wake_pipe[0], 1176 &rs)) 1177 (void) read (wake_pipe[0], 1178 drain_buf, 1179 sizeof (drain_buf)); 1180 MHD_run_from_select (daemon, 1181 &rs, 1182 &ws, 1183 &es); 1184 reap_jobs (); 1185 } 1186 1187 stop_all_jobs (); 1188 reap_jobs (); 1189 MHD_stop_daemon (daemon); 1190 close (wake_pipe[0]); 1191 close (wake_pipe[1]); 1192 return ret; 1193 }