// UnicastZenPing.ping函数
protected void ping(final Consumer<PingCollection> resultsConsumer,
                    final TimeValue scheduleDuration,
                    final TimeValue requestDuration) {
    final List<DiscoveryNode> seedNodes;
    try {
        /*
         * 从配置文件中获取 seed nodes
         */
        seedNodes = resolveHostsLists(
            unicastZenPingExecutorService,
            logger,
            configuredHosts,
            limitPortCounts,
            transportService,
            UNICAST_NODE_PREFIX,
            resolveTimeout);
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }

    /*
     * 加入File-Based Discovery Plugin发现的 seed nodes
     */
    seedNodes.addAll(hostsProvider.buildDynamicNodes());
    final DiscoveryNodes nodes = contextProvider.clusterState().nodes();
    // add all possible master nodes that were active in the last known cluster configuration
    for (ObjectCursor<DiscoveryNode> masterNode : nodes.getMasterNodes().values()) {
        seedNodes.add(masterNode.value);
    }

    final ConnectionProfile connectionProfile =
        ConnectionProfile.buildSingleChannelProfile(TransportRequestOptions.Type.REG, requestDuration, requestDuration);
    /*
     * pingingRound 是一个关键的类, 负责向所有的 seedNodes 发送 ping, 并把结果记录在 pingCollections 中
     */
    final PingingRound pingingRound = new PingingRound(pingingRoundIdGenerator.incrementAndGet(), seedNodes, resultsConsumer,
        nodes.getLocalNode(), connectionProfile);
    activePingingRounds.put(pingingRound.id(), pingingRound);
    final AbstractRunnable pingSender = new AbstractRunnable() {
        @Override
        public void onFailure(Exception e) {
            if (e instanceof AlreadyClosedException == false) {
                logger.warn("unexpected error while pinging", e);
            }
        }

        @Override
        protected void doRun() throws Exception {
            sendPings(requestDuration, pingingRound);
        }
    };
    /*
     * 一共发送三次 pingRound, pingCollections 是一个 map, 多次发送自动去重, 增加了网络抖动情况下的鲁棒性
     */
    threadPool.generic().execute(pingSender);
    threadPool.schedule(TimeValue.timeValueMillis(scheduleDuration.millis() / 3), ThreadPool.Names.GENERIC, pingSender);
    threadPool.schedule(TimeValue.timeValueMillis(scheduleDuration.millis() / 3 * 2), ThreadPool.Names.GENERIC, pingSender);

    /*
     * finishPingingRound 执行 pingConsumer 的 accept 函数
     * 外边传进来的是 CompleteableFuture的complete, 所以此处CompleteableFuture.get()停止阻塞返回pingCollection
     * 见ZenDiscovery.java pingAndWait函数
     */
    threadPool.schedule(scheduleDuration, ThreadPool.Names.GENERIC, new AbstractRunnable() {
        @Override
        protected void doRun() throws Exception {
            finishPingingRound(pingingRound);
        }

        @Override
        public void onFailure(Exception e) {
            logger.warn("unexpected error while finishing pinging round", e);
        }
    });
}

