import org.apache.ignite.compute.ComputeJobResult; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public void invalidate(Iterable<Metadata> subPath, final IgniteBiInClosure<String, Set<Object>> action) {
    Map<String, List<Metadata>> names = getSnapshotsByCache(subPath);
    if (!names.isEmpty()) {
        ignite.compute().execute(new ComputeTaskSplitAdapter<Map<String, List<Metadata>>, Void>() {
            /** {@inheritDoc} */
            @Override protected Collection<? extends ComputeJob> split(int gridSize,
                Map<String, List<Metadata>> byCache) throws IgniteException {
                List<ComputeJob> result = new ArrayList<>();
                for (Map.Entry<String, List<Metadata>> entry : byCache.entrySet()) {
                    String cacheName = entry.getKey();
                    for (Metadata metadata : entry.getValue()) {
                        result.add(new ProcessAllKeysJob(cacheName, metadata, action));
                    }
                }
                return result;
            }

            /** {@inheritDoc} */
            @Override public Void reduce(List<ComputeJobResult> results) throws IgniteException {
                return null;
            }
        }, names);
    }
}
 

import org.apache.ignite.compute.ComputeJobResult; //导入依赖的package包/类
/**
 * Performs restore operaton on given resource names.
 *
 * @param source base path to existing backup.
 * @param names of resources included in this backup.
 */
private void restoreData(final URI source, Iterable<String> names) {
    failOnExistingTransactions();
    ignite.compute().execute(new ComputeTaskSplitAdapter<Iterable<String>, Object>() {
        /** {@inheritDoc} */
        @Override protected Collection<? extends ComputeJob> split(int gridSize,
            Iterable<String> arg) throws IgniteException {
            List<ComputeJob> result = new ArrayList<>();
            for (String name : arg) {
                result.add(new RestoreJob(source, name));
            }
            return result;
        }

        /** {@inheritDoc} */
        @Nullable @Override public Object reduce(List<ComputeJobResult> results) throws IgniteException {
            return null;
        }
    }, names);
}
 

import org.apache.ignite.compute.ComputeJobResult; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public Integer reduce(List<ComputeJobResult> results) {
    if (log.isInfoEnabled())
        log.info("Reducing job [job=" + this + ", results=" + results + ']');

    if (results.size() < SPLIT_COUNT)
        fail();

    int sum = 0;

    for (ComputeJobResult result : results) {
        if (result.getData() != null)
            sum += (Integer)result.getData();
    }

    return sum;
}
 

import org.apache.ignite.compute.ComputeJobResult; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public Collection<Line> reduce(List<ComputeJobResult> results) {
    Collection<Line> lines = new TreeSet<>(new Comparator<Line>() {
        @Override public int compare(Line line1, Line line2) {
            return line1.rangePosition() < line2.rangePosition() ? -1 :
                line1.rangePosition() > line2.rangePosition() ? 1 : line1.lineIndex() - line2.lineIndex();
        }
    });

    for (ComputeJobResult res : results) {
        if (res.getException() != null)
            throw res.getException();

        Collection<Line> line = res.getData();

        if (line != null)
            lines.addAll(line);
    }

    return lines;
}
 

import org.apache.ignite.compute.ComputeJobResult; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public Object reduce(List<ComputeJobResult> results) {
    assert results.size() == 2;

    for (ComputeJobResult res : results) {
        log.info("Job result: " + res.getData());
    }

    String name1 = results.get(0).getData();
    String name2 = results.get(1).getData();

    assert name1.equals(name2);

    assert !name1.equals(ignite1.name());
    assert name1.equals(ignite2.name());

    return null;
}
 

import org.apache.ignite.compute.ComputeJobResult; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public Integer reduce(List<ComputeJobResult> results) {
    if (log.isInfoEnabled())
        log.info("Reducing job [job=" + this + ", results=" + results + ']');

    if (results.size() < SPLIT_COUNT)
        fail("Results size is less than split count: " + results.size());

    int sum = 0;

    for (ComputeJobResult res : results) {
        if (res.getData() == null)
            fail("Got null result data: " + res);
        else
            log.info("Reducing result: " + res.getData());

        sum += (Integer)res.getData();
    }

    return sum;
}
 

import org.apache.ignite.compute.ComputeJobResult; //导入依赖的package包/类
/**
 * @param taskRes Task result.
 * @param results Results.
 * @return Data collector task result.
 */
protected VisorNodeDataCollectorTaskResult reduce(VisorNodeDataCollectorTaskResult taskRes,
    List<ComputeJobResult> results) {
    for (ComputeJobResult res : results) {
        VisorNodeDataCollectorJobResult jobRes = res.getData();

        if (jobRes != null) {
            UUID nid = res.getNode().id();

            IgniteException unhandledEx = res.getException();

            if (unhandledEx == null)
                reduceJobResult(taskRes, jobRes, nid);
            else {
                // Ignore nodes that left topology.
                if (!(unhandledEx instanceof ClusterGroupEmptyException))
                    taskRes.getUnhandledEx().put(nid, new VisorExceptionWrapper(unhandledEx));
            }
        }
    }

    taskRes.setActive(ignite.active());

    return taskRes;
}
 

import org.apache.ignite.compute.ComputeJobResult; //导入依赖的package包/类
/** {@inheritDoc} */
@Nullable @Override protected VisorLogSearchTaskResult reduce0(List<ComputeJobResult> results) {
    List<VisorLogSearchResult> searchRes = new ArrayList<>();
    Map<Exception, UUID> exRes = U.newHashMap(0);

    // Separate successfully executed results and exceptions.
    for (ComputeJobResult result : results) {
        if (result.getException() != null)
            exRes.put(result.getException(), result.getNode().id());
        else if (result.getData() != null) {
            Collection<VisorLogSearchResult> data = result.getData();

            searchRes.addAll(data);
        }
    }

    return new VisorLogSearchTaskResult(exRes.isEmpty() ? null : exRes, searchRes.isEmpty() ? null : searchRes);
}
 

