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

    proc-fns))

(defn prepare-in-fn
  [options scenario in-data ^Counter in-cntr 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))
                                 (sleep batch-delay)))
        create-flood-in-fn (fn [in-fn]
                             (fn []
                               (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-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))]

    (println "Using options:" options)
    (println "Using args:" arguments)
    (let [in-cntr (Counter.)
          pre-in-fn (if (:latency options)
                      (do
                        (println "Creating latency measurement pre-in-fn.")
                        (fn [in-obj]
                          (.inc in-cntr)
                          (LatencyProbe. in-obj)))
                      (do
                        (println "Creating default measurement pre-in-fn.")
                        (fn [in-obj]
                          (.inc in-cntr)
                          in-obj)))
          out-cntr (Counter.)
          latency-probe-collector (LatencyProbeCollector.)
          out-fn (if (:latency options)
