function sOut = mpiprofile(varargin)
%MPIPROFILE Profile parallel communication and execution times
%   MPIPROFILE enables or disables the parallel profiler data collection on a
%   MATLAB worker running in a parallel pool. You can use MPIPROFILE either
%   from the MATLAB client or directly from the worker from within an SPMD
%   block. When you run MPIPROFILE from the MATLAB client, MPIPROFILE
%   carries out the action on the MATLAB workers.
%
%   MPIPROFILE aggregates statistics on execution time and communication
%   times. MPIPROFILE collects statistics in a manner similar to running the
%   PROFILE command on each MATLAB worker. By default, the parallel profiling
%   extensions include array fields that collect information on communication
%   with each of the other workers.
%
%   MPIPROFILE ON <options> starts the parallel profiler and clears previously
%   recorded profile statistics.
%
%   MPIPROFILE takes the following options:
%
%    -MESSAGEDETAIL LEVEL
%         LEVEL can be 'default', 'simplified' or 'off'. This option specifies
%         the detail at which communication information is stored. By default,
%         information is collected per worker instance.  If you have a very
%         large cluster, you may want to change the LEVEL to
%         'simplified'. This will reduce the profiling overhead.  However, you
%         will not get all the detailed inter-worker communication plots in the
%         viewer. Setting LEVEL to 'simplified' will turn off collection for
%         PerLab data fields. Note that changing the messagedetail will clear
%         any previously stored data.
%
%   MPIPROFILE also accepts the following options which have the same meaning
%   as for the PROFILE command:
%
%    -HISTORY
%    -NOHISTORY
%    -TIMESTAMP
%    -HISTORYSIZE SIZE
%    -TIMER
%
%   No other PROFILE options are supported by MPIPROFILE.
%
%   MPIPROFILE OFF stops the parallel profiler. To reset the state of the
%   profiler and disable collecting communication information, you should also
%   call MPIPROFILE RESET.
%
%   MPIPROFILE RESET turns off the parallel profiler and resets the data
%   collection back to the standard profiler. If you do not call reset,
%   subsequent PROFILE commands will collect MPI information.
%
%   MPIPROFILE VIEWER stops the parallel profiler and opens the graphical profile
%   browser with parallel options. The output is an HTML report displayed in
%   the Profiler window. The file listing at the bottom of the function
%   profile page shows several columns to the left of each line of code. In the
%   summary page:
%     * Column 1 indicates the number of calls to that line.
%     * Column 2 indicates total time spent on the line in seconds.
%     * Columns 3-6 contain the communication information specific to the
%       parallel profiler
%
%   MPIPROFILE RESUME restarts the profiler without clearing previously
%   recorded function statistics.
%
%   MPIPROFILE CLEAR clears the profile information.
%
%   S = MPIPROFILE('STATUS') returns a structure containing information about
%   the current parallel profiler state. S contains these fields:
%
%       ProfilerStatus   -- 'on' or 'off'
%       DetailLevel      -- 'mmex'
%       Timer            -- 'cpu', 'real', 'performance', or 'processor'
%       HistoryTracking  -- 'on', 'off', or 'timestamp'
%       HistorySize      -- 1000000 (default)
%       MpiMessageDetail -- 'default', 'simplified', or 'off'
%
%   STATS = MPIPROFILE('INFO') stops the parallel profiler and returns a
%   structure containing the profiler statistics. STATS contains the same
%   fields as returned by PROFILE('INFO'), with the following additional
%   fields in the FunctionTable:
%
%     BytesSent           -- Records the quantity of data sent
%     BytesReceived       -- Records the quantity of data received
%     TimeWasted          -- Records communication waiting time
%     CommTime            -- Records the communication time
%     CommTimePerLab      -- Vector of communication receive time for each worker
%     TimeWastedPerLab    -- Vector of communication waiting time for each worker
%     BytesReceivedPerLab -- Vector of data received from each worker
%
%   The three PerLab fields are collected only on a per-function basis, and
%   can be turned off by using the following command:
%   MPIPROFILE on -messagedetail simplified
%
%   When you run it from the MATLAB client, STATS = MPIPROFILE('INFO')
%   returns information from all workers. When you run it on a worker,
%   MPIPROFILE('INFO') returns the profile information specific to that worker.
%
%   MPIPROFILE('VIEWER', STATS) opens the graphical profile browser
%   showing the profiling information contained in STATS. You can use
%   STATS = MPIPROFILE('INFO') on the client to create the structure
%   array.
%
% Example:
%   % Turn on the profiler. With default preferences, turning on the profiler
%   % will create a parallel pool automatically if there is not one already
%   % created.
%   mpiprofile on
%   % Run some distributed array code
%   A = rand(1000, 'distributed');
%   b = sum(A, 2);
%   x = A\b;
%   % Show the collected profile information.
%   mpiprofile viewer
%
% See also profile, mpiprofview, mpiSettings, spmd, distributed.

