任务并行问题中的资源争用

delete(gcp("nocreate")); % Close any existing parallel pools
p = parpool("Processes");

poolSize = p.NumWorkers;

sz = 2048;
m = rand(sz*sz,1);

tsum = timingFcn(@() sumOp(m), 1:poolSize);

tfft = timingFcn(@() fftOp(m),1:poolSize);

m = reshape(m,sz,sz);
tmtimes = timingFcn(@() multOp(m),1:poolSize);

clear m

allTimes = [tsum(:), tfft(:), tmtimes(:)];
speedup = (allTimes(1,:)./allTimes).*((1:poolSize)');

bar(speedup)
legend('Vector Sum', 'Vector FFT', 'Matrix Mult.', ...
    'Location', 'NorthWest')
xlabel('Number of Concurrent Processes');
ylabel('Speedup')
title(['Effect of No. Concurrent Processes on ', ...
    'Resource Contention and Speedup']);

szs = [128, 256, 512, 1024, 2048];
operations = {'Vector Sum', 'Vector FFT', 'Matrix Mult.', 'Matrix LU', ...
    'Matrix SVD', 'Matrix EIG'};

speedup = zeros(length(szs), length(operations));

% Loop over the data sizes
for i = 1:length(szs)
    sz = szs(i);
    fprintf('Using matrices of size %d-by-%d.\n', sz, sz);
    j = 1;

    % Loop over the different operations
    for f = [{@sumOp; sz^2; 1}, {@fftOp; sz^2; 1}, {@multOp; sz; sz}, ...
            {@lu; sz; sz}, {@svd;  sz; sz}, {@eig;  sz; sz}]
        op = f{1};
        nrows = f{2};
        ncols = f{3};
        m = rand(nrows, ncols);
        % Compare sequential execution to execution on all workers
        tcurr = timingFcn(@() op(m), [1, poolSize]);
        speedup(i, j) = tcurr(1)/tcurr(2)*poolSize;
        j = j + 1;
    end

end

figure
ax = axes;
plot(speedup)
set(ax.Children, {'Marker'}, {'+', 'o', '*', 'x', 's', 'd'}')
hold on

plot([1 length(szs)],[poolSize poolSize], '--')

xticks(1:length(szs));
xticklabels(szs + "^2")
xlabel('Number of Elements per Process')

ylim([0 poolSize+0.5])
ylabel('Speedup')

legend([operations, {'Ideal Speedup'}], 'Location', 'SouthWest')
title('Effect of Data Size on Resource Contention and Speedup')

hold off

function time = timingFcn(fcn, numConcurrent)

time = zeros(1,length(numConcurrent));
numTests = 5;

% Record the execution time on 1 to numConcurrent workers
for ind = 1:length(numConcurrent)
    n = numConcurrent(ind);

    spmd(n)
        tconcurrent = inf;

        % Time the function numTests times, and record the minimum time
        for itr = 1:numTests
            % Measure only task parallel runtime
            spmdBarrier;
            tic;
            fcn();
            % Record the time for all to complete
            tAllDone = spmdReduce(@max, toc);  
            tconcurrent = min(tconcurrent, tAllDone);
        end
        
    end

    time(ind) = tconcurrent{1};
    clear tconcurrent itr tAllDone
    if ind == 1
        fprintf('Execution times: %f', time(ind));
    else
        fprintf(', %f', time(ind));
    end
end

fprintf('\n');
end

function sumOp(m)
s = 0;

for itr = 1:100
    s = s + sum(m);
end

end

function fftOp(m)

for itr = 1:10
    fft(m);
end

end

function multOp(m)
m*m;
end