import org.apache.ignite.cluster.ClusterNode; //导入依赖的package包/类
/**
 * Helper method to help assign ComputeJobs to respective ClusterNodes
 * 
 * @param clusterNode
 * @param keys
 * @param map
 * @return Map<ComputeJob, ClusterNode>
 */

private Map<ComputeJob, ClusterNode> setupCrossOver(ClusterNode clusterNode, List<Long> keys,
    Map<ComputeJob, ClusterNode> map) {
    // Calculate number of Jobs = keys / 2
    // as we desire pairs of Chromosomes to be swapped
    int numberOfJobs = keys.size() / 2;
    int k = 0;
    for (int i = 0; i < numberOfJobs; i++) {
        Long key1 = keys.get(k);
        Long key2 = keys.get(k + 1);

        CrossOverJob job = new CrossOverJob(key1, key2, this.config.getCrossOverRate());
        map.put(job, clusterNode);
        k = k + 2;
    }
    return map;
}
 

import org.apache.ignite.cluster.ClusterNode; //导入依赖的package包/类
/**
 * @param cacheName Cache name.
 * @param fun An operation that accepts a cache entry and processes it.
 * @param ignite Ignite.
 * @param keysGen Keys generator.
 * @param <K> Cache key object type.
 * @param <V> Cache value object type.
 */
public static <K, V> void update(String cacheName, Ignite ignite,
    IgniteBiFunction<Ignite, Cache.Entry<K, V>, Stream<Cache.Entry<K, V>>> fun, IgniteSupplier<Set<K>> keysGen) {
    bcast(cacheName, ignite, () -> {
        Ignite ig = Ignition.localIgnite();
        IgniteCache<K, V> cache = ig.getOrCreateCache(cacheName);

        Affinity<K> affinity = ig.affinity(cacheName);
        ClusterNode locNode = ig.cluster().localNode();

        Collection<K> ks = affinity.mapKeysToNodes(keysGen.get()).get(locNode);

        if (ks == null)
            return;

        Map<K, V> m = new ConcurrentHashMap<>();

        ks.parallelStream().forEach(k -> {
            V v = cache.localPeek(k);
            if (v != null)
                (fun.apply(ignite, new CacheEntryImpl<>(k, v))).forEach(ent -> m.put(ent.getKey(), ent.getValue()));
        });

        cache.putAll(m);
    });
}
 

import org.apache.ignite.cluster.ClusterNode; //导入依赖的package包/类
/**
 * @param ignite Grid
 * @param cnt    Count
 * @throws IgniteCheckedException If failed.
 */
protected void waitForRemoteNodes(Ignite ignite, int cnt) throws IgniteCheckedException {
    while (true) {
        Collection<ClusterNode> nodes = ignite.cluster().forRemotes().nodes();

        if (nodes != null && nodes.size() == cnt)
            return;

        try {
            Thread.sleep(100);
        } catch (InterruptedException ignored) {
            throw new IgniteCheckedException("Interrupted while waiting for remote nodes [gridName=" + ignite.name() +
                    ", count=" + cnt + ']');
        }
    }
}
 

import org.apache.ignite.cluster.ClusterNode; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public void sendMessage(
    final ClusterNode node,
    final Message msg,
    final IgniteInClosure<IgniteException> ackC
) throws IgniteSpiException {
    if (msg instanceof GridIoMessage) {
        Message msg0 = ((GridIoMessage)msg).message();

        if (msg0 instanceof GridNearTxPrepareRequest && delay) {
            try {
                U.sleep(TX_TIMEOUT * 2);
            }
            catch (IgniteInterruptedCheckedException e) {
                e.printStackTrace();
            }
        }
    }

    super.sendMessage(node, msg, ackC);
}
 

import org.apache.ignite.cluster.ClusterNode; //导入依赖的package包/类
/**
 * Actual map logic.
 *
 * @param arg Task execution argument.
 * @param subgrid Nodes available for this task execution.
 * @return Map of grid jobs assigned to subgrid node.
 * @throws IgniteException If mapping could not complete successfully.
 */