%   Copyright 2007-2023 The MathWorks, Inc.

import matlab.internal.capability.Capability
import parallel.internal.profiler.types.MessageDetail

try
    % Convert any string inputs to character vectors
    [varargin{:}] = convertStringsToChars( varargin{:} );

    [action, argsForProfile, comDetailLevel, option] = iParseInputs( varargin{:});
    dctSchedulerMessage(4, 'mpiprofile: action=%s argsForProfile={%s} comDetailLevel=%s option=%s', ....
        formattedDisplayText(action), formattedDisplayText(argsForProfile), ...
        formattedDisplayText(comDetailLevel), formattedDisplayText(option));
catch err
    % Hide argument parsing stack details
    throw(err);
end

% ensure output is initialized to empty if user request output argument
if nargout > 1
    sOut = [];
end

action = lower(action);
% send the arguments that are set by profile to profile command
if ~isempty(argsForProfile)
    iRunOnWorkers(@profile, argsForProfile{:});
end

% If the `-messagedetail` option has been passed with a valid value, 
% change the message detail level on the workers.
if ~isempty(comDetailLevel)
    iRunOnWorkers(@iChangeMessageDetailOnWorker, comDetailLevel);
end

switch action
  case 'on'
    notifyUI('start');
    iRunOnWorkers(@iOnAction);
  case 'off'
    iRunOnWorkers(@iOffAction);
    notifyUI('stop');
  case 'resume'
    notifyUI('resume');
    iRunOnWorkers(@iResumeAction);
  case 'clear'
    notifyUI('clear');
    % clear profiler info
    iRunOnWorkers(@profile, 'clear');
  case 'reset'
    iRunOnWorkers(@iResetAction);
  case 'status'
    if iIsOnClient() && isempty(gcp('nocreate'))
        % On client and no pool - return fabricated status (nothing should really need this).
        sOut = struct('ProfilerStatus', 'off', ...
                      'Timer', iDefaultTimer());
    elseif iIsOnClient()
        % Return status from worker 1.
        statusComposite = iRunOnWorkers(@iStatus);
        sOut = statusComposite{1};
    else
        % On a worker - return the real thing.
        sOut = iStatus();
    end
  case 'undoc:timer'
    % Undocumented option: get the timer from the workers. This is used only by
    % mpiprofview. It has the same flaw that PROFILE has - it might be
    % completely wrong. Unfortunately, the return of "profile('info')" doesn't
    % specify the timer that was used, so we are obliged to check the current
    % value, which might be wrong. We also wish to avoid creating a parallel
    % pool simply to return this value.
    if iIsOnClient() && isempty(gcp('nocreate'))
        % return whatever the default would be
        sOut = iDefaultTimer();
    else
        statusComposite = iRunOnWorkers(@iStatus);
        status1 = statusComposite{1};
        sOut = status1.Timer;
    end
  case 'undoc:invoked'
    % Undocumented option: get the `callstats('has_run')` value from the workers.
    % This is used only by the profiler viewer.
    if iIsOnClient() && isempty(gcp('nocreate'))
        sOut = false;
    else
        statusComposite = iRunOnWorkers(@iStatus);
        status1 = statusComposite{1};
        sOut = status1.ProfilerInvokedStatus;
    end
  case 'info'
    notifyUI('stop');
    if iIsOnClient()
        % Concatenate all the info structs and gather.
        sOut = iRunOnWorkers(@() spmdCat(profile('info'), 1, 1));
        sOut = sOut{1};
    else
        sOut = profile('info');
    end
  case 'viewer'
    % mpiprofile viewer and mpiprofview are not supported in MATLAB clients which are not local (see g2175913)
    Capability.require([Capability.LocalClient]);

    % gather the data and display in Profiler window
    if iIsOnClient()
        if ~isempty(option) && isstruct(option)
            profileData = option;
            mpiprofview(0, profileData);
        else
            % The profile viewer specific to MPI will collect internally
            % the stats from the workers.
            iRunOnWorkers(@iOffAction);
            notifyUI('stop');
            parallel.internal.profileviewer.invokeMpiProfilerViewer();
        end
    else
        if ~isempty(option)
            % Workers cannot specify the profile info to "mpiprofile('viewer')".
            warning(message('parallel:lang:mpi:ProfInvalidViewerArgOnWorker'));
        end
        if spmdIndex == 1
            try
                % Simply defer "mpiprofile('viewer')" to the client.
                runCmdOnClient(sprintf('mpiprofile(''viewer'');'));
            catch err
                throw(err);
            end
        end
    end

  case ''
    % no caught action input in command line so we
    % do nothing and pass the arguments to profile at end of switch
    % function

  otherwise
    % error unknown action
    error(message('parallel:lang:mpi:ProfInvalidArguments'));
