Задачи

Task(func)

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

Примеры

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 завершается неудачно.

@sync

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

Особое примечание для Threads.Condition:

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

wait(r::Future)

Ожидание значения, которое станет доступным для указанной Future.

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

Ожидание значения, которое станет доступным в указанном RemoteChannel.

wait([x])

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

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

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

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

fetch(t::Task)

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

timedwait(callback::Function, timeout::Real; pollint::Real=0.1)

Ожидает, пока callback вернёт true или пройдёт timeout секунд, в зависимости от того, что наступит раньше. callback опрошается каждые pollint секунды. Минимальное значение для timeout и pollint составляет 0.001, то есть 1 миллисекунда.

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

Condition()

Создает источник событий, на который могут ждать задачи. Задачи, которые вызывают wait на Condition, приостанавливаются и помещаются в очередь. Задачи разбуживаются, когда позже вызывается notify на Condition. Триггерирование по краю означает, что разбужены только те задачи, которые ожидали на момент вызова 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)

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

Semaphore(sem_size)

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

Это обеспечивает согласование памяти при вызовах acquire/release.

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 её.

unlock(lock)

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

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

trylock(lock) -> Success (Boolean)

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

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

islocked(lock) -> Status (Boolean)

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

ReentrantLock()

Создаёт рекурсивный замок для синхронизации Taskов. Такая же задача может захватывать замок столько раз, сколько нужно. Каждый lock должен быть сопоставлен с unlock.

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

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

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

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)

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

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

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

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

Вернуть 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! другой задачей.

isready(c::Channel)

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

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

fetch(c::Channel)

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

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

Закрыть канал. Исключение (возможно, заданное excp), выбрасывается:

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

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

Свяжите срок жизни chnl с задачей. Channel chnl автоматически закрывается при завершении задачи. Любое необработанное исключение в задаче передаётся всем ожидающим на 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
[...]