                                  (in-fn in-data)
                                  (.inc in-cntr)))
           direct-proc-fn (create-direct-proc-fn scenario)
+          collection-size (:collection-size options)
+          partition-size (:partition-size options)
           in-loop (ProcessingLoop.
                     "DataGenerationLoop"
                     (cond

    "The number of data instances to be generated for one batch."
    :default 2000
    :parse-fn #(Integer/parseInt %)]
   ["-d" "--batch-delay BATCH-DELAY"
    "The delay in ms between generating batches."
    :default 1

                          (direct-proc-fn in-data)
                          (.inc out-cntr))
                      (.endsWith scenario "-pmap")
                        (let [d (repeat collection-size in-data)]
                          (doall d)
                          (fn []
                            (doseq [_ (pmap direct-proc-fn d)]
                              (.inc out-cntr))))
                      (.endsWith scenario "-reducers-map")
                        (let [in-vec (vec (repeat collection-size in-data))]
                          (doall in-vec)
                          (fn []
                            (doseq [
                                    ;_ (reducers/foldcat (reducers/map direct-proc-fn in-vec))
                                    _ (reducers/fold partition-size reducers/cat reducers/append! (reducers/map direct-proc-fn in-vec))
                                    ]
                              (.inc out-cntr))))
                      (.endsWith scenario "-async-pipeline")
