luciano/babashka
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

[#847] Better error handling for parallel tasks

Browse source
This commit is contained in:
Michiel Borkent 2021-05-18 12:39:21 +02:00 committed by GitHub
parent 00091136d3
commit e18f4302a8
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 118 additions and 38 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

51
src/babashka/impl/tasks.clj
View file

@ -3,12 +3,17 @@
[babashka.impl.common :refer [ctx bb-edn debug]]
[babashka.impl.deps :as deps]
[babashka.process :as p]
[clojure.core.async :refer [chan <!! alts!! thread]]
[clojure.java.io :as io]
[clojure.string :as str]
[rewrite-clj.node :as node]
[rewrite-clj.parser :as parser]
[rewrite-clj.zip :as zip]
[sci.core :as sci]))
[sci.core :as sci])
(:import [clojure.core.async.impl.channels ManyToManyChannel]))
(defn chan? [x]
(instance? ManyToManyChannel x))
(def sci-ns (sci/create-ns 'babashka.tasks nil))
(def default-log-level :error)
@ -118,8 +123,13 @@
(defn -wait [res]
(when res
(if (future? res)
@res
(if (chan? res)
(let [[_task-name res] (<!! res)]
(if (instance? Throwable res)
(throw (ex-info (ex-message res)
{:babashka/exit 1
:data (ex-data res)}))
res))
res)))
(defn depends-map [tasks target-name]
@ -127,13 +137,21 @@
m [target-name deps]]
(into {} (cons m (map #(depends-map tasks %) deps)))))
(defmacro -err-thread [name & body]
`(clojure.core.async/thread
(try [~name ~@body]
(catch Throwable e#
[~name (ex-info (str "Error in task: " ~name
"\n" (ex-message e#))
(or (ex-data e#) {}))]))))
(defn wrap-body [task-map prog parallel?]
(format "(binding [
babashka.tasks/*task* '%s]
%s)"
(pr-str task-map)
(if parallel?
(format "(future %s)" prog)
(format "(babashka.tasks/-err-thread \"%s\" %s)" (:name task-map) prog)
prog)))
(defn wrap-def [task-map prog parallel? last?]
@ -144,8 +162,24 @@
(format "(babashka.tasks/-wait %s)" task-name)
task-name))))
(defn deref-task [dep]
(format "(def %s (babashka.tasks/-wait %s))" dep dep))
(defn wait-tasks [deps]
(if deps
(format "
(let [chans %s]
(loop [cs chans]
(let [[v p] (clojure.core.async/alts!! cs)
[task-name v] v
cs (filterv #(not= p %%) cs)
;; _ (.println System/err (str \"n: \" task-name \" v: \" v))
_ (intern *ns* (symbol task-name) v)]
(when (instance? Throwable v)
(throw (ex-info (ex-message v)
{:babashka/exit 1
:data (ex-data v)})))
(when (seq cs)
(recur cs)))))" deps)
"")
#_(format "(def %s (babashka.tasks/-wait %s))" dep dep))
(defn wrap-enter-leave [task-name prog enter leave]
(str (pr-str enter) "\n"
@ -159,7 +193,7 @@
(defn wrap-depends [prog depends parallel?]
(if parallel?
(format "(do %s)" (str (str/join "\n" (map deref-task depends)) "\n" prog))
(format "(do %s)" (str (str "\n" (wait-tasks depends)) "\n" prog))
prog))
(defn assemble-task-1
@ -312,7 +346,7 @@
(println "No such task:" t)) 1])
(if-let [task (get tasks t)]
(let [prog (str prog "\n"
(apply str (map deref-task depends))
#_(wait-tasks depends) #_(apply str (map deref-task depends))
"\n"
(assemble-task-1 task-map task parallel? true))
extra-paths (concat extra-paths (:extra-paths task))
@ -413,6 +447,7 @@
{'shell (sci/copy-var shell sci-ns)
'clojure (sci/copy-var clojure sci-ns)
'-wait (sci/copy-var -wait sci-ns)
'-err-thread (sci/copy-var -err-thread sci-ns)
'*task* task
'current-task current-task
'current-state state

23
test-resources/coffee-tasks.edn Normal file
View file

@ -0,0 +1,23 @@
{:tasks
{coffeep {:depends [groundsp hot-waterp filterp mugp]
:task (do (Thread/sleep 300)
[:made-coffee [groundsp hot-waterp filterp mugp]])}
groundsp {:depends [beansp]
:task (do
(Thread/sleep 200)
[:ground-beans [beansp]])}
hot-waterp {:depends [waterp]
:task (do (Thread/sleep 200)
[:heated-water [waterp]])}
filterp {:task (do
(Thread/sleep 100)
:filter)}
mugp {:task (do
(Thread/sleep 100)
:mug)}
waterp {:task (do
(Thread/sleep 100)
:poured-water)}
beansp {:task (do
(Thread/sleep 100)
:measured-beans)}}}

24
test/babashka/bb_edn_test.clj
View file

@ -203,6 +203,28 @@
:out)}}
(let [s (bb "run" "--prn" "a")]
(is (= "hello\n" s)))))
(testing "parallel test"
(test-utils/with-config (edn/read-string (slurp "test-resources/coffee-tasks.edn"))
(let [tree [:made-coffee [[:ground-beans [:measured-beans]] [:heated-water [:poured-water]] :filter :mug]]
t0 (System/currentTimeMillis)
s (bb "run" "--prn" "coffeep")
t1 (System/currentTimeMillis)
delta-sequential (- t1 t0)]
(is (= tree s))
(test-utils/with-config (edn/read-string (slurp "test-resources/coffee-tasks.edn"))
(let [t0 (System/currentTimeMillis)
s (bb "run" "--parallel" "--prn" "coffeep")
t1 (System/currentTimeMillis)
delta-parallel (- t1 t0)]
(is (= tree s))
(is (< delta-parallel delta-sequential))))))
(testing "exception"
(test-utils/with-config '{:tasks {a (Thread/sleep 10000)
b (do (Thread/sleep 10)
(throw (ex-info "0 noes" {})))
c {:depends [a b]}}}
(is (thrown-with-msg? Exception #"0 noes"
(bb "run" "--parallel" "c")))))))
(deftest list-tasks-test
(test-utils/with-config {}
@ -235,7 +257,7 @@
:task (+ 1 2 3)}}}"
(let [res (test-utils/bb nil "tasks")]
(is (= "The following tasks are available:\n\ntask1 task1 doc\n"
res)))))))
res))))))
(deftest task-priority-test
(when-not test-utils/native?