     proc-fns))
 
 (defn prepare-in-fn
-  [options scenario in-data ^Counter in-cntr out-fn in-chan pipeline queue-size]
+  [options scenario in-data pre-in-fn out-fn in-chan pipeline queue-size]
   (let [batch-delay (:batch-delay options)
         batch-size (:batch-size options)
         create-batched-in-fn (fn [in-fn]
                                (fn []
                                  (doseq [i (repeat batch-size 0)]
-                                   (in-fn in-data)
-                                   (.inc in-cntr))
+                                   (in-fn (pre-in-fn in-data)))
                                  (sleep batch-delay)))
         create-flood-in-fn (fn [in-fn]
                              (fn []
-                               (in-fn in-data)
-                               (.inc in-cntr)))
+                               (in-fn (pre-in-fn in-data))))
         direct-proc-fn (create-direct-proc-fn scenario)
         collection-size (:collection-size options)
         partition-size (:partition-size options)
         in-fn (cond
                 (.endsWith scenario "-direct")
                   (fn []
-                    (out-fn (direct-proc-fn in-data)))
+                    (out-fn
+                      (direct-proc-fn
+                        (pre-in-fn in-data))))
                 (.endsWith scenario "-direct-pmap")
                   (let [d (repeat collection-size in-data)]
                     (doall d)
                     (fn []
-                      (doseq [obj (pmap direct-proc-fn d)]
+                      (doseq [obj (pmap (fn [in-obj] (direct-proc-fn (pre-in-fn in-obj))) d)]
                         (out-fn obj))))
                 (.endsWith scenario "-direct-reducers-map")
                   (let [in-vec (vec (repeat collection-size in-data))]

          (println "Starting async-pipeline-go input creation.")
          (async/go
            (loop []
              (async/>! in-chan in-data)
              (.inc ^Counter in-cntr)
              (recur)))
          (sleep 130000))
        (do

                    (async/chan queue-size))
          out-chan (if (.contains scenario "async-pipeline")
                     (async/chan queue-size))
          in-fn (prepare-in-fn options scenario in-data pre-in-fn out-fn in-chan pipeline queue-size)
          in-loop (ProcessingLoop.
                    "DataGenerationLoop"
                    in-fn)
