Задачи

Task(func)

Создайте Task (т.е. сопрограмму) для выполнения заданной функции func (которая должна быть вызываемой без аргументов). Задача завершается, когда эта функция возвращает значение. Задача будет выполняться в «возрасте мира» родительского элемента при создании, когда schedule.

Примеры

julia> a() = sum(i for i in 1:1000);

julia> b = Task(a);

В этом примере b является выполнимой Task задачей, которая еще не началась.

@task

Оборачивает выражение в Task, не выполняя его, и возвращает Task. Это только создает задачу, а не запускает ее.

Примеры

julia> a1() = sum(i for i in 1:1000);

julia> b = @task a1();

julia> istaskstarted(b)
false

julia> schedule(b);

julia> yield();

julia> istaskdone(b)
true

@async

Оборачивает выражение в Task и добавляет его в очередь планировщика локальной машины.

Значения могут быть интерполированы в @async с помощью $, которая копирует значение непосредственно в созданный базовый закрытый метод. Это позволяет вставлять значение переменной, изолируя асинхронный код от изменений значения переменной в текущей задаче.

asyncmap(f, c...; ntasks=0, batch_size=nothing)

Использует несколько одновременных задач для отображения f по коллекции (или нескольким коллекциям равной длины). При нескольких аргументах коллекции f применяется поэлементно.

ntasks указывает количество задач для одновременного выполнения. В зависимости от длины коллекций, если ntasks не указано, для одновременного отображения будет использоваться до 100 задач.

ntasks также может быть задано как функция без аргументов. В этом случае количество задач, выполняемых параллельно, проверяется перед обработкой каждого элемента, и новая задача запускается, если значение ntasks_func больше текущего количества задач.

Если batch_size указано, коллекция обрабатывается в пакетном режиме. f должна быть функцией, которая должна принимать Vector кортежей аргументов и возвращать вектор результатов. Вектор входных данных будет иметь длину batch_size или меньше.

Следующие примеры показывают выполнение в разных задачах, возвращая objectid задач, в которых выполняется функция отображения.

Сначала, при неопределённом значении ntasks, каждый элемент обрабатывается в отдельной задаче.

julia> tskoid() = objectid(current_task());

julia> asyncmap(x->tskoid(), 1:5)
5-element Array{UInt64,1}:
 0x6e15e66c75c75853
 0x440f8819a1baa682
 0x9fb3eeadd0c83985
 0xebd3e35fe90d4050
 0x29efc93edce2b961

julia> length(unique(asyncmap(x->tskoid(), 1:5)))
5

Со значением ntasks=2 все элементы обрабатываются в 2 задачах.

julia> asyncmap(x->tskoid(), 1:5; ntasks=2)
5-element Array{UInt64,1}:
 0x027ab1680df7ae94
 0xa23d2f80cd7cf157
 0x027ab1680df7ae94
 0xa23d2f80cd7cf157
 0x027ab1680df7ae94

julia> length(unique(asyncmap(x->tskoid(), 1:5; ntasks=2)))
2

При определённом значении batch_size, функция отображения должна быть изменена на прием массива кортежей аргументов и возвращение массива результатов. map используется в изменённой функции отображения для достижения этого.

julia> batch_func(input) = map(x->string("args_tuple: ", x, ", element_val: ", x[1], ", task: ", tskoid()), input)
batch_func (generic function with 1 method)

julia> asyncmap(batch_func, 1:5; ntasks=2, batch_size=2)
5-element Array{String,1}:
 "args_tuple: (1,), element_val: 1, task: 9118321258196414413"
 "args_tuple: (2,), element_val: 2, task: 4904288162898683522"
 "args_tuple: (3,), element_val: 3, task: 9118321258196414413"
 "args_tuple: (4,), element_val: 4, task: 4904288162898683522"
 "args_tuple: (5,), element_val: 5, task: 9118321258196414413"

asyncmap!(f, results, c...; ntasks=0, batch_size=nothing)