end
end

%--------------------------------------------------------------------------
%--------------------------------------------------------------------------
function sOut = iStatus()
sOut = profile('status');
sOut.MpiMessageDetail = char(iGetParallelMsgDetail());
sOut.ProfilerInvokedStatus = callstats('has_run');
end

%--------------------------------------------------------------------------
%--------------------------------------------------------------------------
function iResetAction()
% resets everything back to normal profiler
dctSchedulerMessage(4, 'mpiprofile: iResetAction.');
profile ('off');
% Reset timer, history tracking and history size to profile's default
profile('-timer', 'performance');
profile('-timestamp');
profile('-historysize', '5000000');
iClearParallelMessageLogging();
iTurnMpiLogOff();
end

%--------------------------------------------------------------------------
%--------------------------------------------------------------------------
function iResumeAction()
iStartProfiling('resume');
end

%--------------------------------------------------------------------------
%--------------------------------------------------------------------------
function iOffAction()
dctSchedulerMessage(4, 'mpiprofile: iOffAction');
iTurnMpiLogOff();
profile('off');
end

%--------------------------------------------------------------------------
%--------------------------------------------------------------------------
function iOnAction()
iStartProfiling('on');
end

%--------------------------------------------------------------------------
%--------------------------------------------------------------------------
function iStartProfiling(action)
% Only set timer and message details option if profiler is
% already off.
if ~iIsProfilerOn()
    iTurnMpiLogOn();
end

% Ensures profiler commands are not profiled and stops profiling.
iAddProfilerFileFiltersAndStop();

% The action can take 'on' and 'resume' values.
dctSchedulerMessage(4, 'mpiprofile: profile(''%s'')', action);
profile(action);
end

%--------------------------------------------------------------------------
%--------------------------------------------------------------------------
function varargout = iRunOnWorkers(fcn, varargin)
if iIsOnClient()
    pool = parallel.internal.pool.getOrCreatePool( ...
        parallel.internal.pool.PoolPurpose.Spmd );
    if isempty(pool)
        % We're on the client, but user's preference disallows automatic creation of a
        % parallel pool. Disallow this case.
        throwAsCaller(MException(message('parallel:lang:mpi:ProfRequiresRunningPool')));
    end
    if pool.NumWorkers == 0
        numOut = nargout;
        % 1. Use parfevalOnAll to run the function on each worker
        % 2. Wrap the function with iWrapCall which sets up MPI just like 'spmd',
        %    and also returns the spmdIndex corresponding to the other outputs
        %    to ensure we get things in the correct order.
        % 3. "transpose" the outputs to make them be cell arrays that mimic the
        %    the Composites that would normally result from 'spmd'.
        fut = parfevalOnAll(@iWrapCall, 1 + numOut, fcn, varargin);
        wait(fut);
        varargout = cell(1, nargout);
        futOut = fut.OutputArguments; % 1 x spmdSize cell of 1 x numOut cells.
        numWorkers = numel(futOut{1});
        for fcnOutIdx = 1:numOut
            varargout{fcnOutIdx} = cell(1, numWorkers);
            for futOutIdx = 1:numWorkers
                workerIdx = futOut{futOutIdx}{1};
                varargout{fcnOutIdx}{workerIdx} = futOut{futOutIdx}{fcnOutIdx+1};
            end
        end
    else
        try
            [varargout{1:nargout}] = spmd_feval_fcn(fcn, varargin);
        catch E
            % If SPMD failed, report errors as mpiprofile.
            throwAsCaller(E);
        end
    end
else
    [varargout{1:nargout}] = fcn(varargin{:});
end
end

function varargout = iWrapCall(fcn, fcnArgsCell)
pctPreRemoteEvaluation('mwmpi');
varargout{1} = spmdIndex;
[varargout{2:nargout}] = fcn(fcnArgsCell{:});
end

