Revert "Use vthread pool instead of core.async pipeline-blocking"

This reverts commit 67c23ea.

Author: ghaskins

src/manetu/sparql_loadtest/core.clj

@@ -5,7 +5,7 @@
             [medley.core :as m]
             [promesa.core :as p]
             [taoensso.timbre :as log]
-            [clojure.core.async :refer [<!! <! >!! go go-loop] :as async]
+            [clojure.core.async :refer [<! go go-loop] :as async]
             [progrock.core :as pr]
             [doric.core :refer [table]]
             [ring.util.codec :as ring.codec]
@@ -34,18 +34,9 @@
                      :duration d)))))))
 
 (defn- pipeline-blocking
-  [nr f in]
+  [nr xf in]
   (let [out (async/chan nr)]
-    (-> (p/all
-         (map (fn [_]
-                (p/vthread
-                 (loop []
-                   (when-let [m (<!! in)]
-                     (>!! out (f m))
-                     (recur)))))
-              (range nr)))
-        (p/then (fn [_]
-                  (async/close! out))))
+    (async/pipeline-blocking nr out xf in)
     out))
 
 (defn async-xform
@@ -66,7 +57,7 @@
   (log/trace "launching with concurrency:" concurrency)
   (let [query (-> query slurp ring.codec/url-encode)]
     (->> (binding-loader/get-bindings bindings nr batch-size)
-         (pipeline-blocking concurrency (partial execute-query ctx query))
+         (pipeline-blocking concurrency (map (partial execute-query ctx query)))
          (async-xform (mapcat (fn [{:keys [success result] :as x}]
                                 (if (true? success)
                                   (map (fn [r] (assoc x :result r)) result)