%ReduceOutputServer
% Internal class that is used when the output location is not on a shared
% filesystem. This permits workers to send data to be written by the
% client.

%   Copyright 2015-2020 The MathWorks, Inc.

classdef ReduceOutputServer < handle
    
    properties (Access = private)
        InputQueue; % A queue of data that is used to communicate data from
                    % the dispatch thread to this instance.
        OutputKeyValueSink; % The actual output KeyValueSink instance.
    end

    properties (Constant, Access = private)
        ID_OF_CLIENT = 1;     % Normally the output writer requires an ID to
                              % differentiate the output file of one worker
                              % from another. Given this is not needed when
                              % the client is the only process outputting,
                              % we use ID 1.
    end
    
    methods
        % Main constructor.
        %
        % This receives the output key-value sink factory that is normally
        % sent to the workers.
        function obj = ReduceOutputServer(sinkFactory, ~)
            import parallel.internal.mapreduce.ReduceOutputServer;
            obj.InputQueue = parallel.pool.PollableDataQueue;
            try
                obj.OutputKeyValueSink = sinkFactory(ReduceOutputServer.ID_OF_CLIENT);
            catch err
                oErr = MException(message('parallel:mapreduce:ParallelOutputError'));
                oErr = oErr.addCause(err);
                throw(oErr);
            end
        end
        
        % Perform one step of the server loop. This will perform a single
        % write or flush if one is available on the internal queue.
        function status = step(obj, timeoutInMillis)
            [data, status] = obj.InputQueue.poll(timeoutInMillis/1000);
            if ~status
                return;
            end
            
            try
                obj.OutputKeyValueSink.add(data{1}, data{2});
            catch err
                oErr = MException(message('parallel:mapreduce:ParallelOutputError'));
                oErr = oErr.addCause(err);
                throw(oErr);
            end
            status = true;
        end
        
        % Commit all results to the output files.
        function commit(obj)
            % Commit is triggered after all parfevals finish and so we
            % ought to have all data messages by this point. Drain anything
            % not yet processed.
            while obj.step(0)
            end
            try
                obj.OutputKeyValueSink.commit();
            catch err
                oErr = MException(message('parallel:mapreduce:ParallelOutputError'));
                oErr = oErr.addCause(err);
                throw(oErr);
            end
        end

        % Get a factory that generates a Sink for each worker to send their
        % data into.
        function outputSinkFactory = getRemoteOutputSinkFactory(obj)
            outputSinkFactory = iCreateRemoteOutputSinkFactory(obj.InputQueue, obj.OutputKeyValueSink.createKeyValueProcessor);
        end

        % Get the output metadata that is used to generate the output
        % datastore.
        function results = getOutputMetadata(obj)
            results = obj.OutputKeyValueSink.getTaskOutputMetadata();
        end
        
        function close(~)
        end
    end    
end

% Create a factory for the output sink that a worker should use to send data
% to this server.
function outputSinkFactory = iCreateRemoteOutputSinkFactory(dataQueue, processor)
outputSinkFactory = @(~)iCreateRemoteOutputSink(dataQueue, processor);
end

% Create the output sink that a worker should use to send data to this server.
function outputSink = iCreateRemoteOutputSink(dataQueue, processor)
import parallel.internal.mapreduce.BufferedKeyValueSink;
import parallel.internal.mapreduce.DataQueueKeyValueSink;
outputSink = BufferedKeyValueSink(DataQueueKeyValueSink(dataQueue, processor));
end
