fix assert on appl thread existing after dds_fini

The thread state management used by the GC and the liveliness monitoring
lazily creates entries for application threads that call (certain)
Cyclone API functions.  It arranges for the entry allocated to such a
thread to be cleared once the thread terminates.

This means that if such a thread still exists once the last participant
is deleted (and Cyclone deinitializes), the corresponding thread entry
still exists as well.  The assertion that all (known) threads must have
stopped on shutting down Cyclone is therefore incorrect.

This commit introduces a special state for thread entries that were
created lazily.  It does monitor that they do not stay in the "awake"
state for too long, but otherwise doesn't care much about them.

Signed-off-by: Erik Boasson <eb@ilities.com>

src/core/ddsi/include/dds/ddsi/q_thread.h

@@ -49,8 +49,9 @@ typedef int32_t svtime_t; /* signed version */
 #define Q_THREAD_SCHEDPRIO_DEFAULT 0
 
 enum thread_state {
-  THREAD_STATE_ZERO,
+  THREAD_STATE_ZERO, /* known to be dead */
-  THREAD_STATE_ALIVE
+  THREAD_STATE_LAZILY_CREATED, /* lazily created in response because an application used it. Reclaimed if the thread terminates, but not considered an error if all of Cyclone is shutdown while this thread hasn't terminated yet */
+  THREAD_STATE_ALIVE /* known to be alive - for Cyclone internal threads */
 };
 
 struct logbuf;
