need per-thread ITC in thread-per-connection mode to avoid races not waking up the thread or the daemon due to sharing of the channel

src/microhttpd/daemon.c

@@ -2057,6 +2057,53 @@ connection_get_wait (struct MHD_Connection *c)
 }
 
 
+/**
+ * Complete the resume of a connection that has its own thread.
+ *
+ * This is the thread-per-connection counterpart of
+ * resume_suspended_connections(): the connection's own thread does the
+ * bookkeeping for itself instead of waiting for the daemon's thread to
+ * do it.  That is what makes the wake-up reliable -- the notification
+ * is sent to the connection's own ITC and is acted upon by the only
+ * thread that reads it, so it can neither be consumed by the daemon's
+ * thread nor be missed between the check of @e resuming and the wait.
+ *
+ * @param connection the suspended connection that is to be resumed,
+ *                   must have @e resuming set
+ */
+static void
+resume_connection_own_thread_ (struct MHD_Connection *connection)
+{
+  struct MHD_Daemon *const daemon = connection->daemon;
+
+  mhd_assert (MHD_D_IS_USING_THREAD_PER_CONN_ (daemon));
+  mhd_assert (! MHD_D_IS_USING_EPOLL_ (daemon));
+#ifdef UPGRADE_SUPPORT
+  mhd_assert (NULL == connection->urh);
+#endif /* UPGRADE_SUPPORT */
+#if defined(MHD_USE_POSIX_THREADS) || defined(MHD_USE_W32_THREADS)
+  MHD_mutex_lock_chk_ (&daemon->cleanup_connection_mutex);
+#endif
+  mhd_assert (connection->suspended);
+  mhd_assert (connection->resuming);
+  DLL_remove (daemon->suspended_connections_head,
+              daemon->suspended_connections_tail,
+              connection);
+  connection->suspended = false;
+  DLL_insert (daemon->connections_head,
+              daemon->connections_tail,
+              connection);
+  connection->resuming = false;
+#if defined(MHD_USE_POSIX_THREADS) || defined(MHD_USE_W32_THREADS)
+  MHD_mutex_unlock_chk_ (&daemon->cleanup_connection_mutex);
+#endif
+  /* Drop the notification that led here, as well as any notification
+     left over from a resume that was undone by a suspend before this
+     thread got to see it. */
+  MHD_itc_clear_ (connection->resume_itc);
+}
+
+
 /**
  * Main function of the thread that handles an individual
  * connection when #MHD_USE_THREAD_PER_CONNECTION is set.
@@ -2109,10 +2156,20 @@ thread_main_handle_connection (void *data)
     {
       /* Connection was suspended, wait for resume. */
       was_suspended = true;
