Utility functions that can replace standard core.async functions and track which writes were stuck. good to debug core async deadlocks

debug-core-async.clj

(ns debug-core-async
  (:require [clojure.core.async :as async])
  (:import (java.util UUID)
           (org.apache.commons.lang.exception ExceptionUtils)))

(defonce chans (atom {}))
(defonce open-writes (atom {}))

(defn reset-state []
  (reset! chans {})
  (reset! open-writes {}))

(defn channels-with-open-writes []
  (let [mapper (fn [[k v]]
                 (assoc v :name (get @chans (:port v))))]
    (map mapper @open-writes)))

(defn open-writes-infos[]
  (doall (map #(println (:name %) (:trace %)) (channels-with-open-writes)))
  (doall (map #(println (:name %)) (channels-with-open-writes))))

(defn chan [name & args]
  (let [ch (apply async/chan args)]
    (swap! chans #(assoc % ch name))
    ch))

(defn stacktrace []
  (ExceptionUtils/getStackTrace (Exception.)))

(defn before-write [write-id port val]
  (let [payload {:port port
                 :trace (stacktrace)}]
    (swap! open-writes #(assoc % write-id payload))))

(defn after-write [write-id port val]
  (swap! open-writes #(dissoc % write-id)))

(defmacro >! [port val]
  `(let [write-id# (UUID/randomUUID)]
     (before-write write-id# ~port ~val)
     (async/>! ~port ~val)
     (after-write write-id# ~port ~val)))

(defn >!! [port val]
  (let [write-id (UUID/randomUUID)]
    (before-write write-id port val)
    (async/>!! port val)
    (after-write write-id port val)))