% ValueStore Data storage shared by MATLAB clients and workers.
% ValueStore is a data storage owned by a specific job. Each
% entry consists of a value and its corresponding key. When the
% owning job is deleted, the store is deleted as well.
%
% Use the ValueStore to send data from workers back to clients 
% during the execution of a job.
%
% Entries added to the ValueStore are stored in the corresponding job
% storage and do not use system memory. Use the ValueStore when your
% workers produce too much data to be held in memory at once.
%
% MATLAB clients and workers are free to read and modify the ValueStore at
% any time. Values added and removed by MATLAB clients will be visible to
% other MATLAB clients and MATLAB workers even while the job is running. Values
% added and removed by MATLAB workers will also be visible to other MATLAB
% clients and workers even while the job is running. The ordering of
% operations executed by different processes is not guaranteed.
%
% Note: For parallel.cluster.Generic clusters with 'HasSharedFileSystem'
% set to false, the visibility of modifications made to the ValueStore while a job
% is running will depend on your specific implementation.
% Without additional synchronization between the client JobStorageLocation
% and the worker JobStorageLocation, changes may only be visible once the
% job has completed.
%
% Examples:
%
%   To add a new entry to a ValueStore:
%       myValueStore("key") = newValue;
%   or to add multiple entries:
%       put(myValueStore, ["key1", "key2"], {value1, value2});
%
%   To extract an entry from the ValueStore:
%       myValue = myValueStore("key");
%   or to extract multiple entries:
%       myValues = get(myValueStore, ["key1", "key2"]);
%
%   To remove an existing entry in a ValueStore:
%       myValueStore("key") = [];
%   or to remove multiple entries:
%       myValueStore.remove(["key1", "key2"]);    
%
% ValueStore methods:
%   isKey   -   Determine whether ValueStore contains given keys
%   keys    -   Return string array of all keys of ValueStore
%   put     -   Add or replace existing entries
%   get     -   Access existing entries
%   remove  -   Remove entries from ValueStore
%
% ValueStore properties:
%   KeyUpdatedFcn  -   Callback executed when an entry is added or replaced.
%   KeyRemovedFcn  -   Callback executed when an entry is removed.
%
% See Also getCurrentValueStore, parallel.FileStore
  
% Copyright 2021-2023 The MathWorks, Inc.
 
