                                (fn []
                                  (in-fn in-data)
                                  (.inc in-cntr)))
+          direct-proc-fn (create-direct-proc-fn scenario)
           in-loop (ProcessingLoop.
                     "DataGenerationLoop"
                     (cond
                       (.endsWith scenario "-direct")
-                        (let [proc-fn (create-direct-proc-fn scenario)]
-                          (fn []
-                            (proc-fn in-data)
+                        (fn []
+                          (direct-proc-fn in-data)
+                          (.inc out-cntr))
+                      (.endsWith scenario "-pmap")
+                        (fn []
+                          (doseq [data (pmap direct-proc-fn (repeat in-data))]
                             (.inc out-cntr)))
-                      (.endsWith scenario "-pmap")
-                        (let [proc-fn (create-direct-proc-fn scenario)]
-                          (fn []
-                            (doseq [data (pmap proc-fn (repeat in-data))]
-                              (.inc out-cntr))))
                       (.endsWith scenario "-async-pipeline")
                         (if
                           (and

                        (> batch-size 0)) (create-batched-in-fn (get-in-fn pipeline))
                      (not (nil? in-data)) (create-flood-in-fn (get-in-fn pipeline))
                      :default (fn [] (.inc out-cntr))))
          self-adaptivity-cfg (options :self-adaptivity-cfg)
          self-adaptivity-controller (if (not (nil? self-adaptivity-cfg))
                                       (create-self-adaptivity-controller

      (println "Starting experiment...")
      (.setName (Thread/currentThread) "Main")
      (.start in-loop)
      (if
        (not (nil? async-out-count-loop))
        (.start async-out-count-loop))
      (run-repeat
        (executor)
        (fn []
          (stats-fn)
          (thread-info-fn) (println)
          )
        1000)
      (run-once
        (executor)
        (fn [] (System/exit 0)) 120000))))

