pool = parpool("Processes",6);

[gpuPool,cpuPool] = partition(pool,"MaxNumWorkersPerGPU",1);

function [r,x] = populationGrowth(N,numIterations)
    r = gpuArray.linspace(0,4,N);
    x = rand(1,N,"gpuArray");
    for n=1:numIterations
        x = r.*x.*(1-x);
    end
end

f = parfeval(gpuPool,@populationGrowth,2,200000,1000);

M = 100;
N = 1e6;
data = rand(M,N);
parfor(idx = 1:M,cpuPool)
    out(idx) = sum(data(idx,:))./N;
end

[x,r] = fetchOutputs(f);

myClusterPool = parpool("myCluster",64);

hostPool = partition(myClusterPool,"MaxNumWorkersPerHost",1);

fSetup = parfevalOnAll(hostPool,@downloadData,1);
fetchOutputs(fSetup)

poolSize = 16;
pool = parpool("myCluster",poolSize);

allWkrs = pool.Workers;
hostnames = {allWkrs.Host}';
[hostnames,ia,ic] = unique(hostnames);
numWorkers = accumarray(ic,1);
workersPerHost = table(hostnames,numWorkers)      

highMemPool = partition(pool,"MaxNumWorkersPerHost",4);
highMemPoolSize = highMemPool.NumWorkers

memoryPerWorker = 8; % In GB
totalMemory = memoryPerWorker*poolSize;
newMemoryPerWorker = totalMemory/highMemPoolSize

maxMemUsagePerWorker = 0.45*newMemoryPerWorker*1024^3; % In bytes.
maxMatSize = round(sqrt(maxMemUsagePerWorker*highMemPoolSize/memoryPerWorker));
fprintf("The `highMemPool` pool can support" + ...
    " a matrix size of up to %d-by-%d\n",maxMatSize,maxMatSize)

spmd(highMemPool)
A = rand(maxMatSize,"codistributed");
b = rand(maxMatSize,1,"codistributed");
x = A\b;
end

poolSize = 16;
pool = parpool("myCluster",poolSize);

allWkrs = pool.Workers;
hostnames = {allWkrs.Host}';
[uniqueHosts,~,hostIndices] = unique(hostnames);
numWorkers = accumarray(hostIndices,1);
hostWorkers = table(uniqueHosts,numWorkers) 

threadsPerWorker = 4;
workersPerHost = ceil(min(hostWorkers.numWorkers/threadsPerWorker))

multiThreadsPool = partition(pool,"MaxNumWorkersPerHost",workersPerHost);

setNumCompThreads = parfevalOnAll(multiThreadsPool, ...
    @maxNumCompThreads,1,threadsPerWorker);
lastThreads = fetchOutputs(setNumCompThreads);

N = randn(5000);
numIterations = 10;
parfor (idx = 1:numIterations,multiThreadsPool)
    out = N*N;
end

setNumCompThreads = parfevalOnAll(multiThreadsPool,@maxNumCompThreads,0,lastThreads(1));
fetchOutputs(setNumCompThreads)

imSize = [500 500 500];

pool = parpool("myCluster",10);

spmd
V = rand(imSize,"single");
BW = false(size(V));
BW(V < 0.000001) = true;
V = bwdist(BW);
V(V <= 20) = 1;
V(V > 20) = 0;
end
volshow(V{1})

spmd
V = spmdCat(V,3,1);
end

spmd
wkrs = getCurrentWorker;
end

dataWkrPool = partition(pool,"Workers",wkrs{1});

V = parallel.pool.Constant(V);

function [numBubbles,bubbleVolumes] = myBubbleFunction(V)
labeledV = bwlabeln(V.Value);
numBubbles = max(labeledV(:));
% Calculate the volume of each bubble 
stats = regionprops3(labeledV,"Volume");
bubbleVolumes = stats.Volume';
end

f = parfeval(dataWkrPool,@myBubbleFunction,2,V);
[numBubbles,bubbleVolumes] = fetchOutputs(f)

pool = parpool("Threads");

allWkrs = pool.Workers;
dBWkrPool = partition(pool,"Workers",allWkrs(1));

