% PoolProfiler Parallel pool profiler

% Copyright 2017-2021 The MathWorks, Inc.

classdef PoolProfiler < handle
    properties (GetAccess = private, SetAccess = immutable)
        ClientSessionLogger
        WorkerSessionLoggerConstant
        Pool
        Session
    end
    
    properties (SetAccess = private)
        StartTime
    end
    
    methods (Access = private)
        function x = drainLogImpl(obj, fcnH)
            xc = fcnH(obj.ClientSessionLogger);
            xc.LabIndex = -ones(height(xc), 1);
            xw = fetchOutputs(parfevalOnAll(...
                obj.Pool, ...
                @(x) fcnH(x.Value), 1, obj.WorkerSessionLoggerConstant), ...
                              'UniformOutput', false);
            for idx = 1:numel(xw)
                xw{idx}.LabIndex = idx * ones(height(xw{idx}), 1);
            end
            x = sortrows(vertcat(xc, xw{:}));
            x = x(:, [1, end, 2:end-1]);
        end
        function x = drainMessageLog(obj)
            x = obj.drainLogImpl(@drainMessageLog);
        end
        function x = drainMatlabLog(obj)
            x = obj.drainLogImpl(@drainMatlabLog);
        end
        function x = drainDispatchLog(obj)
            x = obj.drainLogImpl(@drainDispatchLog);
        end
        function x = drainLanguageConstructLog(obj)
            x = obj.drainLogImpl(@drainLanguageConstructLog);
        end
        function waitUntilPoolIdle(obj, timeoutMillis, onTimeoutFcn)
        % Block until workers idle as far as the resource manager is
        % concerned, calling onTimeoutFcn if the timeout expired. We expect
        % this to happen 'quickly'. During construction, a
        % parallel.pool.Constant has just been created and initialized on
        % the workers, so the pool should become idle almost immediately.
        % During 'drainLog', the caller *should* have left the pool idle.
        % (drainLog needs to use all workers, so "parfeval(@pause, 0, Inf)"
        % or similar will cause problems).
            didBecomeIdle = obj.Session.waitUntilPoolIdle(timeoutMillis);
            if ~didBecomeIdle
                onTimeoutFcn();
            end
        end
    end
    
    methods
        function obj = PoolProfiler()
            pool = parallel.internal.pool.PoolArrayManager.getCurrentWithCleanup;
            if isempty(pool)
                error('parallel:internal:profiling:PoolProfiler:NoPool',...
                      'PoolProfiler cannot be created as no parallel pool is open');
            end
            obj.Pool = pool;
            obj.Session = pool.hGetSession();
            assert(isa(obj.Session, 'parallel.internal.pool.JavaBackedSession'), ...
                   'PoolProfiler works only with JavaBackedSession.');
            obj.WorkerSessionLoggerConstant = parallel.pool.Constant(...
                @parallel.internal.profiling.SessionLogger);
            % Access the Constant value on the workers to force the
            % initialization of the SessionLoger
            wait(parfevalOnAll(@(x) x.Value, 0, obj.WorkerSessionLoggerConstant));
            longTimeoutMillis = 10000;
            obj.waitUntilPoolIdle(longTimeoutMillis, ...
                                  @() error('parallel:internal:profiling:PoolProfile:WorkersNotIdle', ...
                                            'The workers did not become idle within %d milliseconds.', ...
                                            longTimeoutMillis));

            obj.StartTime = iTimestamp();

            % Set up the client session logger after everything else is set up. This
            % should ensure that even if the preceding parfevalOnAll calls
            % have a finish time that matches 'obj.StartTime', they are not
            % recorded at the client, and thus the LanguageConstructLog will
            % not contain those entries.
            obj.ClientSessionLogger = parallel.internal.profiling.SessionLogger(obj.Session);
        end
        function [out, outFull] = drainLog(obj)
            import parallel.internal.profiling.PoolProfileResults
            
            assert(isvalid(obj.WorkerSessionLoggerConstant), ...
                   'PoolProfiler has already been drained.');

            % Ensure pool is idle before we proceed. In this case, we simply warn and
            % attempt to proceed anyway.
            longTimeoutMillis = 10000;
            obj.waitUntilPoolIdle(longTimeoutMillis, ...
                                  @() warning('parallel:internal:profiling:PoolProfile:WorkersNotIdle', ...
                                              'The workers did not become idle within %d milliseconds.', ...
                                              longTimeoutMillis));

            % Before going any further, stop the client logger to ensure that anything we
            % do henceforth will not be logged.
            stop(obj.ClientSessionLogger);
            
            stopTime = iTimestamp();
            % Use a 'closed' interval to ensure we don't drop any trailing events that happen to have the same
            % timestamp as stopTime.
            range = timerange(obj.StartTime, stopTime, 'closed');
            dctSchedulerMessage(2, 'PoolProfiler.drainLog() logged timerange: %s --> %s', ...
                                string(obj.StartTime), string(stopTime));
            
            % Get the unfiltered logs
            unfilteredLogs = cellfun(@(f) f(obj), {@drainMessageLog, @drainMatlabLog, ...
                                @drainDispatchLog, @drainLanguageConstructLog}, ...
                                     'UniformOutput', false);
            % Filter the logs
            filteredLogs = cellfun(@(oneLog) oneLog(range, :), unfilteredLogs, ...
                                   'UniformOutput', false);

            % Delete the loggers, and force the pool to become idle to avoid polluting
            % future PoolProfilers. Note the apparently null parfevalOnAll is
            % required to ensure the Constant destructor has completed.
            delete(obj.WorkerSessionLoggerConstant);
            delete(obj.ClientSessionLogger);
            wait(parfevalOnAll(obj.Pool, @() [], 0));
            
            % Build filtered and unfiltered results. Unfiltered results are useful
            % only for diagnostic purposes.
            out = PoolProfileResults(filteredLogs{:});
            outFull = PoolProfileResults(unfilteredLogs{:});
        end
     end
end

function ts = iTimestamp()
% Create a datetime with the same precision as a Java timestamp so that 
% it can be more directly compared with other Java timestamps.
ts = datetime(java.lang.System.currentTimeMillis() / 1000, ...
              'TimeZone', 'local', ...
              'ConvertFrom', 'posixtime', ...
              'Format', 'dd-MMM-uuuu HH:mm:ss.SSS');
end