protected Map<? extends ComputeJob, ClusterNode> map0(List<ClusterNode> subgrid, VisorTaskArgument<A> arg) {
    Collection<UUID> nodeIds = jobNodes(arg);

    Map<ComputeJob, ClusterNode> map = U.newHashMap(nodeIds.size());

    try {
        for (ClusterNode node : subgrid)
            if (nodeIds.contains(node.id()))
                map.put(job(taskArg), node);

        return map;
    }
    finally {
        if (debug)
            logMapped(ignite.log(), getClass(), map.values());
    }
}
 

import org.apache.ignite.cluster.ClusterNode; //导入依赖的package包/类
/**
 * Process response for DML request.
 *
 * @param node Node.
 * @param msg Message.
 */
private void onDmlResponse(final ClusterNode node, GridH2DmlResponse msg) {
    try {
        long reqId = msg.requestId();

        DistributedUpdateRun r = updRuns.get(reqId);

        if (r == null) {
            U.warn(log, "Unexpected dml response (will ignore). [localNodeId=" + ctx.localNodeId() + ", nodeId=" +
                node.id() + ", msg=" + msg.toString() + ']');

            return;
        }

        r.handleResponse(node.id(), msg);
    }
    catch (Exception e) {
        U.error(log, "Error in dml response processing. [localNodeId=" + ctx.localNodeId() + ", nodeId=" +
            node.id() + ", msg=" + msg.toString() + ']', e);
    }
}
 

import org.apache.ignite.cluster.ClusterNode; //导入依赖的package包/类
/**
 * Release remote resources if needed.
 *
 * @param nodes Query nodes.
 * @param r Query run.
 * @param qryReqId Query id.
 * @param distributedJoins Distributed join flag.
 */
public void releaseRemoteResources(Collection<ClusterNode> nodes, ReduceQueryRun r, long qryReqId,
    boolean distributedJoins) {
    // For distributedJoins need always send cancel request to cleanup resources.
    if (distributedJoins)
        send(nodes, new GridQueryCancelRequest(qryReqId), null, false);
    else {
        for (GridMergeIndex idx : r.indexes()) {
            if (!idx.fetchedAll()) {
                send(nodes, new GridQueryCancelRequest(qryReqId), null, false);

                break;
            }
        }
    }

    if (!runs.remove(qryReqId, r))
        U.warn(log, "Query run was already removed: " + qryReqId);
}
 

import org.apache.ignite.cluster.ClusterNode; //导入依赖的package包/类
/** {@inheritDoc} */
@Override protected GridCommunicationClient createTcpClient(ClusterNode node, int connIdx)
    throws IgniteCheckedException {
    if (PRED.apply(node)) {
        Map<String, Object> attrs = new HashMap<>(node.attributes());

        attrs.put(createAttributeName(ATTR_ADDRS), Collections.singleton("127.0.0.1"));
        attrs.put(createAttributeName(ATTR_PORT), 47200);
        attrs.put(createAttributeName(ATTR_EXT_ADDRS), Collections.emptyList());
        attrs.put(createAttributeName(ATTR_HOST_NAMES), Collections.emptyList());

        ((TcpDiscoveryNode)node).setAttributes(attrs);
    }

    return super.createTcpClient(node, connIdx);
}
 

import org.apache.ignite.cluster.ClusterNode; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public CacheMetrics clusterMetrics(ClusterGroup grp) {
    List<CacheMetrics> metrics = new ArrayList<>(grp.nodes().size());

    for (ClusterNode node : grp.nodes()) {
        Map<Integer, CacheMetrics> nodeCacheMetrics = ((TcpDiscoveryNode)node).cacheMetrics();

        if (nodeCacheMetrics != null) {
            CacheMetrics e = nodeCacheMetrics.get(context().cacheId());

            if (e != null)
                metrics.add(e);
        }
    }

    return new CacheMetricsSnapshot(ctx.cache().localMetrics(), metrics);
}
 

import org.apache.ignite.cluster.ClusterNode; //导入依赖的package包/类
/**
 * @throws Exception If failed.
 */
