[i/items] cancellation

src/hyperfiddle/incseq/items_eager_impl.cljc

@@ -5,15 +5,27 @@
   (:import #?(:clj [clojure.lang IDeref IFn])
            [missionary Cancelled]))
 
-(def ps-field-count (a/deffields -input-ps -input-stepper -input-doner -diff -item*))
+(def ps-field-count (a/deffields -stepped -cancelled -input-ps -input-stepper -input-doner -diff -item*))
+(declare cleanup-ps)
 (deftype Ps [step done state-]
-  IFn (#?(:clj invoke :cljs -invoke) [_] ((a/get state- -input-ps)) (done))
-  IDeref (#?(:clj deref :cljs -deref) [_] (a/get state- -diff)))
+  IFn (#?(:clj invoke :cljs -invoke) [^Ps this]
+        ((a/get state- -input-ps))
+        (let [cancelled? (a/fgetset this -cancelled true)]
+          (when (not (or (a/fgetset this -stepped true) cancelled?)) (step))))
+  IDeref (#?(:clj deref :cljs -deref) [^Ps this]
+           (a/fset this -stepped false)
+           (if (a/fget this -cancelled)
+             (do (cleanup-ps this done) (throw (Cancelled.)))
+             (a/get state- -diff))))
+(defn cleanup-ps [^Ps ps done]
+  (when-not (identical? ps (a/fgetset ps -diff ps))
+    (a/fset ps -input-ps nil, -input-stepper nil, -input-doner nil, -diff nil, -item* nil)
+    (done)))
 
 (def item-field-count (a/deffields -v -flow -ps* -dead))
 (deftype Item [state-])
 
-(def item-ps-field-count (a/deffields -stepped -cache -cancelled))
+(def item-ps-field-count (a/deffields -stepped -cancelled -cache))
 
 (defn remove-item-ps [^Item item ps] (let [ps* (a/fget item -ps*)] (ps* (disj (ps*) ps))))
 
@@ -97,7 +109,10 @@
   (or (seq (:permutation d)) (pos? (:grow d)) (pos? (:shrink d)) (seq (:freeze d))))
 
 (defn consume-input-step [^Ps ps]
-  (fn [] (when (needed-diff? (a/fswap ps -diff merge (transfer-input ps)))  ((.-step ps)))))
+  (fn []
+    (when (needed-diff? (a/fswap ps -diff merge (transfer-input ps)))
+      (when-not (a/fgetset ps -stepped true)
+        ((.-step ps))))))
 (defn consume-input-done [^Ps ps] (fn []))
 
 (defn flow [input]
@@ -106,4 +121,4 @@
       (a/fset ps -input-stepper #() -input-doner #(), -item* (object-array 8))
       (a/fset ps -input-ps (input (fn [] ((a/fget ps -input-stepper))) (fn [] ((a/fget ps -input-doner)))))
       (a/fset ps -input-stepper (consume-input-step ps), -input-doner (consume-input-done ps))
-      (a/fswap ps -diff merge (transfer-input ps)) (step) ps)))
+      (a/fswap ps -diff merge (transfer-input ps)) (a/fset ps -stepped true) (step) ps)))

test/hyperfiddle/incseq/items_eager_impl_test.cljc

@@ -272,7 +272,9 @@
         _                   (item0)
         _                   (t/is (= :item0-step (q)))
         _                   (t/is (thrown? Cancelled @item0))
-        _                   (t/is (= :item0-done (q)))]))
+        _                   (t/is (= :item0-done (q)))
+        _                   (q ::none)
+        _                   (t/is (= ::none (q)))]))
 
 (t/deftest item-ps-cancellation-stepped
   (let [q                   (->mq)
@@ -286,9 +288,53 @@
         _                   (t/is (= :item0-step (q)))
         _                   (item0)
         _                   (t/is (thrown? Cancelled @item0))
-        _                   (t/is (= :item0-done (q)))]))
+        _                   (t/is (= :item0-done (q)))
+        _                   (q ::none)
+        _                   (t/is (= ::none (q)))]))
+
+(t/deftest cancellation-idle
+  (let [q                   (->mq)
+        _                   (q (assoc (d/empty-diff 1) :grow 1 :change {0 :foo})) ; what input will return on transfer
+        items               (spawn-ps q)
+        [_in-step _in-done] (q)
+        _                   (t/is (= :items-step (q)))
+        diff                @items
+        _                   (t/is (= (assoc (d/empty-diff 1) :grow 1) (assoc diff :change {})))
+        _                   (items)
+        _                   (t/is (= :input-cancel (q)))
+        _                   (t/is (= :items-step (q)))
+        _                   (t/is (thrown? Cancelled @items))
+        _                   (t/is (= :items-done (q)))
+        _                   (q ::none)
+        _                   (t/is (= ::none (q)))]))
+
+(t/deftest cancellation-stepped
+  (let [q                   (->mq)
+        _                   (q (assoc (d/empty-diff 1) :grow 1 :change {0 :foo})) ; what input will return on transfer
+        items               (spawn-ps q)
+        [_in-step _in-done] (q)
+        _                   (t/is (= :items-step (q)))
+        _                   (items)
+        _                   (t/is (= :input-cancel (q)))
+        _                   (t/is (thrown? Cancelled @items))
+        _                   (t/is (= :items-done (q)))
+        _                   (q ::none)
+        _                   (t/is (= ::none (q)))]))
 
 ;; missing tests
+;; - items reentrant transfer
 ;; - input terminate
 ;; - failures
+;; - double cancel before termination
+;;   - item-ps
+;;   - dead-item-ps
+;;   - items
+;; - double cancel after termination
+;;   - item-ps
+;;   - dead-item-ps
+;;   - items
+;; - double transfer
+;;   - item-ps
+;;   - dead-item-ps
+;;   - items
 ;; - thread safety