dispatcher work

Author: garlic0x1

package.lisp

@@ -1,5 +1,7 @@
 (defpackage #:thespis
   (:use #:cl)
+  (:local-nicknames
+   (#:a #:alexandria))
   (:export
    ;; thespis.lisp
    #:actor                 ; STRUCT

test/dispatcher.lisp

@@ -0,0 +1,37 @@
+(fiasco:define-test-package #:thespis/test/dispatcher
+  (:use #:thespis))
+(in-package #:thespis/test/dispatcher)
+
+(defun ~= (a b)
+  (>= 2 (abs (- a b))))
+
+(deftest test-dispatcher-balancing ()
+  "Make sure all the workers share."
+  (define-actor counter ((c 0)) (increment)
+    (sleep 0.025)
+    (incf c increment))
+
+  (let ((actor (counter :workers 2)))
+    (dotimes (i 40) (send actor 1))
+    (close-actor actor)
+    (let ((stores (join-actor actor)))
+      (is (~= (first stores) (second stores)))))
+
+  (define-actor sleeper ((c 0)) (time)
+    (sleep time)
+    (incf c))
+
+  (let ((actor (sleeper :workers 2)))
+    (dotimes (i 5)
+      (send (first (thespis::dispatcher-workers actor)) 0.2))
+    (dotimes (i 20) (sleep 0.01) (send actor 0))
+    (is (equal '((5 20)) (close-and-join-actors actor)))))
+
+(deftest test-dispatcher-registry ()
+  (define-actor counter ((c 0)) (increment)
+    (incf c increment))
+
+  (counter :name :my-counter :workers 2)
+  (send :my-counter 1)
+  (is (ask :my-counter 1))
+  (close-actor :my-counter))

thespis.asd

@@ -2,7 +2,7 @@
   :author "garlic0x1"
   :description "Threaded actors for Common Lisp"
   :license "MIT"
-  :depends-on (#:bordeaux-threads #:queues.simple-cqueue)
+  :depends-on (#:alexandria #:bordeaux-threads #:queues.simple-cqueue)
   :components ((:file "package")
                (:file "thespis"))
   :in-order-to ((test-op (test-op #:thespis/test))))

thespis.lisp

@@ -26,14 +26,14 @@
 (defstruct dispatcher
   (name    nil             :type (or nil symbol keyword))
   (workers nil             :type list)
+  (openp   t               :type boolean)
   (lock    (bt2:make-lock) :type bt2:lock))
 
 (defgeneric resolve-actor (actor)
   (:method ((actor dispatcher))
-    (reduce #'min
-            (dispatcher-workers actor)
-            :initial-value -1
-            :key (lambda (actor) (q:qsize (actor-queue actor)))))
+    (a:extremum (dispatcher-workers actor)
+                #'<
+                :key (lambda (actor) (q:qsize (actor-queue actor)))))
   (:method ((actor symbol))
     (gethash actor *registry*)))
 
@@ -70,6 +70,11 @@
   (:method ((actor actor))
     (send-signal actor (make-close-signal))
     (setf (actor-openp actor) nil))
+  (:method ((actor dispatcher))
+    (remhash (dispatcher-name actor) *registry*)
+    (dolist (worker (dispatcher-workers actor))
+      (close-actor worker))
+    (setf (dispatcher-openp actor) nil))
   (:method ((actor t))
     (close-actor (resolve-actor actor))))
 
@@ -78,6 +83,8 @@
   (:method ((actor actor))
     (bt2:join-thread (actor-thread actor))
     (apply #'values (actor-store actor)))
+  (:method ((actor dispatcher))
+    (mapcar #'join-actor (dispatcher-workers actor)))
   (:method ((actor t))
     (join-actor (resolve-actor actor))))
 
@@ -86,12 +93,15 @@
   (:method ((actor actor))
     (remhash (actor-name actor) *registry*)
     (bt2:destroy-thread (actor-thread actor)))
+  (:method ((actor dispatcher))
+    (remhash (dispatcher-name actor) *registry*)
+    (mapcar #'destroy-actor (dispatcher-workers actor)))
   (:method ((actor t))
     (destroy-actor (resolve-actor actor))))
 
 (defun close-and-join-actors (&rest actors)
   (mapc #'close-actor actors)
-  (mapc #'join-actor actors))
+  (mapcar #'join-actor actors))
 
 (defgeneric send (actor &rest args)
   (:documentation "Asyncronously send a message to an actor.")