startmulticoremaster.m

Multicore - Parallel Processing on Multiple Cores

function resultCell = startmulticoremaster(functionHandleCell, parameterCell, settings)
%STARTMULTICOREMASTER  Start multi-core processing master process.
%   RESULTCELL = STARTMULTICOREMASTER(FHANDLE, PARAMETERCELL)
%   starts a multi-core processing master process. The function specified
%   by the given function handle is evaluated with the parameters saved in
%   each cell of PARAMETERCELL. Each cell may include parameters in any
%   form or another cell array which is expanded to an argument list using
%   the {:} notation to pass multiple input arguments. The outputs of the
%   function are returned in cell array RESULTCELL of the same size as
%   PARAMETERCELL. Only the first output argument of the function is
%   returned. If you need to get multiple outputs, write a small adapter
%   that puts the outputs of your function into a single cell array.
%
%   To make use of multiple cores/machines, function STARTMULTICOREMASTER
%   saves files with the function handle and the parameters to a temporary
%   directory (default: <TEMPDIR2>/multicorefiles, where <TEMPDIR2> is the
%   directory returned by function TEMPDIR2). These files are loaded by
%   function STARTMULTICORESLAVE running in other Matlab processes which
%   have access to the temporary directory. The slave processes evaluate
%   the given function with the saved parameters and save the result in
%   another file. The results are later collected by the master process.
%
%   Note that you can make use of multiple cores on a single machine or on
%   different machines with a commonly accessible directory/network share
%   or a combination of both.
%
%   RESULTCELL = STARTMULTICOREMASTER(FHANDLE, PARAMETERCELL, SETTINGS)
%   The additional input structure SETTINGS may contain any of the
%   following fields:
%
%   settings.multicoreDir:
%     Directory for temporary files (standard directory is used if empty)
%   settings.nrOfEvalsAtOnce:
%     Number of function evaluations gathered to a single job.
%   settings.maxEvalTimeSingle:
%     Timeout for a single function evaluation. Choose this parameter
%     appropriately to get optimum performance.
%   settings.masterIsWorker:
%     If true, master process acts as worker and coordinator, if false the
%     master acts only as coordinator.
%   settings.useWaitbar:
%     If true, a waitbar is opened to inform about the overall progress.
%
%   Please refer to the heavily commented demo function MULTICOREDEMO for
%   details and explanations of the settings.
%
%   RESULTCELL = STARTMULTICOREMASTER(FHANDLECELL, PARAMETERCELL, ...),
%   with a cell array FHANDLECELL including function handles, allows to
%   evaluate different functions.
%
%   Example: If you have your parameters saved in parameter cell
%   PARAMETERCELL, the for-loop
%
%   	for k=1:numel(PARAMETERCELL)
%   		RESULTCELL{k} = FHANDLE(PARAMETERCELL{k});
%   	end
%
%   which you would run in a single process can be run in parallel on
%   different cores/machines using STARTMULTICOREMASTER and
%   STARTMULTICORESLAVE. Run
%
%   	RESULTCELL = STARTMULTICOREMASTER(FHANDLE, PARAMETERCELL, DIRNAME)
%
%   in one Matlab process and
%
%   	STARTMULTICORESLAVE(DIRNAME)
%
%   in one or more other Matlab processes.
%
%		Markus Buehren
%		Last modified 17.03.2009
%
%   See also STARTMULTICORESLAVE, FUNCTION_HANDLE.

debugMode = 0;
showFileAccessWarnings = 0;

% parameters
startPauseTime = 0.1;
maxPauseTime   = 2;

% default settings
settingsDefault.multicoreDir      = '';
settingsDefault.nrOfEvalsAtOnce   = 1;
settingsDefault.maxEvalTimeSingle = 60;
settingsDefault.masterIsWorker    = 1;
settingsDefault.useWaitbar        = 0;

if debugMode
  disp(sprintf('*********** Start of function %s **********', mfilename));
end

%%%%%%%%%%%%%%%%
% check inputs %
%%%%%%%%%%%%%%%%
error(nargchk(2, 3, nargin, 'struct'))

% check function handle cell
if isa(functionHandleCell, 'function_handle')
  % expand to cell array
  functionHandleCell = repmat({functionHandleCell}, size(parameterCell));
else
  if ~iscell(functionHandleCell)
    error('First input argument must be a function handle or a cell array of function handles.');
  elseif any(size(functionHandleCell) ~= size(parameterCell))
    error('Input cell arrays functionHandleCell and parameterCell must be of the same size.');
  end
end

% check parameter cell
if ~iscell(parameterCell)
  error('Second input argument must be a cell array.');
end

% get settings
if ~exist('settings', 'var')
  % use default settings
  settings = settingsDefault;
elseif ~isstruct(settings)
  error('Third input argument must be struct.');
else
  % set default values where fields are missing
  fieldNames = fieldnames(settingsDefault);
  for k=1:length(fieldNames)
    if ~isfield(settings, fieldNames{k})
      settings.(fieldNames{k}) = settingsDefault.(fieldNames{k});
    end
  end
end

% check number of evaluations at once
nrOfEvals = numel(parameterCell);
nrOfEvalsAtOnce = settings.nrOfEvalsAtOnce;
if nrOfEvalsAtOnce > nrOfEvals
  nrOfEvalsAtOnce = nrOfEvals;
elseif nrOfEvalsAtOnce < 1
  error('Parameter nrOfEvalsAtOnce must be greater or equal one.');
end
nrOfEvalsAtOnce = round(nrOfEvalsAtOnce);

% check slave file directory
if isempty(settings.multicoreDir)
  % create default slave file directory if not existing
  multicoreDir = fullfile(tempdir2, 'multicorefiles');
  if ~exist(multicoreDir, 'dir')
    try
      mkdir(multicoreDir);
    catch
      error('Unable to create slave file directory %s.', multicoreDir);
    end
  end
