Now message-box/bt class will restart thread if it was aborted because of a non-local exit from processing loop.

I've measured performance before these changes and after them and didn't notice significant changes.
Results of the benchmark along with the baseline results are available in this gist:

https://gist.github.com/svetlyak40wt/a1097ab7d501087ca366d5addb410ebe

Author: svetlyak40wt

bench.lisp

@@ -92,7 +92,8 @@
                       (wait-if-queue-larger-than 10000 wait-if-queue-larger-than-given-p)
                       (duration 10)
                       (num-iterations 60)
-                      (load-threads 8))
+                      (load-threads 8)
+                      (time-out nil))
 
   (log:config :warn)
 
@@ -124,7 +125,8 @@
                 :async-ask-p  async-ask-p 
                 :num-shared-workers num-shared-workers
                 :queue-size queue-size
-                :wait-if-queue-larger-than wait-if-queue-larger-than))
+                :wait-if-queue-larger-than wait-if-queue-larger-than
+                :time-out time-out))
   (force-output)
 
   (with-timing (num-iterations
@@ -255,33 +257,72 @@
 
 
 (defun run-all (&key
-                (num-iterations 10)
-                (duration 10))
+                (num-iterations 60)
+                (duration 10)
+                (queue-size 100)
+                (time-out 3)
+                &aux (started-at (get-internal-real-time)))
   (run-benchmark :num-iterations num-iterations
-                 :duration duration)
+                 :duration duration
+                 :with-reply-p nil
+                 :async-ask-p nil)
 
-  (format t "Running ~A:~%" '(run-benchmark :with-reply-p t :async-ask-p nil))
   (run-benchmark :num-iterations num-iterations
                  :duration duration
-                 :with-reply-p t :async-ask-p nil)
+                 :with-reply-p t
+                 :async-ask-p nil)
+  
+  (run-benchmark :num-iterations num-iterations
+                 :duration duration
+                 :with-reply-p t
+                 :async-ask-p t)
+
+  
+  (format t "With queue size limited to ~A:~2%"
+          queue-size)
 
-  (format t "Running ~A:~%" '(run-benchmark :with-reply-p t :async-ask-p t))
   (run-benchmark :num-iterations num-iterations
                  :duration duration
-                 :with-reply-p t :async-ask-p t)
+                 :with-reply-p nil
+                 :async-ask-p nil
+                 :queue-size queue-size)
 
-  (format t "Running ~A:~%" '(run-benchmark :queue-size 100))
   (run-benchmark :num-iterations num-iterations
                  :duration duration
-                 :queue-size 100)
+                 :with-reply-p t
+                 :async-ask-p nil
+                 :queue-size queue-size)
 
-  (format t "Running ~A:~%" '(run-benchmark :with-reply-p t :async-ask-p nil :queue-size 100))
   (run-benchmark :num-iterations num-iterations
                  :duration duration
-                 :with-reply-p t :async-ask-p nil :queue-size 100)
+                 :with-reply-p t
+                 :async-ask-p t
+                 :queue-size queue-size)
+
+  
+  (format t "With time-out ~A:~2%"
+          time-out)
 
-  (format t "Running ~A:~%" '(run-benchmark :with-reply-p t :async-ask-p t :queue-size 100))
   (run-benchmark :num-iterations num-iterations
                  :duration duration
-                 :with-reply-p t :async-ask-p t :queue-size 100))
+                 :with-reply-p nil
+                 :async-ask-p nil
+                 ;; This should not make sense for with-reply-p = nil
+                 :time-out time-out)
+
+  (run-benchmark :num-iterations num-iterations
+                 :duration duration
+                 :with-reply-p t
+                 :async-ask-p nil
+                 :time-out time-out)
+  
+  (run-benchmark :num-iterations num-iterations
+                 :duration duration
+                 :with-reply-p t
+                 :async-ask-p t
+                 :time-out time-out)
+
+  (format t "All tests are performed in ~,2f seconds.~%"
+          (/ (- (get-internal-real-time) started-at)
+             internal-time-units-per-second)))
 

sento.asd