Подобно asyncmap, но хранит выходные данные в results вместо возврата коллекции.

current_task()

Получить текущую выполняющуюся Task.

istaskdone(t::Task) -> Bool

Определить, завершилась ли задача.

Примеры

julia> a2() = sum(i for i in 1:1000);

julia> b = Task(a2);

julia> istaskdone(b)
false

julia> schedule(b);

julia> yield();

julia> istaskdone(b)
true

istaskstarted(t::Task) -> Bool

Определить, начала ли выполняться задача.

Примеры

julia> a3() = sum(i for i in 1:1000);

julia> b = Task(a3);

julia> istaskstarted(b)
false

istaskfailed(t::Task) -> Bool

Определить, завершилась ли задача из-за возникновения исключения.

Примеры

julia> a4() = error("task failed");

julia> b = Task(a4);

julia> istaskfailed(b)
false

julia> schedule(b);

julia> yield();

julia> istaskfailed(b)
true

task_local_storage(key)

Искать значение ключа в локальном хранилище текущей задачи.

task_local_storage(key, value)

Присвоить значение ключу в локальном хранилище текущей задачи.

task_local_storage(body, key, value)

Вызвать функцию body с изменённым локальным хранилищем задачи, в котором value назначено key; предыдущее значение key, или его отсутствие, восстанавливается после этого. Полезно для эмуляции динамического охвата.

yield()

Переключение на планировщик, чтобы позволить выполнение другой запланированной задачи. Задача, которая вызывает эту функцию, все еще может быть выполнена и будет немедленно перезапущена, если нет других готовых задач.

yield(t::Task, arg = nothing)

Быстрая, нечестная версия планирования schedule(t, arg); yield(), которая немедленно уступает t перед вызовом планировщика.

yieldto(t::Task, arg = nothing)

Переключение на заданную задачу. При первом переключении на задачу, функция задачи вызывается без аргументов. При последующих переключениях, arg возвращается из последнего вызова задачи к yieldto. Это вызов низкого уровня, который только переключает задачи, не учитывая состояния или планирование каким-либо образом. Его использование не рекомендуется.

sleep(seconds)

Заблокировать текущую задачу на указанное количество секунд. Минимальное время ожидания — 1 миллисекунда или значение 0.001.

schedule(t::Task, [val]; error=false)

Добавить Task в очередь планировщика. Это заставляет задачу постоянно выполняться, когда система в противном случае простаивает, если задача не выполняет блокирующую операцию, например, wait.

Если указан второй аргумент val, он будет передан задаче (через возвращаемое значение yieldto) при её повторном выполнении. Если error имеет значение true, это значение будет поднято как исключение в разбуженной задаче.

Примеры

julia> a5() = sum(i for i in 1:1000);

julia> b = Task(a5);

julia> istaskstarted(b)
false

julia> schedule(b);

julia> yield();

julia> istaskstarted(b)
true

julia> istaskdone(b)
true

errormonitor(t::Task)

Вывести журнал ошибок в stderr, если задача t завершится с ошибкой.

Примеры

julia> Base._wait(errormonitor(Threads.@spawn error("task failed")))
Unhandled Task ERROR: task failed
Stacktrace:
[...]

@sync

Подождать, пока все лексически вложенные вызовы @async, @spawn, Distributed.@spawnat и Distributed.@distributed завершатся. Все исключения, сгенерированные вложенными асинхронными операциями, будут собраны и сгенерированы как CompositeException.

Примеры

julia> Threads.nthreads()
4

julia> @sync begin
           Threads.@spawn println("Thread-id $(Threads.threadid()), task 1")
           Threads.@spawn println("Thread-id $(Threads.threadid()), task 2")
       end;
Thread-id 3, task 1
Thread-id 1, task 2

Примечание для Threads.Condition:

Вызывающая функция должна удерживать lock, владеющую Threads.Condition, перед вызовом этого метода. Вызывающая задача будет заблокирована, пока какая-то другая задача не разбудит её, обычно вызвав notify для того же объекта Threads.Condition. Блокировка будет атомарно освобождена во время блокировки (даже если она была заблокирована рекурсивно), и будет повторно захвачена перед возвратом.

wait(r::Future)

Дождаться появления значения для указанного Future.

wait(r::RemoteChannel, args...)

Дождаться появления значения на указанном RemoteChannel.

wait([x])

Заблокировать текущую задачу до наступления какого-либо события, в зависимости от типа аргумента:

• Channel: Дождаться добавления значения в канал.
• Condition: Дождаться notify на условии и вернуть параметр val, переданный в notify. Ожидание на условии также позволяет передать first=true, что приводит к размещению ожидающего в начале очереди на пробуждение по notify, вместо обычного поведения "первый пришёл - первый обслужен".
• Process: Подождать завершения процесса или цепочки процессов. Поле exitcode процесса можно использовать для определения успеха или неудачи.
• Task: Подождать завершения Task. Если задача завершится с ошибкой, будет выброшено исключение TaskFailedException (которое оборачивает завершившуюся с ошибкой задачу).
• RawFD: Подождать изменений на дескрипторе файла (см. пакет FileWatching).

Если аргумент не указан, задача блокируется на неопределённое время. Задачу можно перезапустить только явно, вызвав schedule или yieldto.

Часто wait вызывается внутри цикла while, чтобы убедиться, что ожидаемое условие выполняется перед продолжением.

wait(c::Channel)

Блокируется, пока Channel не станет isready.

julia> c = Channel(1);

julia> isready(c)
false

julia> task = Task(() -> wait(c));

julia> schedule(task);

julia> istaskdone(task)  # task is blocked because channel is not ready
false

julia> put!(c, 1);

julia> istaskdone(task)  # task is now unblocked
true

fetch(t::Task)

Подождать завершения Task, затем вернуть значение результата. Если задача завершится с ошибкой, будет выброшено исключение TaskFailedException (которое оборачивает завершившуюся с ошибкой задачу).

fetch(x::Any)

Вернуть x.

timedwait(testcb, timeout::Real; pollint::Real=0.1)

Подождать, пока testcb() вернёт true или пройдёт timeout секунд, в зависимости от того, что произойдёт раньше. Функция-тест вызывается каждые pollint секунды. Минимальное значение для pollint составляет 0,001 секунды, то есть 1 миллисекунда.

Возвращает :ok или :timed_out.

Примеры

julia> cb() = (sleep(5); return);

julia> t = @async cb();

julia> timedwait(()->istaskdone(t), 1)
:timed_out

julia> timedwait(()->istaskdone(t), 6.5)
:ok

Condition()

Создать источник событий с триггером по краю, для которого задачи могут ожидать. Задачи, вызывающие wait на Condition, приостанавливаются и помещаются в очередь. Задачи возобновляются, когда позже вызывается notify для Condition. Ожидание на условии может вернуть значение или сгенерировать ошибку, если используются необязательные аргументы notify. Триггер по краю означает, что разбужены только задачи, ожидающие в момент вызова notify. Для триггеров по уровню нужно сохранять дополнительное состояние, чтобы отслеживать, произошло ли уведомление. Типы Channel и Threads.Event делают это и могут использоваться для событий с триггером по уровню.

Этот объект НЕ потокобезопасен. См. Threads.Condition для потокобезопасной версии.

Threads.Condition([lock])

Потокобезопасная версия Base.Condition.

Чтобы вызвать wait или notify на Threads.Condition, необходимо сначала вызвать lock на нём. При вызове wait, блокировка атомарно освобождается во время ожидания, и будет повторно захвачена перед возвратом wait. Поэтому характерное использование Threads.Condition c выглядит следующим образом:

lock(c)
try
    while !thing_we_are_waiting_for
        wait(c)
    end
finally
    unlock(c)
end

Event([autoreset=false])