%--------------------------------------------------------------------------
%--------------------------------------------------------------------------
function str = iDefaultTimer()
str = 'performance';
end
%--------------------------------------------------------------------------
%--------------------------------------------------------------------------
function iTurnMpiLogOn()
% Get the current message detail (or initialise to the default one if this
% is the first time `userMessageDetail` is called).
comDetailLevel = userMessageDetail();
detailNum = comDetailLevel.toMessageDetailNumber();
dctSchedulerMessage(4, 'mpiprofile: callstats(''message'',%d).', detailNum);
callstats('message', detailNum);

% It's possible to trigger the MPI profiler on the client but not from within
% an SPMD block, but there will never be any MPI information to record, so
% don't bother enabling the collection. So, if MPI itself has not been
% initialized, don't attempt to enable the MPI-specific profiling collection.
if mpiInitialized
    % set callback from gateway to profiler
    mpigateway( 'logprof' );
    % turn on logging of mpi primitives
    mpigateway( 'logon' );
end
end

%--------------------------------------------------------------------------
%--------------------------------------------------------------------------
function iTurnMpiLogOff
if mpiInitialized
    mpigateway( 'logoff' );
end
end

%--------------------------------------------------------------------------
% parses the inputs and gets the action and arguments to pass to profile
%--------------------------------------------------------------------------
function [action, argsForProfile, comDetailLevel, vieweroptions] = iParseInputs(varargin)
% default outputs
argsForProfile = {};
comDetailLevel = [];
vieweroptions = '';
action = '';
narginchk(1, Inf);
isPairedArg = false;

% The last action will take precedence
% no error is thrown just like profile on off;
for k = 1:length(varargin)
    if isPairedArg
        % second argument has already been consumed
        isPairedArg = false;
        % hence skip this argument
        continue;
    end
    arg = varargin{k};
    if ( (~ischar(arg) && ~isPairedArg) || isempty(arg))
        error(message('parallel:lang:mpi:MpiprofileInvalidInput'));
    end
    switch lower(arg)

        case 'viewer'
            % This is the only action we deal with that is also dealt with
            % by parser since it can be paired.
            % Here we look for a potential followup profile info argument
            action = arg;
            if numel(varargin)> k
                vieweroptions =  varargin{k+1};
            end
            % if its a profiler structure
            if isstruct(vieweroptions)
                % we have a profile info vector input
                % so consume the following argument
                isPairedArg = true;
            end

        case '-messagedetail'
            comDetailLevel = iVerifyNumAndTypeOfArgs(varargin, k, k + 1, 'detaillevel');
            isPairedArg = true;

            % *******arguments to pass to standard profile command*******
        case {'-historysize', '-detail'}
            % lets profile ensure the type of the second argument
            iVerifyNumAndTypeOfArgs(varargin, k, k+1, '');
            % type empty means the type of the second(k+1) argument is not checked
            argsForProfile{end+1} = arg; %#ok
            argsForProfile{end+1} =  varargin{k+1}; %#ok
            isPairedArg = true;

        case {'-history', '-nohistory', '-timestamp'}
            argsForProfile{end+1} = arg; %#ok
            % *******END arguments to pass to standard profile command*******

        case '-timer'
            % lets profile ensure the type of the second argument
            iVerifyNumAndTypeOfArgs(varargin, k, k+1, '');
            % type empty means the type of the second(k+1) argument is not checked
            argsForProfile{end+1} = arg; %#ok
            argsForProfile{end+1} =  varargin{k+1}; %#ok
            isPairedArg = true;

        otherwise
            isOption = arg(1)=='-';
            if isOption
                % don't store the argument as action and error out
                error(message('parallel:lang:mpi:MpiprofileUnsupportedOrInvalidOption'));
            else
                if ~isempty(action)
                    warning(message('parallel:lang:mpi:ProfUnsupportedNumberOfActions', upper( action ), upper( arg )));
                end
                action = arg;
            end
    end
end
end

%--------------------------------------------------------------------------
%--------------------------------------------------------------------------
function iChangeMessageDetailOnWorker(comDetailLvl)
    if ~iIsProfilerOn()
        detailNum = comDetailLvl.toMessageDetailNumber();
        dctSchedulerMessage(4, 'mpiprofile: callstats(''message'',%d).', detailNum);
        callstats('message', detailNum);
        userMessageDetail(comDetailLvl);
    else
        % If the profiler is running the message detail cannot be changed.
        error(message('parallel:lang:mpi:MpiprofileAlreadyStartedCannotChangeMessageDetail'));
    end
