%NonCommRunPartitionFunction
% Run one partition of a partitioned array operation on a remote worker
% without communication. All non-side-effect output of running the
% partition will be returned via output arguments. This is intended to
% be invoked within a parfeval as part of a NonCommPoolExecutor job.
%

%   Copyright 2019 The MathWorks, Inc.

classdef (Sealed) NonCommRunPartitionFunction < handle
    properties (GetAccess = private, SetAccess = immutable)
        % The factory to use to generate the underlying DataProcessor
        % instance.
        DataProcessorFactory (1,1)
        
        % Partition Index for this task.
        PartitionIndex (1,1) double
        
        % The number of partitions over the entire job.
        NumPartitions (1,1) double
        
        % The underlying partitioning strategy.
        PartitionStrategy (1,1)
        
        % Input broadcasts to be passed to underlying data.
        BroadcastMap (1,1)
        
        % Broadcast keys to be returned to the client.
        OutputKeys (:, 1) string
    end
    
    methods
        function obj = NonCommRunPartitionFunction(dataProcessorFactory, ...
                partitionIndex, numPartitions, partitionStrategy, broadcastMap, outputKeys)
            % Build a NonCommRunPartitionFunction with all necessary
            % information to evaluate one partition.
            obj.DataProcessorFactory = dataProcessorFactory;
            obj.PartitionIndex = partitionIndex;
            obj.NumPartitions = numPartitions;
            obj.PartitionStrategy = partitionStrategy;
            obj.BroadcastMap = broadcastMap;
            obj.OutputKeys = outputKeys;
        end
    end
    
    methods
        function [readFailureSummary, outputBroadcastMap] = feval(obj)
            % Evaluate the partition. This is expected to be invoked on a
            % worker.
            partitionIndex = obj.PartitionIndex;
            partition = obj.PartitionStrategy.createPartition(partitionIndex, obj.NumPartitions);
            partitionContext = matlab.bigdata.internal.executor.PartitionContext(partition, obj.BroadcastMap);
            dataProcessor = feval(obj.DataProcessorFactory, partitionContext);
            
            while ~dataProcessor.IsFinished
                process(dataProcessor, false(0));
            end
            
            readFailureSummary = partitionContext.getReadFailureSummary();
            % This BroadcastMap has been modified by the data processor. It
            % now contains both input broadcasts passed in at construction
            % as well as output broadcasts generated by execution. We only
            % need to return the output broadcasts.
            outputBroadcastMap = subset(obj.BroadcastMap, obj.OutputKeys);
        end
    end
end