import org.apache.ignite.compute.ComputeJobResult; //导入依赖的package包/类
/**
 * Initializes failover context.
 *
 * @param taskSes Grid task session.
 * @param jobRes Failed job result.
 * @param loadMgr Load manager.
 * @param partId Partition.
 * @param affCacheName Affinity cache name.
 * @param topVer Affinity topology version.
 */
public GridFailoverContextImpl(GridTaskSessionImpl taskSes,
    ComputeJobResult jobRes,
    GridLoadBalancerManager loadMgr,
    int partId,
    @Nullable String affCacheName,
    @Nullable AffinityTopologyVersion topVer) {
    assert taskSes != null;
    assert jobRes != null;
    assert loadMgr != null;

    this.taskSes = taskSes;
    this.jobRes = jobRes;
    this.loadMgr = loadMgr;
    this.partId = partId;
    this.affCacheName = affCacheName;
    this.topVer = topVer;
}
 

import org.apache.ignite.compute.ComputeJobResult; //导入依赖的package包/类
/** {@inheritDoc} */
@Nullable @Override public Void reduce(List<ComputeJobResult> results) {
    assert results.isEmpty() : "Should not cache result in java for interop task";

    lock.readLock().lock();

    try {
        assert !done;

        ctx.gateway().computeTaskReduce(taskPtr);
    }
    finally {
        lock.readLock().unlock();
    }

    return null;
}
 

import org.apache.ignite.compute.ComputeJobResult; //导入依赖的package包/类
/** {@inheritDoc} */
@SuppressWarnings("ThrowableResultOfMethodCallIgnored")
@Nullable @Override public GridRestResponse reduce(List<ComputeJobResult> results) throws IgniteException {
    Map<String, GridCacheSqlMetadata> map = new HashMap<>();

    for (ComputeJobResult r : results) {
        if (!r.isCancelled() && r.getException() == null) {
            for (GridCacheSqlMetadata m : r.<Collection<GridCacheSqlMetadata>>getData()) {
                if (!map.containsKey(m.cacheName()))
                    map.put(m.cacheName(), m);
            }
        }
    }

    Collection<GridCacheSqlMetadata> metas = new ArrayList<>(map.size());

    metas.addAll(map.values());

    return new GridRestResponse(metas);
}
 

import org.apache.ignite.compute.ComputeJobResult; //导入依赖的package包/类
@Override
public String reduce(List<ComputeJobResult> results) throws IgniteException {
    StringBuilder answer = new StringBuilder();
    for (ComputeJobResult res : results) {
        Object data = res.getData();
        answer.append(data).append(",");
    }
    answer.deleteCharAt(answer.length() - 1);
    return answer.toString();
}
 

import org.apache.ignite.compute.ComputeJobResult; //导入依赖的package包/类
/** {@inheritDoc} */
@Nullable @Override public R reduce(List<ComputeJobResult> results) {
    if (!F.isEmpty(results)) {
        ComputeJobResult res = results.get(0);

        return res.getData();
    }
    else
        return null;
}
 

import org.apache.ignite.compute.ComputeJobResult; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public String reduce(List<ComputeJobResult> results) {
    if (log.isInfoEnabled())
        log.info("Aggregating job [job=" + this + ", results=" + results + ']');

    if (results.size() != SPLIT_COUNT)
        fail("Invalid results size.");

    return "interrupt-task-data";
}
 

import org.apache.ignite.compute.ComputeJobResult; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public ComputeJobResultPolicy result(ComputeJobResult res, List<ComputeJobResult> received) {
    if (res.getException() != null)
        throw res.getException();

    return received.size() == SPLIT_COUNT ? ComputeJobResultPolicy.REDUCE : ComputeJobResultPolicy.WAIT;
}
 

import org.apache.ignite.compute.ComputeJobResult; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public Integer reduce(List<ComputeJobResult> results) {
    if (log.isInfoEnabled())
        log.info("Reducing job [job=" + this + ", results=" + results + ']');

    int res = 0;

    for (ComputeJobResult result : results)
        res += (Integer)result.getData();

    return res;
}
 

import org.apache.ignite.compute.ComputeJobResult; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public Integer reduce(List<ComputeJobResult> results) {
    int sum = 0;

    for (ComputeJobResult res : results)
        sum += res.<Integer>getData();

    return sum;
}
 

import org.apache.ignite.compute.ComputeJobResult; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public ComputeJobResultPolicy result(ComputeJobResult res, List<ComputeJobResult> received) {
    log.info("Task result state: " + state);

    assert state != null;
    assert state == testState;

    return super.result(res, received);
}
 

import org.apache.ignite.compute.ComputeJobResult; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public R reduce(List<ComputeJobResult> res) {
    for (ComputeJobResult r : res)
        if (r.getException() == null)
            return r.getData();

    throw new IgniteException("Failed to find successful job result: " + res);
}
 

import org.apache.ignite.compute.ComputeJobResult; //导入依赖的package包/类
@SuppressWarnings({"ProhibitedExceptionThrown"})
@Override public Serializable reduce(List<ComputeJobResult> results) {
    assert results != null;

    if (failType == FailType.REDUCE)
        throw new RuntimeException("Failed out of reduce method.");

    return (Serializable)results;
}