@@ -65,6 +65,7 @@
   :author "Manfred Bergmann"
   :depends-on ("sento"
                "fiveam"
+               "serapeum"
                "lparallel"
                "cl-mock")
   :components ((:module "tests"
@@ -95,7 +96,9 @@
                  (:file "actor-system-test")
                  (:file "actor-tree-test")
                  (:file "spawn-in-receive-test")
-                 )))
+                 (:file "test-utils")
+                 (:file "message-box-test"
+                     :depends-on ("test-utils")))))
   :description "Test system for sento"
   :perform (test-op (op c) (symbol-call :fiveam :run!
                                         (uiop:find-symbol* '#:test-suite

src/mbox/message-box.lisp

@@ -113,18 +113,41 @@ This is used to break the environment possibly captured as closure at 'submit' s
 (defclass message-box/bt (message-box-base)
   ((queue-thread :initform nil
                  :documentation
-                 "The thread that pops queue items."))
+                 "The thread that pops queue items.")
+   (thread-is-running-p :initform nil
+                        :type boolean
+                        :documentation
+                        "Will be set to NIL if processing loop will be broken because of an error or a restart invocation."))
   (:documentation
    "Bordeaux-Threads based message-box with a single thread operating on a message queue.
 This is used when the actor is created using a `:pinned` dispatcher type.
 There is a limit on the maximum number of actors/agents that can be created with
 this kind of queue because each message-box (and with that each actor) requires exactly one thread."))
 
+
+(declaim (ftype (function (message-box/bt &key (:thread-name (or null string)))
+                          (values &optional))
+                start-thread))
+
+(defun start-thread (msgbox &key thread-name)
+  (with-slots (name queue-thread thread-is-running-p)
+      msgbox
+    (flet ((run-processing-loop ()
+             (setf thread-is-running-p t)
+             (unwind-protect
+                  (message-processing-loop msgbox)
+               (setf thread-is-running-p
+                     nil))))
+      (setf queue-thread
+            (bt2:make-thread #'run-processing-loop
+                             :name (or thread-name
+                                       (mkstr "message-thread-" name))))))
+  (values))
+
+
 (defmethod initialize-instance :after ((self message-box/bt) &key)
-  (with-slots (name queue-thread) self
-    (setf queue-thread (bt2:make-thread
-                        (lambda () (message-processing-loop self))
-                        :name  (mkstr "message-thread-" name))))
+  (start-thread self)
+  
   (when (next-method-p)
     (call-next-method)))
 
@@ -179,6 +202,24 @@ This function sets the result as `handler-result' in `item'. The return of this
               (bt2:condition-notify withreply-cvar)))
           (handler-fun)))))
 
+
+(declaim (ftype (function (message-box/bt)
+                          (values &optional))
+                ensure-thread-is-running))
+
+(defun ensure-thread-is-running (msgbox)
+  (with-slots (queue-thread thread-is-running-p should-run)
+      msgbox
+    (when (and (not thread-is-running-p)
+               should-run)
+      ;; Just to be sure that thread is not alive:
+      (unless (bt2:thread-alive-p queue-thread)
+        (let ((thread-name (bt2:thread-name queue-thread)))
+          (log:warn "Restarting thread" thread-name)
+          (start-thread msgbox
+                        :thread-name thread-name))))
+    (values)))
+
 (defmethod submit ((self message-box/bt) message withreply-p time-out handler-fun-args)
   "The `handler-fun-args` argument must contain a handler function as first list item.
 It will be apply'ed with the rest of the args when the message was 'popped' from queue."
@@ -200,15 +241,21 @@ It will be apply'ed with the rest of the args when the message was 'popped' from
                      :time-out time-out
                      :handler-fun-args handler-fun-args
                      :handler-result 'no-result)))
-    (log:trace "~a: withreply: waiting for arrival of result..." (name msgbox))
+    
     (bt2:with-lock-held (withreply-lock)
       (log:trace "~a: pushing item to queue: ~a" (name msgbox) push-item)
       (queue:pushq queue push-item)
-
-      (if time-out
-          (wait-and-probe-for-msg-handler-result msgbox push-item)
-          (bt2:condition-wait withreply-cvar withreply-lock)))
-
+      (ensure-thread-is-running msgbox)
+
+      (log:trace "~a: withreply: waiting for arrival of result..." (name msgbox))
+      
+      (unless (bt2:condition-wait withreply-cvar withreply-lock
+                                  :timeout time-out)
+        (log:warn "~a: time-out elapsed but result not available yet!" (name msgbox))
+        (setf (slot-value push-item 'cancelled-p) t)
+        (error 'ask-timeout
+               :wait-time time-out)))
+    
     (with-slots (handler-result) push-item
       (log:trace "~a: withreply: result should be available: ~a" (name msgbox) handler-result)
       handler-result)))