classdef (ConstructOnLoad = true) ValueStore < matlab.mixin.Scalar & parallel.internal.keyvalue.Store

    methods (Static, Hidden)
        function store = createDummyForDisplay()
            store = parallel.ValueStore([], '', false, false, false, false);
        end

        function store = hCreateFileBased(path)
            impl = parallel.internal.keyvalue.CJSKeyValueStore(path);
            store = parallel.ValueStore(impl, '', true, true, true, true);
        end
    end

    methods (Hidden, Access = {?parallel.internal.cluster.CJSJobMixin, ...
            ?parallel.internal.cluster.MJSJobMixin, ...
            ?parallel.internal.pool.ThreadsKeyValueStoreAdapter})
        function obj = ValueStore(impl, owner, ...
                workersSeeClientUpdates, workersSeeClientRemoves, ...
                clientsSeeWorkerUpdates, clientsSeeWorkerRemoves)
            parallel.ValueStore.loadCheck(nargin);
            obj = obj@parallel.internal.keyvalue.Store(impl, ...
                workersSeeClientUpdates, workersSeeClientRemoves, ...
                clientsSeeWorkerUpdates, clientsSeeWorkerRemoves);
            % Invalidate this store when the parent job is destroyed
            if ~isempty(owner)
                addlistener(owner , 'ObjectBeingDestroyed', @(~, ~) delete(obj));
            end
        end
    end

    methods
        function tf = isKey(obj, keys)
            %ISKEY Determine whether ValueStore contains given keys
            %   TF = ISKEY(VS, KEYS) returns a logical indicating whether the
            %   ValueStore object, VS, contains the keys, KEYS. TF is a
            %   N element logical array, where N is the number of elements in 
            %   KEYS.
            arguments
                obj (1,1) parallel.ValueStore
                keys string {mustBeText}
            end
            tf = obj.doIsKey(keys);
        end

        function allKeys = keys(obj)
            %KEYS Return a string array of all keys in ValueStore
            %   ALLKEYS = KEYS(VS) returns a N x 1 string array of all the keys
            %   in the ValueStore object, VS.
            arguments
                obj (1,1) parallel.ValueStore
            end
            allKeys = obj.doKeys();
        end

        function put(obj, keys, values)
            %PUT Put MATLAB values in ValueStore
            %   PUT(VS, KEYS, VALUES) copies MATLAB values to the 
            %   ValueStore object, VS. KEYS is a string array of N keys.
            %   VALUES is a cell array of the N MATLAB values which is added to
            %   the ValueStore using the corresponding key in KEYS.
            arguments
                obj (1,1) parallel.ValueStore
                keys string {mustBeText}
                values cell
            end
            iValidateVectorisedInputSizes(keys, values);
            if isempty(keys)
                return
            end

            try
                obj.Impl.putVariable(keys, values);
            catch E
                err = MException(message('parallel:job:StoreWriteError', strjoin(keys, ', ')));
                err = err.addCause(E);
                throwAsCaller(err);
            end
        end

        function values = get(obj, keys)
            %GET Get values from ValueStore
            %   VALUES = GET(VS, KEYS) return MATLAB values from the 
            %   ValueStore object, VS. KEYS is a string array of N keys.
            %   VALUES is a cell array of the N MATLAB values stored at the
            %   corresponding key in KEYS.
            arguments
                obj (1,1) parallel.ValueStore
                keys string {mustBeText}
            end
            if isempty(keys)
                values = {};
                return
            end

            try
                values = obj.Impl.getVariable(keys);
                values = iMatchShape(values, keys);
            catch E
                err = MException(message('parallel:job:StoreReadError', strjoin(keys, ', ')));
                err = err.addCause(E);
                throwAsCaller(err);
            end
        end

        function remove(obj, keys)
            %REMOVE Remove entries from ValueStore
            %   REMOVE(VS, KEYS) removes one or more entries from the
            %   ValueStore object, VS. KEYS is a string array of N entry
            %   keys to remove from the ValueStore.
            arguments
                obj (1,1) parallel.ValueStore
                keys string {mustBeText}
            end
            if isempty(keys)
                return
            end

            obj.Impl.remove(keys);
        end
    end

    methods (Access = protected)  % Called from matlab.mixin.Scalar
        function varargout = parenReference(obj, indexOp)
            if isempty(indexOp(1).Indices)
                % Occurs when accessing as ValueStore(), so just return the
                % store itself.
                varargout{1} = obj;
                return;
            end
            keys = convertCharsToStrings(indexOp(1).Indices{:});
            if ~isscalar(keys)
                % Only support referencing a single key so we can always
                % return a scalar value and not a cell array
                error(message('parallel:job:StoreParenVectorisedGetNotSupported'));
            end
            value = obj.get(keys);
            % Extract single value from cell array output
            value = value{1};
            if isscalar(indexOp)
                % e.g. val = store('key')
                varargout{1} = value;
                return
            end
            % Forward indexing to stored value. For example:
            % val = store('key').MyProp;
            % [val1, val2] = store('key').MyProp{:};
            [varargout{1:nargout}] = value.(indexOp(2:end));
        end

        function obj = parenAssign(obj, indexOp, varargin)
            if isscalar(indexOp)
                assert(nargin==3);
                rhs = varargin{1};
                % e.g. store('key') = val;
                keys = convertCharsToStrings(indexOp.Indices{:});
                if ~isscalar(keys)
                    % Only support assigning a single key
                    error(message('parallel:job:StoreParenVectorisedSetNotSupported'));
                end
                obj.put(keys, {rhs});
                return
            end
            % Forward indexing to stored value. For example:
            % store('key').MyProp = val;
            % This is done by fetching, modifying, then reinserting the
            % stored value.
            value = obj.parenReference(indexOp(1));
            value.(indexOp(2:end)) = varargin{:};
            [obj.(indexOp(1))] = value;
        end

        function obj = parenDelete(obj, indexOp)
            %For example: store('key') = [];
            keys = convertCharsToStrings(indexOp.Indices{:});
            if ~isscalar(keys)
                % Only support removing a single key
                error(message('parallel:job:StoreParenVectorisedRemoveNotSupported'));
            end
            obj.remove(keys);
        end

        function n = parenListLength(obj, indexOp, ctx)
            containedObj = obj.(indexOp(1));
            if iscell(containedObj)
                % If the indexed object is a cell array, the only valid
                % workflow is to return it.
                n = 1;
            else
                % If the indexed object is a scalar, fetch the number of
                % expected outputs by applying subsequent operations on it.
                n = listLength(containedObj, indexOp(2:end), ctx);
            end
        end
    end

    methods (Access = private, Static)
        function loadCheck(ctorNargin)
        %loadCheck if the constructor is being called with zero arguments,
        %assumption is that the store is being loaded. Therefore error.
            if ctorNargin == 0
                error(message('parallel:job:CannotLoadValueStoreFromMatFile'));
            end
        end
    end
end

function iValidateVectorisedInputSizes(in1, in2)
    if numel(in2) ~= numel(in1)
        error(message('parallel:job:StoreNumberOfVariablesMismatch'));
    end
end

function x = iMatchShape(x, input)
    x = reshape(x , size(input));
end