@@ -101,8 +102,6 @@ DDS_EXPORT dds_retcode_t create_thread (struct thread_state1 **ts, const char *n
 DDS_EXPORT struct thread_state1 *lookup_thread_state_real (void);
 DDS_EXPORT int join_thread (struct thread_state1 *ts1);
 DDS_EXPORT void log_stack_traces (void);
-DDS_EXPORT struct thread_state1 *get_thread_state (ddsrt_thread_t id);
-DDS_EXPORT struct thread_state1 * init_thread_state (const char *tname);
 DDS_EXPORT void reset_thread_state (struct thread_state1 *ts1);
 DDS_EXPORT int thread_exists (const char *name);
 

src/core/ddsi/src/ddsi_threadmon.c

@@ -62,7 +62,7 @@ static uint32_t threadmon_thread (struct ddsi_threadmon *sl)
     if (ddsrt_cond_waitfor (&sl->cond, &sl->lock, config.liveliness_monitoring_interval))
       continue;
 
-    unsigned n_alive = 0;
+    unsigned n_alive = 0, n_unused = 0;
     nn_mtime_t tnow = now_mt ();
 
     LOG_THREAD_CPUTIME (next_thread_cputime);
@@ -81,8 +81,8 @@ static uint32_t threadmon_thread (struct ddsi_threadmon *sl)
       tlast = tnow;
       for (i = 0; i < thread_states.nthreads; i++)
       {
-        if (thread_states.ts[i].state != THREAD_STATE_ALIVE)
+        if (thread_states.ts[i].state == THREAD_STATE_ZERO)
-          n_alive++;
+          n_unused++;
         else
         {
           vtime_t vt = thread_states.ts[i].vtime;
@@ -105,7 +105,7 @@ static uint32_t threadmon_thread (struct ddsi_threadmon *sl)
       }
     }
 
-    if (n_alive == thread_states.nthreads)
+    if (n_alive + n_unused == thread_states.nthreads)
     {
       DDS_TRACE(": [%u] OK\n", n_alive);
       was_alive = true;

src/core/ddsi/src/q_thread.c

@@ -45,6 +45,8 @@ extern inline void thread_state_asleep (struct thread_state1 *ts1);
 extern inline void thread_state_awake (struct thread_state1 *ts1);
 extern inline void thread_state_awake_to_awake_no_nest (struct thread_state1 *ts1);
 
+static struct thread_state1 *init_thread_state (const char *tname, enum thread_state state);
+
 void *ddsrt_malloc_aligned_cacheline (size_t size)
 {
   /* This wastes some space, but we use it only once and it isn't a
@@ -115,16 +117,6 @@ void thread_states_fini (void)
   thread_states.ts = NULL;
 }
 
-static void cleanup_thread_state (void *data)
-{
-  struct thread_state1 *ts = get_thread_state(ddsrt_thread_self());
-  (void)data;
-  assert(ts->state == THREAD_STATE_ALIVE);
-  assert(vtime_asleep_p(ts->vtime));
-  reset_thread_state(ts);
-  ddsrt_fini();
-}
-
 static struct thread_state1 *find_thread_state (ddsrt_thread_t tid)
 {
   if (thread_states.ts) {
@@ -138,6 +130,19 @@ static struct thread_state1 *find_thread_state (ddsrt_thread_t tid)
   return NULL;
 }
 
+static void cleanup_thread_state (void *data)
+{
+  struct thread_state1 *ts = find_thread_state(ddsrt_thread_self());
+  (void)data;
+  if (ts)
+  {
+    assert(ts->state == THREAD_STATE_LAZILY_CREATED);
+    assert(vtime_asleep_p(ts->vtime));
+    reset_thread_state(ts);
+  }
+  ddsrt_fini();
+}
+
 static struct thread_state1 *lazy_create_thread_state (ddsrt_thread_t self)
 {
   /* This situation only arises for threads that were not created using
@@ -147,7 +152,7 @@ static struct thread_state1 *lazy_create_thread_state (ddsrt_thread_t self)
   char name[128];
   ddsrt_thread_getname (name, sizeof (name));
   ddsrt_mutex_lock (&thread_states.lock);
-  if ((ts1 = init_thread_state (name)) != NULL) {
+  if ((ts1 = init_thread_state (name, THREAD_STATE_LAZILY_CREATED)) != NULL) {
     ddsrt_init ();
     ts1->extTid = self;
     ts1->tid = self;
@@ -191,18 +196,11 @@ static uint32_t create_thread_wrapper (void *ptr)
 
 static int find_free_slot (const char *name)
 {
-  unsigned i;
+  for (unsigned i = 0; i < thread_states.nthreads; i++)
-  int cand;
-  for (i = 0, cand = -1; i < thread_states.nthreads; i++)
-  {
-    if (thread_states.ts[i].state != THREAD_STATE_ALIVE)
-      cand = (int) i;
     if (thread_states.ts[i].state == THREAD_STATE_ZERO)
-      break;
+      return (int) i;
-  }
-  if (cand == -1)
   DDS_FATAL("create_thread: %s: no free slot\n", name ? name : "(anon)");
-  return cand;
+  return -1;
 }
 
 void upgrade_main_thread (void)
@@ -215,7 +213,7 @@ void upgrade_main_thread (void)
   ts1 = &thread_states.ts[cand];
   if (ts1->state == THREAD_STATE_ZERO)
     assert (vtime_asleep_p (ts1->vtime));
-  ts1->state = THREAD_STATE_ALIVE;
+  ts1->state = THREAD_STATE_LAZILY_CREATED;
   ts1->tid = ddsrt_thread_self ();
   ts1->name = main_thread_name;
   ddsrt_mutex_unlock (&thread_states.lock);
@@ -231,7 +229,7 @@ const struct config_thread_properties_listelem *lookup_thread_properties (const
   return e;
 }
 
-struct thread_state1 *init_thread_state (const char *tname)
+static struct thread_state1 *init_thread_state (const char *tname, enum thread_state state)
 {
   int cand;
   struct thread_state1 *ts;
@@ -240,12 +238,9 @@ struct thread_state1 *init_thread_state (const char *tname)
     return NULL;
 
   ts = &thread_states.ts[cand];
-  if (ts->state == THREAD_STATE_ZERO)
-  {
   assert (vtime_asleep_p (ts->vtime));
-  }
   ts->name = ddsrt_strdup (tname);
-  ts->state = THREAD_STATE_ALIVE;
+  ts->state = state;
 
   return ts;
 }
@@ -259,7 +254,7 @@ dds_retcode_t create_thread (struct thread_state1 **ts1, const char *name, uint3
   ctxt = ddsrt_malloc (sizeof (*ctxt));
   ddsrt_mutex_lock (&thread_states.lock);
 
-  *ts1 = init_thread_state (name);
+  *ts1 = init_thread_state (name, THREAD_STATE_ALIVE);
   if (*ts1 == NULL)
     goto fatal;
 
@@ -336,28 +331,12 @@ void downgrade_main_thread (void)
   thread_states_init_static ();
 }
 
-struct thread_state1 *get_thread_state (ddsrt_thread_t id)
-{
-  unsigned i;
-  struct thread_state1 *ts = NULL;
-
-  for (i = 0; i < thread_states.nthreads; i++)
-  {
-    if (ddsrt_thread_equal (thread_states.ts[i].extTid, id))
-    {
-      ts = &thread_states.ts[i];
-      break;
-    }
-  }
-  return ts;
-}
-
 void log_stack_traces (void)
 {
   unsigned i;
   for (i = 0; i < thread_states.nthreads; i++)
   {
-    if (thread_states.ts[i].state == THREAD_STATE_ALIVE)
+    if (thread_states.ts[i].state != THREAD_STATE_ZERO)
     {
       log_stacktrace (thread_states.ts[i].name, thread_states.ts[i].tid);
     }