Trampolined Style in Clojure

trampoline.clj

#!/usr/bin/env boot

;;
;; Threads & concurrency without hardware threads or continuations
;;
;; Most of this is from the paper "Trampolined Style" by Ganz, Friendman, and Wand
;;

(defrecord Done [value])
(defrecord Doing [thunk])

(defn return [v]
  (Done. v))

(defn bounce [t]
  (Doing. t))

;; run a single thread
(defn pogo-stick [thread]
  (condp instance? thread
    Done (:value thread)
    Doing (recur ((:thunk thread)))
    :error))

;; alternates between two threads
(defn seesaw [down-thread up-thread]
  (condp instance? down-thread
    Done (:value down-thread)
    Doing (recur up-thread ((:thunk down-thread)))
    :error))

;; uses a queue of threads
(defn my-trampoline [[thread & threads]]
  (condp instance? thread
    Done (:value thread)
    Doing (recur (conj (vec threads) ((:thunk thread))))
    :error))

(defn fact-acc [n acc]
  (if (zero? n)
    (return acc)
    (bounce
     #(fact-acc (- n 1) (* acc n)))))

(defn mem? [n ls]
  (cond
    (nil? ls) (return false)
    (= (first ls) n) (return true)
    :else (bounce (fn []
                    (mem? n (rest ls))))))

(defn search [pred [x & xs] status-fn]
  (status-fn x)
  (if (pred x)
    (return x)
    (bounce
     #(search pred xs status-fn))))

;; Basically three threads scheduled round-robin
;; returning the value of the first one to finish
(defn simple-queue-ex []
  (my-trampoline
   [(search #(= 5 %) (range) #(println "search1 " %))
    (search #(= 200 %) (range 100 1000) #(println "search2 " %))
    (fact-acc -1 0) ;; would never return
    ]))

;; prints...
;; search1  0
;; search2  100
;; search1  1
;; search2  101
;; search1  2
;; search2  102
;; search1  3
;; search2  103
;; search1  4
;; search2  104
;; search1  5
;; search2  105

;; returns: 5


;; Using this technique to implement a reduce function
;; similar to Clojure's that can be terminated early
;; by wrapping a return value

(def reduced' return)

(defn reduce' [f acc coll]
  (if (empty? coll)
    (return acc)
    (let [[x & xs] coll
          res (f acc x)]
      (if (instance? Done res)
        res
        (bounce
         #(reduce' f res xs))))))

(defn ex-reduced []
  (pogo-stick
   (reduce' #(if (> %1 20)
               (reduced' %1)
               %2)
            0
            (range 1 100))))

;; returns: 21

;; simple breakpoint mechanism
(defn break [thread]
  (let [loop' (fn loop'' []
                (let [input (read)
                      s (first input)]
                  (println thread)
                  (condp = s
                    'resume thread
                    'return (return (second s))
                    (bounce loop''))))]
    (loop')))

(defn fact-acc-break [n acc]
  (if (zero? n)
    (return acc)
    (break (bounce
            #(fact-acc-break (- n 1) (* acc n))))))

;; > (pogo-stick (fact-acc-break 3 1))
;; (resume)
;; #boot.user.Doing{:thunk #function[boot.user/fact-acc-break$fn--11999]}
;; (resume)
;; #boot.user.Doing{:thunk #function[boot.user/fact-acc-break$fn--11999]}
;; (resume)
;; #boot.user.Doing{:thunk #function[boot.user/fact-acc-break$fn--11999]}
;; (resume)
;; 6

;;;;;;;;;;;;;
;;
;; Threads that can spawn more threads
;;
;; If you see something hacky it was probably from
;; me translating scheme into Clojure
;; 
;;;;;;;;;;;;;

(defn return+ [v]
  [(Done. v)])

(defn bounce+ [t]
  [(Doing. t)])

(defn trampoline+ [thread-queue]
  (if (and (seq? thread-queue)
           (not-empty thread-queue))
    (let [x (first thread-queue)
          y (rest thread-queue)]
      (condp instance? x
        Done (:value x)
        Doing (trampoline+
               (concat
                y
                ((:thunk x))))))
    "No thread returned a value"))

(defn die []
  [])

(defn spawn [threads1 threads2]
  (concat threads1 threads2))

(defn mapcan [f coll]
  (if (empty? coll)
    []
    (let [[x & xs] coll]
      (concat (f x) (mapcan f xs)))))

(defn sequence+ [f threads]
  (mapcan
   (fn [thread]
     (condp instance? thread
       Done (f (:value thread))
       Doing (bounce+
              (fn []
                (sequence+ f ((:thunk thread)))))))
   threads))

(defn search-x [t]
  (cond
    (and
     (seq? t)
     (not-empty t)) (spawn
                       (bounce+
                        (fn []
                          (search-x (first t))))
                       (bounce+
                        (fn []
                          (search-x (rest t)))))
    (nil? t) (die)
    (= t 'x) (return+ t)
    (symbol? t) (do (print (format "^%s " t)) (die))))

(defn ex1 []
  (trampoline+
   (sequence+
    (fn [v]
      (if (= v 'x)
        (return+ 'yes)
        (return+ 'no)))
    (search-x '(((a b c d) (x e) (g h)))))))
;; > (ex1)
;; ^a ^b ^c
;; yes

(defn ex2 []
  (trampoline+
   (sequence+
    (fn [v]
      (if (= v 'x)
        (return+ 'yes)
        (return+ 'no)))
    (search-x '(((a d) (y e) (g h)))))))

;; > (ex2)
;; ^a ^d ^y ^e ^g ^h
;; "No thread returned a value"

(defn ex3 []
  (trampoline+
   (sequence+
    (fn [v]
      (if (= v 'x)
        (return+ 'yes)
        (return+ 'no)))
    (search-x '(((a d) (y e (o i c (x))) (g h)))))))

;; > (ex3)
;; ^a ^d ^y ^e ^g ^h ^o ^i ^c
;; yes

(defn -main [& args]
  (boot (repl)))