// UnicastZenPing.sendPings 函数
protected void sendPings(final TimeValue timeout, final PingingRound pingingRound) {
    final ClusterState lastState = contextProvider.clusterState();
    /*
     * 构造 pingRequest, 包含本身节点信息
     */
    final UnicastPingRequest pingRequest = new UnicastPingRequest(pingingRound.id(), timeout, createPingResponse(lastState));

    Set<DiscoveryNode> nodesFromResponses = temporalResponses.stream().map(pingResponse -> {
        assert clusterName.equals(pingResponse.clusterName()) :
            "got a ping request from a different cluster. expected " + clusterName + " got " + pingResponse.clusterName();
        return pingResponse.node();
    }).collect(Collectors.toSet());

    /*
     * 构造要发送请求的节点集合, 包含两部分:
     * 1. seed nodes, 在配置文件中配置
     * 2. 所有给自己发送过 ping 请求的节点
     */
    // dedup by address
    final Map<TransportAddress, DiscoveryNode> uniqueNodesByAddress =
        Stream.concat(pingingRound.getSeedNodes().stream(), nodesFromResponses.stream())
            .collect(Collectors.toMap(DiscoveryNode::getAddress, Function.identity(), (n1, n2) -> n1));


    // resolve what we can via the latest cluster state
    final Set<DiscoveryNode> nodesToPing = uniqueNodesByAddress.values().stream()
        .map(node -> {
            DiscoveryNode foundNode = lastState.nodes().findByAddress(node.getAddress());
            if (foundNode == null) {
                return node;
            } else {
                return foundNode;
            }
        }).collect(Collectors.toSet());

    nodesToPing.forEach(node -> sendPingRequestToNode(node, timeout, pingingRound, pingRequest));
}

// UnicastZenPing.sendPingRequestToNode函数
private void sendPingRequestToNode(final DiscoveryNode node, TimeValue timeout, final PingingRound pingingRound,
                                   final UnicastPingRequest pingRequest) {
    submitToExecutor(new AbstractRunnable() {
        @Override
        protected void doRun() throws Exception {
            Connection connection = null;
            if (transportService.nodeConnected(node)) {
                try {
                    // concurrency can still cause disconnects
                    connection = transportService.getConnection(node);
                } catch (NodeNotConnectedException e) {
                    logger.trace("[{}] node [{}] just disconnected, will create a temp connection", pingingRound.id(), node);
                }
            }

            if (connection == null) {
                connection = pingingRound.getOrConnect(node);
            }

            logger.trace("[{}] sending to {}", pingingRound.id(), node);
            transportService.sendRequest(connection, ACTION_NAME, pingRequest,
                TransportRequestOptions.builder().withTimeout((long) (timeout.millis() * 1.25)).build(),
                getPingResponseHandler(pingingRound, node));
        }

        @Override
        public void onFailure(Exception e) {
            if (e instanceof ConnectTransportException || e instanceof AlreadyClosedException) {
                // can't connect to the node - this is more common path!
                logger.trace(() -> new ParameterizedMessage("[{}] failed to ping {}", pingingRound.id(), node), e);
            } else if (e instanceof RemoteTransportException) {
                // something went wrong on the other side
                logger.debug(() -> new ParameterizedMessage(
                    "[{}] received a remote error as a response to ping {}", pingingRound.id(), node), e);
            } else {
                logger.warn(() -> new ParameterizedMessage("[{}] failed send ping to {}", pingingRound.id(), node), e);
            }
        }

        @Override
        public void onRejection(Exception e) {
            // The RejectedExecutionException can come from the fact unicastZenPingExecutorService is at its max down in sendPings
            // But don't bail here, we can retry later on after the send ping has been scheduled.
            logger.debug("Ping execution rejected", e);
        }
    });
}

//UnicastZenPing.getPingResponseHandler函数
protected TransportResponseHandler<UnicastPingResponse> getPingResponseHandler(final PingingRound pingingRound,
                                                                               final DiscoveryNode node) {
    return new TransportResponseHandler<UnicastPingResponse>() {

        @Override
        public UnicastPingResponse read(StreamInput in) throws IOException {
            return new UnicastPingResponse(in);
        }

        @Override
        public String executor() {
            return ThreadPool.Names.SAME;
        }

        @Override
        public void handleResponse(UnicastPingResponse response) {
            logger.trace("[{}] received response from {}: {}", pingingRound.id(), node, Arrays.toString(response.pingResponses));
            if (pingingRound.isClosed()) {
                if (logger.isTraceEnabled()) {
                    logger.trace("[{}] skipping received response from {}. already closed", pingingRound.id(), node);
                }
            } else {
                Stream.of(response.pingResponses).forEach(pingingRound::addPingResponseToCollection);
            }
        }

        @Override
        public void handleException(TransportException exp) {
            if (exp instanceof ConnectTransportException || exp.getCause() instanceof ConnectTransportException ||
                exp.getCause() instanceof AlreadyClosedException) {
                // ok, not connected...
                logger.trace(() -> new ParameterizedMessage("failed to connect to {}", node), exp);
            } else if (closed == false) {
                logger.warn(() -> new ParameterizedMessage("failed to send ping to [{}]", node), exp);
            }
        }
    };
}

