Joined threads now terminate, not park forever (and other TSAN-friendly fixes)

include/dmtcp.h

@@ -16,6 +16,8 @@
 #define DMTCP_H
 
 #include <netinet/ip.h>
+#include <assert.h>
+#include <pthread.h>
 #include <stddef.h>
 #include <stdio.h>
 #include <sys/socket.h>
@@ -43,8 +45,8 @@
 # define EXTERNC
 #endif // ifdef __cplusplus
 
-/* Define to the version of this package. */
-#define DMTCP_PLUGIN_API_VERSION "3"
+/* Bump when DmtcpPluginDescriptor_t changes ABI. */
+#define DMTCP_PLUGIN_API_VERSION "4"
 
 #ifdef __cplusplus
 namespace dmtcp {
@@ -145,6 +147,11 @@ typedef union _DmtcpEventData_t {
   struct {
     char *path;
   } realToVirtualPath, virtualToRealPath;
+
+  struct {
+    pthread_t pthread;
+    pid_t tid;
+  } pthreadInfo;
 } DmtcpEventData_t;
 
 typedef void (*HookFunctionPtr_t)(DmtcpEvent_t, DmtcpEventData_t *);
@@ -238,12 +245,78 @@ void dmtcp_initialize_plugin(void) __attribute((weak));
 typedef struct DmtcpUniqueProcessId {
   uint64_t _hostid;  // gethostid()
   uint64_t _time; // time()
-  pid_t _pid; // getpid()
+
+  union {
+    pid_t _pid; // getpid()
+    int32_t _;
+  };
+
   uint32_t _computation_generation; // computationGeneration()
 } DmtcpUniqueProcessId;
 
 int dmtcp_unique_pids_equal(DmtcpUniqueProcessId a, DmtcpUniqueProcessId b);
 
+typedef struct DmtcpInfo {
+  int argc;
+  const char **argv;
+} DmtcpInfo;
+
+enum ElfType {
+  Elf_32,
+  Elf_64
+};
+
+typedef struct {
+  uint64_t startAddr;
+  uint64_t endAddr;
+} MemRegion;
+
+typedef void (*PostRestartFnPtr_t)(double, int);
+#define DMTCP_CKPT_SIGNATURE "DMTCP_CHECKPOINT_IMAGE_v4.0\n"
+typedef struct {
+  char ckptSignature[32];
+
+  DmtcpUniqueProcessId upid;
+  DmtcpUniqueProcessId uppid;
+  DmtcpUniqueProcessId compGroup;
+
+  pid_t pid;
+  pid_t ppid;
+  pid_t sid;
+  pid_t gid;
+  pid_t fgid;
+  uint32_t isRootOfProcessTree;
+
+  uint32_t numPeers;
+  uint32_t elfType;
+
+  uint64_t clock_gettime_offset;
+  uint64_t getcpu_offset;
+  uint64_t gettimeofday_offset;
+  uint64_t time_offset;
+
+  // Reserve 3 * 30MB for restore buffer.
+#define RESTORE_BUF_TOTAL_SIZE (90 * 1024 * 1024)
+  MemRegion restoreBuf;
+
+  MemRegion vdso;
+  MemRegion vvar;
+  MemRegion vvarVClock;
+
+  uint64_t savedBrk;
+  uint64_t endOfStack;
+
+  uint64_t postRestartAddr;
+  //void (*post_restart)(double, int);
+
+  char procname[1024];
+  char procSelfExe[1024];
+
+  char padding[1792];
+} DmtcpCkptHeader;
+
+static_assert(sizeof(DmtcpCkptHeader) == 4096, "DmtcpCkptHeader must be 4096 bytes");
+
 // FIXME:
 // If a plugin is not compiled with defined(__PIC__) and we can verify
 // that we're using DMTCP (environment variables), and dmtcp_is_enabled
@@ -329,8 +402,6 @@ const char *dmtcp_get_ckpt_files_subdir(void);
 int dmtcp_should_ckpt_open_files(void);
 int dmtcp_allow_overwrite_with_ckpted_files(void);
 int dmtcp_skip_truncate_file_at_restart(const char* path);
-void dmtcp_set_restore_buf_addr(void *new_addr, uint64_t len);
-uint64_t dmtcp_restore_buf_len();
 
 int dmtcp_get_ckpt_signal(void);
 const char *dmtcp_get_uniquepid_str(void) __attribute__((weak));
@@ -428,13 +499,22 @@ int dmtcp_protected_environ_fd(void);
  *  discovers a pid without going through a system call (e.g., through
  *  the proc filesystem), use this to virtualize the pid.
  */
-pid_t dmtcp_real_to_virtual_pid(pid_t realPid) __attribute((weak));
-pid_t dmtcp_virtual_to_real_pid(pid_t virtualPid) __attribute((weak));
-
+pid_t dmtcp_pid_real_to_virtual(pid_t realPid) __attribute((weak));
+pid_t dmtcp_pid_virtual_to_real(pid_t virtualPid) __attribute((weak));
+
+// Returns 1 if virtual_tid names TSAN's own background thread, else 0.
+int dmtcp_is_tsan_background_thread(int virtual_tid) __attribute((weak));
+#define dmtcp_is_tsan_background_thread(virtual_tid) \
+  (dmtcp_is_tsan_background_thread ? \
+   dmtcp_is_tsan_background_thread(virtual_tid) : 0)
+
+// McMini helpers: translate pids only when running under DMTCP.
+// (DMTCP plugin API v4 renamed dmtcp_{real_to_virtual,virtual_to_real}_pid to
+// dmtcp_pid_{real_to_virtual,virtual_to_real}.)
 #define mcmini_virtual_pid(PID) \
-  (dmtcp_is_enabled() ? dmtcp_real_to_virtual_pid((PID)) : (PID))
+  (dmtcp_is_enabled() ? dmtcp_pid_real_to_virtual((PID)) : (PID))
 #define mcmini_real_pid(PID) \
-  (dmtcp_is_enabled() ? dmtcp_virtual_to_real_pid((PID)) : (PID))
+  (dmtcp_is_enabled() ? dmtcp_pid_virtual_to_real((PID)) : (PID))
 
 // bq_file -> "batch queue file"; used only by batch-queue plugin
 int dmtcp_is_bq_file(const char *path) __attribute((weak));
@@ -445,7 +525,7 @@ int dmtcp_bq_restore_file(const char *path,
                           int type) __attribute((weak));
 
 /*  These next two functions are defined in contrib/ckptfile/ckptfile.cpp
- *  But they are currently used only in src/plugin/ipc/file/fileconnection.cpp
+ *  But they are currently used only in src/plugin/file/fileconnection.cpp
  *    and in a trivial fashion.  These are intended for future extensions.
  */
 int dmtcp_must_ckpt_file(const char *path) __attribute((weak));

include/mcmini/spy/intercept/interception.h

@@ -29,6 +29,11 @@ int libdmtcp_pthread_create(pthread_t *thread, const pthread_attr_t *attr,
 int pthread_join(pthread_t thread, void**);
 int libpthread_pthread_join(pthread_t thread, void**);
 int libdmtcp_pthread_join(pthread_t thread, void**);
+// TSan-safe (libtsan-bypassing) handle for pthread_timedjoin_np, used by
+// mc_pthread_join's RECORD loop. Calling pthread_timedjoin_np directly resolves
+// to libtsan's interceptor, which trips a thread-registry CHECK. See
+// TSAN-McMini-DMTCP.txt.
+int libpthread_timedjoin_np(pthread_t thread, void**, const struct timespec*);
 
 int libpthread_mutex_init(pthread_mutex_t *, const pthread_mutexattr_t *);
 int libpthread_mutex_lock(pthread_mutex_t *);

src/common/multithreaded_fork.c

@@ -177,7 +177,7 @@ pid_t get_tid_from_pthread_descriptor(pthread_t pthread_descriptor) {
   int offset = pthreadDescriptorTidOffset();
   pid_t ctid = *(pid_t*)((char*)(pthread_descriptor) + offset);
 #ifdef DMTCP
-  pid_t virttid = dmtcp_real_to_virtual_pid(ctid);
+  pid_t virttid = dmtcp_pid_real_to_virtual(ctid);
   ctid = (virttid ? virttid : ctid);
 #endif
   return ctid;
@@ -209,7 +209,7 @@ int get_child_threads(int child_threads[]) {
       if (atoi(entry->d_name) != 0) {
         pid_t nexttid = atoi(entry->d_name);
 #ifdef DMTCP
-        pid_t virttid = dmtcp_real_to_virtual_pid(nexttid);
+        pid_t virttid = dmtcp_pid_real_to_virtual(nexttid);
         nexttid = (virttid ? virttid : nexttid);
 #endif
         child_threads[i++] = nexttid;

src/lib/dmtcp-callback.c

@@ -1,6 +1,5 @@
 #define _GNU_SOURCE
 #include <assert.h>
-#include <dirent.h>
 #include <errno.h>
 #include <fcntl.h>  // man 2 open
 #include <pthread.h>
@@ -321,23 +320,27 @@ static void *template_thread(void *unused) {
   // to ensure a stable recorded state is pointless: we're not going to read it
   // anyway! This is an only a potential optimization for later though.
 
+  // Counting via a live /proc/self/task scan is a TOCTOU race: DMTCP
+  // recreates checkpointed threads asynchronously via clone(), so a scan
+  // that runs before it has finished recreating all of them undercounts,
+  // and this barrier then releases before every thread has actually
+  // restarted (confirmed empirically: one thread's own restart-completion
+  // signal can arrive after this barrier already declared a "consistent
+  // state").
+  //
+  // head_record_mode instead gives an exact, race-free count: it only ever
+  // gets THREAD entries for genuine target threads (the template thread and
+  // the checkpoint thread never go through libmcmini's wrapped
+  // pthread_create(), so neither is ever recorded here), and -- since a
+  // DMTCP checkpoint is a full memory snapshot -- this list is preserved
+  // exactly as it was at record time across every restart, with no
+  // dependence on restart-time scheduling.
   int thread_count = 0;
-  struct dirent *entry;
-  DIR *dp = opendir("/proc/self/task");
-  if (dp == NULL) {
-    perror("opendir");
-    mc_exit(EXIT_FAILURE);
-  }
-
-  while ((entry = readdir(dp)))
-    if (strcmp(entry->d_name, ".") != 0 && strcmp(entry->d_name, "..") != 0)
+  for (rec_list *entry = head_record_mode; entry != NULL; entry = entry->next) {
+    if (entry->vo.type == THREAD && entry->vo.thrd_state.status == ALIVE) {
       thread_count++;
-
-  // We don't want to count the template thread nor
-  // the checkpoint thread, but these will appear in
-  // `/proc/self/tasks`
-  thread_count -= 2;
-  closedir(dp);
+    }
+  }
   log_debug(
       "There are %d threads... waiting for them to get into a consistent "
       "state...\n",

src/lib/interception.c

@@ -16,6 +16,7 @@ typeof(&pthread_create) libpthread_pthread_create_ptr;
 typeof(&pthread_create) libdmtcp_pthread_create_ptr;
 typeof(&pthread_join) libpthread_pthread_join_ptr;
 typeof(&pthread_join) libdmtcp_pthread_join_ptr;
+typeof(&pthread_timedjoin_np) libpthread_timedjoin_np_ptr;
 typeof(&pthread_mutex_init) pthread_mutex_init_ptr;
 typeof(&pthread_mutex_lock) pthread_mutex_lock_ptr;
 typeof(&pthread_mutex_trylock) pthread_mutex_trylock_ptr;
@@ -69,6 +70,7 @@ void mc_load_intercepted_pthread_functions(void) {
 
   libpthread_pthread_create_ptr = dlsym(libpthread_handle, "pthread_create");
   libpthread_pthread_join_ptr = dlsym(libpthread_handle, "pthread_join");
+  libpthread_timedjoin_np_ptr = dlsym(libpthread_handle, "pthread_timedjoin_np");
   pthread_mutex_init_ptr = dlsym(libpthread_handle, "pthread_mutex_init");
   pthread_mutex_lock_ptr = dlsym(libpthread_handle, "pthread_mutex_lock");
   pthread_mutex_trylock_ptr = dlsym(libpthread_handle, "pthread_mutex_trylock");
@@ -228,6 +230,11 @@ int libpthread_pthread_join(pthread_t thread, void **rv) {
   libmcmini_init();
   return (*libpthread_pthread_join_ptr)(thread, rv);
 }
+int libpthread_timedjoin_np(pthread_t thread, void **rv,
+                            const struct timespec *abstime) {
+  libmcmini_init();
+  return (*libpthread_timedjoin_np_ptr)(thread, rv, abstime);
+}
 int libdmtcp_pthread_join(pthread_t thread, void **rv) {
   libmcmini_init();
   return (*libdmtcp_pthread_join_ptr)(thread, rv);

src/lib/wrappers.c

@@ -18,6 +18,11 @@
 typedef struct pthread_map {
     pthread_t thread;
     runner_id_t value;
+    // Posted by whichever thread eventually calls pthread_join() on this
+    // one (mc_pthread_join()'s TARGET_BRANCH case), waited on by this
+    // thread itself before it may really terminate (mc_exit_thread_in_child()).
+    // See the two functions for the full rationale.
+    sem_t exit_permission_sem;
     struct pthread_map *next;
 } pthread_map_t;
 
@@ -29,6 +34,7 @@ void insert_pthread_map(pthread_t t, runner_id_t v) {
     pthread_map_t *n = malloc(sizeof *n);
     n->thread = t;
     n->value = v;
+    libpthread_sem_init(&n->exit_permission_sem, 0, 0);
     n->next = head;
     head = n;
     pthread_rwlock_unlock(&pthread_map_lock);
@@ -39,12 +45,27 @@ runner_id_t search_pthread_map(pthread_t t) {
     pthread_map_t *cur = head;
     while (cur) {
         if (pthread_equal(cur->thread, t)) {
-            return cur->value;
+            break;
         }
         cur = cur->next;
     }
     pthread_rwlock_unlock(&pthread_map_lock);
-    return RID_INVALID;
+    return cur == NULL ? RID_INVALID : cur->value;
+}
+
+// Returns the given thread's own exit_permission_sem (see pthread_map_t),
+// or NULL if the thread is not (yet) registered.
+sem_t *find_exit_permission_sem(pthread_t t) {
+  pthread_rwlock_rdlock(&pthread_map_lock);
+  pthread_map_t *cur = head;
+  while (cur) {
+    if (pthread_equal(cur->thread, t)) {
+      break;
+    }
+    cur = cur->next;
+  }
+  pthread_rwlock_unlock(&pthread_map_lock);
+  return cur == NULL ? NULL : &cur->exit_permission_sem;
 }
 
 
@@ -335,7 +356,16 @@ void mc_exit_thread_in_child(void) {
   thread_get_mailbox()->type = THREAD_EXIT_TYPE;
   thread_wake_scheduler_and_wait();
   thread_awake_scheduler_for_thread_finish_transition();
-  thread_block_indefinitely();
+
+  // Wait for whichever thread eventually calls pthread_join() on this one
+  // (see mc_pthread_join()'s TARGET_BRANCH case) to grant permission before
+  // this thread is allowed to really terminate. This keeps this thread's
+  // pthread_t/tid valid for exactly as long as a real, unjoined POSIX
+  // thread's would be -- no longer -- rather than parking it forever
+  // regardless of whether anyone ever joins it.
+  sem_t *exit_permission = find_exit_permission_sem(pthread_self());
+  assert(exit_permission != NULL);
+  libpthread_sem_wait(exit_permission);
 }
 
 void mc_exit_main_thread_in_child(void) {
@@ -726,7 +756,10 @@ int mc_pthread_join(pthread_t t, void **rv) {
 
       struct timespec time = {.tv_sec = 2, .tv_nsec = 0};
       while (1) {
-        int rc = pthread_timedjoin_np(t, rv, &time);
+        // Use the libtsan-bypassing handle: a direct pthread_timedjoin_np would
+        // hit libtsan's interceptor and trip its thread-registry CHECK under
+        // DMTCP. See TSAN-McMini-DMTCP.txt.
+        int rc = libpthread_timedjoin_np(t, rv, &time);
         if (rc == 0) {  // Join succeeded
           libpthread_mutex_lock(&rec_list_lock);
           thread_record->vo.thrd_state.status = EXITED;
@@ -769,7 +802,16 @@ int mc_pthread_join(pthread_t t, void **rv) {
       memcpy_v(thread_get_mailbox()->cnts, &rid, sizeof(runner_id_t));
       thread_get_mailbox()->type = THREAD_JOIN_TYPE;
       thread_wake_scheduler_and_wait();
-      return 0;
+
+      // Grant the target thread permission to really terminate now that
+      // it's been joined (see mc_exit_thread_in_child()), then perform a
+      // genuine join, so its OS thread actually dies -- and *rv is
+      // genuinely populated -- before we return, instead of only
+      // simulating success at the model level.
+      sem_t *exit_permission = find_exit_permission_sem(t);
+      assert(exit_permission != NULL);
+      libpthread_sem_post(exit_permission);
+      return libpthread_pthread_join(t, rv);
     }
     default: {
       libc_abort();

src/mcmini/mcmini.cpp

@@ -142,14 +142,23 @@ void found_abnormal_termination(
   for (const auto& t : program_model.get_trace()) {
     ss << "thread " << t->get_executor() << ": " << t->to_string() << "\n";
   }
+  // `ub.culprit` may not have a pending transition in the model's current
+  // view: this callback can also fire for a termination_error raised while
+  // `coordinator::return_to_depth()` is replaying already-recorded history
+  // against a freshly restarted process, and by the replay's target depth
+  // the culprit thread may already have exited in the model.
   const transition* terminator =
       program_model.get_pending_transition_for(ub.culprit);
-  ss << "thread " << terminator->get_executor() << ": "
-     << terminator->to_string() << "\n";
+  if (terminator != nullptr) {
+    ss << "thread " << terminator->get_executor() << ": "
+       << terminator->to_string() << "\n";
+  } else {
+    ss << "thread " << ub.culprit << ": (no longer pending)\n";
+  }
 
   ss << "\nNEXT THREAD OPERATIONS\n";
   for (const auto& tpair : program_model.get_pending_transitions()) {
-    if (tpair.first == terminator->get_executor()) {
+    if (terminator != nullptr && tpair.first == terminator->get_executor()) {
       ss << "thread " << tpair.first << ": executing"
          << "\n";
     } else {