%MapTask
% Class for holding all the details necessary for performing a Map Task.
% This should be both serializable and immutable.

% Copyright 2014-2017 The MathWorks, Inc.

classdef (Sealed, Hidden) MapTask < parallel.internal.mapreduce.AbstractTask
properties (GetAccess = private, SetAccess = immutable)
MapTaskID; % Number representing the current map task index.

InputSplit; % Information specifying which bit of the data to operate on.
InputFormat; % Information specifying how to interpret inputSplit and read in the data.

MapFunction; % Function handle to the customers Map function.
end

methods
% Construct a map task.
% Syntax:
% MapTask(mapTaskID, inputSplit, inputFormat, mapFunction, shuffleSortStrategy)
% MapTask(mapTaskID, inputSplit, remoteMapReduceDetails)
function obj = MapTask(mapTaskID, inputSplit, varargin)
if nargin == 3
inputFormat = [];
mapFunction = [];
baseInputs = varargin(1);
elseif nargin == 5
inputFormat = varargin{1};
mapFunction = varargin{2};
baseInputs = varargin(3);
else
assert(false, 'MapTask requires 3 or 5 inputs');
end

obj = obj@parallel.internal.mapreduce.AbstractTask(baseInputs{:});
obj.MapTaskID = mapTaskID;
obj.InputSplit = inputSplit;
obj.InputFormat = inputFormat;
obj.MapFunction = mapFunction;
end

% Runs the given map task.
function results = run(obj)
import parallel.mapreduce.KeyValueOutputStore;

mapFunction = obj.getMapFunction();

shuffleSortStrategy = obj.getShuffleSortStrategy();
mapOutputSink = shuffleSortStrategy.createMapOutputSink(obj.MapTaskID);
keyValueOutputStore = KeyValueOutputStore.hBuild(mapOutputSink);

try
inputDatastore = obj.pReconstituteDatastore();

while hasdata(inputDatastore)
[data, info] = read(inputDatastore);
mapFunction(data, info, keyValueOutputStore);
end
mapOutputSink.commit();

results = mapOutputSink.getTaskOutputMetadata();
catch err
% ParallelException shortens the stack trace such that only
% frames from the map function are sent back to the client.
% However, we do not want to hide UndefinedFunction as this
% allows pool auto-attach to work.
if strcmp(err.identifier, 'MATLAB:UndefinedFunction')
err.rethrow();
else
pErr = ParallelException.hBuildFromRemoteException(err, 'MapTask');
pErr.throw();
end
end
end
end

methods (Access = private)
% Build a datastore object from the given split and format
% information describing the input data of this task.
function inputDatastore = pReconstituteDatastore(obj)
inputDatastore = obj.getInputFormat();

if iscell(obj.InputSplit)
inputDatastore = partition(inputDatastore, obj.InputSplit{:});
else
inputDatastore = copy(inputDatastore);
matlab.io.datastore.internal.shim.initializeDatastore(inputDatastore, obj.InputSplit);
end
end

% Helper function that returns the input format.
function inputFormat = getInputFormat(obj)
if isempty(obj.MapReduceRemoteCache)
inputFormat = obj.InputFormat;
else
inputFormat = obj.MapReduceRemoteCache.getInputDatastore();
end
end

% Helper function that returns the map function.
function mapFunction = getMapFunction(obj)
if isempty(obj.MapReduceRemoteCache)
mapFunction = obj.MapFunction;
else
mapFunction = obj.MapReduceRemoteCache.getMapFunction();
end

% The map function handle is empty only if we could not load it.
if isempty(mapFunction)
error(message('parallel:mapreduce:InvalidMapFunction'));
end
end
end

methods (Hidden)
% Launch a task on the given pool.
% Visitor will be an object/struct with both a runOnSingle and a
% runOnAll method. This should call one of those and return the
% result.
function future = hLaunchTask(obj, visitor)
future = visitor.runOnSingle(obj);
end
end
end