Задачи

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