     (println "Using args:" arguments)
     (let [in-cntr (Counter.)
           out-cntr (Counter.)
+          out-fn (fn [_] (.inc out-cntr))
           delta-cntr (delta-counter)
           ^String scenario (:scenario options)
           in-data (prepare-in-data options)

    proc-fns))

(defn prepare-in-fn
  [options scenario in-data ^Counter in-cntr ^Counter out-cntr in-chan pipeline queue-size]
  (let [batch-delay (:batch-delay options)
        batch-size (:batch-size options)
        create-batched-in-fn (fn [in-fn]

        in-fn (cond
                (.endsWith scenario "-direct")
                  (fn []
                    (out-fn (direct-proc-fn in-data)))
                (.endsWith scenario "-direct-pmap")
                  (let [d (repeat collection-size in-data)]
                    (doall d)
                    (fn []
                      (doseq [obj (pmap direct-proc-fn d)]
                        (out-fn obj))))
                (.endsWith scenario "-direct-reducers-map")
                  (let [in-vec (vec (repeat collection-size in-data))]
                    (doall in-vec)
                    (fn []
                      (doseq [obj (reducers/fold
                                    partition-size
                                    reducers/cat
                                    reducers/append!
                                    (reducers/map direct-proc-fn in-vec))]
                        (out-fn obj))))
                (.endsWith scenario "-simple-pmap")
                  (let [pmap-proc (create-simple-pmap-processor direct-proc-fn collection-size out-fn)]
                    (fn []
                      (pmap-proc in-data)
                      (.inc in-cntr)))
                (.endsWith scenario "-simple-reducers-map")
                  (let [red-proc (create-simple-reducers-map-processor direct-proc-fn collection-size partition-size out-fn)]
                    (fn []
                      (red-proc in-data)
                      (.inc in-cntr)))
