Skip to content
Permalink
master
Switch branches/tags
Go to file
 
 
Cannot retrieve contributors at this time

Tasks

Core.Task
Base.@task
Base.@async
Base.asyncmap
Base.asyncmap!
Base.current_task
Base.istaskdone
Base.istaskstarted
Base.istaskfailed
Base.task_local_storage(::Any)
Base.task_local_storage(::Any, ::Any)
Base.task_local_storage(::Function, ::Any, ::Any)

Scheduling

Base.yield
Base.yieldto
Base.sleep
Base.schedule

[Synchronization](@id lib-task-sync)

Synchronization

Base.errormonitor
Base.@sync
Base.wait
Base.fetch(t::Task)
Base.timedwait

Base.Condition
Base.Threads.Condition
Base.Threads.Event
Base.notify

Base.Semaphore
Base.acquire
Base.release

Base.AbstractLock
Base.lock
Base.unlock
Base.trylock
Base.islocked
Base.ReentrantLock

Channels

Base.Channel
Base.Channel(::Function)
Base.put!(::Channel, ::Any)
Base.take!(::Channel)
Base.isready(::Channel)
Base.fetch(::Channel)
Base.close(::Channel)
Base.bind(c::Channel, task::Task)

[Low-level synchronization using schedule and wait](@id low-level-schedule-wait)

The easiest correct use of schedule is on a Task that is not started (scheduled) yet. However, it is possible to use schedule and wait as a very low-level building block for constructing synchronization interfaces. A crucial pre-condition of calling schedule(task) is that the caller must "own" the task; i.e., it must know that the call to wait in the given task is happening at the locations known to the code calling schedule(task). One strategy for ensuring such pre-condition is to use atomics, as demonstrated in the following example:

@enum OWEState begin
    OWE_EMPTY
    OWE_WAITING
    OWE_NOTIFYING
end

mutable struct OneWayEvent
    @atomic state::OWEState
    task::Task
    OneWayEvent() = new(OWE_EMPTY)
end

function Base.notify(ev::OneWayEvent)
    state = @atomic ev.state
    while state !== OWE_NOTIFYING
        # Spin until we successfully update the state to OWE_NOTIFYING:
        state, ok = @atomicreplace(ev.state, state => OWE_NOTIFYING)
        if ok
            if state == OWE_WAITING
                # OWE_WAITING -> OWE_NOTIFYING transition means that the waiter task is
                # already waiting or about to call `wait`. The notifier task must wake up
                # the waiter task.
                schedule(ev.task)
            else
                @assert state == OWE_EMPTY
                # Since we are assuming that there is only one notifier task (for
                # simplicity), we know that the other possible case here is OWE_EMPTY.
                # We do not need to do anything because we know that the waiter task has
                # not called `wait(ev::OneWayEvent)` yet.
            end
            break
        end
    end
    return
end

function Base.wait(ev::OneWayEvent)
    ev.task = current_task()
    state, ok = @atomicreplace(ev.state, OWE_EMPTY => OWE_WAITING)
    if ok
        # OWE_EMPTY -> OWE_WAITING transition means that the notifier task is guaranteed to
        # invoke OWE_WAITING -> OWE_NOTIFYING transition.  The waiter task must call
        # `wait()` immediately.  In particular, it MUST NOT invoke any function that may
        # yield to the scheduler at this point in code.
        wait()
    else
        @assert state == OWE_NOTIFYING
        # Otherwise, the `state` must have already been moved to OWE_NOTIFYING by the
        # notifier task.
    end
    return
end

ev = OneWayEvent()
@sync begin
    @async begin
        wait(ev)
        println("done")
    end
    println("notifying...")
    notify(ev)
end

# output
notifying...
done

OneWayEvent lets one task to wait for another task's notify. It is a limited communication interface since wait can only be used once from a single task (note the non-atomic assignment of ev.task)

In this example, notify(ev::OneWayEvent) is allowed to call schedule(ev.task) if and only if it modifies the state from OWE_WAITING to OWE_NOTIFYING. This lets us know that the task executing wait(ev::OneWayEvent) is now in the ok branch and that there cannot be other tasks that tries to schedule(ev.task) since their @atomicreplace(ev.state, state => OWE_NOTIFYING) will fail.