// UnicastZenPing.handlePingRequest
private UnicastPingResponse handlePingRequest(final UnicastPingRequest request) {
    assert clusterName.equals(request.pingResponse.clusterName()) :
        "got a ping request from a different cluster. expected " + clusterName + " got " + request.pingResponse.clusterName();

    /*
     * 每收到一个请求加到 temporalResponses 中
     */
    temporalResponses.add(request.pingResponse);
    // add to any ongoing pinging

    /*
     * 加到 pingingCollection 中
     */
    activePingingRounds.values().forEach(p -> p.addPingResponseToCollection(request.pingResponse));

    /*
     * temporalResponse 中的 response 会定时过期
     */
    threadPool.schedule(TimeValue.timeValueMillis(request.timeout.millis() * 2), ThreadPool.Names.SAME,
        () -> temporalResponses.remove(request.pingResponse));

    /*
     * 返回给请求节点的 responses 集合包含两部分
     * 1. 自己曾经收到ping 请求的节点
     * 2. 本身节点
     */
    List<PingResponse> pingResponses = CollectionUtils.iterableAsArrayList(temporalResponses);
    pingResponses.add(createPingResponse(contextProvider.clusterState()));

    return new UnicastPingResponse(request.id, pingResponses.toArray(new PingResponse[pingResponses.size()]));
}

//ZenDiscovery.findMaster 函数
private DiscoveryNode findMaster() {
    logger.trace("starting to ping");
    List<ZenPing.PingResponse> fullPingResponses = pingAndWait(pingTimeout).toList();
    if (fullPingResponses == null) {
        logger.trace("No full ping responses");
        return null;
    }
    if (logger.isTraceEnabled()) {
        StringBuilder sb = new StringBuilder();
        if (fullPingResponses.size() == 0) {
            sb.append(" {none}");
        } else {
            for (ZenPing.PingResponse pingResponse : fullPingResponses) {
                sb.append("\n\t--> ").append(pingResponse);
            }
        }
        logger.trace("full ping responses:{}", sb);
    }

    final DiscoveryNode localNode = transportService.getLocalNode();

    /*
     * 没有把自己加到 pingCollections 中
     */
    // add our selves
    assert fullPingResponses.stream().map(ZenPing.PingResponse::node)
        .filter(n -> n.equals(localNode)).findAny().isPresent() == false;

    fullPingResponses.add(new ZenPing.PingResponse(localNode, null, this.clusterState()));

    // filter responses
    final List<ZenPing.PingResponse> pingResponses = filterPingResponses(fullPingResponses, masterElectionIgnoreNonMasters, logger);

    /*
     * 找到当前集群的 master
     */
    List<DiscoveryNode> activeMasters = new ArrayList<>();
    for (ZenPing.PingResponse pingResponse : pingResponses) {
        // We can't include the local node in pingMasters list, otherwise we may up electing ourselves without
        // any check / verifications from other nodes in ZenDiscover#innerJoinCluster()
        if (pingResponse.master() != null && !localNode.equals(pingResponse.master())) {
            activeMasters.add(pingResponse.master());
        }
    }

    /*
     * 找到当前集群的 master candidate
     */
    // nodes discovered during pinging
    List<ElectMasterService.MasterCandidate> masterCandidates = new ArrayList<>();
    for (ZenPing.PingResponse pingResponse : pingResponses) {
        if (pingResponse.node().isMasterNode()) {
            masterCandidates.add(new ElectMasterService.MasterCandidate(pingResponse.node(), pingResponse.getClusterStateVersion()));
        }
    }

    /*
     * 当前集群没有 master
     */
    if (activeMasters.isEmpty()) {
        /*
         * 当前节点与超过 minimum candidate 个节点保持联系
         * 从候选节点中选出 master
         */
        if (electMaster.hasEnoughCandidates(masterCandidates)) {
            final ElectMasterService.MasterCandidate winner = electMaster.electMaster(masterCandidates);
            logger.trace("candidate {} won election", winner);
            return winner.getNode();
        } else {

            /*
             * 无法选出 master 返回 null
             */
            // if we don't have enough master nodes, we bail, because there are not enough master to elect from
            logger.warn("not enough master nodes discovered during pinging (found [{}], but needed [{}]), pinging again",
                masterCandidates, electMaster.minimumMasterNodes());
            return null;
        }
    } else {
        assert !activeMasters.contains(localNode) : "local node should never be elected as master when other nodes indicate an active master";
        // lets tie break between discovered nodes
        /*
         * 集群存在多个 master 节点, 选择 id 较小的当做 master
         */
        return electMaster.tieBreakActiveMasters(activeMasters);
    }
}