end

%--------------------------------------------------------------------------
%--------------------------------------------------------------------------
function detailLvl = iGetParallelMsgDetail()
import parallel.internal.profiler.types.MessageDetail
% get the message detail level from the profiler
detailLvl = MessageDetail(callstats('message'));
end

%--------------------------------------------------------------------------
%--------------------------------------------------------------------------
function iClearParallelMessageLogging()
% stops profiler from collecting communication information
import parallel.internal.profiler.types.MessageDetail

try
    callstats('message', MessageDetail.Off.toMessageDetailNumber());
    err = [];
catch exception
    err = exception;
end

% reset the user persistent messagedetail setting back to default which currently is 2.
userMessageDetail(MessageDetail.Reset);
if ~isempty(err)
    rethrow(err);
end
end

%--------------------------------------------------------------------------
%--------------------------------------------------------------------------
function isOn = iIsProfilerOn()
import parallel.internal.profiler.types.MessageDetail
% Error if we are not on the worker. This is a programmer error.
assert(~iIsOnClient);
pStatus = profile('status');
isOn = strcmpi(pStatus.ProfilerStatus, 'on');
% throw warning if its on but no communication fields are being recorded
% because `profile on` was used.

if isOn && iGetParallelMsgDetail() == MessageDetail.Off ...
        && userMessageDetail() ~= MessageDetail.Off
    warning(message('parallel:lang:mpi:ProfNotInParallelMode'));
end
end

%--------------------------------------------------------------------------
%--------------------------------------------------------------------------
function verifiedVal = iVerifyNumAndTypeOfArgs(argsToParse, argNum, valNum, valType)
% iVerifyNumAndTypeOfArgs(argsToParse, inK, Type)
% verifies the type and number of options to output a relevant message.
import parallel.internal.profiler.types.MessageDetail

ERROR = -1;
if numel(argsToParse) < valNum
    error(message('parallel:lang:mpi:ProfInvalidNumArgs', ...
                  upper(argsToParse{argNum}), valType));
else
    switch(valType)
            % case deal with message detail enumeration
        case 'detaillevel'
            msgDetail = argsToParse{valNum};
            if any(strcmpi(msgDetail, {'simplified', 'novectorfields'}))
                verifiedVal = MessageDetail.Simplified;
            elseif any(strcmpi(msgDetail, {'default', 'withvectorfields'}))
                verifiedVal = MessageDetail.Default;
            elseif strcmpi(msgDetail, 'off')
                verifiedVal = MessageDetail.Off;
            else
                % indicate error
                verifiedVal = ERROR;
                valType = 'default | simplified | off';
            end

        case 'string'
            if ischar( argsToParse{argNum})
                verifiedVal = 1;
            else
                verifiedVal = ERROR;
            end

        otherwise
            verifiedVal = true;
    end

    if verifiedVal == ERROR
        error(message('parallel:lang:mpi:ProfInvalidInputType', upper( argsToParse{ argNum } ), upper( valType )));
    end
end
end

%--------------------------------------------------------------------------
%--------------------------------------------------------------------------
function onclient = iIsOnClient()
% It needs to be revisited what means "being on the client" (g2610654).
if system_dependent('isdmlworker') || parallel.internal.pool.isPoolThreadWorker()
    % On a worker, we consider the worker with a pool available to be the
    % 'client'.
    onclient = ~isempty(gcp('nocreate'));
else
    % Not a worker process, so always true.
    onclient = true;
end
end

%--------------------------------------------------------------------------
% This function ensures the mpiprofile commands are not profiled, also various
% boring bits of infrastructure are filtered.
%--------------------------------------------------------------------------
function iAddProfilerFileFiltersAndStop()
% make sure the profiler filters out functions and files used M profiler
% interface. This implementation follows the pattern of
% profile>filterFilesInProfile.
% Warning: callstats('ffilter', ...) stops the collection of data. For consistency we
% stop the profiling infrastructure here **always**, in case the condition
% below is not met.
dctSchedulerMessage(4, 'mpiprofile: invoking callstats(''stop'').');
callstats('stop');
if ~any(contains(callstats('ffilter'), 'mpiprofile.m'))
    files = {'mpiprofile', 'mpiprofview', 'private/userMessageDetail'};
    cellfun(@(f) callstats('ffilter', 'add', which(f)), files);
end
end

function notifyUI(action)
    if iIsOnClient
        matlab.internal.profileviewer.notifyUI(action, matlab.internal.profileviewer.ProfilerType.MPI);
    end
end