@@ -222,6 +269,7 @@ The submitting code has to await the side-effect and possibly handle a timeout."
                     :handler-fun-args handler-fun-args)))
     (log:trace "~a: pushing item to queue: ~a" (name msgbox) push-item)
     (queue:pushq queue push-item)
+    (ensure-thread-is-running msgbox)
     t))
 
 (defmethod stop ((self message-box/bt) &optional (wait nil))
@@ -301,6 +349,7 @@ Returns the handler-result if `withreply-p' is eq to `T', otherwise the return i
                processed-messages
                dispatcher) self
     (incf processed-messages)
+    
     (let ((push-item (make-message-item/dp
                       :message message
                       :handler-fun-args handler-fun-args

tests/message-box-test.lisp

@@ -0,0 +1,111 @@
+(defpackage :sento.message-box-test
+  (:use :cl :fiveam :cl-mock :sento.actor :sento.future)
+  (:shadow #:! #:?)
+  (:import-from #:miscutils
+                #:assert-cond
+                #:await-cond
+                #:filter)
+  (:import-from #:timeutils
+                #:ask-timeout)
+  (:import-from #:sento.messageb
+                #:message-box/bt
+                #:submit
+                #:no-result
+                #:queue-thread
+                #:stop)
+  (:import-from #:sento.test-utils
+                #:parametrized-test)
+  (:import-from #:ac
+                #:actor-of))
+
+(in-package :sento.message-box-test)
+
+(def-suite message-box-tests
+  :description "message-box tests"
+  :in sento.tests:test-suite)
+
+(in-suite message-box-tests)
+
+
+(defun wait-while-thread-will-die (msgbox &key (timeout 10))
+  (let ((wait-until (+ (get-internal-real-time) (* timeout
+                                                   internal-time-units-per-second))))
+    (with-slots (queue-thread)
+        msgbox
+      (loop :while (bt2:thread-alive-p queue-thread)
+            :do (sleep 0.1)
+                (when (< wait-until
+                         (get-internal-real-time))
+                  (error "Thread didn't die in ~A seconds."
+                         timeout))))))
+
+
+(parametrized-test bt-box-resurrects-thread-after-abort-if-handler-catches-all-signals
+    ((withreply-p timeout)
+     (nil         nil)
+     (t           1)
+     (t           nil))
+  
+  "Simulates a situation when error has happened during message processing, and ABORT restart was invoked.
+   Usually this kill a thread, but here we ensure that by the thread is resurrected when we submit a
+   subsequent message."
+
+  (flet ((kill-by-restart-invoke (msg)
+           (declare (ignore msg))
+           (handler-case
+               ;; This way we are simulating that the user choose
+               ;; an ABORT restart in the IDE during debug session:
+               (handler-bind ((serious-condition #'abort))
+                 (error "Die, thread, die!"))
+             ;; This part the same as error handling code in the
+             ;; SENTO.ACTOR-CELL:HANDLE-MESSAGE function:
+             ;; 
+             ;; TODO: t was used to check if it is able to
+             ;; catch stack unwinding because of INVOKE-RESTART,
+             ;; but it can't.
+             (t (c)
+               (log:error "error condition was raised: ~%~a~%"
+                          c)
+               (cons :handler-error c)))))
+    
+    (let ((box (make-instance 'message-box/bt
+                              :name "foo")))
+      (unwind-protect
+           (progn
+             (let ((first-reply
+                     (submit box "The Message"
+                             t
+                             ;; Don't wait for result here, because we are
+                             ;; intentionally raise error here and will never
+                             ;; return a result:
+                             nil
+                             (list #'kill-by-restart-invoke))))
+               (is (equal first-reply
+                          'no-result)))
+
+             (wait-while-thread-will-die box)
+
+             (is (not
+                  (bt2:thread-alive-p
+                   (slot-value box 'queue-thread))))
+
+             (let ((result (handler-case
+                               (submit box "The Message"
+                                       withreply-p
+                                       timeout
+                                       (list (lambda (msg)
+                                               (reverse msg))))
+                             (ask-timeout ()
+                               :timeout))))
+
+               (cond
+                 (withreply-p
+                  (is (string= "egasseM ehT" result)))
+                 (t
+                  (is (eql result t)))))
+
+             (is (bt2:thread-alive-p
+                  (slot-value box 'queue-thread))))
+                         
+        ;; Cleanup a thread:
+        (stop box t)))))