Задачи

Core.TaskТип

Task(func)

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

Примеры

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

julia> b = Task(a);

В этом примере b — это исполняемая Task задача, которая ещё не запущена.

Base.current_taskФункция

current_task()

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

Base.istaskdoneФункция

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

Base.istaskstartedФункция

istaskstarted(t::Task) -> Bool

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

Примеры

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

julia> b = Task(a3);

julia> istaskstarted(b)
false

Base.yieldФункция

yield()

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

yield(t::Task, arg = nothing)

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

Base.yieldtoФункция

yieldto(t::Task, arg = nothing)

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

Base.task_local_storageМетод

task_local_storage(key)

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

Base.task_local_storageМетод

task_local_storage(key, value)

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

Base.task_local_storageМетод

task_local_storage(body, key, value)

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

Base.ConditionТип

Condition()

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

Base.notifyФункция

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

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

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

Base.scheduleФункция

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

Base.@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

Base.sleepФункция

sleep(seconds)

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

Base.ChannelТип

Channel{T}(sz::Int)

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

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

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

• Channel(Inf): эквивалентно Channel{Any}(typemax(Int))
• Channel(sz): эквивалентно Channel{Any}(sz)

Base.put!Метод

put!(c::Channel, v)

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

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

Base.take!Метод

take!(c::Channel)

Удалить и вернуть значение из Channel. Блокируется до тех пор, пока данные не станут доступны.

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

Base.isreadyМетод

isready(c::Channel)

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

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

Base.fetchМетод

fetch(c::Channel)

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

Base.closeМетод

close(c::Channel)

Закрыть канал. Исключение выброшено:

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

Base.bindМетод

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: foo
Stacktrace:
[...]

Base.asyncmapФункция

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"

Base.asyncmap!Функция

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

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