Add timeout parameter to wait(::Condition)

base/boot.jl

@@ -224,7 +224,7 @@ export
     InterruptException, InexactError, OutOfMemoryError, ReadOnlyMemoryError,
     OverflowError, StackOverflowError, SegmentationFault, UndefRefError, UndefVarError,
     TypeError, ArgumentError, MethodError, AssertionError, LoadError, InitError,
-    UndefKeywordError, ConcurrencyViolationError, FieldError,
+    UndefKeywordError, ConcurrencyViolationError, FieldError, TimeoutError,
     # AST representation
     Expr, QuoteNode, LineNumberNode, GlobalRef,
     # object model functions
@@ -473,6 +473,10 @@ end
 
 struct PrecompilableError <: Exception end
 
+struct TimeoutError <: Exception
+    timeout::Real
+end
+
 String(s::String) = s  # no constructor yet
 
 const Cvoid = Nothing

base/condition.jl

@@ -126,19 +126,61 @@ proceeding.
 function wait end
 
 """
-    wait(c::GenericCondition; first::Bool=false)
+    wait(c::GenericCondition; first::Bool=false, timeout::Real=0.0)
 
 Wait for [`notify`](@ref) on `c` and return the `val` parameter passed to `notify`.
 
 If the keyword `first` is set to `true`, the waiter will be put _first_
 in line to wake up on `notify`. Otherwise, `wait` has first-in-first-out (FIFO) behavior.
+
+If `timeout` is specified, cancel the `wait` when it expires and throw a
+`TimeoutError` to the waiting task. The minimum value for `timeout`
+is 0.001 seconds, i.e. 1 millisecond.
 """
-function wait(c::GenericCondition; first::Bool=false)
+function wait(c::GenericCondition; first::Bool=false, timeout::Real=0.0)
+    timeout == 0.0 || timeout ≥ 1e-3 || throw(ArgumentError("timeout must be ≥ 1 millisecond"))
+
     ct = current_task()
     _wait2(c, ct, first)
     token = unlockall(c.lock)
+
+    timer::Union{Timer, Nothing} = nothing
+    waiter_left = Threads.Atomic{Bool}(false)
+    if timeout > 0.0
+        timer = Timer(timeout)
+        # start a task to wait on the timer
+        t = Task() do
+            try
+                wait(timer)
+            catch e
+                # if the timer was closed, the waiting task has been scheduled; do nothing
+                e isa EOFError && return
+            end
+            dosched = false
+            lock(c.lock)
+            # Confirm that the waiting task is still in the wait queue and remove it. If
+            # the task is not in the wait queue, it must have been notified already so we
+            # don't do anything here.
+            if !waiter_left[] && ct.queue == c.waitq
+                dosched = true
+                Base.list_deletefirst!(c.waitq, ct)
+            end
+            unlock(c.lock)
+            # send the waiting task a timeout
+            dosched && schedule(ct, TimeoutError(timeout); error=true)
+        end
+        t.sticky = false
+        Threads._spawn_set_thrpool(t, :interactive)
+        schedule(t)
+    end
+
     try
-        return wait()
+        res = wait()
+        if timer !== nothing
+            close(timer)
+            waiter_left[] = true
+        end
+        return res
     catch
         q = ct.queue; q === nothing || Base.list_deletefirst!(q::IntrusiveLinkedList{Task}, ct)
         rethrow()

test/channels.jl

@@ -1,6 +1,7 @@
 # This file is a part of Julia. License is MIT: https://julialang.org/license
 
 using Random
+using Base.Threads
 using Base: Experimental
 using Base: n_avail
 
@@ -39,6 +40,22 @@ end
     @test fetch(t) == "finished"
 end
 
+@testset "timed wait on Condition" begin
+    a = Condition()
+    @test_throws ArgumentError wait(a; timeout=0.0005)
+    @test_throws TimeoutError wait(a; timeout=0.1)
+    @spawn begin
+        sleep(0.01)
+        notify(a)
+    end
+    @test try
+        wait(a; timeout=2)
+        true
+    catch
+        false
+    end
+end
+
 @testset "various constructors" begin
     c = Channel()
     @test eltype(c) == Any