public void testNodeSingletonProxy() throws Exception {
    String name = "testNodeSingletonProxy";

    Ignite ignite = randomGrid();

    ignite.services().deployNodeSingleton(name, new CounterServiceImpl());

    CounterService svc = ignite.services().serviceProxy(name, CounterService.class, false);

    for (int i = 0; i < 10; i++)
        svc.increment();

    assertEquals(10, svc.get());
    assertEquals(10, svc.localIncrements());
    assertEquals(10, ignite.services(ignite.cluster().forLocal()).
        serviceProxy(name, CounterService.class, false).localIncrements());

    // Make sure that remote proxies were not called.
    for (ClusterNode n : ignite.cluster().forRemotes().nodes()) {
        CounterService rmtSvc =
                ignite.services(ignite.cluster().forNode(n)).serviceProxy(name, CounterService.class, false);

        assertEquals(0, rmtSvc.localIncrements());
    }
}
 

import org.apache.ignite.cluster.ClusterNode; //导入依赖的package包/类
/**
 * Starts statically configured caches received from remote nodes during exchange.
 *
 * @param nodeId Joining node ID.
 * @param exchTopVer Current exchange version.
 * @return Started caches descriptors.
 * @throws IgniteCheckedException If failed.
 */
public Collection<DynamicCacheDescriptor> startReceivedCaches(UUID nodeId, AffinityTopologyVersion exchTopVer)
    throws IgniteCheckedException {
    List<DynamicCacheDescriptor> started = cachesInfo.cachesReceivedFromJoin(nodeId);

    for (DynamicCacheDescriptor desc : started) {
        IgnitePredicate<ClusterNode> filter = desc.groupDescriptor().config().getNodeFilter();

        if (CU.affinityNode(ctx.discovery().localNode(), filter)) {
            prepareCacheStart(
                desc.cacheConfiguration(),
                desc,
                null,
                exchTopVer,
                false);
        }
    }

    return started;
}
 

import org.apache.ignite.cluster.ClusterNode; //导入依赖的package包/类
/**
 * @param order Node order.
 * @return Node index.
 */
private int serverNodeBinarySearch(long order) {
    int low = 0;
    int high = srvNodes.size() - 1;

    while (low <= high) {
        int mid = (low + high) >>> 1;

        ClusterNode midVal = srvNodes.get(mid);

        int cmp = Long.compare(midVal.order(), order);

        if (cmp < 0)
            low = mid + 1;
        else if (cmp > 0)
            high = mid - 1;
        else
            return mid;
    }

    return -(low + 1);
}
 

import org.apache.ignite.cluster.ClusterNode; //导入依赖的package包/类
/**
 * @param keyToRmv Removed key.
 */
protected void checkLocalRemovedKey(String keyToRmv) {
    for (int i = 0; i < 500; ++i) {
        String key = "key" + i;

        boolean found = primaryIgnite(key).cache(DEFAULT_CACHE_NAME).localPeek(key) != null;

        if (keyToRmv.equals(key)) {
            Collection<ClusterNode> nodes = grid(0).affinity(DEFAULT_CACHE_NAME).mapKeyToPrimaryAndBackups(key);

            for (int j = 0; j < gridCount(); ++j) {
                if (nodes.contains(grid(j).localNode()) && grid(j) != primaryIgnite(key))
                    assertTrue("Not found on backup removed key ", grid(j).cache(DEFAULT_CACHE_NAME).localPeek(key) != null);
            }

            assertFalse("Found removed key " + key, found);
        }
        else
            assertTrue("Not found key " + key, found);
    }
}
 

import org.apache.ignite.cluster.ClusterNode; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, String arg) {
    assert ignite != null;

    UUID locNodeId = ignite.configuration().getNodeId();

    assert locNodeId != null;

    ClusterNode remoteNode = null;

    for (ClusterNode node : subgrid) {
        if (!node.id().equals(locNodeId))
            remoteNode = node;
    }

    return Collections.singletonMap(new ComputeJobAdapter(arg) {
        @Override public Serializable execute() {
            throw new IgniteException("Job exception.");
        }
    }, remoteNode);
}
 

import org.apache.ignite.cluster.ClusterNode; //导入依赖的package包/类
/**
 * @throws Exception If failed.
 */