Создайте источник событий с уровнем срабатывания. Задачи, которые вызывают wait на Event приостанавливаются и помещаются в очередь до тех пор, пока не будет вызван notify на Event. После вызова notify, Event остаётся в сигнализированном состоянии, и задачи больше не будут блокироваться при ожидании его, пока не будет вызван reset.

Если autoreset имеет значение true, то не более одной задачи будет освобождена из wait для каждого вызова notify.

Это обеспечивает порядок памяти для приобретения и освобождения при уведомлении/ожидании.

notify(condition, val=nothing; all=true, error=false)

Разбудить задачи, ожидающие условия, передав им val. Если all имеет значение true (по умолчанию), все ожидающие задачи будятся, в противном случае только одна. Если error имеет значение true, переданное значение поднимается как исключение в разбуженных задачах.

Возвращает количество разбуженных задач. Возвращает 0, если на condition нет ожидающих задач.

reset(::Event)

Сбросить Event обратно в неинициализированное состояние. Тогда любые последующие вызовы wait будут блокироваться до тех пор, пока снова не будет вызван notify.

Semaphore(sem_size)

Создайте счётчик семафора, который позволяет не более sem_size приобретений быть в использовании в любой момент. Каждое приобретение должно быть сопоставлено с освобождением.

Это обеспечивает порядок памяти для приобретения и освобождения при вызовах приобретения/освобождения.

acquire(s::Semaphore)

Подождите, пока один из sem_size разрешений не станет доступным, блокируясь до тех пор, пока одно не будет приобретено.

acquire(f, s::Semaphore)

Выполните f после приобретения из семафора s, и release по завершении или ошибке.

Например, блок do, который гарантирует, что только 2 вызова foo будут активны одновременно:

s = Base.Semaphore(2)
@sync for _ in 1:100
    Threads.@spawn begin
        Base.acquire(s) do
            foo()
        end
    end
end

release(s::Semaphore)

Возвращает одно разрешение в пул, возможно, позволяя другой задаче его приобрести и возобновить выполнение.

AbstractLock

Абстрактный супертип, описывающий типы, которые реализуют примитивы синхронизации: lock, trylock, unlock и islocked.

lock(lock)

Приобретает lock при его доступности. Если блокировка уже заблокирована другой задачей/потоком, дождитесь, пока она станет доступной.

Каждое lock должно быть сопоставлено с unlock.

lock(f::Function, lock)

Приобретает lock, выполняет f с захваченной lock, и освобождает lock при возвращении f. Если блокировка уже заблокирована другой задачей/потоком, дождитесь, пока она станет доступной.

При возвращении этой функции lock была освобождена, поэтому вызывающая сторона не должна пытаться unlock её.

См. также: @lock.

lock(f::Function, l::Lockable)

Приобретает блокировку, связанную с l, выполняет f с захваченной блокировкой и освобождает блокировку при возвращении f. f получит один позиционный аргумент: значение, обернутое l. Если блокировка уже заблокирована другой задачей/потоком, дождитесь, пока она станет доступной. При возвращении этой функции lock была освобождена, поэтому вызывающая сторона не должна пытаться unlock её.

unlock(lock)

Освобождает владение lock.

Если это рекурсивная блокировка, которая была приобретена ранее, уменьшается внутренний счётчик и возвращается немедленно.

trylock(lock) -> Success (Boolean)

Приобрести блокировку, если она доступна, и вернуть true в случае успеха. Если блокировка уже заблокирована другой задачей/потоком, вернуть false.

Каждое успешное trylock должно быть сопоставлено с unlock.

Функция trylock в сочетании с islocked может использоваться для написания алгоритмов «проверка и установка» или экспоненциального отката, если это поддерживается typeof(lock) (прочитайте документацию).

islocked(lock) -> Status (Boolean)

Проверить, удерживается ли lock какой-либо задачей/потоком. Эта функция сама по себе не должна использоваться для синхронизации. Однако, islocked в сочетании с trylock может использоваться для написания алгоритмов «проверка и установка» или экспоненциального отката, если это поддерживается typeof(lock) (прочитайте документацию).

