xsignal

c-core/CMakeLists.txt

@@ -1,5 +1,5 @@
 cmake_minimum_required(VERSION 3.20.2 FATAL_ERROR)
-project(deciphon VERSION 0.15.7 LANGUAGES C)
+project(deciphon VERSION 0.15.8 LANGUAGES C)
 
 include(cmake/warnings.cmake)
 include(cmake/sanitizers.cmake)

c-core/scan.c

@@ -1,9 +1,3 @@
-#if !defined(_POSIX_C_SOURCE) || _POSIX_C_SOURCE < 200809L
-#undef _POSIX_C_SOURCE
-#define _POSIX_C_SOURCE 200809L
-#endif
-#include <signal.h>
-
 #include "scan.h"
 #include "database_reader.h"
 #include "debug.h"
@@ -15,6 +9,8 @@
 #include "sequence_queue.h"
 #include "thread.h"
 #include "thread_params.h"
+#include "xsignal.h"
+#include <omp.h>
 #include <stdlib.h>
 
 struct scan
@@ -116,14 +112,13 @@ int scan_run(struct scan *x, char const *product_dir)
   int rc = 0;
   int num_threads = x->params.num_threads;
   x->interrupted = false;
+  struct xsignal *xsignal = xsignal_new();
+
   debug("%d thread(s)", num_threads);
 
   if ((rc = product_open(&x->product, num_threads, product_dir)))
     defer_return(rc);
 
-  sigset_t signal_mask;
-  sigprocmask(SIG_BLOCK, NULL, &signal_mask);
-
   for (int i = 0; i < num_threads; ++i)
   {
     struct thread_params params = {&x->db.protein,
@@ -135,17 +130,24 @@ int scan_run(struct scan *x, char const *product_dir)
     if ((rc = thread_setup(x->threads + i, params))) defer_return(rc);
   }
 
-#pragma omp parallel for default(none) shared(x, num_threads, rc)
+#pragma omp parallel for default(none) shared(x, num_threads, rc, xsignal)
   for (int i = 0; i < num_threads; ++i)
   {
-    int r = thread_run(x->threads + i, &x->sequences, &x->done_proteins);
+    struct xsignal *signal = omp_get_thread_num() == 0 ? xsignal : NULL;
+    int *proteins = &x->done_proteins;
+    int r = thread_run(x->threads + i, &x->sequences, proteins, signal);
+    if (!r)
+    {
+      for (int i = 0; i < num_threads; ++i)
+        x->threads[i].interrupted = true;
+    }
 
 #pragma omp critical
     if (r && !rc) rc = r;
   }
 
 defer:
-  sigprocmask(SIG_SETMASK, &signal_mask, NULL);
+  xsignal_del(xsignal);
 
   for (int i = 0; i < num_threads; ++i)
   {

c-core/thread.c

@@ -1,9 +1,3 @@
-#if !defined(_POSIX_C_SOURCE) || _POSIX_C_SOURCE < 200809L
-#undef _POSIX_C_SOURCE
-#define _POSIX_C_SOURCE 200809L
-#endif
-#include <signal.h>
-
 #include "thread.h"
 #include "chararray.h"
 #include "database_reader.h"
@@ -24,6 +18,7 @@
 #include "trellis.h"
 #include "viterbi.h"
 #include "window.h"
+#include "xsignal.h"
 
 void thread_init(struct thread *x)
 {
@@ -85,20 +80,10 @@ static int process_window(struct thread *, int protein_idx,
                           struct window const *);
 
 int thread_run(struct thread *x, struct sequence_queue const *sequences,
-               int *done_proteins)
+               int *done_proteins, struct xsignal *xsignal)
 {
   int rc = 0;
 
-  sigset_t sigmask;
-  sigemptyset(&sigmask);
-  sigaddset(&sigmask, SIGINT);
-  sigaddset(&sigmask, SIGTERM);
-  sigprocmask(SIG_BLOCK, &sigmask, NULL);
-
-  sigset_t sigpend;
-  sigemptyset(&sigpend);
-  int signal = 0;
-
   struct protein_iter *protein_iter = &x->iter;
 
   if ((rc = protein_iter_rewind(protein_iter))) goto cleanup;
@@ -116,18 +101,11 @@ int thread_run(struct thread *x, struct sequence_queue const *sequences,
       int last_hit_pos = -1;
       while (window_next(&w, last_hit_pos))
       {
+        if (x->interrupted) goto cleanup;
         int protein_idx = protein_iter_idx(protein_iter);
         if ((rc = process_window(x, protein_idx, &w))) break;
 
-        sigpending(&sigpend);
-        if (sigismember(&sigpend, SIGINT) || sigismember(&sigpend, SIGTERM))
-        {
-          if (sigwait(&sigmask, &signal) == 0)
-          {
-            x->interrupted = true;
-            goto cleanup;
-          }
-        }
+        if (xsignal && xsignal_interrupted(xsignal)) x->interrupted = true;
       }
     }
 

c-core/thread.h

@@ -13,6 +13,7 @@
 struct product_thread;
 struct sequence_queue;
 struct viterbi;
+struct xsignal;
 
 struct thread
 {
@@ -34,7 +35,7 @@ struct thread
 void thread_init(struct thread *);
 int  thread_setup(struct thread *, struct thread_params);
 void thread_cleanup(struct thread *);
-int  thread_run(struct thread *, struct sequence_queue const *, int *done_proteins);
+int  thread_run(struct thread *, struct sequence_queue const *, int *done_proteins, struct xsignal *);
 bool thread_interrupted(struct thread const *);
 
 #endif

c-core/xsignal.c

@@ -0,0 +1,45 @@
+#if !defined(_POSIX_C_SOURCE) || _POSIX_C_SOURCE < 200809L
+#undef _POSIX_C_SOURCE
+#define _POSIX_C_SOURCE 200809L
+#endif
+#include <signal.h>
+
+#include "xsignal.h"
+#include <stdlib.h>
+
+struct xsignal
+{
+  sigset_t old_mask;
+  sigset_t new_sigmask;
+};
+
+
+struct xsignal *xsignal_new(void)
+{
+  struct xsignal *x = malloc(sizeof(*x));
+  sigemptyset(&x->new_sigmask);
+  sigaddset(&x->new_sigmask, SIGINT);
+  sigaddset(&x->new_sigmask, SIGTERM);
+  sigprocmask(SIG_BLOCK, &x->new_sigmask, &x->old_mask);
+  return x;
+}
+
+bool xsignal_interrupted(struct xsignal *x)
+{
+  sigset_t sigpend;
+  int signal = 0;
+
+  sigemptyset(&sigpend);
+  sigpending(&sigpend);
+
+  if (sigismember(&sigpend, SIGINT) || sigismember(&sigpend, SIGTERM))
+  {
+    if (sigwait(&x->new_sigmask, &signal) == 0) return true;
+  }
+  return false;
+}
+
+void xsignal_del(struct xsignal *x)
+{
+  sigprocmask(SIG_SETMASK, &x->old_mask, NULL);
+}

c-core/xsignal.h

@@ -0,0 +1,12 @@
+#ifndef XSIGNAL_H
+#define XSIGNAL_H
+
+#include <stdbool.h>
+
+struct xsignal;
+
+struct xsignal *xsignal_new(void);
+bool            xsignal_interrupted(struct xsignal *);
+void            xsignal_del(struct xsignal *);
+
+#endif