Make loops cancellable: poll interrupts on back edges, and add MP:WITH-TIMEOUT

src/core/bytecode.cc

@@ -217,6 +217,22 @@ static unsigned char* long_dispatch(VirtualMachine&, unsigned char*, MultipleVal
                                     core::T_O**, core::T_O**, size_t, core::T_O**, uint8_t);
 
 SYMBOL_EXPORT_SC_(KeywordPkg, name);
+
+// Poll interrupts on backward branches so loops of pure opcodes stay cancellable.
+static inline unsigned char* vm_branch(VirtualMachine& vm, ThreadLocalState* thread, unsigned char* pc, core::T_O** sp,
+                                       int32_t rel) {
+  pc += rel;
+  if (rel < 0) {
+    core::Cons_sp head = thread->_PendingInterruptsHead.load(std::memory_order_acquire);
+    if (thread->pending_signals_p() || (static_cast<bool>(head) && head->cdr().notnilp())) {
+      // The handler can cons or unwind, so the GC must see the current frame.
+      vm._pc = pc;
+      vm._stackPointer = sp;
+      gctools::handle_all_queued_interrupts();
+    }
+  }
+  return pc;
+}
 #ifdef DEBUG_VIRTUAL_MACHINE
 __attribute__((optnone))
 #endif
