Добавить работу в очередь

Создание параллельного пула p с двумя рабочими.

p = parpool(2);

Starting parallel pool (parpool) using the 'local' profile ...
Connected to the parallel pool (number of workers: 2).

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

Использовать parfeval для создания массива фьючерсов f инструктирование работников для выполнения функции pause. Использовать аргумент 1 для третьего будущего и аргумент Inf для всех остальных фьючерсов.

for n = 1:5
    if n == 3
        f(n) = parfeval(@pause,0,1);
    else
        f(n) = parfeval(@pause,0,Inf);
    end
end

Каждое использование parfeval возвращает будущий объект, представляющий выполнение функции на работнике. За исключением третьего будущего, для вычисления каждого будущего потребуется бесконечное количество времени. Будущее, созданное parfeval(@pause,0,Inf) это крайний случай будущего, который может замедлить очередь.

Отменить фьючерсы напрямую

Вы можете использовать State имущество для получения статуса фьючерса. Построение массива ячеек состояния каждого будущего в f.

{f.State}

ans = 1×5 cell
    {'running'}    {'running'}    {'queued'}    {'queued'}    {'queued'}

Каждая задача, кроме третьей, останавливается навсегда.

Отмена второго будущего непосредственно с помощью cancel.

cancel(f(2));
{f.State}

ans = 1×5 cell
    {'running'}    {'finished'}    {'running'}    {'queued'}    {'queued'}

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

wait(f(3));
{f.State}

ans = 1×5 cell
    {'running'}    {'finished'}    {'finished'}    {'running'}    {'queued'}

Третье будущее теперь за государством 'finished'.

Проверка ошибок завершения

По завершении будущего State свойство становится 'finished'. Чтобы различать фьючерсы, которые были отменены и завершены нормально, используйте Error собственность.

fprintf("f(2): %s\n", f(2).Error.message)

f(2): Execution of the future was cancelled.

fprintf("f(3): %s\n", f(3).Error.message)

f(3): 

Код отменяет второе будущее, как указывает свойство сообщения. Второе будущее было отменено, как указано в message собственность. Третье будущее завершается без ошибок и поэтому не имеет сообщения об ошибке.

Отмена фьючерсов в очереди пула

Вы можете использовать FevalQueue для доступа к фьючерсам в очереди пула.

p.FevalQueue

ans = 
 FevalQueue with properties: 

        Number Queued: 1
       Number Running: 2

Очередь имеет два свойства: RunningFutures и QueuedFutures. RunningFutures свойство представляет собой массив фьючерсов, соответствующий выполняемым в настоящее время задачам.

disp(p.FevalQueue.RunningFutures)

 1x2 FevalFuture array:
 
         ID              State  FinishDateTime  Function  Error
       --------------------------------------------------------
    1    12            running                    @pause       
    2    15            running                    @pause       

QueuedFutures - это массив фьючерсов, соответствующий задачам, которые в данный момент поставлены в очередь и не выполняются.

disp(p.FevalQueue.QueuedFutures)

 FevalFuture with properties: 

                   ID: 16
             Function: @pause
       CreateDateTime: 15-Jul-2020 17:29:37
        StartDateTime: 
     Running Duration: 0 days 0h 0m 0s
                State: queued
                Error: none

Можно отменить одно будущее или массив фьючерсов. Отменить все фьючерсы в QueuedFutures.

cancel(p.FevalQueue.QueuedFutures);
{f.State}

ans = 1×5 cell
    {'running'}    {'finished'}    {'finished'}    {'running'}    {'finished'}

RunningFutures и QueuedFutures сортируются с самых новых на самые старые независимо от того, f в порядке от самых новых к самым старым. Каждое будущее имеет уникальный ID для времени жизни клиента. Проверьте ID имущество каждого из фьючерсов в f.

disp(f)

 1x5 FevalFuture array:
 
         ID              State        FinishDateTime  Function  Error
       --------------------------------------------------------------
    1    12            running                          @pause       
    2    13  finished (unread)  15-Jul-2020 17:29:37    @pause  Error
    3    14  finished (unread)  15-Jul-2020 17:29:39    @pause       
    4    15            running                          @pause       
    5    16  finished (unread)  15-Jul-2020 17:29:39    @pause  Error

Сравните результат с ID свойство каждого из RunningFutures.

for j = 1:length(p.FevalQueue.RunningFutures)
    rf = p.FevalQueue.RunningFutures(j);
    fprintf("p.FevalQueue.RunningFutures(%i): ID = %i\n", j, rf.ID)
end

p.FevalQueue.RunningFutures(1): ID = 12
p.FevalQueue.RunningFutures(2): ID = 15

Здесь, RunningFutures является массивом, содержащим f(1) и f(4). Если отменить RunningFutures(2), вы отменяете четвертое будущее f(4).

Иногда фьючерсы недоступны в рабочей области, например, если один и тот же фрагмент кода выполняется дважды до его завершения или если используется parfeval в функции. Можно отменить фьючерсы, недоступные в рабочей области.

Ясно f из рабочей области.

clear f

Вы можете использовать RunningFutures и QueuedFutures для доступа к фьючерсам, которые еще не завершены. Использовать RunningFutures отменить f(4).

rf2 = p.FevalQueue.RunningFutures(2);
cancel(rf2)
rf2.State

ans = 
'finished'

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

cancel([p.FevalQueue.RunningFutures p.FevalQueue.QueuedFutures])