Дополнительная помощь

Например, экспоненциальный откат может быть реализован следующим образом, если реализация lock удовлетворяет свойствам, описанным ниже.

nspins = 0
while true
    while islocked(lock)
        GC.safepoint()
        nspins += 1
        nspins > LIMIT && error("timeout")
    end
    trylock(lock) && break
    backoff()
end

Реализация

Рекомендуется, чтобы реализация блокировки определяла islocked со следующими свойствами и указанием об этом в своей документации.

• islocked(lock) является свободной от гонок данных.
• Если islocked(lock) возвращает false, непосредственный вызов trylock(lock) должен быть успешным (возвращает true) в случае отсутствия помех от других задач.

ReentrantLock()

Создаёт рекурсивную блокировку для синхронизации Task. Одна и та же задача может приобретать блокировку сколько угодно раз (отсюда и «Reentrant» в названии). Каждое lock должно быть сопоставлено с unlock.

Вызов lock также запретит выполнение финализаторов в этом потоке до соответствующего unlock. Использование стандартного шаблона блокировки, проиллюстрированного ниже, должно поддерживаться, но будьте осторожны при инвертировании порядка try/lock или пропуска блоков try (например, попытка возврата с ещё удерживаемой блокировкой):

Это обеспечивает порядок памяти для приобретения/освобождения при вызовах lock/unlock.

lock(l)
try
    <atomic work>
finally
    unlock(l)
end

Если !islocked(lck::ReentrantLock) выполняется, trylock(lck) имеет успех, если нет других задач, пытающихся удерживать блокировку «одновременно».

@lock l expr

Макроверсия lock(f, l::AbstractLock), но с функцией expr вместо f. Развертывается в:

lock(l)
try
    expr
finally
    unlock(l)
end

Это аналогично использованию lock с блоком do, но избегает создания замыкания, что может улучшить производительность.

Lockable(value, lock = ReentrantLock())

Создаёт объект Lockable, который оборачивает value и связывает его с предоставленным lock. Этот объект поддерживает @lock, lock, trylock, unlock. Для доступа к значению необходимо использовать индексацию объекта Lockable, удерживая замок.

Пример

julia> locked_list = Base.Lockable(Int[]);

julia> @lock(locked_list, push!(locked_list[], 1)) # must hold the lock to access the value
1-element Vector{Int64}:
 1

julia> lock(summary, locked_list)
"1-element Vector{Int64}"

AbstractChannel{T}

Представление канала, передающего объекты типа T.

Channel{T=Any}(size::Int=0)

Создаёт канал Channel с внутренней буфером, который может содержать максимальное количество size объектов типа T. Вызовы put! на полном канале будут блокироваться до тех пор, пока объект не будет удалён с помощью take!.

Channel(0) создаёт небуферизованный канал. put! блокируется до тех пор, пока не будет вызван соответствующий take!. И наоборот.

Другие конструкторы:

• Channel(): конструктор по умолчанию, эквивалентный Channel{Any}(0)
• Channel(Inf): эквивалентный Channel{Any}(typemax(Int))
• Channel(sz): эквивалентный Channel{Any}(sz)

Channel{T=Any}(func::Function, size=0; taskref=nothing, spawn=false, threadpool=nothing)

Создаёт новую задачу из func, связывает её с новым каналом типа T и размером size, и планирует выполнение задачи, всё в одном вызове. Канал автоматически закрывается при завершении задачи.

func должен принимать привязанный канал в качестве единственного аргумента.

Если вам нужна ссылка на созданную задачу, передайте объект Ref{Task} через ключевой аргумент taskref.

Если spawn=true, задача, созданная для Task может быть запланирована на другой потоке параллельно, что эквивалентно созданию задачи через Threads.@spawn.

Если spawn=true и аргумент threadpool не задан, он по умолчанию равен :default.

