<commit_before>(ns lambdacd.control-flow
  "control flow elements for a pipeline: steps that control the way their child-steps are being run"
  (:require [lambdacd.execution :as execution]
            [clojure.core.async :as async]))

(defn- parallel-step-result-producer [args steps-and-ids]
  (pmap (partial execution/execute-step args) steps-and-ids))

(defn- execute-steps-in-parallel [steps args step-id]
  (execution/execute-steps parallel-step-result-producer steps args step-id))

(defn ^{:display-type :parallel} in-parallel [& steps]
  (fn [args ctx]
    (execute-steps-in-parallel steps args (execution/new-base-context-for ctx))))


(defn- wait-for-success-on [channels]
  (let [merged (async/merge channels)
        filtered-by-success (async/filter< #(= :success (:status %)) merged)]
    (async/<!! filtered-by-success)))

(defn- step-producer-returning-with-first-successful [args steps-and-ids]
  (let [step-result-channels (map #(async/go (execution/execute-step args %)) steps-and-ids)
        result (wait-for-success-on step-result-channels)]
    (if (nil? result)
      [{:status :failure}]
      [result]))
  )

(defn ^{:display-type :parallel} either [& steps]
  (fn [args ctx]
    (let [execute-output (execution/execute-steps step-producer-returning-with-first-successful steps args (execution/new-base-context-for ctx))]
      (if (= :success (:status execute-output))
        (first (vals (:outputs execute-output)))
        execute-output))))



  (defn ^{:display-type :container} in-cwd [cwd & steps]
  (fn [args ctx]
    (execution/execute-steps steps (assoc args :cwd cwd) (execution/new-base-context-for ctx))))<commit_msg>fix whitespaces<commit_after>(ns lambdacd.control-flow
  "control flow elements for a pipeline: steps that control the way their child-steps are being run"
  (:require [lambdacd.execution :as execution]
            [clojure.core.async :as async]))

(defn- parallel-step-result-producer [args steps-and-ids]
  (pmap (partial execution/execute-step args) steps-and-ids))

(defn- execute-steps-in-parallel [steps args step-id]
  (execution/execute-steps parallel-step-result-producer steps args step-id))

(defn ^{:display-type :parallel} in-parallel [& steps]
  (fn [args ctx]
    (execute-steps-in-parallel steps args (execution/new-base-context-for ctx))))


(defn- wait-for-success-on [channels]
  (let [merged (async/merge channels)
        filtered-by-success (async/filter< #(= :success (:status %)) merged)]
    (async/<!! filtered-by-success)))

(defn- step-producer-returning-with-first-successful [args steps-and-ids]
  (let [step-result-channels (map #(async/go (execution/execute-step args %)) steps-and-ids)
        result (wait-for-success-on step-result-channels)]
    (if (nil? result)
      [{:status :failure}]
      [result]))
  )

(defn ^{:display-type :parallel} either [& steps]
  (fn [args ctx]
    (let [execute-output (execution/execute-steps step-producer-returning-with-first-successful steps args (execution/new-base-context-for ctx))]
      (if (= :success (:status execute-output))
        (first (vals (:outputs execute-output)))
        execute-output))))



(defn ^{:display-type :container} in-cwd [cwd & steps]
  (fn [args ctx]
    (execution/execute-steps steps (assoc args :cwd cwd) (execution/new-base-context-for ctx))))
{'fraction_non_alphanumeric': 0.1550995775497888, 'fraction_numerical': 0.0, 'mean_word_length': 3.8043478260869565, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2242710', 'n_tokens_mistral': 1104, 'n_tokens_neox': 1070, 'n_words': 266}