libmicrohttpd

asyncresponses.inc (19104B)

---

 The previous chapter left one question open.  A content reader that
 has no data ready may return @code{0}, but doing so turns the daemon
 into a busy loop, and the honest answer---suspend the connection---was
 only sketched there.  This chapter builds the thing itself: a server
 whose slow work happens in threads of its own, whose connections are
 parked while that work is under way, and which is driven by an
 external @code{select()} loop so that the application stays in charge
 of its own event handling.
 
 The combination is a common one.  Most requests are cheap and can be
 answered on the spot; a few of them start a report, a backup or a
 query that runs for seconds.  Handling the slow ones inline would
 block every other client for as long as they take.  Giving every
 connection its own thread---@code{MHD_USE_THREAD_PER_CONNECTION}---is
 no answer either, both because it scales badly and because suspending
 is not available in that mode at all.  What is wanted is a single
 event loop that never blocks, plus a worker per slow job.
 
 
 @heading Who may call what
 
 Almost every difficulty in this design comes from one rule, so it is
 worth stating before any code:
 
 @table @asis
 @item @code{MHD_queue_response()}
 Must be called from the thread that runs the daemon, from inside the
 access handler.  A worker thread must never queue a response.
 
 @item @code{MHD_suspend_connection()}
 May only be called from the access handler or from a content reader
 callback---that is, from @emph{MHD}'s own thread, while @emph{MHD} is
 calling into the application.
 
 @item @code{MHD_resume_connection()}
 May be called from any thread at any time, and this is the exception
 that makes the whole design work.  It is the only @emph{MHD} function
 a worker thread ever needs.
 @end table
 
 @noindent
 So the split is: the worker computes and, when it has something,
 resumes.  Everything that touches the connection or the response
 happens on the daemon's thread, where @emph{MHD} calls the application
 back.  The worker and the daemon share exactly one object, a
 per-request context, and the only thing the worker does with the
 connection handle it holds is pass it to
 @code{MHD_resume_connection()}.
 
 The daemon has to be started with @code{MHD_ALLOW_SUSPEND_RESUME}; both
 suspend and resume call @code{MHD_PANIC()} and abort the process
 otherwise.  Note also that suspending does not work with a thread pool:
 @code{MHD_resume_connection()} asserts that the daemon has none.
 
 
 @heading The event loop
 
 Leaving out @code{MHD_USE_INTERNAL_POLLING_THREAD} puts @emph{MHD}
 into external polling mode: it starts no thread of its own and does
 nothing until the application asks it to.  The loop is the usual one,
 and the example adds its own descriptor to the very same sets---there
 is nothing special about @emph{MHD}'s:
 
 @verbatim
   FD_ZERO (&rs); FD_ZERO (&ws); FD_ZERO (&es);
   max = 0;
   if (MHD_YES != MHD_get_fdset (daemon, &rs, &ws, &es, &max))
     break;                      /* fatal internal error */
   FD_SET (wake_pipe[0], &rs);   /* our own wakeup pipe */
   if (max < wake_pipe[0])
     max = wake_pipe[0];
 
   if (MHD_YES == MHD_get_timeout64 (daemon, &mhd_timeout))
     { ...fill tv...; tvp = &tv; }
   else
     tvp = NULL;                 /* nothing to wait for */
 
   select ((int) max + 1, &rs, &ws, &es, tvp);
   MHD_run_from_select (daemon, &rs, &ws, &es);
 @end verbatim
 @noindent
 
 The interesting branch is the one that sets @code{tvp} to @code{NULL}.
 @code{MHD_get_timeout64()} returns @code{MHD_NO} when @emph{MHD} has
 nothing whatsoever to wait for, and with every connection suspended
 that is precisely the situation: suspended connections do not time
 out, so there is no deadline to wake up for.  The loop then blocks in
 @code{select()} indefinitely.  Whether the server works at all depends
 on something waking it up.
 
 
 @heading Waking the loop up
 
 That something is @emph{MHD}'s inter-thread communication channel.
 @code{MHD_ALLOW_SUSPEND_RESUME} is defined as
 @code{8192 | MHD_USE_ITC}, so requesting suspend support requests the
 ITC as well; @code{MHD_get_fdset()} puts the reading end of that
 channel into the read set, ahead of everything else; and
 @code{MHD_resume_connection()} writes a byte to it.  A worker thread
 that resumes a connection therefore breaks the main thread out of
 @code{select()}, even though the two share nothing but the daemon
 handle.
 
 This deserves emphasis because the header file is misleading about it:
 the comment on @code{MHD_USE_ITC} says the flag is ignored with
 external polling.  That is true for the flag in isolation, but not for
 the channel: @code{MHD_ALLOW_SUSPEND_RESUME} creates it regardless of
 the polling mode, and in external mode it is the application's own
 @code{select()} that watches it.  Without that, a resumed connection
 would sit untouched until some unrelated event happened to wake the
 loop.
 
 There is no race to guard against here.  If the worker resumes in the
 window between @code{MHD_get_fdset()} and @code{select()}, the byte is
 already in the channel when @code{select()} is entered, and
 @code{select()} returns immediately---the channel is level triggered,
 so a wakeup cannot be missed.  The same property makes it safe for a
 worker to resume a connection that has not finished suspending yet:
 @emph{MHD} remembers the request and cancels the pending suspension
 instead of parking the connection.  Neither side needs to know what
 the other is doing.
 
 
 @heading Answering in one piece
 
 The simplest of the two patterns is for an answer that is only useful
 as a whole---a generated report, the result of a query.  Nothing can be
 sent before it is finished, so the connection is suspended in the
 access handler and the response is queued when @emph{MHD} comes back:
 
 @verbatim
 static enum MHD_Result
 handle_slow (struct MHD_Connection *connection, void **req_cls)
 {
   struct Job *job = *req_cls;
 
   if (NULL == job)
     {
       /* First call for this request: start the work and ask MHD to
          come back to us. */
       job = job_create (connection, 0);
       if (NULL == job)
         return MHD_NO;
       if (MHD_NO == job_start (job))
         { job_unref (job); job_unref (job); return MHD_NO; }
       *req_cls = job;
       return MHD_YES;
     }
   pthread_mutex_lock (&job->lock);
   if (0 == job->finished)
     {
       job->suspended = 1;
       MHD_suspend_connection (connection);
       pthread_mutex_unlock (&job->lock);
       return MHD_YES;
     }
   ...copy the result out from under the lock...
   pthread_mutex_unlock (&job->lock);
   ...MHD_create_response_from_buffer_copy() and MHD_queue_response()...
 }
 @end verbatim
 @noindent
 
 The control flow is worth following, because it is not obvious that it
 terminates.  @emph{MHD} calls the access handler for a request more
 than once; the first call is the one where @code{*req_cls} is still
 @code{NULL}, and returning @code{MHD_YES} without queueing anything
 tells @emph{MHD} to carry on.  On the second call the worker is
 running but not finished, so the connection is suspended and the
 handler returns again.  When the worker resumes the connection,
 @emph{MHD} calls the handler a third time---on its own thread, with
 the same @code{*req_cls}---and this time @code{finished} is set and the
 response is queued.
 
 That third call is the answer to the question this pattern exists for.
 The worker never queues anything; it merely makes the daemon call the
 application back at a moment when the answer is ready.
 
 
 @heading Answering as it is produced
 
 The second pattern is the more useful one when the answer is long, and
 it is what makes a browser show results as they appear rather than
 after a long blank pause.  Here the response is queued
 @emph{immediately}, with a content reader and an unknown size, so the
 headers go out at once and the body follows as the worker fills it in:
 
 @verbatim
   job = job_create (connection, 1);
   response = MHD_create_response_from_callback (MHD_SIZE_UNKNOWN,
                                                 IO_BLOCK_SIZE,
                                                 &stream_reader,
                                                 job,
                                                 &stream_done);
   ...
   MHD_add_response_header (response, MHD_HTTP_HEADER_CONTENT_TYPE,
                            "text/event-stream");
   ret = MHD_queue_response (connection, MHD_HTTP_OK, response);
   MHD_destroy_response (response);
 @end verbatim
 @noindent
 
 Because the size is unknown, @emph{MHD} uses chunked transfer encoding,
 and every piece the content reader hands back is framed as its own
 chunk and goes out on the wire straight away.  The content reader is
 where suspending happens:
 
 @verbatim
 static ssize_t
 stream_reader (void *cls, uint64_t pos, char *buf, size_t max)
 {
   struct Job *job = cls;
   size_t ready;
 
   pthread_mutex_lock (&job->lock);
   if (job->off == job->fill)
     {
       job->off = 0;
       job->fill = 0;
       if (0 != job->finished)
         {
           pthread_mutex_unlock (&job->lock);
           return MHD_CONTENT_READER_END_OF_STREAM;
         }
       /* Nothing to send yet.  Take the connection out of the event
          loop; the worker will put it back in. */
       job->suspended = 1;
       MHD_suspend_connection (job->connection);
       pthread_mutex_unlock (&job->lock);
       return 0;
     }
   ready = job->fill - job->off;
   if (ready > max)
     ready = max;
   memcpy (buf, &job->payload[job->off], ready);
   job->off += ready;
   pthread_mutex_unlock (&job->lock);
   return (ssize_t) ready;
 }
 @end verbatim
 @noindent
 
 The return value is still @code{0}, exactly as in the busy-waiting
 version the previous chapter warned about.  The difference is the line
 above it.  Suspending removes the connection from the event loop, so
 @emph{MHD} has no reason to come back until the worker says so, and
 the process goes to sleep.  The example ships with a test that measures
 this: streaming a document that takes one second to produce costs the
 server no measurable CPU time at all while the connection is suspended,
 and very nearly a full second---one core, saturated---when the
 @code{MHD_suspend_connection()} call is removed and only the
 @code{return 0} is left.  The bytes that arrive are identical in both
 cases, which is what makes this such an easy mistake to keep.
 
 
 @heading The shared state
 
 Everything the two threads share sits in one structure per request:
 
 @verbatim
 struct Job
 {
   pthread_mutex_t lock;
   pthread_cond_t cond;
   struct MHD_Connection *connection;
   char payload[MAX_PAYLOAD];
   size_t fill;
   size_t off;
   unsigned int rc;
   int finished;
   int suspended;
   int abandoned;
   ...
 };
 @end verbatim
 @noindent
 
 There is no registry of pending requests involved in answering them,
 and no lookup: the pointer travels in @code{*req_cls} for the first
 pattern and in the content reader's @code{crc_cls} for the second, and
 both the daemon and the worker are handed it directly.  A shared list
 of all jobs is still useful, but only for shutting down in an orderly
 fashion, and it is protected by a lock of its own that is always taken
 before any job's lock, never after.
 
 The structure is reference counted, with one reference held by
 @emph{MHD} and one by the worker.  The @emph{MHD} side lets go in the
 free callback of the response for a streamed answer, and in the
 @code{MHD_OPTION_NOTIFY_COMPLETED} callback for the other pattern; the
 worker lets go when it returns.  Dropping the last reference does not
 free the job, though, because the worker still has to be joined and
 only the main thread may do that.  The main loop therefore ends with a
 sweep that joins and frees whatever has no references left.
 
 
 @heading When the client goes away
 
 This is the part that is easy to get wrong, because it does not happen
 during development and does happen constantly in production.  A client
 that closes the connection halfway through a download leaves a worker
 running with a pointer to a connection @emph{MHD} is about to free.
 
 The two callbacks that tell the application a request is over---the
 response's free callback and @code{MHD_OPTION_NOTIFY_COMPLETED}---are
 therefore where the worker is cut loose:
 
 @verbatim
 static void
 job_detach (struct Job *job)
 {
   pthread_mutex_lock (&job->lock);
   job->abandoned = 1;
   job->connection = NULL;
   pthread_cond_signal (&job->cond);
   pthread_mutex_unlock (&job->lock);
   job_unref (job);
 }
 @end verbatim
 @noindent
 
 and the worker checks that flag before every resume:
 
 @verbatim
 static void
 job_resume_locked (struct Job *job)
 {
   if ( (0 == job->suspended) || (0 != job->abandoned) )
     return;
   job->suspended = 0;
   MHD_resume_connection (job->connection);
 }
 @end verbatim
 @noindent
 
 Note that this function is called with the job's lock @emph{held}, and
 calls into @emph{MHD} without releasing it.  That is deliberate and it
 is what makes the pattern safe.  @emph{MHD} destroys the connection
 only after the completion callback has returned, and that callback
 cannot run while the worker holds the lock; so for as long as the
 worker is inside @code{MHD_resume_connection()}, the connection it
 passes is guaranteed to still exist.  Dropping the lock first and
 resuming afterwards would open exactly the window this is meant to
 close.
 
 The reverse order never occurs: @emph{MHD} invokes the free and
 completion callbacks without holding the internal locks that
 @code{MHD_resume_connection()} acquires, so there is no way for the
 two to deadlock.
 
 The condition variable in the same function is a convenience rather
 than a necessity.  Signalling it wakes a worker that is asleep between
 steps, so that the work stops within microseconds of the client
 disappearing instead of running to completion for nobody.
 
 
 @heading Shutting down
 
 Stopping a daemon that still has suspended connections is an API
 violation, and a suspended connection is one that only a worker can
 release.  The order is therefore fixed:
 
 @enumerate
 @item
 Stop handing out new jobs.
 
 @item
 Tell every worker to give up, and @emph{join} them.  After this point
 no connection can be suspended any more: the workers resume before
 they exit, and the content reader only ever suspends while a worker is
 running.
 
 @item
 Keep running the event loop until the connections have drained, so
 that the last bytes actually reach the clients.
 
 @item
 @code{MHD_stop_daemon()}.
 @end enumerate
 
 @noindent
 Skipping the join in step two is the tempting shortcut, and it is the
 one that leaves a connection parked forever with nothing left alive to
 wake it.
 
 
 @heading Example code
 
 The complete program is available as @code{asyncresponse.c}.  It runs
 one external @code{select()} loop and serves four URLs:
 
 @table @code
 @item /
 A small page whose JavaScript opens the stream below and appends a row
 for every event as it arrives.  The rows visibly trickle in, one per
 step, which is the most direct demonstration of what the chapter is
 about.
 
 @item /events
 The stream: a response is queued at once, and the content reader
 suspends between steps.  The body is a sequence of server-sent events,
 which needs no library on the browser side.
 
 @item /slow
 The other pattern: the connection is suspended in the access handler
 and a complete answer is queued once the worker is done.
 
 @item /fast
 Answered on the spot.  The page has a button that requests it and
 reports the round trip time; it stays in the low milliseconds while
 jobs are running, which is the point of suspending rather than
 blocking.
 @end table
 
 @noindent
 The program takes the port, the duration of a step in milliseconds and
 the number of steps on the command line.  Passing @code{0} as the port
 makes the operating system choose a free one, which the program then
 reports via @code{MHD_DAEMON_INFO_BIND_PORT}:
 
 @verbatim
 $ ./asyncresponse 0 250 20
 Listening on port 44321
 @end verbatim
 @noindent
 
 The accompanying @code{test_asyncresponse.c} starts exactly that
 binary and checks over HTTP that @code{/slow} waits for its worker,
 that the body of @code{/events} really arrives in pieces spread over
 the lifetime of the request rather than in one burst at the end, that
 @code{/fast} is still served promptly while jobs are parked, that a
 client which hangs up in the middle takes neither the server nor the
 worker with it, and that the process exits cleanly when it is signalled
 while a connection is suspended.  It also compares the server's CPU
 time against the wall clock time of a transfer, which is the only one
 of those checks that notices the difference between a suspended
 connection and a busy-waiting one.
 
 
 @heading Remarks
 
 @emph{MHD} does not detect that a client has disconnected while a
 connection is suspended, and the connection timeout does not run
 either.  A suspended connection stays suspended until somebody resumes
 it, without exception.  Any timeout on the work itself is therefore the
 application's responsibility.
 
 Suspended connections still count against the connection limits, both
 the global one and the per-IP one.  A server that parks connections for
 minutes needs its limits sized for the number of jobs it expects to
 have in flight, not for the number it expects to be transmitting.
 
 If the client vanishes in the middle of a chunked response, the next
 write fails and @emph{MHD} logs it---with
 @code{MHD_USE_ERROR_LOG} enabled the example prints @code{Failed to
 send the chunked response body ... The socket is no longer available
 for sending}.  That is the normal course of events for an aborted
 download and not a sign of a problem.
 
 Finally, none of this is specific to @code{select()}.  The same
 argument holds for @code{poll()} and for @code{epoll}, where the
 daemon's descriptor can be obtained with
 @code{MHD_DAEMON_INFO_EPOLL_FD} and added to an epoll set of the
 application's own; @code{suspend_resume_epoll.c} among the
 distribution's examples shows that variant.
 
 
 @heading Exercises
 
 @itemize @bullet
 
 @item
 Give the jobs a deadline.  Since a suspended connection never times
 out on its own, add a check to the main loop that resumes---and
 answers with @code{503}---any job that has been running for too long,
 and confirm with a step duration long enough to trigger it.
 
 @item
 Remove the @code{MHD_suspend_connection()} call from the content
 reader, leaving only the @code{return 0}, and watch the server process
 with @code{top} while a single client downloads the stream.  Then put
 it back and watch again.  Both servers send the same bytes.
 
 @item
 Take the resume out of the worker's final block, so that a finished job
 leaves its connection parked, and then stop the server while a job is
 in flight.  Watch what the shutdown does---and what @code{valgrind}
 has to say about it---and work out which of the four steps above was
 violated.
 
 @item
 Replace the per-request thread with a fixed pool of worker threads and
 a queue of pending jobs.  The reference counting and the
 @code{abandoned} flag should not need to change at all; if they do, the
 ownership rules were not as clean as they looked.
 
 @item
 Serve the stream to two browser windows at once and confirm from the
 timestamps that the two jobs run concurrently and are interleaved by a
 single-threaded event loop.
 
 @end itemize