Задачи

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, все элементы обрабатываются в 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 вместо возврата коллекции.