// ZenDiscovery.innerJoinCluster函数
private void innerJoinCluster() {
    DiscoveryNode masterNode = null;
    final Thread currentThread = Thread.currentThread();
    nodeJoinController.startElectionContext();
    while (masterNode == null && joinThreadControl.joinThreadActive(currentThread)) {
        masterNode = findMaster();
    }

    if (!joinThreadControl.joinThreadActive(currentThread)) {
        logger.trace("thread is no longer in currentJoinThread. Stopping.");
        return;
    }

    /*
     * 当前节点被选为 master
     */
    if (transportService.getLocalNode().equals(masterNode)) {
        final int requiredJoins = Math.max(0, electMaster.minimumMasterNodes() - 1); // we count as one
        logger.debug("elected as master, waiting for incoming joins ([{}] needed)", requiredJoins);
        nodeJoinController.waitToBeElectedAsMaster(requiredJoins, masterElectionWaitForJoinsTimeout,
            new NodeJoinController.ElectionCallback() {
                @Override
                public void onElectedAsMaster(ClusterState state) {
                    synchronized (stateMutex) {
                        joinThreadControl.markThreadAsDone(currentThread);
                    }
                }

                @Override
                public void onFailure(Throwable t) {
                    logger.trace("failed while waiting for nodes to join, rejoining", t);
                    synchronized (stateMutex) {
                        joinThreadControl.markThreadAsDoneAndStartNew(currentThread);
                    }
                }
            }

        );
    } else {
        // process any incoming joins (they will fail because we are not the master)
        nodeJoinController.stopElectionContext(masterNode + " elected");

        // send join request
        /*
         * 此处会阻塞等待,并有多次重试,
         */
        final boolean success = joinElectedMaster(masterNode);

        synchronized (stateMutex) {
            if (success) {
                DiscoveryNode currentMasterNode = this.clusterState().getNodes().getMasterNode();
                if (currentMasterNode == null) {
                    // Post 1.3.0, the master should publish a new cluster state before acking our join request. we now should have
                    // a valid master.
                    logger.debug("no master node is set, despite of join request completing. retrying pings.");
                    joinThreadControl.markThreadAsDoneAndStartNew(currentThread);
                } else if (currentMasterNode.equals(masterNode) == false) {
                    // update cluster state
                    joinThreadControl.stopRunningThreadAndRejoin("master_switched_while_finalizing_join");
                }

                joinThreadControl.markThreadAsDone(currentThread);
            } else {
                // failed to join. Try again...
                joinThreadControl.markThreadAsDoneAndStartNew(currentThread);
            }
        }
    }
}