@@ -564,19 +580,19 @@ bytecode_vm(VirtualMachine& vm, T_O** literals, T_O** closed, Closure_O* closure
     case vm_code::jump_8: {
       int8_t rel = *(pc + 1);
       DBG_VM1("jump %" PRId8 "\n", rel);
-      pc += rel;
+      pc = vm_branch(vm, thread, pc, sp, rel);
       break;
     }
     case vm_code::jump_16: {
       int16_t rel = read_s16(pc + 1);
       DBG_VM("jump %" PRId16 "\n", rel);
-      pc += rel;
+      pc = vm_branch(vm, thread, pc, sp, rel);
       break;
     }
     case vm_code::jump_24: {
       int32_t rel = read_label(pc, 3);
       DBG_VM("jump %" PRId32 "\n", rel);
-      pc += rel;
+      pc = vm_branch(vm, thread, pc, sp, rel);
       break;
     }
     case vm_code::jump_if_8: {
@@ -585,7 +601,7 @@ bytecode_vm(VirtualMachine& vm, T_O** literals, T_O** closed, Closure_O* closure
       T_sp tval((gctools::Tagged)vm.pop(sp));
       VM_RECORD_PLAYBACK(tval.raw_(), "vm_jump_if_8");
       if (tval.notnilp())
-        pc += rel;
+        pc = vm_branch(vm, thread, pc, sp, rel);
       else
         pc += 2;
       break;
@@ -595,7 +611,7 @@ bytecode_vm(VirtualMachine& vm, T_O** literals, T_O** closed, Closure_O* closure
       DBG_VM("jump-if %" PRId16 "\n", rel);
       T_sp tval((gctools::Tagged)vm.pop(sp));
       if (tval.notnilp())
-        pc += rel;
+        pc = vm_branch(vm, thread, pc, sp, rel);
       else
         pc += 3;
       break;
@@ -605,7 +621,7 @@ bytecode_vm(VirtualMachine& vm, T_O** literals, T_O** closed, Closure_O* closure
       DBG_VM("jump-if %" PRId32 "\n", rel);
       T_sp tval((gctools::Tagged)vm.pop(sp));
       if (tval.notnilp())
-        pc += rel;
+        pc = vm_branch(vm, thread, pc, sp, rel);
       else
         pc += 4;
       break;
@@ -618,7 +634,7 @@ bytecode_vm(VirtualMachine& vm, T_O** literals, T_O** closed, Closure_O* closure
         pc += 2;
       } else {
         vm.push(sp, tval.raw_());
-        pc += rel;
+        pc = vm_branch(vm, thread, pc, sp, rel);
       }
       break;
     }
@@ -630,7 +646,7 @@ bytecode_vm(VirtualMachine& vm, T_O** literals, T_O** closed, Closure_O* closure
         pc += 3;
       } else {
         vm.push(sp, tval.raw_());
-        pc += rel;
+        pc = vm_branch(vm, thread, pc, sp, rel);
       }
       break;
     }

src/core/lispStream.cc

@@ -1353,7 +1353,9 @@ CL_DEFUN T_mv core__read_fd(int filedes, SimpleBaseString_sp buffer) {
   size_t buffer_length = cl__length(buffer);
   unsigned char* buffer_data = &(*buffer)[0];
   while (1) {
-    int num = read(filedes, buffer_data, buffer_length);
+    // Park: read() blocks until data arrives. Safe to hold BUFFER_DATA across it
+    // because no GC variant relocates (NON_MOVING_GC).
+    int num = BEGIN_PARK { return (int)read(filedes, buffer_data, buffer_length); } END_PARK;
     if (!(num < 0 && errno == EINTR)) {
       if (num < 0) {
         return Values(make_fixnum(num), make_fixnum(errno));

src/core/unixfsys.cc

@@ -37,6 +37,7 @@ THE SOFTWARE.
 */
 
 #include <clasp/core/foundation.h>
+#include <clasp/gctools/park.h>
 
 #include <string.h>
 #include <stdio.h>
@@ -294,7 +295,8 @@ The status can be passed to //core:wifexited// and //core:wifsignaled//. )")
 DOCGROUP(clasp);
 CL_DEFUN T_mv core__wait() {
   int status;
-  pid_t p = wait(&status);
+  // Park: wait() blocks until a child exits, which may be never.
+  pid_t p = BEGIN_PARK { return wait(&status); } END_PARK;
   return Values(make_fixnum(p), make_fixnum(status));
 };
 
@@ -1707,7 +1709,9 @@ DOCGROUP(clasp);
 CL_DEFUN T_mv ext__system(String_sp cmd) {
   ASSERT(cl__stringp(cmd));
   string command = cmd->get_std_string();
-  int ret = system(command.c_str());
+  // Park: system() blocks for an unbounded time, so the thread must be GC-safe
+  // and marked blocking, which is what lets an interrupt wake it with SIGCONT.
+  int ret = BEGIN_PARK { return system(command.c_str()); } END_PARK;
   if (ret == 0) {
     return Values(core::make_fixnum(0));
   } else {
@@ -1828,7 +1832,7 @@ CL_DEFUN T_mv ext__vfork_execvp(List_sp call_and_arguments, T_sp return_stream)
       while (b_done == false) {
         errno = 0;
 
-        wait_ret = wait(&status);
+        wait_ret = BEGIN_PARK { return wait(&status); } END_PARK;
 
         if (WIFEXITED(status)) {
           child_exit_status = WEXITSTATUS(status);
@@ -1946,7 +1950,7 @@ CL_DEFUN T_mv ext__fork_execvp(List_sp call_and_arguments, T_sp return_stream) {
     } else {
       // Parent
       int status;
-      pid_t wait_ret = wait(&status);
+      pid_t wait_ret = BEGIN_PARK { return wait(&status); } END_PARK;
       // Clean up args
       for (int i(0); i < execvp_args.size() - 1; ++i)
         free((void*)execvp_args[i]);
@@ -2003,7 +2007,10 @@ CL_DEFUN T_mv core__select(int nfds, FdSet_sp readfds, FdSet_sp writefds, FdSet_
   struct timeval timeout;
   timeout.tv_sec = seconds;
   timeout.tv_usec = microseconds;
-  int num = select(nfds, &readfds->_fd_set, &writefds->_fd_set, &errorfds->_fd_set, &timeout);
+  // Park: select() blocks for up to the caller's timeout.
+  int num = BEGIN_PARK {
+    return select(nfds, &readfds->_fd_set, &writefds->_fd_set, &errorfds->_fd_set, &timeout);
+  } END_PARK;
   if (num < 0) {
     return Values(make_fixnum(num), make_fixnum(errno));
   }

src/lisp/kernel/cleavir/translate.lisp

@@ -255,12 +255,44 @@ function-or-placeholder - the llvm function or a placeholder for
                                          (inst-source instruction)
                                          999902))
     (call-next-method)))
+(defvar *laid-out-iblocks*)
+(defvar *safepoint-counter* nil)
+
+;; Iblocks are laid out in forward flow order, so a branch to one already laid
+;; out is a back edge - the native counterpart of the VM's negative jump offset.
+(defun back-edge-p (instruction)
+  (and (boundp '*laid-out-iblocks*)
+       (loop for succ in (bir:next instruction)
+             thereis (gethash succ *laid-out-iblocks*))))
+
+;; cc_safepoint is an out-of-line unwinding call, so polling every back edge
+;; costs ~3x on a tight loop; sample one edge in +safepoint-sample+ instead.
+(defparameter +safepoint-sample+ 255)
+
+(defun emit-back-edge-safepoint ()
+  (let* ((n (cmp:irc-typed-load cmp:%size_t% *safepoint-counter*))
+         (n1 (cmp:irc-add n (cmp:jit-constant-size_t 1)))
+         (poll (cmp:irc-basic-block-create "safepoint-poll"))
+         (cont (cmp:irc-basic-block-create "safepoint-cont")))
+    (cmp:irc-store n1 *safepoint-counter*)
+    (cmp:irc-cond-br (cmp:irc-icmp-eq
+                      (cmp:irc-and n1 (cmp:jit-constant-size_t +safepoint-sample+))
+                      (cmp:jit-constant-size_t 0))
+                     poll cont)
+    (cmp:irc-begin-block poll)
+    (%intrinsic-invoke-if-landing-pad-or-call "cc_safepoint" ())
+    (cmp:irc-br cont)
+    (cmp:irc-begin-block cont)))
+
 (defmethod translate-terminator :around
     ((instruction bir:instruction) abi next)
   (declare (ignore abi next))
   (cmp:with-debug-info-source-position ((ensure-origin
                                          (inst-source instruction)
                                          999903))
+    ;; Poll interrupts on back edges so native loops stay cancellable.
+    (when (and *safepoint-counter* (back-edge-p instruction))
+      (emit-back-edge-safepoint))
     (call-next-method)))
 
 (defmethod translate-terminator ((instruction bir:unreachable)
@@ -1876,6 +1908,8 @@ function-or-placeholder - the llvm function or a placeholder for
               do (setf (gethash phi *datum-values*) dat))))))
 
 (defun layout-iblock (iblock abi)
+  (when (boundp '*laid-out-iblocks*)
+    (setf (gethash iblock *laid-out-iblocks*) t))
   (cmp:irc-begin-block (iblock-tag iblock))
   (cmp:with-landing-pad (maybe-entry-landing-pad
                          (bir:dynamic-environment iblock) *tags*)
@@ -1991,11 +2025,18 @@ function-or-placeholder - the llvm function or a placeholder for
                                          (arguments llvm-function-info))
                   when lexical ; skip unused fixed
                     do (setf (gethash lexical *datum-values*) arg)))
-          ;; Branch to the start block.
-          (cmp:irc-br (iblock-tag (bir:start ir)))
-          ;; Lay out blocks.
-          (bir:do-iblocks (ib ir)
-            (layout-iblock ib abi))))))
+          ;; Counter for sampled back-edge safepoints; must live in the alloca block.
+          (let ((*safepoint-counter*
+                  (cmp:with-irbuilder (cmp:*irbuilder-function-alloca*)
+                    (let ((c (cmp:alloca-size_t "safepoint-counter")))
+                      (cmp:irc-store (cmp:jit-constant-size_t 0) c)
+                      c)))
+                (*laid-out-iblocks* (make-hash-table :test #'eq)))
+            ;; Branch to the start block.
+            (cmp:irc-br (iblock-tag (bir:start ir)))
+            ;; Lay out blocks.
+            (bir:do-iblocks (ib ir)
+              (layout-iblock ib abi)))))))
   ;; Finish up by jumping from the entry block to the body block
   (cmp:with-irbuilder (cmp:*irbuilder-function-alloca*)
     (cmp:irc-br body-block))

src/lisp/kernel/lsp/mp-package.lisp

@@ -21,5 +21,7 @@
           signal-pending-interrupts raise
           without-interrupts with-interrupts with-local-interrupts
           with-restored-interrupts allow-with-interrupts interruptiblep
+          ;; deadlines
+          timeout timeout-seconds with-timeout call-with-timeout
           ))
 ) ; eval-when