Если аргумент threadpool задан (в :default или :interactive), это означает, что spawn=true и новая задача запускаются в указанном пуле потоков.

Возвращает Channel.

Примеры

julia> chnl = Channel() do ch
           foreach(i -> put!(ch, i), 1:4)
       end;

julia> typeof(chnl)
Channel{Any}

julia> for i in chnl
           @show i
       end;
i = 1
i = 2
i = 3
i = 4

Ссылка на созданную задачу:

julia> taskref = Ref{Task}();

julia> chnl = Channel(taskref=taskref) do ch
           println(take!(ch))
       end;

julia> istaskdone(taskref[])
false

julia> put!(chnl, "Hello");
Hello

julia> istaskdone(taskref[])
true

julia> chnl = Channel{Char}(1, spawn=true) do ch
           for c in "hello world"
               put!(ch, c)
           end
       end
Channel{Char}(1) (2 items available)

julia> String(collect(chnl))
"hello world"

put!(c::Channel, v)

Добавляет элемент v в канал c. Блокируется, если канал заполнен.

Для небуферизованных каналов блокируется до тех пор, пока другая задача не выполнит take!.

take!(c::Channel)

Удаляет и возвращает значение из канала Channel в порядке очереди. Блокируется до тех пор, пока данные не станут доступными. Для небуферизованных каналов блокируется до тех пор, пока другая задача не выполнит put!.

Примеры

Буферизованный канал:

julia> c = Channel(1);

julia> put!(c, 1);

julia> take!(c)
1

Небуферизованный канал:

julia> c = Channel(0);

julia> task = Task(() -> put!(c, 1));

julia> schedule(task);

julia> take!(c)
1

isready(c::Channel)

Определяет, есть ли значение в Channel. Возвращает значение немедленно, не блокируется.

Для небуферизованных каналов возвращает true если есть задачи, ожидающие put!.

Примеры

Буферизованный канал:

julia> c = Channel(1);

julia> isready(c)
false

julia> put!(c, 1);

julia> isready(c)
true

Небуферизованный канал:

julia> c = Channel();

julia> isready(c)  # no tasks waiting to put!
false

julia> task = Task(() -> put!(c, 1));

julia> schedule(task);  # schedule a put! task

julia> isready(c)
true

fetch(c::Channel)

Ожидает и возвращает (без удаления) первый доступный элемент из Channel. Примечание: fetch не поддерживается для небуферизованного (0-размерного) Channel.

Примеры

Буферизованный канал:

julia> c = Channel(3) do ch
           foreach(i -> put!(ch, i), 1:3)
       end;

julia> fetch(c)
1

julia> collect(c)  # item is not removed
3-element Vector{Any}:
 1
 2
 3

close(c::Channel[, excp::Exception])

Закрывает канал. Исключение (если указано в excp):

• put! на закрытом канале.
• take! и fetch на пустом, закрытом канале.

bind(chnl::Channel, task::Task)

Связывает срок жизни chnl с задачей. Channel chnl автоматически закрывается при завершении задачи. Любое необработанное исключение в задаче передаётся всем ожидающим задачам на chnl.

Объект канала можно явным образом закрыть независимо от завершения задачи. Завершение задач не влияет на уже закрытые объекты каналов Channel.

Когда канал связан с несколькими задачами, первая завершающаяся задача закрывает канал. Когда несколько каналов связаны с одной задачей, завершение задачи закрывает все связанные каналы.

Примеры

julia> c = Channel(0);

julia> task = @async foreach(i->put!(c, i), 1:4);

julia> bind(c,task);

julia> for i in c
           @show i
       end;
i = 1
i = 2
i = 3
i = 4

julia> isopen(c)
false

julia> c = Channel(0);

julia> task = @async (put!(c, 1); error("foo"));

julia> bind(c, task);

julia> take!(c)
1

julia> put!(c, 1);
ERROR: TaskFailedException
Stacktrace:
[...]
    nested task error: foo
[...]