mirrors/libmicrohttpd
2
0
Fork 0
mirror of https://git.gnunet.org/libmicrohttpd.git synced 2026-09-25 04:09:31 +03:00
Code Issues Packages Projects Releases Wiki Activity

add tutorial for #5657

Browse Source
This commit is contained in:
Christian Grothoff
2026-07-28 22:47:32 +02:00
parent 6418a0b3b5
commit 8dbbdd77fe
9 changed files with 2441 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
doc/Makefile.am
+1
View File
@@ -22,6 +22,7 @@ microhttpd_TEXINFOS = \
gpl-2.0.texi \ gpl-2.0.texi \
fdl-1.3.texi fdl-1.3.texi
microhttpd_tutorial_TEXINFOS = \ microhttpd_tutorial_TEXINFOS = \
chapters/asyncresponses.inc \
chapters/basicauthentication.inc \ chapters/basicauthentication.inc \
chapters/bibliography.inc \ chapters/bibliography.inc \
chapters/callbackresponse.inc \ chapters/callbackresponse.inc \
doc/chapters/asyncresponses.inc
+501
View File
@@ -0,0 +1,501 @@
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
doc/chapters/callbackresponse.inc
+10 -1
View File
@@ -367,6 +367,12 @@ million times and burns a full core; the same callback suspending the
connection is entered exactly once more and the daemon uses no connection is entered exactly once more and the daemon uses no
measurable CPU time at all. measurable CPU time at all.
Suspending brings requirements of its own---on who may call what from
which thread, on how a worker thread wakes up a blocked event loop, and
on who releases the shared state when a client disappears halfway
through. @ref{Answering from another thread} works all of that out on
a complete program.
@heading Trailers @heading Trailers
@@ -456,7 +462,10 @@ handle in the context struct, and suspend the connection whenever the
buffer is empty. Do not forget @code{MHD_ALLOW_SUSPEND_RESUME} when buffer is empty. Do not forget @code{MHD_ALLOW_SUSPEND_RESUME} when
starting the daemon. Then remove the call to starting the daemon. Then remove the call to
@code{MHD_suspend_connection()}, leaving only the @code{return 0}, and @code{MHD_suspend_connection()}, leaving only the @code{return 0}, and
compare the CPU usage of the server process. compare the CPU usage of the server process. A complete solution,
together with the ownership rules that a shared buffer between a
worker and the event loop needs, is the subject of
@ref{Answering from another thread}.
@item @item
Compute a checksum over the generated document while it is being sent Compute a checksum over the generated document while it is being sent
doc/examples/.gitignore
+5
View File
@@ -1,3 +1,5 @@
/asyncresponse
/test_asyncresponse
/upgrade /upgrade
/tlsauthentication /tlsauthentication
/simplepost /simplepost
@@ -12,3 +14,6 @@
*.o *.o
*.lo *.lo
*.la *.la
# Produced by the automake test harness
*.log
*.trs
doc/examples/Makefile.am
+28 -1
View File
@@ -3,7 +3,7 @@ SUBDIRS = .
AM_CPPFLAGS = \ AM_CPPFLAGS = \
-I$(top_srcdir)/src/include \ -I$(top_srcdir)/src/include \
$(CPPFLAGS_ac) $(CPPFLAGS_ac) $(LIBCURL_CPPFLAGS)
AM_CFLAGS = $(CFLAGS_ac) @LIBGCRYPT_CFLAGS@ AM_CFLAGS = $(CFLAGS_ac) @LIBGCRYPT_CFLAGS@
@@ -44,10 +44,37 @@ if HAVE_POSTPROCESSOR
noinst_PROGRAMS += simplepost largepost sessions noinst_PROGRAMS += simplepost largepost sessions
endif endif
if HAVE_POSIX_THREADS
noinst_PROGRAMS += asyncresponse
if HAVE_FORK_WAITPID
if RUN_LIBCURL_TESTS
check_PROGRAMS = test_asyncresponse
TESTS = $(check_PROGRAMS)
endif
endif
endif
if HAVE_W32 if HAVE_W32
AM_CPPFLAGS += -DWINDOWS AM_CPPFLAGS += -DWINDOWS
endif endif
asyncresponse_SOURCES = \
asyncresponse.c
asyncresponse_CFLAGS = \
$(AM_CFLAGS) $(PTHREAD_CFLAGS)
asyncresponse_LDADD = \
$(PTHREAD_LIBS) \
$(top_builddir)/src/microhttpd/libmicrohttpd.la
test_asyncresponse_SOURCES = \
test_asyncresponse.c
test_asyncresponse_CFLAGS = \
$(AM_CFLAGS) $(PTHREAD_CFLAGS)
test_asyncresponse_LDADD = \
$(PTHREAD_LIBS) \
$(top_builddir)/src/microhttpd/libmicrohttpd.la \
@LIBCURL@
basicauthentication_SOURCES = \ basicauthentication_SOURCES = \
basicauthentication.c basicauthentication.c
basicauthentication_LDADD = \ basicauthentication_LDADD = \
doc/examples/asyncresponse.c
+1193
View File
@@ -0,0 +1,1193 @@
/* Feel free to use this example code in any way
you see fit (Public Domain) */
/**
* @file asyncresponse.c
* @brief Example for answering a request from another thread while
* the daemon is driven by an external select() loop. The
* main thread owns libmicrohttpd and does nothing but poll;
* every slow request gets a worker thread of its own and the
* connection is suspended for as long as that worker has
* nothing to say.
*
* "GET /" serves a small page whose JavaScript renders the
* data as it trickles in, "GET /events" is the stream behind
* it (a response is queued immediately and the content reader
* suspends between chunks), "GET /slow" suspends in the access
* handler until a complete answer is ready, and "GET /fast" is
* answered on the spot so that it can be used to show that the
* event loop never blocks.
*
* This example needs POSIX threads.
* @author Christian Grothoff
*/
#include <sys/types.h>
#include <sys/select.h>
#include <sys/socket.h>
#include <microhttpd.h>
#include <errno.h>
#include <pthread.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
/**
* Port to listen on if none was given on the command line. Pass 0 to
* let the operating system pick a free port; the port that was
* actually bound is then printed on standard output.
*/
#define DEFAULT_PORT 8888
/**
* Time the worker thread pretends to need for one step, in
* milliseconds.
*/
#define DEFAULT_STEP_MS 250
/**
* Number of steps that make up one job.
*/
#define DEFAULT_STEPS 20
/**
* Size of the buffer in which a job collects the bytes that the
* content reader has not picked up yet.
*/
#define MAX_PAYLOAD 4096
/**
* Block size we ask MHD to use when it queries our content reader.
*/
#define IO_BLOCK_SIZE 1024
/**
* How long we are willing to wait for the last connections to go away
* when the server is shutting down, in milliseconds.
*/
#define DRAIN_TIMEOUT_MS 2000
/**
* The front page. Its JavaScript opens the "/events" stream and adds
* a row for every server-sent event as it arrives, so that the delay
* between the steps is plainly visible. The "ping" button fires a
* request against "/fast" and reports the round trip time; it stays in
* the low milliseconds even while several jobs are running, which is
* the whole point of suspending instead of blocking.
*/
static const char *PAGE =
"<!DOCTYPE html>\n"
"<html lang='en'>\n"
"<head>\n"
"<meta charset='utf-8'>\n"
"<title>libmicrohttpd: answering from another thread</title>\n"
"<style>\n"
"body { font-family: sans-serif; max-width: 40em; margin: 2em auto; }\n"
"#bar { height: 1em; border: 1px solid #888; margin: 1em 0; }\n"
"#fill { height: 100%; width: 0; background: #4a90d9; transition: width .2s; }\n"
"#log { font-family: monospace; font-size: 90%; list-style: none; padding: 0; }\n"
"#log li { border-bottom: 1px solid #eee; padding: 2px 0; }\n"
"button { margin-right: .5em; }\n"
"</style>\n"
"</head>\n"
"<body>\n"
"<h1>Answering from another thread</h1>\n"
"<p>The server runs a single external <code>select()</code> loop. The job\n"
"below is computed by a thread of its own, and the connection carrying it is\n"
"suspended whenever that thread has nothing to say. Nothing is buffered: each\n"
"row appears the moment the worker produced it.</p>\n"
"<button id='go'>Start a job</button>\n"
"<button id='ping'>Ping /fast</button>\n"
"<div id='bar'><div id='fill'></div></div>\n"
"<p id='status'>idle</p>\n"
"<ul id='log'></ul>\n"
"<script>\n"
"const $ = (id) => document.getElementById(id);\n"
"let es = null;\n"
"let t0 = 0;\n"
"const stop = () => { if (es) { es.close(); es = null; } };\n"
"$('go').onclick = () => {\n"
" stop ();\n"
" $('log').innerHTML = '';\n"
" $('fill').style.width = '0';\n"
" $('status').textContent = 'running';\n"
" t0 = performance.now ();\n"
" es = new EventSource ('/events');\n"
" es.addEventListener ('step', (e) => {\n"
" const d = JSON.parse (e.data);\n"
" const li = document.createElement ('li');\n"
" li.textContent = '+' + Math.round (performance.now () - t0) +\n"
" ' ms ' + d.label;\n"
" $('log').appendChild (li);\n"
" $('fill').style.width = (100 * d.n / d.total) + '%';\n"
" });\n"
" es.addEventListener ('done', () => {\n"
" stop ();\n"
" $('status').textContent = 'done after ' +\n"
" Math.round (performance.now () - t0) + ' ms';\n"
" });\n"
" es.onerror = () => { stop (); $('status').textContent = 'connection lost'; };\n"
"};\n"
"$('ping').onclick = async () => {\n"
" const t = performance.now ();\n"
" await fetch ('/fast', { cache: 'no-store' });\n"
" $('status').textContent = '/fast answered in ' +\n"
" Math.round (performance.now () - t) + ' ms';\n"
"};\n"
"</script>\n"
"</body>\n"
"</html>\n";
/**
* State shared between the thread that runs libmicrohttpd and the
* worker thread that produces the answer. Everything below @e lock is
* protected by it.
*
* The structure is reference counted, with one reference held by MHD
* (dropped in #stream_done() or #request_completed()) and one held by
* the worker (dropped when the worker function returns). Reaching zero
* does not free the job: only the main thread does that, in
* #reap_jobs(), because it is also the thread that has to join the
* worker.
*/
struct Job
{
/**
* Kept in a doubly linked list of all jobs, so that the shutdown
* code can reach every worker. The list is protected by
* #jobs_lock, never by @e lock.
*/
struct Job *next;
/**
* See @e next.
*/
struct Job *prev;
/**
* Protects every field below, and---just as importantly---keeps the
* connection alive across #MHD_resume_connection().
*/
pthread_mutex_t lock;
/**
* Used to wake the worker out of its sleep when the client goes away
* or the server is shutting down.
*/
pthread_cond_t cond;
/**
* The worker thread. Only touched by the main thread, under
* #jobs_lock.
*/
pthread_t tid;
/**
* The connection this job answers. Set to NULL by the MHD side as
* soon as MHD is done with the connection; the worker must not touch
* it after that, which is why @e abandoned is checked under @e lock
* before every #MHD_resume_connection().
*/
struct MHD_Connection *connection;
/**
* Bytes produced by the worker that the content reader has not
* picked up yet.
*/
char payload[MAX_PAYLOAD];
/**
* Number of valid bytes in @e payload.
*/
size_t fill;
/**
* Number of bytes of @e payload already handed to MHD.
*/
size_t off;
/**
* Number of references, see the comment on the structure.
*/
unsigned int rc;
/**
* Non-zero if the answer is streamed as it is produced ("/events"),
* zero if the client only ever sees the finished result ("/slow").
* A streaming job resumes the connection after every step, the other
* kind only once, at the very end.
*/
int stream;
/**
* Non-zero once the worker has produced everything it is going to
* produce.
*/
int finished;
/**
* Non-zero while the connection is suspended (or is just about to
* be, which under @e lock is the same thing).
*/
int suspended;
/**
* Non-zero once MHD is done with the connection. The worker must
* not call any MHD function on @e connection any more.
*/
int abandoned;
/**
* Non-zero if the worker should give up early.
*/
int stop;
/**
* Non-zero once the worker thread was started. Main thread only.
*/
int started;
/**
* Non-zero once the worker thread was joined. Main thread only.
*/
int joined;
};
/**
* Head of the list of all jobs.
*/
static struct Job *jobs_head;
/**
* Protects #jobs_head and the list links of every job. Lock order is
* #jobs_lock before any `struct Job`'s @e lock, never the other way
* round.
*/
static pthread_mutex_t jobs_lock = PTHREAD_MUTEX_INITIALIZER;
/**
* Set by the signal handler, read by the main loop.
*/
static volatile sig_atomic_t shutdown_requested;
/**
* Written to by the signal handler so that a blocking select() returns
* at once. MHD's own wakeup goes through its inter-thread
* communication channel, but our signal is our own business.
*/
static int wake_pipe[2] = { -1, -1 };
/**
* Milliseconds the worker sleeps per step.
*/
static unsigned int step_ms = DEFAULT_STEP_MS;
/**
* Number of steps that make up one job.
*/
static unsigned int total_steps = DEFAULT_STEPS;
/**
* Signal handler for SIGINT and SIGTERM. Does the two things that are
* safe to do here: set a flag and poke a pipe.
*
* @param sig the signal that was received
*/
static void
signal_handler (int sig)
{
static const char c = 'x';
(void) sig; /* Unused. Silent compiler warning. */
shutdown_requested = 1;
if (0 > write (wake_pipe[1],
&c,
1))
{
/* Nothing useful can be done about this here. */
}
}
/**
* Allocate a job and put it on the global list.
*
* @param connection the connection the job answers
* @param stream non-zero to deliver the steps as they are produced
* @return the new job with a reference count of two (one for MHD, one
* for the worker that the caller is about to start), NULL on
* error
*/
static struct Job *
job_create (struct MHD_Connection *connection,
int stream)
{
struct Job *job;
job = malloc (sizeof (struct Job));
if (NULL == job)
return NULL;
memset (job,
0,
sizeof (struct Job));
if (0 != pthread_mutex_init (&job->lock,
NULL))
{
free (job);
return NULL;
}
if (0 != pthread_cond_init (&job->cond,
NULL))
{
pthread_mutex_destroy (&job->lock);
free (job);
return NULL;
}
job->connection = connection;
job->stream = stream;
job->rc = 2;
pthread_mutex_lock (&jobs_lock);
job->next = jobs_head;
if (NULL != jobs_head)
jobs_head->prev = job;
jobs_head = job;
pthread_mutex_unlock (&jobs_lock);
return job;
}
/**
* Drop a reference. Note that this never frees the job: the main
* thread has to join the worker first, and only it may do that. See
* #reap_jobs().
*
* @param job the job to release
*/
static void
job_unref (struct Job *job)
{
pthread_mutex_lock (&job->lock);
job->rc--;
pthread_mutex_unlock (&job->lock);
}
/**
* Called by the MHD side once MHD is done with the connection. After
* this returns, the worker will not touch the connection again.
*
* @param job the job to detach from its connection
*/
static void
job_detach (struct Job *job)
{
pthread_mutex_lock (&job->lock);
job->abandoned = 1;
job->connection = NULL;
/* Wake the worker out of its sleep: if the client hung up in the
middle of the transfer there is no point in producing the rest. */
pthread_cond_signal (&job->cond);
pthread_mutex_unlock (&job->lock);
job_unref (job);
}
/**
* Resume the connection of @a job if it is suspended.
*
* Must be called with @a job's lock held, and it deliberately calls
* into MHD while holding it: the MHD thread has to take the very same
* lock in #job_detach() before it can finish the connection, so as
* long as we hold the lock the connection cannot go away underneath
* us. The reverse order never occurs---MHD invokes our callbacks
* without holding any of its own locks that #MHD_resume_connection()
* would need---so this cannot deadlock.
*
* @param job the job whose connection to resume
*/
static void
job_resume_locked (struct Job *job)
{
if ( (0 == job->suspended) ||
(0 != job->abandoned) )
return;
job->suspended = 0;
MHD_resume_connection (job->connection);
}
/**
* Sleep for one step, or until the job is woken up.
*
* @param job the job to sleep on
* @return non-zero if the worker should stop early
*/
static int
job_sleep (struct Job *job)
{
struct timespec ts;
int stop;
if (0 != clock_gettime (CLOCK_REALTIME,
&ts))
return 1;
ts.tv_sec += (time_t) (step_ms / 1000);
ts.tv_nsec += (long) (step_ms % 1000) * 1000000L;
if (1000000000L <= ts.tv_nsec)
{
ts.tv_nsec -= 1000000000L;
ts.tv_sec++;
}
pthread_mutex_lock (&job->lock);
while ( (0 == job->stop) &&
(0 == job->abandoned) )
{
if (ETIMEDOUT == pthread_cond_timedwait (&job->cond,
&job->lock,
&ts))
break;
}
stop = (0 != job->stop) || (0 != job->abandoned);
pthread_mutex_unlock (&job->lock);
return stop;
}
/**
* Append @a len bytes from @a data to the job's payload buffer,
* compacting the buffer first. Must be called with @a job's lock
* held. A real application would need a policy for the case that the
* client cannot keep up; here we simply drop the step, which cannot
* happen with the sizes used in this example.
*
* @param job the job to append to
* @param data the bytes to append
* @param len the number of bytes to append
*/
static void
job_append_locked (struct Job *job,
const char *data,
size_t len)
{
if (0 != job->off)
{
memmove (job->payload,
&job->payload[job->off],
job->fill - job->off);
job->fill -= job->off;
job->off = 0;
}
if (len > MAX_PAYLOAD - job->fill)
return;
memcpy (&job->payload[job->fill],
data,
len);
job->fill += len;
}
/**
* The worker. This stands in for whatever expensive computation,
* database cursor or remote query a real application would run: it
* produces one server-sent event per step and resumes the connection
* whenever it has something new to offer.
*
* Note that the only MHD function this thread ever calls is
* #MHD_resume_connection(). Building and queueing the response is the
* business of the thread that runs the daemon.
*
* @param cls the `struct Job` to work on
* @return always NULL
*/
static void *
worker (void *cls)
{
struct Job *job = cls;
char event[256];
unsigned int i;
int len;
for (i = 1; i <= total_steps; i++)
{
if (0 != job_sleep (job))
break;
if (0 != job->stream)
len = snprintf (event,
sizeof (event),
"event: step\ndata: {\"n\":%u,\"total\":%u,"
"\"label\":\"step %u of %u done\"}\n\n",
i,
total_steps,
i,
total_steps);
else
len = snprintf (event,
sizeof (event),
"step %u of %u done\n",
i,
total_steps);
if ( (0 >= len) ||
(sizeof (event) <= (size_t) len) )
break;
pthread_mutex_lock (&job->lock);
job_append_locked (job,
event,
(size_t) len);
/* Only a streaming job has anything to show yet. The other kind
stays suspended until the very last step. */
if (0 != job->stream)
job_resume_locked (job);
pthread_mutex_unlock (&job->lock);
}
if (0 != job->stream)
len = snprintf (event,
sizeof (event),
"event: done\ndata: {\"n\":%u,\"total\":%u}\n\n",
total_steps,
total_steps);
else
len = snprintf (event,
sizeof (event),
"all %u steps done\n",
total_steps);
pthread_mutex_lock (&job->lock);
if ( (0 < len) &&
(sizeof (event) > (size_t) len) )
job_append_locked (job,
event,
(size_t) len);
job->finished = 1;
/* The last resume is not optional: if the connection were left
suspended, nothing would ever wake it up again, and the daemon
could not be stopped. */
job_resume_locked (job);
pthread_mutex_unlock (&job->lock);
job_unref (job);
return NULL;
}
/**
* Start the worker thread of @a job.
*
* @param job the job whose worker to start
* @return #MHD_YES on success
*/
static enum MHD_Result
job_start (struct Job *job)
{
pthread_mutex_lock (&jobs_lock);
if (0 != pthread_create (&job->tid,
NULL,
&worker,
job))
{
pthread_mutex_unlock (&jobs_lock);
return MHD_NO;
}
job->started = 1;
pthread_mutex_unlock (&jobs_lock);
return MHD_YES;
}
/**
* Content reader for the "/events" stream. This is the interesting
* one: when the worker has produced nothing since the last call we
* suspend the connection and return zero, instead of returning zero on
* its own---which would make MHD ask again immediately and burn a CPU
* core for the whole duration of the transfer.
*
* @param cls our `struct Job`
* @param pos number of bytes already returned for this response
* @param buf where to copy the data
* @param max maximum number of bytes to copy to @a buf
* @return number of bytes written to @a buf, 0 if the connection was
* suspended, MHD_CONTENT_READER_END_OF_STREAM at the end
*/
static ssize_t
stream_reader (void *cls,
uint64_t pos,
char *buf,
size_t max)
{
struct Job *job = cls;
size_t ready;
(void) pos; /* Unused. Silent compiler warning. */
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;
}
/**
* Called by MHD when the response object of an "/events" request is
* destroyed, which happens for a completed transfer just as well as
* for a client that went away in the middle of one. This is the MHD
* side of the job's reference count.
*
* @param cls our `struct Job`
*/
static void
stream_done (void *cls)
{
job_detach (cls);
}
/**
* Handle "GET /events": queue a streaming response right away and let
* the content reader deal with the fact that the data does not exist
* yet.
*
* @param connection the connection to answer
* @return #MHD_YES on success
*/
static enum MHD_Result
handle_events (struct MHD_Connection *connection)
{
struct MHD_Response *response;
struct Job *job;
enum MHD_Result ret;
job = job_create (connection,
1);
if (NULL == job)
return MHD_NO;
response = MHD_create_response_from_callback (MHD_SIZE_UNKNOWN,
IO_BLOCK_SIZE,
&stream_reader,
job,
&stream_done);
if (NULL == response)
{
/* No response object exists, so nobody will call stream_done() for
us; drop both references by hand. The worker was never
started. */
job_unref (job);
job_unref (job);
return MHD_NO;
}
if (MHD_NO == job_start (job))
{
MHD_destroy_response (response);
job_unref (job);
return MHD_NO;
}
(void) MHD_add_response_header (response,
MHD_HTTP_HEADER_CONTENT_TYPE,
"text/event-stream");
(void) MHD_add_response_header (response,
MHD_HTTP_HEADER_CACHE_CONTROL,
"no-cache");
ret = MHD_queue_response (connection,
MHD_HTTP_OK,
response);
MHD_destroy_response (response);
return ret;
}
/**
* Handle "GET /slow": the answer is only useful as a whole, so instead
* of streaming we suspend the connection until the worker is done.
* MHD then calls this function again---on its own thread---and that is
* where the response is queued. #MHD_queue_response() is never called
* from the worker.
*
* @param connection the connection to answer
* @param req_cls the per-request pointer, holding our `struct Job`
* @return #MHD_YES on success
*/
static enum MHD_Result
handle_slow (struct MHD_Connection *connection,
void **req_cls)
{
struct MHD_Response *response;
struct Job *job = *req_cls;
char answer[MAX_PAYLOAD];
size_t len;
enum MHD_Result ret;
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)
{
/* Still working. Park the connection; the worker resumes it when
it is done, and MHD will then enter this function once more. */
job->suspended = 1;
MHD_suspend_connection (connection);
pthread_mutex_unlock (&job->lock);
return MHD_YES;
}
len = job->fill - job->off;
if (len > sizeof (answer))
len = sizeof (answer);
memcpy (answer,
&job->payload[job->off],
len);
pthread_mutex_unlock (&job->lock);
response = MHD_create_response_from_buffer_copy (len,
answer);
if (NULL == response)
return MHD_NO;
(void) MHD_add_response_header (response,
MHD_HTTP_HEADER_CONTENT_TYPE,
"text/plain");
ret = MHD_queue_response (connection,
MHD_HTTP_OK,
response);
MHD_destroy_response (response);
return ret;
}
/**
* Queue a complete response that is already sitting in memory.
*
* @param connection the connection to answer
* @param status the HTTP status code to use
* @param mime the value for the "Content-Type" header
* @param body the body to send
* @return #MHD_YES on success
*/
static enum MHD_Result
queue_static (struct MHD_Connection *connection,
unsigned int status,
const char *mime,
const char *body)
{
struct MHD_Response *response;
enum MHD_Result ret;
response = MHD_create_response_from_buffer_copy (strlen (body),
body);
if (NULL == response)
return MHD_NO;
(void) MHD_add_response_header (response,
MHD_HTTP_HEADER_CONTENT_TYPE,
mime);
(void) MHD_add_response_header (response,
MHD_HTTP_HEADER_CACHE_CONTROL,
"no-store");
ret = MHD_queue_response (connection,
status,
response);
MHD_destroy_response (response);
return ret;
}
static enum MHD_Result
answer_to_connection (void *cls,
struct MHD_Connection *connection,
const char *url,
const char *method,
const char *version,
const char *upload_data,
size_t *upload_data_size,
void **req_cls)
{
(void) cls; /* Unused. Silent compiler warning. */
(void) version; /* Unused. Silent compiler warning. */
(void) upload_data; /* Unused. Silent compiler warning. */
(void) upload_data_size; /* Unused. Silent compiler warning. */
if (0 != strcmp (method,
MHD_HTTP_METHOD_GET))
return MHD_NO;
if (0 == strcmp (url,
"/"))
return queue_static (connection,
MHD_HTTP_OK,
"text/html",
PAGE);
if (0 == strcmp (url,
"/fast"))
return queue_static (connection,
MHD_HTTP_OK,
"text/plain",
"pong\n");
if ( (0 == strcmp (url,
"/events")) ||
(0 == strcmp (url,
"/slow")) )
{
if (0 != shutdown_requested)
return queue_static (connection,
MHD_HTTP_SERVICE_UNAVAILABLE,
"text/plain",
"shutting down\n");
if (0 == strcmp (url,
"/events"))
return handle_events (connection);
return handle_slow (connection,
req_cls);
}
return queue_static (connection,
MHD_HTTP_NOT_FOUND,
"text/plain",
"not found\n");
}
/**
* Called by MHD when a request is done. For "/slow" this is where the
* MHD side of the reference count is dropped; the other handlers leave
* @a req_cls alone, so there is nothing to do for them.
*
* @param cls closure, unused
* @param connection the connection that finished
* @param req_cls the per-request pointer
* @param toe why the request ended
*/
static void
request_completed (void *cls,
struct MHD_Connection *connection,
void **req_cls,
enum MHD_RequestTerminationCode toe)
{
struct Job *job = *req_cls;
(void) cls; /* Unused. Silent compiler warning. */
(void) connection; /* Unused. Silent compiler warning. */
(void) toe; /* Unused. Silent compiler warning. */
if (NULL == job)
return;
*req_cls = NULL;
job_detach (job);
}
/**
* Free every job that nobody references any more, joining its worker
* on the way. Called from the main loop, and thus always from the
* thread that also runs the daemon.
*/
static void
reap_jobs (void)
{
struct Job *job;
struct Job *next;
unsigned int rc;
pthread_mutex_lock (&jobs_lock);
for (job = jobs_head; NULL != job; job = next)
{
next = job->next;
pthread_mutex_lock (&job->lock);
rc = job->rc;
pthread_mutex_unlock (&job->lock);
if (0 != rc)
continue;
if ( (0 != job->started) &&
(0 == job->joined) )
{
pthread_join (job->tid,
NULL);
job->joined = 1;
}
if (NULL != job->prev)
job->prev->next = job->next;
else
jobs_head = job->next;
if (NULL != job->next)
job->next->prev = job->prev;
pthread_cond_destroy (&job->cond);
pthread_mutex_destroy (&job->lock);
free (job);
}
pthread_mutex_unlock (&jobs_lock);
}
/**
* Ask every worker to stop and wait until they all did.
*
* This has to happen before the daemon is stopped: a worker that is
* still asleep may be holding the only promise to resume a suspended
* connection, and stopping a daemon that has suspended connections is
* an API violation. Once every worker has returned, no connection can
* be suspended any more, because the workers resume before they exit
* and the content reader only ever suspends while a worker is running.
*/
static void
stop_all_jobs (void)
{
struct Job *job;
pthread_mutex_lock (&jobs_lock);
for (job = jobs_head; NULL != job; job = job->next)
{
pthread_mutex_lock (&job->lock);
job->stop = 1;
pthread_cond_signal (&job->cond);
pthread_mutex_unlock (&job->lock);
}
for (job = jobs_head; NULL != job; job = job->next)
{
if ( (0 != job->started) &&
(0 == job->joined) )
{
pthread_join (job->tid,
NULL);
job->joined = 1;
}
}
pthread_mutex_unlock (&jobs_lock);
}
/**
* Number of connections the daemon is currently handling.
*
* @param daemon the daemon to query
* @return the number of connections, 0 if it cannot be determined
*/
static unsigned int
connection_count (struct MHD_Daemon *daemon)
{
const union MHD_DaemonInfo *info;
info = MHD_get_daemon_info (daemon,
MHD_DAEMON_INFO_CURRENT_CONNECTIONS);
if (NULL == info)
return 0;
return info->num_connections;
}
int
main (int argc,
char *const *argv)
{
struct MHD_Daemon *daemon;
const union MHD_DaemonInfo *info;
struct sigaction sa;
fd_set rs;
fd_set ws;
fd_set es;
struct timeval tv;
struct timeval *tvp;
MHD_socket max;
uint64_t mhd_timeout;
unsigned long port = DEFAULT_PORT;
unsigned int drained_ms = 0;
char drain_buf[64];
int draining = 0;
int ret = 0;
if (1 < argc)
port = strtoul (argv[1],
NULL,
10);
if (2 < argc)
step_ms = (unsigned int) strtoul (argv[2],
NULL,
10);
if (3 < argc)
total_steps = (unsigned int) strtoul (argv[3],
NULL,
10);
if ( (65535 < port) ||
(0 == total_steps) )
{
fprintf (stderr,
"Usage: %s [PORT [STEP_MS [STEPS]]]\n"
"Pass 0 as PORT to let the system pick a free one.\n",
argv[0]);
return 1;
}
if (0 > pipe (wake_pipe))
{
fprintf (stderr,
"Failed to create wakeup pipe: %s\n",
strerror (errno));
return 1;
}
memset (&sa,
0,
sizeof (sa));
sa.sa_handler = &signal_handler;
sigaction (SIGINT,
&sa,
NULL);
sigaction (SIGTERM,
&sa,
NULL);
sa.sa_handler = SIG_IGN;
sigaction (SIGPIPE,
&sa,
NULL);
/* No MHD_USE_INTERNAL_POLLING_THREAD: the loop below is ours.
MHD_ALLOW_SUSPEND_RESUME implies MHD_USE_ITC, and that channel is
what lets a worker thread break us out of the select() call. */
daemon = MHD_start_daemon (MHD_ALLOW_SUSPEND_RESUME | MHD_USE_ERROR_LOG,
(uint16_t) port,
NULL, NULL,
&answer_to_connection, NULL,
MHD_OPTION_NOTIFY_COMPLETED,
&request_completed, NULL,
MHD_OPTION_END);
if (NULL == daemon)
{
fprintf (stderr,
"Failed to start daemon.\n");
return 1;
}
info = MHD_get_daemon_info (daemon,
MHD_DAEMON_INFO_BIND_PORT);
if ( (NULL == info) ||
(0 == info->port) )
{
fprintf (stderr,
"Failed to determine bound port.\n");
MHD_stop_daemon (daemon);
return 1;
}
printf ("Listening on port %u\n",
(unsigned int) info->port);
fflush (stdout);
while (1)
{
if ( (0 != shutdown_requested) &&
(0 == draining) )
{
/* Every worker has to be gone before the daemon may be stopped. */
stop_all_jobs ();
draining = 1;
}
if ( (0 != draining) &&
( (0 == connection_count (daemon)) ||
(DRAIN_TIMEOUT_MS <= drained_ms) ) )
break;
FD_ZERO (&rs);
FD_ZERO (&ws);
FD_ZERO (&es);
max = 0;
if (MHD_YES != MHD_get_fdset (daemon,
&rs,
&ws,
&es,
&max))
{
fprintf (stderr,
"Failed to obtain the file descriptor set.\n");
ret = 1;
break;
}
/* Our own descriptors go into the very same sets. */
FD_SET (wake_pipe[0],
&rs);
if (max < wake_pipe[0])
max = wake_pipe[0];
if (0 != draining)
{
/* Look at the drain deadline regularly. */
tv.tv_sec = 0;
tv.tv_usec = 50 * 1000;
tvp = &tv;
drained_ms += 50;
}
else if (MHD_YES == MHD_get_timeout64 (daemon,
&mhd_timeout))
{
tv.tv_sec = (time_t) (mhd_timeout / 1000);
tv.tv_usec = ((long) (mhd_timeout % 1000)) * 1000;
tvp = &tv;
}
else
{
/* MHD has nothing to wait for. With every connection suspended
this is the normal case, and the loop simply sleeps until a
worker resumes one---which reaches us through MHD's ITC, as it
is part of the read set above. */
tvp = NULL;
}
if (-1 == select ((int) max + 1,
&rs,
&ws,
&es,
tvp))
{
if (EINTR == errno)
continue;
fprintf (stderr,
"Aborting due to error during select: %s\n",
strerror (errno));
ret = 1;
break;
}
if (FD_ISSET (wake_pipe[0],
&rs))
(void) read (wake_pipe[0],
drain_buf,
sizeof (drain_buf));
MHD_run_from_select (daemon,
&rs,
&ws,
&es);
reap_jobs ();
}
stop_all_jobs ();
reap_jobs ();
MHD_stop_daemon (daemon);
close (wake_pipe[0]);
close (wake_pipe[1]);
return ret;
}
doc/examples/test_asyncresponse.c
+671
View File
@@ -0,0 +1,671 @@
/* Feel free to use this example code in any way
you see fit (Public Domain) */
/**
* @file test_asyncresponse.c
* @brief Test for the "asyncresponse" tutorial example. The example
* is started as a child process on an ephemeral port and then
* exercised over HTTP, so that what is tested is exactly the
* code the tutorial prints.
*
* Beyond checking that the right bytes come back, the test
* verifies the two properties that the chapter is actually
* about: that the body of "/events" really trickles in rather
* than arriving in one piece at the end, and that the server
* burns no CPU while it waits. The latter is what tells a
* properly suspended connection apart from a content reader
* that returns zero in a loop---both deliver the same bytes.
* @author Christian Grothoff
*/
#include <sys/types.h>
#include <sys/wait.h>
#include <curl/curl.h>
#include <errno.h>
#include <pthread.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
/**
* Milliseconds the example should spend per step.
*/
#define STEP_MS 50
/**
* Number of steps per job, and thus (STEPS * STEP_MS) milliseconds of
* work for one request.
*/
#define STEPS 20
/**
* Nominal duration of one job in milliseconds.
*/
#define JOB_MS (STEPS * STEP_MS)
/**
* Upper bound for any single request, in seconds.
*/
#define REQUEST_TIMEOUT 30
/**
* Largest response body we are prepared to keep.
*/
#define MAX_BODY 65536
/**
* What one HTTP request collected.
*/
struct Fetch
{
/**
* The body, as far as it was received.
*/
char body[MAX_BODY];
/**
* Number of valid bytes in @e body.
*/
size_t len;
/**
* Number of times the write callback was invoked.
*/
unsigned int chunks;
/**
* Time of the first invocation of the write callback, in
* milliseconds since the start of the request.
*/
long first_ms;
/**
* Time of the last invocation of the write callback, in
* milliseconds since the start of the request.
*/
long last_ms;
/**
* Start of the request.
*/
struct timespec start;
/**
* Abort the transfer after this many callbacks; zero to never abort.
*/
unsigned int abort_after;
/**
* Result of curl_easy_perform().
*/
CURLcode res;
};
/**
* The example's process ID.
*/
static pid_t server_pid;
/**
* The port the example bound.
*/
static unsigned int server_port;
/**
* Number of checks that failed.
*/
static unsigned int failures;
/**
* Report the outcome of one check.
*
* @param ok non-zero if the check passed
* @param what what was being checked
*/
static void
check (int ok,
const char *what)
{
if (! ok)
failures++;
fprintf (stderr,
"%s: %s\n",
ok ? "PASS" : "FAIL",
what);
}
/**
* @return the current value of the monotonic clock, in milliseconds
*/
static long
now_ms (void)
{
struct timespec ts;
if (0 != clock_gettime (CLOCK_MONOTONIC,
&ts))
return 0;
return (long) ts.tv_sec * 1000 + ts.tv_nsec / 1000000;
}
/**
* @param since the earlier point in time
* @return milliseconds elapsed since @a since
*/
static long
elapsed_ms (const struct timespec *since)
{
struct timespec ts;
if (0 != clock_gettime (CLOCK_MONOTONIC,
&ts))
return 0;
return (long) (ts.tv_sec - since->tv_sec) * 1000
+ (ts.tv_nsec - since->tv_nsec) / 1000000;
}
/**
* Read the CPU time consumed by process @a pid so far. Only
* implemented for Linux; everywhere else the test simply skips the
* check that uses it.
*
* @param pid the process to look at
* @param[out] ms where to store the sum of user and system time in
* milliseconds
* @return non-zero on success
*/
static int
cpu_time_ms (pid_t pid,
long *ms)
{
#ifdef __linux__
char path[64];
char buf[1024];
char *p;
FILE *f;
size_t got;
unsigned long utime;
unsigned long stime;
long ticks;
snprintf (path,
sizeof (path),
"/proc/%ld/stat",
(long) pid);
f = fopen (path,
"r");
if (NULL == f)
return 0;
got = fread (buf,
1,
sizeof (buf) - 1,
f);
fclose (f);
if (0 == got)
return 0;
buf[got] = '\0';
/* The second field is the executable name and may contain spaces and
parentheses, so start parsing behind its closing one. */
p = strrchr (buf,
')');
if (NULL == p)
return 0;
if (2 != sscanf (p + 1,
" %*c %*d %*d %*d %*d %*d %*u %*u %*u %*u %*u %lu %lu",
&utime,
&stime))
return 0;
ticks = sysconf (_SC_CLK_TCK);
if (0 >= ticks)
return 0;
*ms = (long) ((utime + stime) * 1000 / (unsigned long) ticks);
return 1;
#else
(void) pid;
(void) ms;
return 0;
#endif
}
static size_t
write_cb (void *ptr,
size_t size,
size_t nmemb,
void *cls)
{
struct Fetch *f = cls;
size_t total = size * nmemb;
f->chunks++;
if (1 == f->chunks)
f->first_ms = elapsed_ms (&f->start);
f->last_ms = elapsed_ms (&f->start);
if ( (0 != f->abort_after) &&
(f->chunks >= f->abort_after) )
return 0; /* makes curl abort the transfer */
if (total > sizeof (f->body) - f->len - 1)
total = sizeof (f->body) - f->len - 1;
memcpy (&f->body[f->len],
ptr,
total);
f->len += total;
f->body[f->len] = '\0';
return size * nmemb;
}
/**
* Perform one GET against the example.
*
* @param path the path to request
* @param[out] f where to store what was received; @e abort_after has to
* be set by the caller, everything else is initialised here
* @return milliseconds the request took
*/
static long
fetch (const char *path,
struct Fetch *f)
{
CURL *curl;
char url[128];
unsigned int abort_after = f->abort_after;
memset (f,
0,
sizeof (struct Fetch));
f->abort_after = abort_after;
f->res = CURLE_FAILED_INIT;
snprintf (url,
sizeof (url),
"http://127.0.0.1:%u%s",
server_port,
path);
curl = curl_easy_init ();
if (NULL == curl)
return 0;
curl_easy_setopt (curl, CURLOPT_URL, url);
curl_easy_setopt (curl, CURLOPT_WRITEFUNCTION, &write_cb);
curl_easy_setopt (curl, CURLOPT_WRITEDATA, f);
curl_easy_setopt (curl, CURLOPT_TIMEOUT, (long) REQUEST_TIMEOUT);
curl_easy_setopt (curl, CURLOPT_NOSIGNAL, 1L);
clock_gettime (CLOCK_MONOTONIC,
&f->start);
f->res = curl_easy_perform (curl);
curl_easy_cleanup (curl);
return elapsed_ms (&f->start);
}
/**
* Thread body that streams "/events" in the background while the main
* thread does something else.
*
* @param cls a `struct Fetch` whose @e abort_after is already set
* @return always NULL
*/
static void *
background_stream (void *cls)
{
(void) fetch ("/events",
cls);
return NULL;
}
/**
* Start the example on an ephemeral port and learn which one it got.
*
* @return non-zero on success
*/
static int
start_server (void)
{
char steps[16];
char step_ms[16];
int fds[2];
FILE *out;
char line[128];
if (0 != pipe (fds))
return 0;
snprintf (step_ms,
sizeof (step_ms),
"%u",
(unsigned int) STEP_MS);
snprintf (steps,
sizeof (steps),
"%u",
(unsigned int) STEPS);
server_pid = fork ();
if (-1 == server_pid)
{
close (fds[0]);
close (fds[1]);
return 0;
}
if (0 == server_pid)
{
close (fds[0]);
if (STDOUT_FILENO != fds[1])
{
dup2 (fds[1],
STDOUT_FILENO);
close (fds[1]);
}
execl ("./asyncresponse",
"asyncresponse",
"0",
step_ms,
steps,
(char *) NULL);
fprintf (stderr,
"Failed to exec ./asyncresponse: %s\n",
strerror (errno));
_exit (77);
}
close (fds[1]);
out = fdopen (fds[0],
"r");
if (NULL == out)
{
close (fds[0]);
return 0;
}
if (NULL == fgets (line,
sizeof (line),
out))
{
fclose (out);
return 0;
}
fclose (out);
if (1 != sscanf (line,
"Listening on port %u",
&server_port))
{
fprintf (stderr,
"Unexpected greeting from the example: %s",
line);
return 0;
}
return 1;
}
/**
* "GET /slow" suspends in the access handler and is answered as a
* whole once the worker is done.
*/
static void
test_slow (void)
{
struct Fetch f;
long ms;
memset (&f,
0,
sizeof (f));
ms = fetch ("/slow",
&f);
check (CURLE_OK == f.res,
"/slow completed");
check (NULL != strstr (f.body,
"all 20 steps done"),
"/slow delivered the complete answer");
/* The worker sleeps STEP_MS per step, so the answer cannot possibly
be ready earlier. Only a lower bound is checked: a loaded machine
may take arbitrarily longer. */
check (ms >= JOB_MS / 2,
"/slow waited for the worker");
}
/**
* "GET /events" streams as the worker produces, and the server stays
* idle in between. This is the check that a content reader returning
* zero in a loop would fail.
*/
static void
test_events_are_incremental (void)
{
struct Fetch f;
long cpu_before;
long cpu_after;
long ms;
int have_cpu;
memset (&f,
0,
sizeof (f));
have_cpu = cpu_time_ms (server_pid,
&cpu_before);
ms = fetch ("/events",
&f);
check (CURLE_OK == f.res,
"/events completed");
check (NULL != strstr (f.body,
"event: done"),
"/events delivered the final event");
check (5 <= f.chunks,
"/events arrived in several pieces");
/* The decisive one: the first piece has to show up long before the
last. A response that is assembled first and sent afterwards would
have first_ms very close to last_ms. */
check (f.last_ms - f.first_ms >= JOB_MS / 2,
"/events trickled in rather than arriving at once");
if (! have_cpu)
{
fprintf (stderr,
"SKIP: no way to read the server's CPU time on this "
"platform\n");
return;
}
if (! cpu_time_ms (server_pid,
&cpu_after))
{
fprintf (stderr,
"SKIP: could not read the server's CPU time\n");
return;
}
/* The server spent almost all of that second waiting. If the
connection were not suspended, MHD would poll the content reader
as fast as it can and this would be close to 100% of one core. */
fprintf (stderr,
"server used %ld ms of CPU during %ld ms of streaming\n",
cpu_after - cpu_before,
ms);
check ((cpu_after - cpu_before) * 4 < ms,
"server stayed idle while the connection was suspended");
}
/**
* The event loop keeps serving other requests while jobs are parked.
*/
static void
test_loop_stays_responsive (void)
{
struct Fetch stream;
struct Fetch ping;
pthread_t tid;
long worst = 0;
long ms;
unsigned int i;
memset (&stream,
0,
sizeof (stream));
if (0 != pthread_create (&tid,
NULL,
&background_stream,
&stream))
{
check (0,
"could not start the background stream");
return;
}
for (i = 0; i < 5; i++)
{
memset (&ping,
0,
sizeof (ping));
ms = fetch ("/fast",
&ping);
if ( (CURLE_OK != ping.res) ||
(NULL == strstr (ping.body,
"pong")) )
worst = REQUEST_TIMEOUT * 1000;
else if (ms > worst)
worst = ms;
usleep (50 * 1000);
}
pthread_join (tid,
NULL);
fprintf (stderr,
"slowest /fast while a job was running: %ld ms\n",
worst);
check (worst < 500,
"/fast was served promptly while a job was suspended");
}
/**
* A client that goes away in the middle must not take the server with
* it, and the worker behind it has to be cleaned up.
*/
static void
test_client_abort (void)
{
struct Fetch f;
struct Fetch ping;
memset (&f,
0,
sizeof (f));
f.abort_after = 3;
(void) fetch ("/events",
&f);
check (CURLE_OK != f.res,
"aborted transfer was reported as such");
memset (&ping,
0,
sizeof (ping));
(void) fetch ("/fast",
&ping);
check ( (CURLE_OK == ping.res) &&
(NULL != strstr (ping.body,
"pong")),
"server survived a client that hung up mid-stream");
}
/**
* Stopping the daemon while a connection is suspended is an API
* violation, so the example has to resume everything first. If it got
* that wrong we would see a panic or a hang here instead of a clean
* exit.
*/
static void
test_clean_shutdown (void)
{
struct Fetch stream;
pthread_t tid;
int status;
pid_t got;
memset (&stream,
0,
sizeof (stream));
if (0 != pthread_create (&tid,
NULL,
&background_stream,
&stream))
{
check (0,
"could not start the background stream");
return;
}
/* Give the job time to get going, so that its connection really is
suspended when the signal arrives. */
usleep ((useconds_t) (STEP_MS * 3) * 1000);
kill (server_pid,
SIGTERM);
pthread_join (tid,
NULL);
got = waitpid (server_pid,
&status,
0);
server_pid = -1;
check (0 < got,
"reaped the server");
check (WIFEXITED (status) && (0 == WEXITSTATUS (status)),
"server shut down cleanly with a suspended connection");
}
int
main (void)
{
long deadline;
if (0 != curl_global_init (CURL_GLOBAL_ALL))
return 77;
if (! start_server ())
{
fprintf (stderr,
"Could not start ./asyncresponse\n");
curl_global_cleanup ();
return 77;
}
fprintf (stderr,
"example listening on port %u\n",
server_port);
/* Wait for the listen socket to be usable. */
deadline = now_ms () + 5000;
while (now_ms () < deadline)
{
struct Fetch f;
memset (&f,
0,
sizeof (f));
(void) fetch ("/fast",
&f);
if (CURLE_OK == f.res)
break;
usleep (50 * 1000);
}
test_slow ();
test_events_are_incremental ();
test_loop_stays_responsive ();
test_client_abort ();
test_clean_shutdown ();
if (-1 != server_pid)
{
kill (server_pid,
SIGKILL);
waitpid (server_pid,
NULL,
0);
}
curl_global_cleanup ();
if (0 != failures)
fprintf (stderr,
"%u check(s) failed\n",
failures);
return (0 == failures) ? 0 : 1;
}
doc/libmicrohttpd-tutorial.texi
+12
View File
@@ -68,6 +68,7 @@ Free Documentation License".
* Improved processing of POST data:: * Improved processing of POST data::
* Session management:: * Session management::
* Generating responses on the fly:: * Generating responses on the fly::
* Answering from another thread::
* Adding a layer of security:: * Adding a layer of security::
* Upgrading connections:: * Upgrading connections::
* Bibliography:: * Bibliography::
@@ -111,6 +112,10 @@ Free Documentation License".
@chapter Generating responses on the fly @chapter Generating responses on the fly
@include chapters/callbackresponse.inc @include chapters/callbackresponse.inc
@node Answering from another thread
@chapter Answering from another thread
@include chapters/asyncresponses.inc
@node Adding a layer of security @node Adding a layer of security
@chapter Adding a layer of security @chapter Adding a layer of security
@include chapters/tlsauthentication.inc @include chapters/tlsauthentication.inc
@@ -138,6 +143,7 @@ Free Documentation License".
* largepost.c:: * largepost.c::
* sessions.c:: * sessions.c::
* callbackresponse.c:: * callbackresponse.c::
* asyncresponse.c::
* tlsauthentication.c:: * tlsauthentication.c::
* upgrade.c:: * upgrade.c::
@end menu @end menu
@@ -190,6 +196,12 @@ Free Documentation License".
@verbatiminclude examples/callbackresponse.c @verbatiminclude examples/callbackresponse.c
@end smalldisplay @end smalldisplay
@node asyncresponse.c
@section asyncresponse.c
@smalldisplay
@verbatiminclude examples/asyncresponse.c
@end smalldisplay
@node tlsauthentication.c @node tlsauthentication.c
@section tlsauthentication.c @section tlsauthentication.c
@smalldisplay @smalldisplay
doc/libmicrohttpd.texi
+17
View File
@@ -2482,10 +2482,27 @@ make sure to run @code{MHD_run} afterwards (before again calling
the set returned by @code{MHD_get_fdset} and you may end up with a the set returned by @code{MHD_get_fdset} and you may end up with a
connection that is stuck until the next network activity. connection that is stuck until the next network activity.
This function may be called from any thread, which is what makes it
usable to hand work to threads of your own. In ``external'' select
mode it writes to the inter-thread communication channel that
@code{MHD_ALLOW_SUSPEND_RESUME} sets up, and since
@code{MHD_get_fdset} adds the reading end of that channel to the read
set, a call from another thread interrupts a @code{select} that your
application is blocked in. That matters because
@code{MHD_get_timeout} reports no timeout at all while every
connection is suspended, so your application would otherwise block
indefinitely. The channel is level triggered, so a resume issued
between @code{MHD_get_fdset} and @code{select} cannot be lost.
You can check whether a connection is currently suspended using You can check whether a connection is currently suspended using
@code{MHD_get_connection_info} by querying for @code{MHD_get_connection_info} by querying for
@code{MHD_CONNECTION_INFO_CONNECTION_SUSPENDED}. @code{MHD_CONNECTION_INFO_CONNECTION_SUSPENDED}.
A worked example of suspending connections while worker threads
produce the answers, driven by an external @code{select} loop, is
given in the chapter ``Answering from another thread'' of the MHD
tutorial.
@table @var @table @var
@item connection @item connection
the connection to resume the connection to resume