src/lisp/kernel/lsp/mp.lisp

@@ -158,3 +158,59 @@ If DATUM is provided, it and ARGUMENTS designate a condition of default type SIM
    (if datum
        (core::coerce-to-condition datum arguments 'simple-error 'abort-process)
        nil)))
+
+#+threads
+;; A subtype of ERROR, not just SERIOUS-CONDITION, so HANDLER-CASE on ERROR and
+;; IGNORE-ERRORS catch it; SBCL and bordeaux-threads use SERIOUS-CONDITION.
+(define-condition timeout (error)
+  ((%seconds :initarg :seconds :reader timeout-seconds))
+  (:report (lambda (condition stream)
+             (format stream "Timed out after ~a second~:p."
+                     (timeout-seconds condition)))))
+
+#+threads
+(defun call-with-timeout (seconds function)
+  "Call FUNCTION, signalling TIMEOUT in this process if it runs longer than SECONDS.
+The timeout is delivered as an interrupt, so it lands at the next safepoint. A body
+blocked in a foreign call reaches no safepoint until the call returns, so there the
+expiry is detected on return instead; either way TIMEOUT is signalled inside the
+dynamic extent of the caller, where handlers are still established."
+  (let* ((lock (make-lock :name 'with-timeout))
+         (cv (make-condition-variable :name 'with-timeout))
+         (target *current-process*)
+         (deadline (+ (get-internal-real-time)
+                      (round (* seconds internal-time-units-per-second))))
+         (donep nil)
+         (firedp nil)
+         (watchdog
+           (process-run-function
+            'with-timeout-watchdog
+            (lambda ()
+              (with-lock (lock)
+                ;; Re-check DONEP because a timedwait may return early.
+                (loop until donep
+                      for remaining = (/ (- deadline (get-internal-real-time))
+                                         internal-time-units-per-second)
+                      while (plusp remaining)
+                      do (condition-variable-timedwait cv lock (float remaining 1d0)))
+                (unless donep
+                  (setf firedp t)
+                  (interrupt-process
+                   target
+                   ;; Re-check at delivery: an interrupt queued while the body was
+                   ;; in a foreign call arrives after the body has already returned.
+                   (lambda () (unless donep (error 'timeout :seconds seconds))))))))))
+    (let ((values (unwind-protect (multiple-value-list (funcall function))
+                    (with-lock (lock)
+                      (setf donep t)
+                      (condition-variable-signal cv))
+                    (process-join watchdog))))
+      ;; The deadline passed but the interrupt could not land in time.
+      (when firedp (error 'timeout :seconds seconds))
+      (values-list values))))
+
+#+threads
+(defmacro with-timeout ((seconds) &body body)
+  "Execute BODY, signalling MP:TIMEOUT in this process if it has not finished
+within SECONDS."
+  `(call-with-timeout ,seconds (lambda () ,@body)))