else
  multicoreDir = settings.multicoreDir;
  if ~exist(multicoreDir, 'dir')
    error('Slave file directory %s not existing.', multicoreDir);
  end
end

% check maxEvalTimeSingle
maxEvalTimeSingle = settings.maxEvalTimeSingle;
if maxEvalTimeSingle < 0
  error('Parameter maxEvalTimeSingle must be greater or equal zero.');
end

% initialize waitbar
if settings.useWaitbar
  waitbar__('init');
end

% compute the maximum waiting time for a complete job
maxMasterWaitTime = maxEvalTimeSingle * nrOfEvalsAtOnce;

% remove all existing temporary multicore files
existingMulticoreFiles = [...
  findfiles(multicoreDir, 'parameters_*.mat', 'nonrecursive'), ...
  findfiles(multicoreDir, 'working_*.mat',    'nonrecursive'), ...
  findfiles(multicoreDir, 'result_*.mat',     'nonrecursive')];
deletewithsemaphores(existingMulticoreFiles);

% build parameter file name (including the date is important because slave
% processes might still be working with old parameters)
dateStr = sprintf('%04d%02d%02d%02d%02d%02d', round(clock));
parameterFileNameTemplate = fullfile(multicoreDir, sprintf('parameters_%s_XX.mat', dateStr));

%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% generate parameter files %
%%%%%%%%%%%%%%%%%%%%%%%%%%%%

% compute number of files/jobs
nrOfFiles = ceil(nrOfEvals / nrOfEvalsAtOnce);
if debugMode
  disp(sprintf('nrOfFiles = %d', nrOfFiles));
end

% save parameter files with all parameter sets
for fileNr = nrOfFiles:-1:1
  curFileNr = fileNr; % for simpler copy&paste
  parameterFileName = strrep(parameterFileNameTemplate, 'XX', sprintf('%04d', curFileNr));
  parIndex = ((curFileNr-1)*nrOfEvalsAtOnce+1) : min(curFileNr*nrOfEvalsAtOnce, nrOfEvals);
  functionHandles = functionHandleCell(parIndex); %#ok
  parameters      = parameterCell     (parIndex); %#ok
  sem = setfilesemaphore(parameterFileName);
  save(parameterFileName, 'functionHandles', 'parameters');
  removefilesemaphore(sem);

  if debugMode
    disp(sprintf('Parameter file nr %d generated.', curFileNr));
  end
end

resultCell = cell(size(parameterCell));

fileNr  = 1;         % start working down the list from top to bottom
fileNr2 = nrOfFiles; % check for results from bottom to top
parameterFileFoundTime  = NaN;
parameterFileRegCounter = 0;
nrOfFilesMaster = 0;
nrOfFilesSlaves = 0;
waitbar__('init2');

masterIsWorker = settings.masterIsWorker;
while 1 % this while-loop will be left if all work is done
  if masterIsWorker
    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
    % work down the file list from top %
    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
    if debugMode
      disp(sprintf('********** 1. Working from top to bottom (file nr %d)', fileNr));
    end
    curFileNr = fileNr; % for simpler copy&paste
    parameterFileName = strrep(parameterFileNameTemplate, 'XX', sprintf('%04d', curFileNr));
    resultFileName    = strrep(parameterFileName, 'parameters', 'result' );
    workingFileName   = strrep(parameterFileName, 'parameters', 'working');
    parIndex = ((curFileNr-1)*nrOfEvalsAtOnce+1) : min(curFileNr*nrOfEvalsAtOnce, nrOfEvals);

    sem = setfilesemaphore(parameterFileName);
    parameterFileExisting = existfile(parameterFileName);
    if parameterFileExisting
      % If the parameter file is existing, no other process has started
      % working on that job --> Remove parameter file, so that no slave
      % process can load it. The master will do the current job.
      mbdelete(parameterFileName, showFileAccessWarnings);
      if debugMode
        disp(sprintf('Parameter file nr %d deleted by master.', curFileNr));
      end
    end

    % check if the current parameter set was evaluated before by a slave process
    resultLoaded = false;
    if parameterFileExisting
      % If the master has taken the parameter file, there is no need to check
      % for a result. Semaphore will be removed below.
      if debugMode
        disp(sprintf('Not checking for result because parameter file nr %d was existing.', curFileNr));
      end

    else
      % Another process has taken the parameter file. This branch is
      % entered if master and slave "meet in the middle", i.e. if a slave
      % has taken the parameter file of the job the master would have done
      % next. In this case, the master will wait until the job was finished
      % by the slave process or until the job has timed out.
      curPauseTime = startPauseTime;
      firstRun = true;
      while 1 % this while-loop will be left if result was loaded or job timed out
        if firstRun
          % use the semaphore generated above
          firstRun = false;
        else
          % set semaphore
          sem = setfilesemaphore(parameterFileName);
        end

        % Check if the result is available. The semaphore file of the
        % parameter file is used for the following file accesses of the
        % result file.
        if existfile(resultFileName)
          [result, resultLoaded] = loadresultfile__(resultFileName, showFileAccessWarnings);
          if resultLoaded && debugMode
            disp(sprintf('Result file nr %d loaded.', curFileNr));
          end
        else
          resultLoaded = false;
          if debugMode
            disp(sprintf('Result file nr %d was not found.', curFileNr));
          end
        end

        if resultLoaded
          % Save result
          resultCell(parIndex) = result;
          nrOfFilesSlaves = nrOfFilesSlaves + 1;
          waitbar__('update', nrOfFiles, nrOfFilesMaster, nrOfFilesSlaves);