+      if (con->resuming)
+      {
+        /* The application resumed this connection.  Complete the resume
+           right here: this thread is the only one that handles this
+           connection, and doing it here is what keeps the wake-up below
+           free of lost-notification races. */
+        resume_connection_own_thread_ (con);
+        continue;
+      }
+      mhd_assert (MHD_ITC_IS_VALID_ (con->resume_itc));
       if (! use_poll)
       {
         FD_ZERO (&rs);
-        if (! MHD_add_to_fd_set_ (MHD_itc_r_fd_ (daemon->itc),
+        if (! MHD_add_to_fd_set_ (MHD_itc_r_fd_ (con->resume_itc),
                                   &rs,
                                   NULL,
                                   FD_SETSIZE))
@@ -2123,7 +2180,7 @@ thread_main_handle_connection (void *data)
   #endif
           goto exit;
         }
-        if (0 > MHD_SYS_select_ (MHD_itc_r_fd_ (daemon->itc) + 1,
+        if (0 > MHD_SYS_select_ (MHD_itc_r_fd_ (con->resume_itc) + 1,
                                  &rs,
                                  NULL,
                                  NULL,
@@ -2146,7 +2203,7 @@ thread_main_handle_connection (void *data)
       else     /* use_poll */
       {
         p[0].events = POLLIN;
-        p[0].fd = MHD_itc_r_fd_ (daemon->itc);
+        p[0].fd = MHD_itc_r_fd_ (con->resume_itc);
         p[0].revents = 0;
         if (0 > MHD_sys_poll_ (p,
                                1,
@@ -2163,7 +2220,7 @@ thread_main_handle_connection (void *data)
         }
       }
 #endif /* HAVE_POLL */
-      MHD_itc_clear_ (daemon->itc);
+      MHD_itc_clear_ (con->resume_itc);
       continue; /* Check again for resume. */
     }           /* End of "suspended" branch. */
 
@@ -2751,6 +2808,29 @@ new_connection_prepare_ (struct MHD_Daemon *daemon,
   if (0 != connection->connection_timeout_ms)
     connection->last_activity = MHD_monotonic_msec_counter ();
 
+  MHD_itc_set_invalid_ (connection->resume_itc);
+  if ( (MHD_D_IS_USING_THREAD_PER_CONN_ (daemon)) &&
+       (0 != (daemon->options & MHD_TEST_ALLOW_SUSPEND_RESUME)) &&
+       (! MHD_itc_init_ (connection->resume_itc)) )
+  {
+    eno = errno;
+#ifdef HAVE_MESSAGES
+    MHD_DLOG (daemon,
+              _ ("Failed to create inter-thread communication channel " \
+                 "for the connection: %s\n"),
+              MHD_itc_last_strerror_ ());
+#endif
+    MHD_socket_close_chk_ (client_socket);
+    MHD_ip_limit_del (daemon,
+                      addr,
+                      addrlen);
+    if (NULL != connection->addr)
+      free (connection->addr);
+    free (connection);
+    errno = eno;
+    return NULL;
+  }
+
   if (0 == (daemon->options & MHD_USE_TLS))
   {
     /* set default connection handlers  */
@@ -2931,6 +3011,8 @@ new_connection_close_ (struct MHD_Daemon *daemon,
   }
 #endif /* HTTPS_SUPPORT */
   MHD_socket_close_chk_ (connection->socket_fd);
+  if (MHD_ITC_IS_VALID_ (connection->resume_itc))
+    MHD_itc_destroy_chk_ (connection->resume_itc);
   MHD_ip_limit_del (daemon,
                     connection->addr,
                     connection->addr_len);
@@ -3460,6 +3542,7 @@ _MHD_EXTERN void
 MHD_resume_connection (struct MHD_Connection *connection)
 {
   struct MHD_Daemon *daemon = connection->daemon;
+  bool own_thread; /**< The connection's own thread completes the resume */
 #if defined(MHD_USE_THREADS)
   mhd_assert (NULL == daemon->worker_pool);
 #endif /* MHD_USE_THREADS */
@@ -3471,12 +3554,35 @@ MHD_resume_connection (struct MHD_Connection *connection)
   MHD_mutex_lock_chk_ (&daemon->cleanup_connection_mutex);
 #endif
   connection->resuming = true;
-  daemon->resuming = true;
+  /* An "upgraded" connection is resumed only to be moved to the cleanup
+     list, which is the daemon thread's job even with a thread per
+     connection. */
+  own_thread = MHD_D_IS_USING_THREAD_PER_CONN_ (daemon)
+#ifdef UPGRADE_SUPPORT
+               && (NULL == connection->urh)
+#endif /* UPGRADE_SUPPORT */
+  ;
+  if (! own_thread)
+    daemon->resuming = true;
 #if defined(MHD_USE_POSIX_THREADS) || defined(MHD_USE_W32_THREADS)
   MHD_mutex_unlock_chk_ (&daemon->cleanup_connection_mutex);
 #endif
-  if ( (MHD_ITC_IS_VALID_ (daemon->itc)) &&
-       (! MHD_itc_activate_ (daemon->itc, "r")) )
+  if (own_thread)
+  {
+    /* Wake the connection's own thread.  See the description of
+       @e resume_itc for why the daemon-wide ITC will not do. */
+    mhd_assert (MHD_ITC_IS_VALID_ (connection->resume_itc));
+    if (! MHD_itc_activate_ (connection->resume_itc, "r"))
+    {
+#ifdef HAVE_MESSAGES
+      MHD_DLOG (daemon,
+                _ ("Failed to signal resume via the connection's " \
+                   "inter-thread communication channel.\n"));
+#endif
+    }
+  }
+  else if ( (MHD_ITC_IS_VALID_ (daemon->itc)) &&
+            (! MHD_itc_activate_ (daemon->itc, "r")) )
   {
 #ifdef HAVE_MESSAGES
     MHD_DLOG (daemon,
@@ -4223,6 +4329,8 @@ MHD_cleanup_connections (struct MHD_Daemon *daemon)
     }
     if (MHD_INVALID_SOCKET != pos->socket_fd)
       MHD_socket_close_chk_ (pos->socket_fd);
+    if (MHD_ITC_IS_VALID_ (pos->resume_itc))
+      MHD_itc_destroy_chk_ (pos->resume_itc);
     if (NULL != pos->addr)
       free (pos->addr);
     free (pos);

src/microhttpd/internal.h

@@ -1641,6 +1641,21 @@ struct MHD_Connection
    */
   bool resumed;
 
+  /**
+   * Inter-thread communication channel used to wake up the thread that
+   * handles this connection when the connection is resumed.
+   *
+   * Only initialised in thread-per-connection mode with
+   * #MHD_ALLOW_SUSPEND_RESUME enabled, invalid otherwise.
+   *
+   * The daemon-wide ITC cannot serve this purpose: the daemon's own
+   * thread waits on it as well, and #MHD_itc_clear_() drains it, so
+   * whichever of the two threads runs first consumes the notification
+   * and the other one sleeps through it.  A channel that only this
+   * connection's thread ever reads cannot lose the wake-up that way.
+   */
+  struct MHD_itc_ resume_itc;
+
   /**
    * Special member to be returned by #MHD_get_connection_info()
    */

src/testcurl/Makefile.am

@@ -59,6 +59,7 @@ endif
 THREAD_ONLY_TESTS += \
   test_get_wait \
   test_get_wait11 \
+  test_suspend_resume_thread \
   $(EMPTY_ITEM)
 
 if HEAVY_TESTS
@@ -272,6 +273,13 @@ test_quiesce_stream_CFLAGS = \
 test_quiesce_stream_LDADD = \
   $(PTHREAD_LIBS) $(LDADD)
 
+test_suspend_resume_thread_SOURCES = \
+  test_suspend_resume_thread.c
+test_suspend_resume_thread_CFLAGS = \
+  $(AM_CFLAGS) $(PTHREAD_CFLAGS)
+test_suspend_resume_thread_LDADD = \
+  $(PTHREAD_LIBS) $(LDADD)
+
 test_callback_SOURCES = \
   test_callback.c
 

src/testcurl/test_suspend_resume_thread.c

@@ -0,0 +1,393 @@
+/*
+     This file is part of libmicrohttpd
+     Copyright (C) 2026 Christian Grothoff
+
+     libmicrohttpd is free software; you can redistribute it and/or modify
+     it under the terms of the GNU General Public License as published
+     by the Free Software Foundation; either version 2, or (at your
+     option) any later version.
+
+     libmicrohttpd is distributed in the hope that it will be useful, but
+     WITHOUT ANY WARRANTY; without even the implied warranty of
+     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+     General Public License for more details.
+
+     You should have received a copy of the GNU General Public License
+     along with libmicrohttpd; see the file COPYING.  If not, write to the
+     Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
+     Boston, MA 02110-1301, USA.
+*/
+/**
+ * @file test_suspend_resume_thread.c
+ * @brief  Testcase for suspend/resume with a thread per connection
+ *
+ * The content reader suspends the connection and returns zero, which is
+ * the pattern microhttpd.h prescribes for "no body data yet".  The
+ * resume comes from a separate thread, as it would from an application
+ * that is waiting on some other I/O.  Several clients run at once and
+ * every response stalls repeatedly, so a single lost resume anywhere
+ * hangs the connection it belongs to and the body arrives short.
+ *
+ * @author Christian Grothoff
+ */
+#include "mhd_options.h"
+#include <stdlib.h>
+#include <stdio.h>
+#include <string.h>
+#include <pthread.h>
+#include <curl/curl.h>
+#include <microhttpd.h>
+
+/* Turn any MHD_PANIC() or failing mhd_assert() reached from this
+   test into a marked, classifiable test error (TESTING.md, P5). */
+#include "mhd_panic_tripwire.h"
+
+#ifndef WINDOWS
+#include <unistd.h>
+#endif
+
+/**
+ * Number of clients to run against the daemon at the same time.
+ */
+#define NUM_CLIENTS 4
+
+/**
+ * Total size of the response body.
+ */
+#define BODY_SIZE 64
+
+/**
+ * Number of bytes the content reader hands out at a time.
+ */
+#define CHUNK_SIZE 8
+
+/**
+ * Number of times the content reader stalls before each chunk.
+ */
+#define NUM_STALLS 3
+
+/**
+ * The byte the response body is made of.
+ */
+#define BODY_FILL 'x'
+
+
+struct ReaderData
+{
+  /**
+   * Connection to suspend and resume.
+   */
+  struct MHD_Connection *connection;
+
+  /**
+   * Number of stalls left before the next chunk is handed out.
+   */
+  unsigned int stalls_left;
+};
+
+
+static uint16_t port;
+
+static volatile unsigned int panicked;
+
+
+_MHD_NORETURN static void
+test_panic_cb (void *cls,
+               const char *file,
+               unsigned int line,
+               const char *reason)
+{
+  (void) cls;
+  fprintf (stderr,
+           "PANIC: %s at %s:%u\n",
+           (NULL != reason) ? reason : "",
+           file,
+           line);
+  panicked = 1;
+  exit (99);
+}
+
+
+static void *
+resume_thread (void *cls)
+{
+  /* Give the connection's thread a chance to actually park itself,
+     so that the resume has to travel between threads. */
+  (void) usleep (1000);
+  MHD_resume_connection (cls);
+  return NULL;
+}
+
+
+static ssize_t
+content_reader (void *cls,
+                uint64_t pos,
+                char *buf,
+                size_t max)
+{
+  struct ReaderData *data = cls;
+  pthread_t tid;
+
+  if (pos >= BODY_SIZE)
+    return MHD_CONTENT_READER_END_OF_STREAM;
+  if (0 != data->stalls_left)
+  {
+    /* No data yet.  Park the connection and report that, as
+       documented for #MHD_ContentReaderCallback. */
+    data->stalls_left--;
+    MHD_suspend_connection (data->connection);
+    if (0 != pthread_create (&tid,
+                             NULL,
+                             &resume_thread,
+                             data->connection))
+      return MHD_CONTENT_READER_END_WITH_ERROR;
+    (void) pthread_detach (tid);
+    return 0;
+  }
+  data->stalls_left = NUM_STALLS;
+  if (max > CHUNK_SIZE)
+    max = CHUNK_SIZE;
+  if (max > (size_t) (BODY_SIZE - pos))
+    max = (size_t) (BODY_SIZE - pos);
+  memset (buf,
+          BODY_FILL,
+          max);
+  return (ssize_t) max;
+}
+
+
+static void
+free_reader_data (void *cls)
+{
+  free (cls);
+}
+
+
+static enum MHD_Result
+ahc_echo (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)
+{
+  static int marker;
+  struct MHD_Response *response;
+  struct ReaderData *data;
+  enum MHD_Result ret;
+  (void) cls; (void) url; (void) method; (void) version;
+  (void) upload_data; (void) upload_data_size;
+
+  if (&marker != *req_cls)
+  {
+    *req_cls = &marker;
+    return MHD_YES;
+  }
+  data = malloc (sizeof (struct ReaderData));
+  if (NULL == data)
+    return MHD_NO;
+  data->connection = connection;
+  data->stalls_left = NUM_STALLS;
+  response = MHD_create_response_from_callback (BODY_SIZE,
+                                                4096,
+                                                &content_reader,
+                                                data,
+                                                &free_reader_data);
+  if (NULL == response)
+  {
+    free (data);
+    return MHD_NO;
+  }
+  ret = MHD_queue_response (connection,
+                            MHD_HTTP_OK,
+                            response);
+  MHD_destroy_response (response);
+  return ret;
+}
+
+
+struct Buffer
+{
+  size_t used;
+  char data[2 * BODY_SIZE];
+};
+
+
+static size_t
+copy_buffer (void *ptr,
+             size_t size,
+             size_t nmemb,
+             void *cls)
+{
+  struct Buffer *buf = cls;
+
+  if (0 == size * nmemb)
+    return 0;
+  if (buf->used + size * nmemb > sizeof (buf->data))
+    return 0; /* overflow */
+  memcpy (&buf->data[buf->used],
+          ptr,
+          size * nmemb);
+  buf->used += size * nmemb;
+  return size * nmemb;
+}
+
+
+/**
+ * Fetch the response once.
+ *
+ * @param cls unused
+ * @return NULL on success, non-NULL on failure
+ */
+static void *
+client_thread (void *cls)
+{
+  static int failure = 1;
+  char url[128];
+  struct Buffer buf;
+  CURL *c;
+  CURLcode errornum;
+  (void) cls;
+
+  memset (&buf, 0, sizeof (buf));
+  c = curl_easy_init ();
+  if (NULL == c)
+    return &failure;
+  snprintf (url,
+            sizeof (url),
+            "http://127.0.0.1:%u/",
+            (unsigned int) port);
+  curl_easy_setopt (c, CURLOPT_URL, url);
+  curl_easy_setopt (c, CURLOPT_WRITEFUNCTION, &copy_buffer);
+  curl_easy_setopt (c, CURLOPT_WRITEDATA, &buf);
+  curl_easy_setopt (c, CURLOPT_FAILONERROR, 1L);
+  curl_easy_setopt (c, CURLOPT_TIMEOUT, 30L);
+  curl_easy_setopt (c, CURLOPT_CONNECTTIMEOUT, 30L);
+  curl_easy_setopt (c, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_1_1);
+  errornum = curl_easy_perform (c);
+  curl_easy_cleanup (c);
+  if (CURLE_OK != errornum)
+  {
+    fprintf (stderr,
+             "curl_easy_perform() failed: `%s'\n",
+             curl_easy_strerror (errornum));
+    return &failure;
+  }
+  if (BODY_SIZE != buf.used)
+  {
+    fprintf (stderr,
+             "Got %u bytes of body, expected %u.\n",
+             (unsigned int) buf.used,
+             (unsigned int) BODY_SIZE);
+    return &failure;
+  }
+  if (BODY_SIZE != strspn (buf.data, "x"))
+  {
+    fprintf (stderr,
+             "Body has unexpected content.\n");
+    return &failure;
+  }
+  return NULL;
+}
+
+
+/**
+ * Run all clients against a daemon started with the given flags.
+ *
+ * @param flags the flags to start the daemon with
+ * @return 0 on success
+ */
+static unsigned int
+test_daemon (unsigned int flags)
+{
+  pthread_t clients[NUM_CLIENTS];
+  struct MHD_Daemon *d;
+  const union MHD_DaemonInfo *dinfo;
+  void *res;
+  unsigned int i;
+  unsigned int started;
+  unsigned int failures;
+
+  d = MHD_start_daemon (flags
+                        | MHD_USE_THREAD_PER_CONNECTION
+                        | MHD_USE_INTERNAL_POLLING_THREAD
+                        | MHD_ALLOW_SUSPEND_RESUME
+                        | MHD_USE_ERROR_LOG,
+                        0,
+                        NULL, NULL,
+                        &ahc_echo, NULL,
+                        MHD_OPTION_END);
+  if (NULL == d)
+  {
+    fprintf (stderr,
+             "Failed to start daemon with flags %x.\n",
+             flags);
+    return 1;
+  }
+  dinfo = MHD_get_daemon_info (d,
+                               MHD_DAEMON_INFO_BIND_PORT);
+  if ( (NULL == dinfo) ||
+       (0 == dinfo->port) )
+  {
+    MHD_stop_daemon (d);
+    fprintf (stderr,
+             "Failed to get the port number.\n");
+    return 1;
+  }
+  port = dinfo->port;
+
+  failures = 0;
+  for (started = 0; started < NUM_CLIENTS; started++)
+  {
+    if (0 != pthread_create (&clients[started],
+                             NULL,
+                             &client_thread,
+                             NULL))
+    {
+      fprintf (stderr,
+               "Failed to create a client thread.\n");
+      failures++;
+      break;
+    }
+  }
+  for (i = 0; i < started; i++)
+  {
+    res = NULL;
+    if (0 != pthread_join (clients[i],
+                           &res))
+    {
+      fprintf (stderr,
+               "Failed to join a client thread.\n");
+      failures++;
+    }
+    else if (NULL != res)
+      failures++;
+  }
+  MHD_stop_daemon (d);
+  return failures;
+}
+
+
+int
+main (int argc,
+      char *const *argv)
+{
+  unsigned int failures = 0;
+  (void) argc; (void) argv;
+
+  MHD_set_panic_func (&test_panic_cb,
+                      NULL);
+  if (0 != curl_global_init (CURL_GLOBAL_WIN32))
+    return 2;
+  /* Without an ITC, and with one; both use select() internally. */
+  failures += test_daemon (0);
+  failures += test_daemon (MHD_USE_ITC);
+  if (MHD_NO != MHD_is_feature_supported (MHD_FEATURE_POLL))
+    failures += test_daemon (MHD_USE_POLL | MHD_USE_ITC);
+  curl_global_cleanup ();
+  if (0 != panicked)
+    return 99;
+  return (0 == failures) ? 0 : 1;
+}