cDbase = parallel.pool.Constant(@() connectToDatabase("Database"));

function imgs = getDbaseData(C,query)
conn = C.Value;
imgs = fetchQuery(conn,query);
end

imageNames = ["outdoors" "indoors" "daytime" "nightime"];
sqlqueries = "SELECT * FROM " + imageNames;
cFilter = parallel.pool.Constant(randn(16,1));

fImport = parfeval(dBWkrPool,@getDbaseData,1,cDbase,sqlqueries(1));

for j = 1:length(sqlqueries)
    % Fetch data asynchronously
    data = fetchOutputs(fImport); % Wait for the data to be fetched
    % Schedule the next parfeval computation to run in the background.
    if j < length(imageNames)
        fImport = parfeval(dBWkrPool,@getDbaseData,1,cDbase,sqlqueries(j+1));
    end
    parfor k=1:size(data,2)
        % Zero-pad filter to the length of data, and transform
        [rows,~] = size(data);
        filterF = fft(cFilter.Value,rows);
        % Transform each column of the input data
        imgFft = fft2(data{k});
        % Multiply each column by filter and compute inverse transform
        out{k} = ifft2(filterF.*imgFft)
    end
    processedImages.(imageNames(j)) = out;
    fprintf("'%s' images processed\n",imageNames(j))
end

pool = gcp;
poolWkrs = pool.Workers;

traderPools(1) = partition(pool,"Workers",poolWkrs(1));
traderPools(2) = partition(pool,"Workers",poolWkrs(2));
traderPools(3) = partition(pool,"Workers",poolWkrs(3));

function decision = tradingWorker(marketData,riskFactor)
% Initialize or update the state
persistent state
if isempty(state)
    state = struct("history",[],"position","none");
end

% Process new market data
state.history = [state.history;marketData];

% Calculate short-term and long-term averages
shortTermAvg = mean(marketData);
longTermAvg = mean(state.history);

% Trading logic with a worker-specific risk factor
randomFactor = 1 + riskFactor*randn; % Worker-specific randomness

if strcmp(state.position,"none") && (shortTermAvg > longTermAvg*randomFactor)
    state.position = "buy";
elseif strcmp(state.position,"buy") && (shortTermAvg < longTermAvg*randomFactor)
    state.position = "sell";
elseif strcmp(state.position,"sell") && (shortTermAvg >= longTermAvg*randomFactor)
    state.position = "none";
end

decision = state.position;
pause(1) % Simulate processing time
end

fig = uifigure(Position=[619 525 443 308]);
decisionTable = table(Size=[0 4], ...
    VariableTypes=["double","string","string","string"], ...
    VariableNames=["Time","Trader 1","Trader 2","Trader 3"]);
uit = uitable(fig,Data=decisionTable,Position=[20 11 350 266]);

riskFactors = [0.1 0.2 0.3];
for t = 1:10
    %   % Simulate real-time market data
    marketData = rand(100,1)*100; % Random price data

    % Create futures to process data in parallel
    futures(1) = parfeval(traderPools(1),@tradingWorker,1,marketData,riskFactors(1));
    futures(2) = parfeval(traderPools(2),@tradingWorker,1,marketData,riskFactors(2));
    futures(3) = parfeval(traderPools(3),@tradingWorker,1,marketData,riskFactors(3));

    decisions = fetchOutputs(futures);
    decisionTable(end+1,:) = {t,decisions(1),decisions(2),decisions(3)};
    uit.Data = decisionTable;
    drawnow limitrate nocallbacks
end

clear tradingWorker

pool = parpool(10);

fs = 44100;
frequencyDuration = 3;
audioDuration = 30;
numAudio = 50;
audioSamples = cell(1,numAudio);
parfor idx = 1:numAudio
audioSamples{idx} = generateAudio(audioDuration,frequencyDuration,fs);
end

poolWkrs = pool.Workers;
smallPool = partition(pool,"Workers",poolWkrs(1:2:end));

futures(1:numAudio) = parallel.FevalFuture;
for a = 1:numAudio
futures(a) = parfeval(smallPool,@extractFeatures,1,audioSamples{a},fs);
end

sftfData = fetchOutputs(futures);