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

Создание параллельного пула 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])

Например, отправьте один запрос в параллельный пул. Извлечение выходных данных с помощью fetchOutputs.

f = parfeval(@magic,1,10);
value = fetchOutputs(f);

Вы также можете отправить вектор из нескольких будущих запросов в for-закольцовывать и собирать результаты по мере их доступности. Для повышения эффективности предварительно распределите массив будущих объектов.

f(1:10) = parallel.FevalFuture;
for idx = 1:10
    f(idx) = parfeval(@magic,1,idx);
end

Извлеките отдельные будущие результаты по мере их доступности с помощью fetchNext.

magicResults = cell(1,10);
for idx = 1:10
    [completedIdx,value] = fetchNext(f);
    magicResults{completedIdx} = value;
    fprintf('Got result with index: %d.\n', completedIdx);
end

Создать сетку параметров

Определите диапазон параметров, которые требуется исследовать при протягивании параметров.

gridSize = 40;
sigma = linspace(5, 45, gridSize);
rho = linspace(50, 100, gridSize);
beta = 8/3;

Создайте 2-D сетку параметров с помощью meshgrid функция.

[rho,sigma] = meshgrid(rho,sigma);

Создание объекта фигуры и установка 'Visible' кому true чтобы она открылась в новом окне, за пределами живого сценария. Для визуализации результатов сдвига параметров создайте график поверхности. Обратите внимание, что инициализация Z компонент поверхности с NaN создает пустой график.

figure('Visible',true);
surface = surf(rho,sigma,NaN(size(sigma)));
xlabel('\rho','Interpreter','Tex')
ylabel('\sigma','Interpreter','Tex')

Настройка параллельной среды

Создание пула параллельных работников с помощью parpool функция.

parpool;

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

Чтобы отправить данные от работников, создайте DataQueue объект. Настройте функцию, которая обновляет график поверхности каждый раз, когда работник отправляет данные с помощью afterEach функция. updatePlot функция является вспомогательной функцией, определенной в конце примера.

Q = parallel.pool.DataQueue;
afterEach(Q,@(data) updatePlot(surface,data));

Выполнение сдвига параллельных параметров

После определения параметров можно выполнить параллельный сдвиг параметров.

parfeval работает более эффективно при распределении рабочей нагрузки. Чтобы распределить рабочую нагрузку, сгруппируйте параметры для изучения в разделы. Для этого примера разбейте на однородные секции размера step с помощью оператора двоеточия (:). Результирующий массив partitions содержит границы секций. Обратите внимание, что необходимо добавить конечную точку последнего раздела.

step = 100;
partitions = [1:step:numel(sigma), numel(sigma)+1]

partitions = 1×17

           1         101         201         301         401         501         601         701         801         901        1001        1101        1201        1301        1401        1501        1601

Для достижения оптимальной производительности попробуйте разбить на следующие разделы:

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

f(1:numel(partitions)-1) = parallel.FevalFuture;

Разгрузка вычислений для параллельных работников с помощью parfeval функция. parameterSweep - вспомогательная функция, определенная в конце этого сценария, которая решает систему Лоренца на разделе исследуемых параметров. Он имеет один выходной аргумент, поэтому необходимо указать 1 как количество выходов в parfeval.

for ii = 1:numel(partitions)-1
    f(ii) = parfeval(@parameterSweep,1,partitions(ii),partitions(ii+1),sigma,rho,beta,Q);
end

parfeval не блокирует MATLAB, поэтому можно продолжить работу во время выполнения вычислений. Рабочие вычисляют параллельно и отправляют промежуточные результаты через DataQueue как только они станут доступны.

Если требуется заблокировать MATLAB до parfeval завершает, используйте wait функция на будущих объектах. Использование wait полезна, когда последующий код зависит от завершения parfeval.

wait(f);

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

results = reshape(fetchOutputs(f),gridSize,[]);
contourf(rho,sigma,results)
xlabel('\rho','Interpreter','Tex')
ylabel('\sigma','Interpreter','Tex')

Если для сдвига параметров требуется больше вычислительных ресурсов и у вас есть доступ к кластеру, вы можете увеличить parfeval вычисления. Дополнительные сведения см. в разделе Масштабирование с рабочего стола на кластер.

Определение вспомогательных функций

Определите вспомогательную функцию, которая решает систему Лоренца на разделе исследуемых параметров. Отправка промежуточных результатов клиенту MATLAB с помощью send функции на DataQueue объект.

function results = parameterSweep(first,last,sigma,rho,beta,Q)
    results = zeros(last-first,1);
    for ii = first:last-1
        lorenzSystem = @(t,a) [sigma(ii)*(a(2) - a(1)); a(1)*(rho(ii) - a(3)) - a(2); a(1)*a(2) - beta*a(3)];
        [t,a] = ode45(lorenzSystem,[0 100],[1 1 1]);
        result = a(end,3);
        send(Q,[ii,result]);
        results(ii-first+1) = result;
    end
end

Определите другую вспомогательную функцию, которая обновляет график поверхности при поступлении новых данных.

function updatePlot(surface,data)
    surface.ZData(data(1)) = data(2);
    drawnow('limitrate');
end