Make some modifications to the wait functions in the cllojure.clr.async.task API

Author: dmiller

Clojure/Clojure.Source/clojure/clr/async/task.clj

@@ -4,19 +4,25 @@
   (:refer-clojure :exclude [await])
   (:import [System.Threading.Tasks Task]))
 
+(alias-type TaskObj |System.Threading.Tasks.Task`1[Object]|)
 
-;; ── Internal helpers ──────────────────────────────────────────────────
+
+;;  ── Extracting a Task result ──────────────────────
 
 ; This would get a reflection warning, so we don't turn on *warn-on-reflection* until after.
 ; We WANT a reflection warning -- we need the DLR callsite mechanism on the call to .GetAwaiter.
 ; If we tag the task parameter as ^Task, we always pick up Task.GetAwaiter(), even for generic Task<TResult> tasks.
 ; That just does not work.
 
-(defn- task-result
-  "Extracts the result from a completed Task<T> via its typed awaiter.
-   Uses GetAwaiter().GetResult() which unwraps exceptions cleanly
-   (no AggregateException wrapping unlike .Result).
-   Returns nil for non-generic Task (void)."
+(defn result
+  "Blocks the calling thread until the task completes and returns its result.
+   For Task<T>, returns T. For void Task (non-generic), returns nil.
+   Unwraps AggregateException to throw the inner exception directly.
+
+   Usage:
+     (t/result (t/->task 42))       ;=> 42
+     (t/result (t/completed-task))   ;=> nil
+     (t/result (t/async (t/await (t/delay-task 100)) \"done\"))  ;=> \"done\""
  [task]
   (let [t (.GetType ^Object task)]
     (when (.IsGenericType t)
@@ -52,41 +58,101 @@
   [& body]
   `((^:async fn* [] ~@body)))
 