public void testIoTestMessages() throws Exception {
    for (Ignite node : G.allGrids()) {
        IgniteKernal ignite = (IgniteKernal)node;

        List<ClusterNode> rmts = new ArrayList<>(ignite.cluster().forRemotes().nodes());

        assertEquals(4, rmts.size());

        for (ClusterNode rmt : rmts) {
            ignite.sendIoTest(rmt, new byte[1024], false);

            ignite.sendIoTest(rmt, new byte[1024], true);

            ignite.sendIoTest(rmts, new byte[1024], false);

            ignite.sendIoTest(rmts, new byte[1024], true);
        }
    }
}
 

import org.apache.ignite.cluster.ClusterNode; //导入依赖的package包/类
/**
 * @param crd Exchange coordinator.
 * @param topVer Exchange topology version.
 */
private void blockExchangeFinish(Ignite crd, long topVer) {
    final AffinityTopologyVersion topVer0 = new AffinityTopologyVersion(topVer);

    TestRecordingCommunicationSpi.spi(crd).blockMessages(new IgniteBiPredicate<ClusterNode, Message>() {
        @Override public boolean apply(ClusterNode node, Message msg) {
            if (msg instanceof GridDhtPartitionsFullMessage) {
                GridDhtPartitionsFullMessage msg0 = (GridDhtPartitionsFullMessage)msg;

                return msg0.exchangeId() != null && msg0.exchangeId().topologyVersion().equals(topVer0);
            }

            return false;
        }
    });
}
 

import org.apache.ignite.cluster.ClusterNode; //导入依赖的package包/类
/**{@inheritDoc} */
@Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, @Nullable GridJobLoadTestParams arg) {
    assert !subgrid.isEmpty();

    Map<ComputeJob, ClusterNode> jobs = new HashMap<>();

    for (int i = 0; i < arg.getJobsCount(); i++)
        jobs.put(
            new GridJobLoadTestJob(
                /*only on the first step*/i == 0,
                arg.getJobFailureProbability(),
                arg.getExecutionDuration(),
                arg.getCompletionDelay()),
            subgrid.get(0));

    return jobs;
}
 

import org.apache.ignite.cluster.ClusterNode; //导入依赖的package包/类
private ClusterGroup getClusterGroupAsSubcollection(Ignite ignite, int size) {
    Collection<ClusterNode> clusterNodes = ignite.cluster().forServers().nodes();

    if (size >= clusterNodes.size()) {
        return ignite.cluster().forServers();
    }
    Iterator<ClusterNode> clusterNodeIter = clusterNodes.iterator();
    List<ClusterNode> groupNodes = new ArrayList<>(size);

    while (groupNodes.size() < size) {
        groupNodes.add(clusterNodeIter.next());
    }
    return ignite.cluster().forNodes(groupNodes);
}
 

import org.apache.ignite.cluster.ClusterNode; //导入依赖的package包/类
@Override public void run() {
    long currentTopologyVersion = ignite.cluster().topologyVersion();
    Set<UUID> currentTopology = new HashSet<>();
    Set<UUID> newNodes = new HashSet<>();

    if ((lastTopologyVersion == null) || (currentTopologyVersion != lastTopologyVersion)) {
        for (ClusterNode node : ignite.cluster().topology(currentTopologyVersion)) {
            currentTopology.add(node.id());
            if ((aggregatorNodeId != node.id()) && !lastTopology.contains(node.id())) {
                newNodes.add(node.id());
            }
        }
        lastTopologyVersion = currentTopologyVersion;
        lastTopology = currentTopology;
    }
    else {
        return;
    }
    Collection<UUID> response = ignite.compute(ignite.cluster().forNodeIds(newNodes)).broadcast(
        new RemoteStatisticsUpdaterStartTask(aggregatorNodeId)
    );
    for (UUID nodeId : response) {
        if (nodeId != null) {
            LOGGER.info("[T] Started statistics collection on node {}", nodeId);
        }
    }
}
 

import org.apache.ignite.cluster.ClusterNode; //导入依赖的package包/类
/** {@inheritDoc} */
@Nullable @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid,
    @Nullable Void arg) throws IgniteException {
    Map<ComputeJob, ClusterNode> res = new HashMap<>();

    for (ClusterNode node : subgrid) {
        BaseLocalProcessorJob<K, V, T, R> job = createJob();
        res.put(job, node);
    }

    return res;
}