in_threads.clj

(defn opening? [sym]
  (when (symbol? sym)
    (.endsWith (name sym) ">")))
 
(defn closing? [sym]
  (when (symbol? sym)
    (.startsWith (name sym) "<")))
 
(defn split-closing [t exprs]
  (split-with
    (fn [[tid expr]]
      (not
        (and (= tid t)
             (closing? expr))))
   exprs))
 
(defn parse-sym [sym]
  (->> (name sym)
       (re-matches #"<?([^<>]+)>?")
       second
       symbol))
 
(defn print-expr [expr]
  `(println ~(str (first expr) ": ") '~(second expr)))
 
(defmacro thread [& body]
  `(doto
     (Thread. (fn [] ~@body))
     (.start)))
 
(defn expand-subtree [t exprs]
  (loop [exprs exprs
         acc []]
    (let [[tid expr] (first exprs)
          acc (conj acc `(Thread/sleep 100))]
      (cond
        (nil? tid)
          (list* acc)
        (not= tid t)
          (recur (next exprs) acc)
        (opening? expr)
          (let [[sub-exprs cont-exprs] (split-closing t (next exprs))]
            (recur
              (next cont-exprs)
              (conj acc
                 (print-expr [tid expr])
                 (list*
                   (parse-sym expr)
                   (expand-subtree t sub-exprs))
                 (print-expr (first cont-exprs)))))
        :else
          (recur (next exprs)
                 (conj acc `(do
                              ~(print-expr [tid expr])
                              ~expr)))))))
 
(defmacro in-threads [& body]
  (let [exprs (partition 2 body)
        threads (->> exprs (map first) set)]
   `(do
      ~@(for [t threads]
          `(thread
              ~@(expand-subtree t exprs))))))

tests.clj

(let [x (ref 0)]
  (in-threads
    1 dosync>
    1 (ref-set x 1)
    1 <dosync
    2 dosync>
    2 (ref-set x 2)
    2 <dosync
    3 (println "x =" @x)
   ))

;;   Nothing special here, transactions do not affect each other
;;
;;   1:  dosync>
;;   1:  (ref-set x 1)
;;   1:  <dosync
;;   2:  dosync>
;;   2:  (ref-set x 2)
;;   2:  <dosync
;;   3:  (println x = (clojure.core/deref x))
;;   x = 2



(let [x (ref 0)]
  (in-threads
    1 dosync>
    2 dosync>
    2 (ref-set x 2)
    2 <dosync
    1 (ref-set x 1)
    1 <dosync
    3 (println "x =" @x)
   ))

;;   Here we see that 1 started transaction, then 2 changed value, then 1 tries to change value and 1 gets restarted
;;
;;   1:  dosync>
;;   2:  dosync>
;;   2:  (ref-set x 2)
;;   2:  <dosync
;;   1:  (ref-set x 1)
;;   3:  (println x = (clojure.core/deref x))
;;   x = 2
;;   1:  (ref-set x 1)
;;   1:  <dosync




(let [x (ref 0)]
  (in-threads
    1 dosync>
    1 (ref-set x 1)
    2 dosync>
    2 (ref-set x 2)
    2 <dosync
    1 <dosync
    3 (println "x =" @x)
   ))

;;   Here we see that 1 started transaction and changed value, then 2 tries to change value and 2 cannot commit before 1 commits
;;
;;   1:  dosync>
;;   1:  (ref-set x 1)
;;   2:  dosync>
;;   2:  (ref-set x 2)
;;   1:  <dosync
;;   2:  (ref-set x 2)
;;   3:  (println x = (clojure.core/deref x))
;;   x = 1
;;   2:  (ref-set x 2)
;;   2:  <dosync