function varargout = findJob( obj, varargin )
%findJob  Find job objects belonging to a cluster object
%   jobs = findJob(cluster) finds all jobs in a cluster, and is identical to
%   accessing the Jobs property on the cluster.
%
%   jobs = findJob(cluster, 'p1', v1, 'p2', v2, ...) gets a 1 x N array of job
%   objects belonging to the cluster. The returned job objects will be only
%   those having the specified property-value pairs.
%
%   When a property value is specified, it must use the same format that the get
%   function returns. For example, if get returns the Name property value as
%   'MyJob', then findJob will not find that object while searching for a Name
%   property value of 'myjob' as the match is case sensitive.
%
%   [pending, queued, running, finished] = findJob(cluster) returns the jobs
%   from the cluster in four output variables, grouped by their current
%   state. (Note that jobs with state 'failed' are also included in the 4th
%   output argument).
%
%   If the cluster is contained in a remote service, findJob will result in a
%   call to the remote service. This could result in findJob taking a long time
%   to complete, depending on the number of jobs retrieved and the network
%   speed. Also, if the remote service is no longer available, an error will be
%   thrown.
%
%   Examples:
%   % Find all jobs on the default cluster that were submitted by user 'myUserName'
%   myCluster = parcluster; 
%   myJobs = findJob(myCluster, 'Username', 'myUserName');
%
%   % Find all jobs grouped by state
%   [pendingJobs, queuedJobs, runningJobs, finishedJobs] = findJob(myCluster);
%
%   % Find all running jobs that were submitted by user 'myUserName'
%   myRunningJobs = findJob(myCluster, 'Username', 'myUserName', 'State', 'running');
%
%   See also parallel.Cluster/createJob, parallel.Cluster/Jobs, parcluster, 
%            parallel.Job/findTask, parallel.Job/submit.

% Copyright 2011-2018 The MathWorks, Inc.

import parallel.internal.customattr.WorkUnitFinder;
if ~any( nargout == [ 0 1 4 ] )
    error(message('parallel:cluster:FindJobNumOutputArgs'));
end

% Convert any string inputs to character vectors
[varargin{:}] = parallel.internal.general.convertPVPairsToChars(varargin{:});

validateattributes( obj, {'parallel.Cluster'}, {'nonempty'}, 'findJob', 'cluster', 1 );

try
    % Make sure the clusters are in a row vector, so we can assume
    % this for concatenation.
    obj = obj(:)';
    if nargout == 4
        [jp, jq, jr, jf] = parallel.internal.cluster.hetfun( @getJobsByState, obj );
        % each output is a cell of cells, need to unpick first to
        % straight cell array
        jp = [jp{:}]; jq = [jq{:}]; jr = [jr{:}]; jf = [jf{:}];
        % And then reconstitute into a 4-cell containing columns
        jobsCell = { [jp{:}].', [jq{:}].', [jr{:}].', [jf{:}].' };
    else
        jl = parallel.internal.cluster.hetfun( @hGetJobs, obj );

        % Concatenate the heterogeneous parts into one array.
        jl = [jl{:}];
        
        % Concatenate all the jobs into one array and reshape
        % as a column.
        jobsCell = { [jl{:}]' };
    end
catch E
    throw( E );
end

try
    % We potentially have a heterogeneous job array, so we need to call
    % find() using hetfun.
    findFcn = @(x) iWorkUnitMatch( x, varargin );
    varargout = cellfun( @(x) findUsingHetfun( findFcn, x ), ...
                         jobsCell, 'UniformOutput', false );
catch E
    if strcmpi(E.identifier, 'parallel:cluster:PropertyValueInPairs')
        rethrow(E);
    end
    
    % See if we have any better luck getting each prop individually.
    matchingJobs = cellfun( @(x) iFindSingleJobFromArrayNoError(x, varargin{:}), ...
                            jobsCell, 'UniformOutput', false );
    varargout = matchingJobs;
end

end

function foundJobs = findUsingHetfun( findFcn, jobsCell )
    if numel( jobsCell ) > 0
        % In order to preserve job order we use the 'stable' flag. To do
        % this, findFcn must return an output for each job in the array. 
        jobMatches = parallel.internal.cluster.hetfun( findFcn, jobsCell, 'stable' );
        jobMatches = vertcat(jobMatches{:});
        foundJobs = jobsCell(jobMatches);
    else
        % In the case the jobsCell is empty, we need to make sure we return
        % the right sort of empty (e.g. 0x1 CJSIndependentJob). If we were
        % to call hetfun above, we would get the wrong sort of empty.
        foundJobs = jobsCell;
    end
end

function matchingJobs = iFindSingleJobFromArrayNoError(jobArray, varargin)
isMatchingJob = arrayfun( @(x) iIsMatchingJobNoError(x, varargin{:}), ...
                         jobArray );
% Remove empties
matchingJobs = jobArray(isMatchingJob);
end

function tf = iIsMatchingJobNoError(job, varargin)
import parallel.internal.customattr.WorkUnitFinder;

tf = false;
try
    tf = ~isempty(WorkUnitFinder.find(job, varargin));
catch err %#ok<NASGU>
    % Swallow the error and return false
end

end

function match = iWorkUnitMatch( jobArray, varargin )
import parallel.internal.customattr.WorkUnitFinder;
[~, match] = WorkUnitFinder.find( jobArray, varargin{:} );
end