Add common threadpool with :thread-count option

When running tests a single threadpool with a configurable thread count
should be used. Previous versions used a threadpool per namespace,
resulting in more threads used than would be performant. Additionally,
on environments that misrepresent the available processors (such as
CircleCI), also result in too many threads being used.

README.md

@@ -94,6 +94,29 @@ Or you can synchronize the entire namespace:
             [foo.core :refer :all]))
 ```
 
+##### Setting the number of threads used
+
+When multithreading is enabled, Eftest uses a single fixed-threadpool
+[`ExecutorService`][executorservice] to run all selected tests.
+
+By default, Eftest will instantiate the threadpool with the number of processors
+(cores) available to the JVM, as reported by
+[`Runtime.availableProcessors`][availableprocessors]. (NB: in some
+circumstances, such as [in a CircleCI test container][resource-class],
+`Runtime.availableProcessors` returns an erroneous value.)
+
+[executorservice]: https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/concurrent/ExecutorService.html
+[availableprocessors]: https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/lang/Runtime.html#availableProcessors()
+[resource-class]: https://circleci.com/docs/2.0/configuration-reference/#resource_class
+
+Users can override the default behavior by including the key `:thread-count`
+in the options map supplied to `run-tests` with the value being any
+positive integer:
+
+```clojure
+user=> (run-tests (find-tests "test") {:thread-count 4})
+```
+
 #### Reporting
 
 You can also change the reporting function used. For example, if you
@@ -177,11 +200,12 @@ To use the Lein-Eftest plugin, just run:
 lein eftest
 ```
 
-You can customize the reporter and set the concurrency by adding an
-`:eftest` key to your project map:
+You can customize the reporter and configure the concurrency settings
+by adding an `:eftest` key to your project map:
 
 ```clojure
-:eftest {:multithread? false
+:eftest {:multithread? :vars
+         :thread-count 4
          :report eftest.report.junit/report
          ;; You can optionally write the output to a file like so:
          :report-to-file "target/junit.xml"}

eftest/src/eftest/runner.clj

@@ -44,37 +44,40 @@
 (defn- ^Callable bound-callback [f]
   (cast Callable (bound-fn* f)))
 
-(defn- ^ExecutorService threadpool-executor []
-  (Executors/newFixedThreadPool (+ 2 (.availableProcessors (Runtime/getRuntime)))))
-
-(defn- pcalls* [fs]
-  (let [executor (threadpool-executor)]
-    (try
-      (->> fs
-           (map #(.submit executor (bound-callback %)))
-           (doall)
-           (map #(.get %))
-           (doall))
-      (finally
-        (.shutdownNow executor)))))
-
-(defn- pmap* [f xs]
-  (pcalls* (map (fn [x] #(f x)) xs)))
+(defn- default-thread-count []
+  (+ 2 (.availableProcessors (Runtime/getRuntime))))
+
+(defn- ^ExecutorService threadpool-executor
+  [{:keys [thread-count] :or {thread-count (default-thread-count)}}]
+  (Executors/newFixedThreadPool thread-count))
+
+(defn- pcalls* [executor fs]
+  (->> fs
+       (map #(.submit executor (bound-callback %)))
+       (doall)
+       (map #(.get %))
+       (doall)))
+
+(defn- pmap* [executor f xs]
+  (pcalls* executor (map (fn [x] #(f x)) xs)))
 
 (defn- multithread-vars? [{:keys [multithread?] :or {multithread? true}}]
   (or (true? multithread?) (= multithread? :vars)))
 
 (defn- multithread-namespaces? [{:keys [multithread?] :or {multithread? true}}]
   (or (true? multithread?) (= multithread? :namespaces)))
 
+(defn- multithread? [opts]
+  (or (multithread-vars? opts) (multithread-namespaces? opts)))
+
 (defn- fixture-exception [throwable]
   {:type    :error
    :message "Uncaught exception during fixture initialization."
    :actual  throwable})
 
 (defn- test-vars
   [ns vars report
-   {:as opts :keys [fail-fast? capture-output? test-warn-time]
+   {:as opts :keys [executor fail-fast? capture-output? test-warn-time]
     :or {capture-output? true}}]
   (let [once-fixtures (-> ns meta ::test/once-fixtures test/join-fixtures)
         each-fixtures (-> ns meta ::test/each-fixtures test/join-fixtures)
@@ -99,8 +102,8 @@
         (once-fixtures
           (fn []
             (if (multithread-vars? opts)
-              (do (->> vars (filter synchronized?) (map test-var)   (dorun))
-                  (->> vars (remove synchronized?) (pmap* test-var) (dorun)))
+              (do (->> vars (filter synchronized?) (map test-var) (dorun))
+                  (->> vars (remove synchronized?) (pmap* executor test-var) (dorun)))
               (doseq [v vars] (test-var v)))))
         (catch Throwable t
           (test/do-report (fixture-exception t)))))))
@@ -113,9 +116,13 @@
       (test/do-report {:type :end-test-ns, :ns ns})
       @test/*report-counters*)))
 
-(defn- test-all [vars {:as opts :keys [capture-output?] :or {capture-output? true}}]
+(defn- test-all [vars {:as opts
+                       :keys [capture-output? executor]
+                       :or {capture-output? true}}]
   (let [report (synchronize test/report)
-        mapf   (if (multithread-namespaces? opts) pmap* map)
+        mapf   (if (multithread-namespaces? opts)
+                 (partial pmap* executor)
+                 map)
         f      #(->> (group-by (comp :ns meta) vars)
                      (mapf (fn [[ns vars]] (test-ns ns vars report opts)))
                      (apply merge-with +))]
@@ -169,6 +176,10 @@
                        in those namespaces are run serially. If set to :vars,
                        the namespaces are run serially, but the vars inside run
                        in parallel.
+    :thread-count    - the number of threads used to run the tests in parallel
+                       (as per :multithread?). If not specified, the number
+                       reported by java.lang.Runtime.availableProcessors (which
+                       is not always accurate) *plus two* will be used.
     :report          - the test reporting function to use
                        (defaults to eftest.report.progress/report)
     :test-warn-time  - print a warning for any test that exceeds this time
@@ -182,7 +193,10 @@
        (binding [report/*context* (atom {})
                  test/report      (:report opts progress/report)]
          (test/do-report {:type :begin-test-run, :count (count vars)})
-         (let [counters (test-all vars opts)
+         (let [executor (when (multithread? opts) (threadpool-executor opts))
+               opts     (assoc opts :executor executor)
+               counters (try (test-all vars opts)
+                             (finally (when executor (.shutdownNow executor))))
                duration (/ (- (System/nanoTime) start-time) 1e6)
                summary  (assoc counters :type :summary, :duration duration)]
            (test/do-report summary)