% ThreadsRemoteResourceSet object.


% Copyright 2021-2022 The MathWorks, Inc.

classdef ThreadsRemoteResourceSet < spmdlang.AbstractRemoteResourceSet

    methods (Access = public, Static)

        % This static method builds the "world" set from the pool, and
        % requires no communication or remote execution to achieve that.
        function worldSet = buildWorldSet(pool)
            validateattributes(pool, {'parallel.ThreadPool'}, {});

            if isempty(pool)
                error(message('parallel:lang:spmd:NoSession'));
            end

            commId = parallel.internal.mpi.buildWorldThreadCommunicator(pool);

            isWorld = true;
            worldSet = spmdlang.ThreadsRemoteResourceSet(pool.NumWorkers, commId, pool, isWorld);

            % Add a listener for pool shutdown, so we can cleanup any
            % outstanding composite data on workers. We add this listener
            % here as a worldSet is built exactly once per pool.
            [~] = addlistener(pool, 'ObjectBeingDestroyed', @(~,~) iCompositeCleanup(pool) );
        end
    end

    methods (Access = public, Hidden)
        function obj = ThreadsRemoteResourceSet(numWorkers, commId, pool, isWorld)
            obj = obj@spmdlang.AbstractRemoteResourceSet(numWorkers, commId, pool, isWorld);
        end

        function invalidate(obj, workersStillAvailable)
            parallel.internal.mpi.invalidateThreadCommunicator(obj.RemoteCommCell);

            % Call the superclass invalidation
            invalidate@spmdlang.AbstractRemoteResourceSet(obj, workersStillAvailable);
        end

        % "Factory method" to build a remote resource set of a specific size from the
        % open pool. For Threads, this can be done without a remote call.
        function subSetObj = buildRemoteResourceSet(obj, sz)

            if sz == obj.NumWorkers
                subSetObj = obj;
            elseif sz < 0
                error(message('parallel:lang:spmd:InvalidResourceSelection', sz));
            elseif sz < obj.NumWorkers
                newCommId = parallel.internal.mpi.buildSubsetThreadCommunicator(obj.RemoteCommCell, sz);

                isWorld = false; % trivially as sz < spmdSize
                subSetObj = spmdlang.ThreadsRemoteResourceSet(sz, newCommId, obj.Pool, isWorld);
            else
                % Should never get here because there's already a check that will not let us
                % build a resource set greater than the size of the base set
                error(message('parallel:lang:spmd:InvalidResourceSelection', sz));
            end
        end
    end
end

function iCompositeCleanup(pool)

try
    f = parfevalOnAll(pool, @spmdlang.ValueStore.clear, 0);
    wait(f);
    spmdlang.hGetCompositeKeysToClearMap('clear');
catch
    % don't throw composite errors on pool cleanup.
end

end