-;; ── Functions ─────────────────────────────────────────────────────────
-
-(def ^:private task-object-type |System.Threading.Tasks.Task`1[System.Object]|)
-
-(defn ^:async await-all
-  "Awaits all tasks in parallel. Takes a collection of any Task types
-   (Task<string>, Task<int>, Task<Object>, etc.).
-   Returns an Object[] of results. Must be called in an async context.
-
-   Usage:
-     (let [results (t/await (t/await-all [task-a task-b task-c]))]
-       (vec results))"
-  [tasks]
-  (let [^|System.Threading.Tasks.Task[]| task-array (into-array Task tasks)]
-    (if (every? #(instance? task-object-type %) task-array)
-      ;; Fast path: all Task<Object> — use generic WhenAll, no per-element reflection
-      (let [^|System.Threading.Tasks.Task`1[System.Object][]| obj-array
-            (into-array task-object-type tasks)]
-        (await* (Task/WhenAll (type-args Object) obj-array)))
-      ;; Slow path: mixed types — await non-generic, extract results via reflection
-      (do
-        (await* (Task/WhenAll task-array))
-        (into-array Object (map task-result task-array))))))
-
-(defn ^:async await-any
-  "Awaits the first task to complete from a collection of any Task types.
-   Returns the first completed Task (not its result).
-   Must be called in an async context.
-
-   Usage:
-     (let [winner (t/await (t/await-any [fast-task slow-task]))]
-       (t/result winner))"
-  [tasks]
-  (let [^|System.Threading.Tasks.Task[]| task-array (into-array Task tasks)]
-    (await* (Task/WhenAny task-array))))
+;; ── Waiting, but staying non-:async ─────────────────────────────────────────────────────────
+
+(defn- convert-timeout
+  "Converts a Clojure timeout value to an int for Task.WaitAll.
+   Accepts nil (no timeout), a number (milliseconds), or a TimeSpan."
+  [timeout]
+  (cond
+    (nil? timeout) -1
+    (number? timeout) (int timeout)
+    :else (int (.TotalMilliseconds ^TimeSpan timeout))))
+
+(defn wait-all 
+  "Waits for all of the provided Tasks to complete execution.  Returns nil.
+
+   Usage: (t/wait-all tasks) 
+          (t/wait-all tasks timeout)
+          (t/wait-all task timeout cancellation-token)
+  
+   timeout values can be any numeric value (cast to int, milliseconds, -1 = no limit) or a TimeSpan"
+  ([tasks] (Task/WaitAll ^Task/1 (into-array Task tasks)))
+  ([tasks timeout] (Task/WaitAll ^Task/1  (into-array Task tasks) ^int (convert-timeout timeout)))
+  ([tasks timeout cancellation-token] (Task/WaitAll ^Task/1  (into-array Task tasks) (convert-timeout timeout) cancellation-token)))
+
+
+(defn wait-any
+  "Waits for any of the provided Task to complete execution.  Returns the task that completed or nil if a timeout occurred.
+
+   Usage: (t/wait-any tasks) 
+          (t/wait-any tasks timeout)
+          (t/wait-any task timeout cancellation-token)
+  
+   timeout values can be any numeric value (cast to int, milliseconds, -1 = no limit) or a TimeSpan"
+  ([tasks]
+    (let [^Task/1  task-array (into-array Task tasks)
+          idx (Task/WaitAny task-array)]
+        (nth tasks idx)))
+  ([tasks timeout]
+    (let [^Task/1 task-array (into-array Task tasks)
+          idx (Task/WaitAny task-array ^int (convert-timeout timeout))]
+        (when-not (= idx -1)
+            (nth tasks idx))))
+  ([tasks timeout cancellation-token]
+    (let [^Task/1 task-array (into-array Task tasks)
+          idx (Task/WaitAny task-array (convert-timeout timeout) cancellation-token)]
+        (when-not (= idx -1)
+            (nth tasks idx)))))
+
+(defn wait-all-results
+  "Waits for all of the provided Tasks to complete execution.  Returns a lazy sequence of the result for each task, with nil with for non-generic Tasks.
+
+   Usage: (t/wait-all-results tasks) 
+          (t/wait-all-results tasks timeout)
+          (t/wait-all-results task timeout cancellation-token)
+  
+   timeout values can be any numeric value (cast to int, milliseconds, -1 = no limit) or a TimeSpan"
+
+  ([tasks]
+    (let [^Task/1 task-array (into-array Task tasks)]
+      (Task/WaitAll task-array)
+      (map result tasks)))
+  ([tasks timeout]
+    (let [^Task/1 task-array (into-array Task tasks)]
+      (Task/WaitAll task-array ^int (convert-timeout timeout))
+      (map result tasks)))
+  ([tasks timeout cancellation-token]
+    (let [^Task/1 task-array (into-array Task tasks)]
+      (Task/WaitAll task-array (convert-timeout timeout) cancellation-token)
+      (map result tasks))))
+
+(defn wait-any-result 
+ "Waits for any of the provided Task to complete execution.  Returns the result of the task that completed or nil if a timeout occurred.
+
+   Usage: (t/wait-any-results tasks) 
+          (t/wait-any-results tasks timeout)
+          (t/wait-any-results task timeout cancellation-token)
+  
+   timeout values can be any numeric value (cast to int, milliseconds, -1 = no limit) or a TimeSpan"
+  ([tasks]
+    (let [^Task/1 task-array (into-array Task tasks)
+          idx (Task/WaitAny task-array)]
+        (when-not (= idx -1)
+            (result (nth tasks idx)))))
+  ([tasks timeout]
+    (let [^Task/1 task-array (into-array Task tasks)
+          idx (Task/WaitAny task-array ^int (convert-timeout timeout))]
+        (when-not (= idx -1)
+            (result (nth tasks idx)))))
+  ([tasks timeout cancellation-token]
+    (let [^Task/1 task-array (into-array Task tasks)
+          idx (Task/WaitAny task-array (convert-timeout timeout) cancellation-token)]
+        (when-not (= idx -1)
+            (result (nth tasks idx))))))
+
+
+;; ── Creating standard tasks
 
 (defn delay-task
   "Returns a Task that completes after the specified duration in milliseconds.
@@ -97,15 +163,6 @@
   [milliseconds]
   (Task/Delay (int milliseconds)))
 
-(defn run
-  "Runs f (zero-arg fn) on the thread pool. Returns Task<object>.
-
-   Usage:
-     (t/await (t/run (fn [] (+ 1 2 3))))"
-  [f]
-  (let [func (gen-delegate |System.Func`1[System.Object]| [] (f))]
-    (Task/Run func)))
-
 (defn ->task
   "Wraps a value in a completed Task<object>.
 
@@ -124,14 +181,4 @@
   [x]
   (instance? Task x))
 
-(defn result
-  "Blocks the calling thread until the task completes and returns its result.
-   For Task<T>, returns T. For void Task, returns nil.
-   Unwraps AggregateException to throw the inner exception directly.
 
-   Usage:
-     (t/result (t/->task 42))       ;=> 42
-     (t/result (t/completed-task))   ;=> nil
-     (t/result (t/async (t/await (t/delay-task 100)) \"done\"))  ;=> \"done\""
-  [task]
-  (task-result task))