/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */
package org.apache.geode.distributed.internal.membership.gms.fd;


import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import org.apache.logging.log4j.Logger;
import org.jgroups.util.UUID;

import org.apache.geode.CancelException;
import org.apache.geode.GemFireConfigException;
import org.apache.geode.SystemConnectException;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.DMStats;
import org.apache.geode.distributed.internal.DistributionConfig;
import org.apache.geode.distributed.internal.DistributionMessage;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.distributed.internal.membership.NetView;
import org.apache.geode.distributed.internal.membership.gms.GMSMember;
import org.apache.geode.distributed.internal.membership.gms.ServiceConfig;
import org.apache.geode.distributed.internal.membership.gms.Services;
import org.apache.geode.distributed.internal.membership.gms.interfaces.HealthMonitor;
import org.apache.geode.distributed.internal.membership.gms.messages.FinalCheckPassedMessage;
import org.apache.geode.distributed.internal.membership.gms.messages.HeartbeatMessage;
import org.apache.geode.distributed.internal.membership.gms.messages.HeartbeatRequestMessage;
import org.apache.geode.distributed.internal.membership.gms.messages.SuspectMembersMessage;
import org.apache.geode.distributed.internal.membership.gms.messages.SuspectRequest;
import org.apache.geode.internal.ConnectionWatcher;
import org.apache.geode.internal.Version;
import org.apache.geode.internal.logging.LoggingExecutors;
import org.apache.geode.internal.net.SocketCreatorFactory;
import org.apache.geode.internal.security.SecurableCommunicationChannel;

/**
 * Failure Detection
 * <p>
 * This class make sure that each member is alive and communicating to this member. To make sure
 * that we create the ring of members based on current view. On this ring, each member make sure
 * that next-member in ring is communicating with it. For that we record last message timestamp from
 * next-member. And if it sees this member has not communicated in last period(member-timeout) then
 * we check whether this member is still alive or not. Based on that we informed probable
 * coordinators to remove that member from view.
 * <p>
 * It has {@link #suspect(InternalDistributedMember, String)} api, which can be used to initiate
 * suspect processing for any member. First is checks whether the member is responding or not. Then
 * it informs probable coordinators to remove that member from view.
 * <p>
 * It has {@link HealthMonitor#checkIfAvailable(InternalDistributedMember, String, boolean)} api to
 * see if that member is
 * alive. Then based on removal flag it initiates the suspect processing for that member.
 */
@SuppressWarnings({"SynchronizationOnLocalVariableOrMethodParameter", "NullableProblems"})
public class GMSHealthMonitor implements HealthMonitor {

  private Services services;
  private volatile NetView currentView;
  private volatile InternalDistributedMember nextNeighbor;

  long memberTimeout;
  private volatile boolean isStopping = false;
  private final AtomicInteger requestId = new AtomicInteger();

  /**
   * membership logger
   */
  private static final Logger logger = Services.getLogger();

  /**
   * The number of recipients of periodic heartbeats. The recipients will be selected from the
   * members that are likely to be monitoring this member.
   */
  private static final int NUM_HEARTBEATS = Integer.getInteger("geode.heartbeat-recipients", 2);

  /**
   * Member activity will be recorded per interval/period. Timer task will set interval's starting
   * time. Each interval will be member-timeout/LOGICAL_INTERVAL. LOGICAL_INTERVAL may be configured
   * via a system property with a default of 2. At least 1 interval is needed.
   */
  public static final int LOGICAL_INTERVAL =
      Integer.getInteger("geode.logical-message-received-interval", 2);

  /**
   * stall time to wait for members leaving concurrently
   */
  public static final long MEMBER_SUSPECT_COLLECTION_INTERVAL =
      Long.getLong("geode.suspect-member-collection-interval", 200);

  private volatile long currentTimeStamp;

  /**
   * this member's ID
   */
  private InternalDistributedMember localAddress;

  /**
   * Timestamp at which we last had contact from a member
   */
  final ConcurrentMap<InternalDistributedMember, TimeStamp> memberTimeStamps =
      new ConcurrentHashMap<>();

  /**
   * Members currently being suspected and the view they were suspected in
   */
  private final ConcurrentHashMap<InternalDistributedMember, NetView> suspectedMemberIds =
      new ConcurrentHashMap<>();

  /**
   * Members undergoing final checks
   */
  private final List<InternalDistributedMember> membersInFinalCheck =
      Collections.synchronizedList(new ArrayList<>(30));

  /**
   * Replies to messages
   */
  private final Map<Integer, Response> requestIdVsResponse = new ConcurrentHashMap<>();

  /**
   * Members suspected in a particular view
   */
  private final Map<NetView, Set<SuspectRequest>> suspectRequestsInView = new HashMap<>();

  private ScheduledExecutorService scheduler;

  private ExecutorService checkExecutor;

  /**
   * to stop check scheduler
   */
  private ScheduledFuture<?> monitorFuture;

  /**
   * test hook
   */
  private volatile boolean playingDead = false;

  /**
   * test hook
   */
  private volatile boolean beingSick = false;

  // For TCP check
  private ExecutorService serverSocketExecutor;
  static final int OK = 0x7B;
  static final int ERROR = 0x00;
  private volatile int socketPort;
  private volatile ServerSocket serverSocket;

  /**
   * Statistics about health monitor
   */
  private DMStats stats;

  /**
   * Interval to run the Monitor task
   */
  private long monitorInterval;

  /**
   * /**
   * this class is to avoid garbage
   */
  private static class TimeStamp {

    private volatile long timeStamp;

    TimeStamp(long timeStamp) {
      this.timeStamp = timeStamp;
    }

    public long getTime() {
      return timeStamp;
    }

    public void setTime(long timeStamp) {
      this.timeStamp = timeStamp;
    }
  }

  /***
   * This class sets start interval timestamp to record the activity of all members. That is used by
   * {@link GMSHealthMonitor#contactedBy(InternalDistributedMember)} to record the activity of
   * member.
   *
   * It initiates the suspect processing for next neighbour if it doesn't see any activity from that
   * member in last interval(member-timeout)
   */
  private class Monitor implements Runnable {
    /**
     * Here we use the same threshold for detecting JVM pauses as the StatSampler
     */
    private final long MONITOR_DELAY_THRESHOLD =
        Long.getLong(DistributionConfig.GEMFIRE_PREFIX + "statSamplerDelayThreshold", 3000);


    final long memberTimeoutInMillis;

    public Monitor(long memberTimeout) {
      memberTimeoutInMillis = memberTimeout;
    }

    @Override
    public void run() {

      if (GMSHealthMonitor.this.isStopping) {
        return;
      }

      InternalDistributedMember neighbour = nextNeighbor;

      long currentTime = System.currentTimeMillis();
      // this is the start of interval to record member activity
      GMSHealthMonitor.this.currentTimeStamp = currentTime;


      long oldTimeStamp = currentTimeStamp;
      currentTimeStamp = System.currentTimeMillis();

      NetView myView = GMSHealthMonitor.this.currentView;
      if (myView == null) {
        return;
      }

      if (currentTimeStamp - oldTimeStamp > monitorInterval + MONITOR_DELAY_THRESHOLD) {
        // delay in running this task - don't suspect anyone for a while
        logger.info(
            "Failure detector has noticed a JVM pause and is giving all members a heartbeat in view {}",
            currentView);
        for (InternalDistributedMember member : myView.getMembers()) {
          contactedBy(member);
        }
        return;
      }

      if (neighbour != null) {
        TimeStamp nextNeighborTS;
        synchronized (GMSHealthMonitor.this) {
          nextNeighborTS = GMSHealthMonitor.this.memberTimeStamps.get(neighbour);
        }

        if (nextNeighborTS == null) {
          TimeStamp customTS = new TimeStamp(currentTime);
          memberTimeStamps.put(neighbour, customTS);
          return;
        }

        long interval = memberTimeoutInMillis / GMSHealthMonitor.LOGICAL_INTERVAL;
        long lastTS = currentTime - nextNeighborTS.getTime();
        if (lastTS + interval >= memberTimeoutInMillis) {
          logger.debug("Checking member {} ", neighbour);
          // now do check request for this member;
          checkMember(neighbour);
        }
      }
    }
  }

  /***
   * Check thread waits on this object for response. It puts requestId in requestIdVsResponse map.
   * Response will have requestId, which is used to get ResponseObject. Then it is used to notify
   * waiting thread.
   */
  private class Response {

    private DistributionMessage responseMsg;

    public DistributionMessage getResponseMsg() {
      return responseMsg;
    }

    public void setResponseMsg(DistributionMessage responseMsg) {
      this.responseMsg = responseMsg;
    }

  }

  class ClientSocketHandler implements Runnable {

    private final Socket socket;

    public ClientSocketHandler(Socket socket) {
      this.socket = socket;
    }

    @Override
    public void run() {
      try {
        socket.setTcpNoDelay(true);
        DataInputStream in = new DataInputStream(socket.getInputStream());
        OutputStream out = socket.getOutputStream();
        @SuppressWarnings("UnusedAssignment")
        short version = in.readShort();
        int vmViewId = in.readInt();
        long uuidLSBs = in.readLong();
        long uuidMSBs = in.readLong();
        GMSHealthMonitor.this.stats.incFinalCheckRequestsReceived();
        GMSHealthMonitor.this.stats.incTcpFinalCheckRequestsReceived();
        GMSMember gmbr = (GMSMember) GMSHealthMonitor.this.localAddress.getNetMember();
        UUID myUUID = gmbr.getUUID();
        // during reconnect or rapid restart we will have a zero viewId but there may still
        // be an old ID in the membership view that we do not want to respond to
        int myVmViewId = gmbr.getVmViewId();
        if (playingDead) {
          logger.debug("HealthMonitor: simulating sick member in health check");
        } else if (uuidLSBs == myUUID.getLeastSignificantBits()
            && uuidMSBs == myUUID.getMostSignificantBits()
            && (vmViewId == myVmViewId || myVmViewId < 0)) {
          logger.debug("HealthMonitor: sending OK reply");
          out.write(OK);
          out.flush();
          socket.shutdownOutput();
          GMSHealthMonitor.this.stats.incFinalCheckResponsesSent();
          GMSHealthMonitor.this.stats.incTcpFinalCheckResponsesSent();
          logger.debug("HealthMonitor: server replied OK.");
        } else {
          if (logger.isDebugEnabled()) {
            logger.debug(
                "HealthMonitor: sending ERROR reply - my UUID is {},{} received is {},{}.  My viewID is {} received is {}",
                Long.toHexString(myUUID.getMostSignificantBits()),
                Long.toHexString(myUUID.getLeastSignificantBits()), Long.toHexString(uuidMSBs),
                Long.toHexString(uuidLSBs), myVmViewId, vmViewId);
          }
          out.write(ERROR);
          out.flush();
          socket.shutdownOutput();
          GMSHealthMonitor.this.stats.incFinalCheckResponsesSent();
          GMSHealthMonitor.this.stats.incTcpFinalCheckResponsesSent();
          logger.debug("HealthMonitor: server replied ERROR.");
        }
      } catch (IOException e) {
        // this is expected if it is a connection-timeout or other failure
        // to connect
      } catch (RuntimeException e) {
        logger.debug("Unexpected runtime exception", e);
        throw e;
      } catch (Error e) {
        logger.debug("Unexpected error", e);
        throw e;
      } finally {
        if (socket != null) {
          try {
            socket.close();
          } catch (IOException e) {
            // expected if the socket is already closed
          }
        }
      }
    }
  }

  public GMSHealthMonitor() {

  }

  @SuppressWarnings("EmptyMethod")
  public static void loadEmergencyClasses() {}

  /*
   * Record the member activity for current time interval.
   */
  @Override
  public void contactedBy(InternalDistributedMember sender) {
    contactedBy(sender, currentTimeStamp);
  }


  /**
   * Record member activity at a specified time
   */
  private void contactedBy(InternalDistributedMember sender, long timeStamp) {
    TimeStamp cTS = new TimeStamp(timeStamp);
    cTS = memberTimeStamps.putIfAbsent(sender, cTS);
    if (cTS != null && cTS.getTime() < timeStamp) {
      cTS.setTime(timeStamp);
    }
    if (suspectedMemberIds.containsKey(sender)) {
      memberUnsuspected(sender);
      setNextNeighbor(currentView, null);
    }
  }


  private HeartbeatRequestMessage constructHeartbeatRequestMessage(
      final InternalDistributedMember mbr) {
    final int reqId = requestId.getAndIncrement();
    final HeartbeatRequestMessage hrm = new HeartbeatRequestMessage(mbr, reqId);
    hrm.setRecipient(mbr);

    return hrm;
  }

  private void checkMember(final InternalDistributedMember mbr) {
    final NetView cv = GMSHealthMonitor.this.currentView;

    // as check may take time
    setNextNeighbor(cv, mbr);

    // we need to check this member
    checkExecutor.execute(() -> {
      boolean pinged;
      try {
        pinged = GMSHealthMonitor.this.doCheckMember(mbr, true);
      } catch (CancelException e) {
        return;
      }

      if (!pinged) {
        String reason = "Member isn't responding to heartbeat requests";
        memberSuspected(localAddress, mbr, reason);
        initiateSuspicion(mbr, reason);
        setNextNeighbor(currentView, null);
      } else {
        logger.trace("Setting next neighbor as member {} has responded.", mbr);
        memberUnsuspected(mbr);
        // back to previous one
        setNextNeighbor(currentView, null);
      }
    });

  }

  private void initiateSuspicion(InternalDistributedMember mbr, String reason) {
    if (services.getJoinLeave().isMemberLeaving(mbr)) {
      return;
    }
    sendSuspectRequest(Collections.singletonList(new SuspectRequest(mbr, reason)));
  }

  /**
   * This method sends heartbeat request to other member and waits for member-timeout time for
   * response. If it doesn't see response then it returns false.
   */
  private boolean doCheckMember(InternalDistributedMember member, boolean waitForResponse) {
    if (playingDead || beingSick) {
      // a member playingDead should not be sending messages to other
      // members, so we avoid sending heartbeat requests or suspect
      // messages by returning true.
      return true;
    }
    long startTime = System.currentTimeMillis();
    logger.debug("Requesting heartbeat from {}", member);
    final HeartbeatRequestMessage hrm = constructHeartbeatRequestMessage(member);
    Response pingResp = null;
    if (waitForResponse) {
      pingResp = new Response();
      requestIdVsResponse.put(hrm.getRequestId(), pingResp);
    } else {
      hrm.clearRequestId();
    }
    try {
      Set<InternalDistributedMember> membersNotReceivedMsg = this.services.getMessenger().send(hrm);
      this.stats.incHeartbeatRequestsSent();
      if (membersNotReceivedMsg != null && membersNotReceivedMsg.contains(member)) {
        // member is not part of current view.
        logger.trace("Member {} is not part of current view.", member);
      } else if (waitForResponse) {
        synchronized (pingResp) {
          if (pingResp.getResponseMsg() == null) {
            pingResp.wait(memberTimeout);
          }
          TimeStamp ts = memberTimeStamps.get(member);
          if (ts != null && ts.getTime() > startTime) {
            return true;
          }
          if (pingResp.getResponseMsg() == null) {
            if (isStopping) {
              return true;
            }
            logger.debug("no heartbeat response received from {} and no recent activity", member);
            return false;
          } else {
            logger.trace("received heartbeat from {}", member);
            this.stats.incHeartbeatsReceived();
            if (ts != null) {
              ts.setTime(System.currentTimeMillis());
            }
            return true;
          }
        }
      }
    } catch (InterruptedException e) {
      logger.debug(
          "GMSHealthMonitor checking thread interrupted, while waiting for response from member: {} .",
          member);
    } finally {
      if (waitForResponse) {
        requestIdVsResponse.remove(hrm.getRequestId());
      }
    }
    return false;
  }

  /**
   * During final check, establish TCP connection between current member and suspect member. And
   * exchange PING/PONG message to see if the suspect member is still alive.
   *
   * @param suspectMember member that does not respond to HeartbeatRequestMessage
   * @return true if successfully exchanged PING/PONG with TCP connection, otherwise false.
   */
  boolean doTCPCheckMember(InternalDistributedMember suspectMember, int port,
      boolean retryIfConnectFails) {
    Socket clientSocket = null;
    // make sure we try to check on the member for the contracted memberTimeout period
    // in case a timed socket.connect() returns immediately
    long giveupTime = System.nanoTime() + TimeUnit.NANOSECONDS.convert(
        services.getConfig().getMemberTimeout(), TimeUnit.MILLISECONDS);
    boolean passed = false;
    int iteration = 0;
    do {
      iteration++;
      if (iteration > 1) {
        try {
          Thread.sleep(100);
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
          return false;
        }
      }
      try {
        logger.debug("Checking member {} with TCP socket connection {}:{}.", suspectMember,
            suspectMember.getInetAddress(), port);
        clientSocket =
            SocketCreatorFactory.getSocketCreatorForComponent(SecurableCommunicationChannel.CLUSTER)
                .connect(suspectMember.getInetAddress(), port, (int) memberTimeout,
                    new ConnectTimeoutTask(services.getTimer(), memberTimeout), false, -1, false);
        clientSocket.setTcpNoDelay(true);
        passed = doTCPCheckMember(suspectMember, clientSocket);
      } catch (IOException e) {
        // this is expected if it is a connection-timeout or other failure
        // to connect
      } catch (IllegalStateException | GemFireConfigException e) {
        if (!isStopping) {
          logger.trace("Unexpected exception", e);
        }
      } finally {
        try {
          if (clientSocket != null) {
            clientSocket.setSoLinger(true, 0); // abort the connection
            clientSocket.close();
          }
        } catch (IOException e) {
          // expected
        }
      }
    } while (retryIfConnectFails && !passed && !this.isShutdown()
        && System.nanoTime() < giveupTime);
    return passed;
  }

  // Package protected for testing purposes
  boolean doTCPCheckMember(InternalDistributedMember suspectMember, Socket clientSocket) {
    try {
      if (clientSocket.isConnected()) {
        clientSocket.setSoTimeout((int) services.getConfig().getMemberTimeout());
        InputStream in = clientSocket.getInputStream();
        DataOutputStream out = new DataOutputStream(clientSocket.getOutputStream());
        GMSMember gmbr = (GMSMember) suspectMember.getNetMember();
        writeMemberToStream(gmbr, out);
        this.stats.incFinalCheckRequestsSent();
        this.stats.incTcpFinalCheckRequestsSent();
        logger.debug("Connected to suspect member - reading response");
        int b = in.read();
        if (logger.isDebugEnabled()) {
          logger.debug("Received {}",
              (b == OK ? "OK" : (b == ERROR ? "ERROR" : "unknown response: " + b)));
        }
        if (b >= 0) {
          this.stats.incFinalCheckResponsesReceived();
          this.stats.incTcpFinalCheckResponsesReceived();
        }
        if (b == OK) {
          TimeStamp ts = memberTimeStamps.get(suspectMember);
          if (ts != null) {
            ts.setTime(System.currentTimeMillis());
          }
          return true;
        } else {
          // received ERROR
          return false;
        }
      } else {// cannot establish TCP connection with suspect member
        return false;
      }
    } catch (SocketTimeoutException e) {
      logger.debug("Availability check TCP/IP connection timed out for suspect member {}",
          suspectMember);
      return false;
    } catch (IOException e) {
      logger.trace("Unexpected exception", e);
    }
    return false;
  }

  void writeMemberToStream(GMSMember gmbr, DataOutputStream out) throws IOException {
    out.writeShort(Version.CURRENT_ORDINAL);
    out.writeInt(gmbr.getVmViewId());
    out.writeLong(gmbr.getUuidLSBs());
    out.writeLong(gmbr.getUuidMSBs());
    out.flush();
  }

  @Override
  public void suspect(InternalDistributedMember mbr, String reason) {
    initiateSuspicion(mbr, reason);
  }

  @Override
  public boolean checkIfAvailable(InternalDistributedMember mbr, String reason,
      boolean initiateRemoval) {
    return inlineCheckIfAvailable(localAddress, currentView, initiateRemoval,
        mbr, reason);
  }

  @Override
  public void start() {
    scheduler = LoggingExecutors.newScheduledThreadPool("Geode Failure Detection Scheduler", 1);
    checkExecutor = LoggingExecutors.newCachedThreadPool("Geode Failure Detection thread ", true);
    Monitor m = this.new Monitor(memberTimeout);
    monitorInterval = memberTimeout / LOGICAL_INTERVAL;
    monitorFuture =
        scheduler.scheduleAtFixedRate(m, monitorInterval, monitorInterval, TimeUnit.MILLISECONDS);
    serverSocketExecutor =
        LoggingExecutors.newCachedThreadPool("Geode Failure Detection Server thread ", true);
  }

  ServerSocket createServerSocket(InetAddress socketAddress, int[] portRange) {
    ServerSocket serverSocket;
    try {
      serverSocket = SocketCreatorFactory
          .getSocketCreatorForComponent(SecurableCommunicationChannel.CLUSTER)
          .createServerSocketUsingPortRange(socketAddress, 50/* backlog */, true/* isBindAddress */,
              false/* useNIO */, 65536/* tcpBufferSize */, portRange, false);
      socketPort = serverSocket.getLocalPort();
    } catch (IOException | SystemConnectException e) {
      throw new GemFireConfigException(
          "Unable to allocate a failure detection port in the membership-port range", e);
    }
    return serverSocket;
  }

  /**
   * start the thread that listens for tcp/ip connections and responds to connection attempts
   */
  private void startTcpServer(ServerSocket ssocket) {
    // allocate a socket here so there are no race conditions between knowing the FD
    // socket port and joining the system

    serverSocketExecutor.execute(() -> {
      logger.info("Started failure detection server thread on {}:{}.", ssocket.getInetAddress(),
          socketPort);
      Socket socket = null;
      try {
        while (!services.getCancelCriterion().isCancelInProgress()
            && !GMSHealthMonitor.this.isStopping) {
          try {
            socket = ssocket.accept();
            if (GMSHealthMonitor.this.playingDead) {
              continue;
            }
            serverSocketExecutor.execute(new ClientSocketHandler(socket));
          } catch (RejectedExecutionException e) {
            // this can happen during shutdown

          } catch (IOException e) {
            if (!isStopping) {
              logger.trace("Unexpected exception", e);
            }
            try {
              if (socket != null) {
                socket.close();
              }
            } catch (IOException ioe) {
              logger.trace("Unexpected exception", ioe);
            }
          }
        }
        logger.info("GMSHealthMonitor server thread exiting");
      } finally {
        // close the server socket
        if (!ssocket.isClosed()) {
          try {
            ssocket.close();
          } catch (IOException e) {
            logger.debug("Unexpected exception", e);
          }
        }
      }
    });
  }

  /**
   * start the thread that periodically sends a message to processes that might be watching this
   * process
   */
  private void startHeartbeatThread() {
    checkExecutor.execute(new Runnable() {
      @Override
      public void run() {
        Thread.currentThread().setName("Geode Heartbeat Sender");
        sendPeriodicHeartbeats();
      }

      private void sendPeriodicHeartbeats() {
        while (!isStopping && !services.getCancelCriterion().isCancelInProgress()) {
          try {
            Thread.sleep(memberTimeout / LOGICAL_INTERVAL);
          } catch (InterruptedException e) {
            return;
          }
          NetView v = currentView;
          if (v != null) {
            List<InternalDistributedMember> mbrs = v.getMembers();
            int index = mbrs.indexOf(localAddress);
            if (index < 0 || mbrs.size() < 2) {
              continue;
            }
            if (!playingDead) {
              sendHeartbeats(mbrs, index);
            }
          }
        }
      }

      private void sendHeartbeats(List<InternalDistributedMember> mbrs, int startIndex) {
        InternalDistributedMember coordinator = currentView.getCoordinator();
        if (coordinator != null && !coordinator.equals(localAddress)) {
          HeartbeatMessage message = new HeartbeatMessage(-1);
          message.setRecipient(coordinator);
          try {
            if (isStopping) {
              return;
            }
            services.getMessenger().sendUnreliably(message);
            GMSHealthMonitor.this.stats.incHeartbeatsSent();
          } catch (CancelException e) {
            return;
          }
        }

        int index = startIndex;
        int numSent = 0;
        for (;;) {
          index--;
          if (index < 0) {
            index = mbrs.size() - 1;
          }
          InternalDistributedMember mbr = mbrs.get(index);
          if (mbr.equals(localAddress)) {
            break;
          }
          if (mbr.equals(coordinator)) {
            continue;
          }
          if (isStopping) {
            return;
          }
          HeartbeatMessage message = new HeartbeatMessage(-1);
          message.setRecipient(mbr);
          try {
            services.getMessenger().sendUnreliably(message);
            GMSHealthMonitor.this.stats.incHeartbeatsSent();
            numSent++;
            if (numSent >= NUM_HEARTBEATS) {
              break;
            }
          } catch (CancelException e) {
            return;
          }
        }
      } // for (;;)
    });
  }

  @Override
  public synchronized void installView(NetView newView) {
    synchronized (suspectRequestsInView) {
      suspectRequestsInView.clear();
    }
    for (Iterator<InternalDistributedMember> it = memberTimeStamps.keySet().iterator(); it
        .hasNext();) {
      if (!newView.contains(it.next())) {
        it.remove();
      }
    }
    for (Iterator<InternalDistributedMember> it = suspectedMemberIds.keySet().iterator(); it
        .hasNext();) {
      if (!newView.contains(it.next())) {
        it.remove();
      }
    }
    currentView = newView;
    setNextNeighbor(newView, null);
  }

  /**
   * this method is primarily for tests. The current view should be pulled from JoinLeave or the
   * MembershipManager (which includes surprise members)
   */
  public synchronized NetView getView() {
    return currentView;
  }

  /***
   * This method sets next neighbour which it needs to watch in current view.
   *
   * if nextTo == null then it watches member next to it.
   *
   * It becomes null when we suspect current neighbour, during that time it watches member next to
   * suspect member.
   */
  protected synchronized void setNextNeighbor(NetView newView, InternalDistributedMember nextTo) {
    if (newView == null) {
      return;
    }
    if (nextTo == null) {
      nextTo = localAddress;
    }

    List<InternalDistributedMember> allMembers = newView.getMembers();

    if (allMembers.size() > 1 && suspectedMemberIds.size() >= allMembers.size() - 1) {
      boolean nonSuspectFound = false;
      for (InternalDistributedMember member : allMembers) {
        if (member.equals(localAddress)) {
          continue;
        }
        if (!suspectedMemberIds.containsKey(member)) {
          nonSuspectFound = true;
          break;
        }
      }
      if (!nonSuspectFound) {
        logger.info("All other members are suspect at this point");
        nextNeighbor = null;
        return;
      }
    }

    int index = allMembers.indexOf(nextTo);
    if (index != -1) {
      int nextNeighborIndex = (index + 1) % allMembers.size();
      InternalDistributedMember newNeighbor = allMembers.get(nextNeighborIndex);
      if (suspectedMemberIds.containsKey(newNeighbor)) {
        setNextNeighbor(newView, newNeighbor);
        return;
      }
      InternalDistributedMember oldNeighbor = nextNeighbor;
      if (oldNeighbor != newNeighbor) {
        logger.debug("Failure detection is now watching " + newNeighbor);
        nextNeighbor = newNeighbor;
      }
    }

    if (nextNeighbor != null && nextNeighbor.equals(localAddress)) {
      if (logger.isDebugEnabled()) {
        logger.debug("Health monitor is unable to find a neighbor to watch.  "
            + "Current suspects are {}", suspectedMemberIds);
      }
      nextNeighbor = null;
    }

  }

  /** test method */
  public InternalDistributedMember getNextNeighbor() {
    return nextNeighbor;
  }

  @Override
  public void init(Services s) {
    isStopping = false;
    services = s;
    memberTimeout = s.getConfig().getMemberTimeout();
    this.stats = services.getStatistics();

    services.getMessenger().addHandler(HeartbeatRequestMessage.class,
        this::processMessage);
    services.getMessenger().addHandler(HeartbeatMessage.class,
        this::processMessage);
    services.getMessenger().addHandler(SuspectMembersMessage.class,
        this::processMessage);
    services.getMessenger().addHandler(FinalCheckPassedMessage.class,
        this::processMessage);
  }

  @Override
  public void started() {
    setLocalAddress(services.getMessenger().getMemberID());
    serverSocket = createServerSocket(localAddress.getInetAddress(),
        services.getConfig().getMembershipPortRange());
    startTcpServer(serverSocket);
    startHeartbeatThread();
  }

  @Override
  public void stop() {
    stopServices();
  }

  private void stopServices() {
    logger.debug("Stopping HealthMonitor");
    isStopping = true;
    if (monitorFuture != null) {
      monitorFuture.cancel(true);
    }
    if (scheduler != null) {
      scheduler.shutdown();
    }

    Collection<Response> val = requestIdVsResponse.values();
    for (Response r : val) {
      synchronized (r) {
        r.notify();
      }
    }

    if (checkExecutor != null) {
      checkExecutor.shutdown();
    }

    stopServer();
  }

  void stopServer() {
    if (serverSocketExecutor != null) {
      if (serverSocket != null && !serverSocket.isClosed()) {
        try {
          serverSocket.close();
        } catch (IOException e) {
          logger.trace("Unexpected exception", e);
        }
      }
      serverSocketExecutor.shutdownNow();
      try {
        serverSocketExecutor.awaitTermination(2000, TimeUnit.MILLISECONDS);
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
      }
    }
  }

  /***
   * test method
   */
  public boolean isShutdown() {
    return scheduler.isShutdown() && checkExecutor.isShutdown()
        && serverSocketExecutor.isShutdown();
  }

  /**
   * Test method - check to see if a member is under suspicion
   */
  public boolean isSuspectMember(InternalDistributedMember m) {
    return this.suspectedMemberIds.containsKey(m);
  }

  @Override
  public void stopped() {

  }

  @Override
  public void memberSuspected(InternalDistributedMember initiator,
      InternalDistributedMember suspect, String reason) {
    synchronized (suspectRequestsInView) {
      suspectedMemberIds.put(suspect, currentView);
      Collection<SuspectRequest> requests = suspectRequestsInView.get(currentView);
      boolean found = false;
      if (requests == null) {
        requests = new HashSet<>();
        requests.add(new SuspectRequest(suspect, reason));
      }
      for (SuspectRequest request : requests) {
        if (suspect.equals(request.getSuspectMember())) {
          found = true;
          break;
        }
      }
      if (!found) {
        requests.add(new SuspectRequest(suspect, reason));
      }
    }
  }

  private void memberUnsuspected(InternalDistributedMember mbr) {
    synchronized (suspectRequestsInView) {
      if (suspectedMemberIds.remove(mbr) != null) {
        logger.info("No longer suspecting {}", mbr);
      }
      Collection<SuspectRequest> suspectRequests = suspectRequestsInView.get(currentView);
      if (suspectRequests != null) {
        Collection<SuspectRequest> removals = new ArrayList<>(suspectRequests.size());
        for (SuspectRequest suspectRequest : suspectRequests) {
          if (mbr.equals(suspectRequest.getSuspectMember())) {
            removals.add(suspectRequest);
          }
        }
        suspectRequests.removeAll(removals);
      }
    }
  }

  @Override
  public void beSick() {
    this.beingSick = true;
  }

  @Override
  public void playDead() {
    this.playingDead = true;
  }

  @Override
  public void beHealthy() {
    this.beingSick = false;
    this.playingDead = false;
  }

  @Override
  public void emergencyClose() {
    stopServices();
  }

  @Override
  public void setLocalAddress(InternalDistributedMember idm) {
    this.localAddress = idm;
  }

  void processMessage(HeartbeatRequestMessage m) {
    if (isStopping) {
      return;
    }
    if (beingSick || playingDead) {
      logger.debug("sick member is ignoring check request");
      return;
    }

    this.stats.incHeartbeatRequestsReceived();

    if (this.isStopping) {
      return;
    }

    // only respond if the intended recipient is this member
    InternalDistributedMember me = localAddress;

    if (me == null || me.getVmViewId() >= 0 && m.getTarget().equals(me)) {
      HeartbeatMessage hm = new HeartbeatMessage(m.getRequestId());
      hm.setRecipient(m.getSender());
      Set<InternalDistributedMember> membersNotReceivedMsg = services.getMessenger().send(hm);
      this.stats.incHeartbeatsSent();
      if (membersNotReceivedMsg != null && membersNotReceivedMsg.contains(m.getSender())) {
        logger.debug("Unable to send heartbeat to member: {}", m.getSender());
      }
    } else {
      logger.debug("Ignoring heartbeat request intended for {}.  My ID is {}", m.getTarget(), me);
    }
  }


  void processMessage(HeartbeatMessage m) {
    if (isStopping) {
      return;
    }
    if (beingSick || playingDead) {
      logger.debug("sick member is ignoring check response");
      return;
    }

    this.stats.incHeartbeatsReceived();
    if (m.getRequestId() >= 0) {
      Response resp = requestIdVsResponse.get(m.getRequestId());
      logger.trace("Got heartbeat from member {}. {}", m.getSender(),
          (resp != null ? "Check thread still waiting" : "Check thread is not waiting"));
      if (resp != null) {
        synchronized (resp) {
          resp.setResponseMsg(m);
          resp.notify();
        }
      }

    }
    // we got heartbeat lets update timestamp
    contactedBy(m.getSender(), System.currentTimeMillis());
  }

  /**
   * Process a Suspect request from another member. This may cause this member to become the new
   * membership coordinator. it will to final check on that member and then it will send remove
   * request for that member
   */
  void processMessage(SuspectMembersMessage incomingRequest) {
    if (isStopping) {
      return;
    }
    if (beingSick || playingDead) {
      logger.debug("sick member is ignoring suspect message");
      return;
    }

    this.stats.incSuspectsReceived();

    NetView cv = currentView;

    if (cv == null) {
      return;
    }

    List<SuspectRequest> suspectRequests = incomingRequest.getMembers();

    InternalDistributedMember sender = incomingRequest.getSender();
    int viewId = sender.getVmViewId();
    if (cv.getViewId() >= viewId && !cv.contains(incomingRequest.getSender())) {
      logger.info("Membership ignoring suspect request for " + incomingRequest + " from non-member "
          + incomingRequest.getSender());
      services.getJoinLeave().remove(sender,
          "this process is initiating suspect processing but is no longer a member");
      return;
    }

    // take care of any suspicion of this member by sending a heartbeat back
    for (Iterator<SuspectRequest> it = incomingRequest.getMembers().iterator(); it.hasNext();) {
      SuspectRequest req = it.next();
      if (req.getSuspectMember().equals(localAddress)) {
        HeartbeatMessage message = new HeartbeatMessage(-1);
        message.setRecipient(sender);
        try {
          services.getMessenger().send(message);
          this.stats.incHeartbeatsSent();
          it.remove();
        } catch (CancelException e) {
          return;
        }
      }
    }

    logger.debug(
        "Processing suspect requests {}\nproposed view is currently {}\nwith coordinator {}",
        suspectRequests, cv, cv.getCoordinator());
    if (cv.getCoordinator().equals(localAddress)) {
      // This process is the membership coordinator and should perform a final check
      logSuspectRequests(incomingRequest, sender);
      checkIfAvailable(sender, suspectRequests, cv);

    } else {
      // Another process has raised suspicion - check to see if
      // this process should become the membership coordinator if
      // all current suspects are gone
      NetView check = new NetView(cv, cv.getViewId() + 1);
      ArrayList<SuspectRequest> membersToCheck = new ArrayList<>();
      synchronized (suspectRequestsInView) {
        recordSuspectRequests(suspectRequests, cv);
        Set<SuspectRequest> suspectsInView = suspectRequestsInView.get(cv);
        logger.debug("Current suspects are {}", suspectsInView);
        for (final SuspectRequest sr : suspectsInView) {
          check.remove(sr.getSuspectMember());
          membersToCheck.add(sr);
        }
      }
      List membersLeaving = new ArrayList();
      for (InternalDistributedMember member : cv.getMembers()) {
        if (services.getJoinLeave().isMemberLeaving(member)) {
          membersLeaving.add(member);
        }
      }
      if (!membersLeaving.isEmpty()) {
        logger.debug("Current leave requests are {}", membersLeaving);
        check.removeAll(membersLeaving);
      }
      logger.trace(
          "Proposed view with suspects & leaving members removed is {}\nwith coordinator {}\nmy address is {}",
          check,
          check.getCoordinator(), localAddress);

      InternalDistributedMember coordinator = check.getCoordinator();
      if (coordinator != null && coordinator.equals(localAddress)) {
        // new coordinator
        logSuspectRequests(incomingRequest, sender);
        checkIfAvailable(sender, membersToCheck, cv);
      }
    }

  }

  void processMessage(FinalCheckPassedMessage m) {
    if (isStopping) {
      return;
    }
    contactedBy(m.getSuspect());
  }


  private void logSuspectRequests(SuspectMembersMessage incomingRequest,
      InternalDistributedMember sender) {
    for (SuspectRequest req : incomingRequest.getMembers()) {
      String who = sender.equals(localAddress) ? "myself" : sender.toString();
      logger.info("received suspect message from {} for {}: {}", who, req.getSuspectMember(),
          req.getReason());
    }
  }

  /***
   * This method make sure that records suspectRequest. We need to make sure this on preferred
   * coordinators, as elder coordinator might be in suspected list next.
   */
  private void recordSuspectRequests(List<SuspectRequest> suspectRequests, NetView cv) {
    // record suspect requests
    Set<SuspectRequest> suspectedMembers;
    synchronized (suspectRequestsInView) {
      suspectedMembers = suspectRequestsInView.get(cv);
      if (suspectedMembers == null) {
        suspectedMembers = new HashSet<>();
        suspectRequestsInView.put(cv, suspectedMembers);
      }
      suspectedMembers.addAll(suspectRequests);
    }
  }

  /**
   * performs a "final" health check on the member. If failure-detection socket information is
   * available for the member (in the view) then we attempt to connect to its socket and ask if it's
   * the expected member. Otherwise we send a heartbeat request and wait for a reply.
   */
  private void checkIfAvailable(final InternalDistributedMember initiator,
      List<SuspectRequest> sMembers, final NetView cv) {

    for (final SuspectRequest sr : sMembers) {
      final InternalDistributedMember mbr = sr.getSuspectMember();

      if (!cv.contains(mbr) || membersInFinalCheck.contains(mbr)) {
        continue;
      }

      if (mbr.equals(localAddress)) {
        continue;// self
      }

      final String reason = sr.getReason();
      logger.debug("Scheduling availability check for member {}; reason={}", mbr, reason);
      // its a coordinator
      checkExecutor.execute(() -> {
        try {
          inlineCheckIfAvailable(initiator, cv, true, mbr, reason);
        } catch (CancelException e) {
          // shutting down
        } catch (Exception e) {
          logger.info("Unexpected exception while verifying member", e);
        }
      });
    }
  }

  protected boolean inlineCheckIfAvailable(final InternalDistributedMember initiator,
      final NetView cv, boolean forceRemovalIfCheckFails, final InternalDistributedMember mbr,
      final String reason) {

    if (services.getJoinLeave().isMemberLeaving(mbr)) {
      return false;
    }

    boolean failed = false;

    logger.info("Performing availability check for suspect member {} reason={}", mbr, reason);
    membersInFinalCheck.add(mbr);
    setNextNeighbor(currentView, mbr);

    try {
      services.memberSuspected(initiator, mbr, reason);
      long startTime = System.currentTimeMillis();
      // for some reason we used to update the timestamp for the member
      // with the startTime, but we don't want to do that because it looks
      // like a heartbeat has been received

      boolean pinged;
      int port = cv.getFailureDetectionPort(mbr);
      if (port <= 0) {
        logger.info("Unable to locate failure detection port - requesting a heartbeat");
        if (logger.isDebugEnabled()) {
          logger.debug("\ncurrent view: {}\nports: {}", cv,
              Arrays.toString(cv.getFailureDetectionPorts()));
        }
        pinged = GMSHealthMonitor.this.doCheckMember(mbr, true);
        GMSHealthMonitor.this.stats.incFinalCheckRequestsSent();
        GMSHealthMonitor.this.stats.incUdpFinalCheckRequestsSent();
        if (pinged) {
          GMSHealthMonitor.this.stats.incFinalCheckResponsesReceived();
          GMSHealthMonitor.this.stats.incUdpFinalCheckResponsesReceived();
        }
      } else {
        // this will just send heartbeat request, it will not wait for response
        // if we will get heartbeat then it will change the timestamp, which we are
        // checking below in case of tcp check failure..
        doCheckMember(mbr, false);
        // now, while waiting for a heartbeat, try connecting to the suspect's failure detection
        // port
        final boolean retryIfConnectFails = forceRemovalIfCheckFails;
        pinged = doTCPCheckMember(mbr, port, retryIfConnectFails);
      }

      if (!pinged && !isStopping) {
        failed = true;
        TimeStamp ts = memberTimeStamps.get(mbr);
        if (ts == null || ts.getTime() < startTime) {
          logger.info("Availability check failed for member {}", mbr);
          // if the final check fails & this VM is the coordinator we don't need to do another final
          // check
          if (forceRemovalIfCheckFails) {
            logger.info("Requesting removal of suspect member {}", mbr);
            services.getJoinLeave().remove(mbr, reason);
            // make sure it is still suspected
            memberSuspected(localAddress, mbr, reason);
          } else {
            // if this node can survive an availability check then initiate suspicion about
            // the node that failed the availability check
            if (doTCPCheckMember(localAddress, this.socketPort, false)) {
              membersInFinalCheck.remove(mbr);
              // tell peers about this member and then perform another availability check
              memberSuspected(localAddress, mbr, reason);
              initiateSuspicion(mbr, reason);
              SuspectMembersMessage suspectMembersMessage =
                  new SuspectMembersMessage(Collections.singletonList(localAddress),
                      Collections
                          .singletonList(new SuspectRequest(mbr, "failed availability check")));
              suspectMembersMessage.setSender(localAddress);
              logger.debug("Performing local processing on suspect request");
              processMessage(suspectMembersMessage);
            } else {
              logger.info(
                  "Self-check for availability failed - will not continue to suspect {} for now",
                  mbr);
              failed = false;
            }
          }
        } else {
          logger.info(
              "Availability check failed but detected recent message traffic for suspect member "
                  + mbr);
          failed = false;
        }
      }

      if (!failed) {
        if (!isStopping && !initiator.equals(localAddress)
            && initiator.getVersionObject().compareTo(Version.GEODE_1_3_0) >= 0) {
          // let the sender know that it's okay to monitor this member again
          FinalCheckPassedMessage message = new FinalCheckPassedMessage(initiator, mbr);
          services.getMessenger().send(message);
        }

        logger.info("Availability check passed for suspect member " + mbr);
      }
    } finally {
      if (!failed) {
        memberUnsuspected(mbr);
        setNextNeighbor(currentView, null);
      }
      membersInFinalCheck.remove(mbr);
    }
    return !failed;
  }

  @Override
  public void memberShutdown(DistributedMember mbr, String reason) {}

  @Override
  public int getFailureDetectionPort() {
    return this.socketPort;
  }

  private void sendSuspectRequest(final List<SuspectRequest> requests) {
    logger.debug("Sending suspect request for members {}", requests);
    List<InternalDistributedMember> recipients;
    if (currentView.size() > ServiceConfig.SMALL_CLUSTER_SIZE) {
      HashSet<InternalDistributedMember> filter = new HashSet<>();
      for (Enumeration<InternalDistributedMember> e = suspectedMemberIds.keys(); e
          .hasMoreElements();) {
        filter.add(e.nextElement());
      }
      filter.addAll(
          requests.stream().map(SuspectRequest::getSuspectMember).collect(Collectors.toList()));
      recipients =
          currentView.getPreferredCoordinators(filter, services.getJoinLeave().getMemberID(),
              ServiceConfig.SMALL_CLUSTER_SIZE + 1);
    } else {
      recipients = currentView.getMembers();
    }

    logger.trace("Sending suspect messages to {}", recipients);
    SuspectMembersMessage smm = new SuspectMembersMessage(recipients, requests);
    smm.setSender(localAddress);
    Set<InternalDistributedMember> failedRecipients;
    try {
      failedRecipients = services.getMessenger().send(smm);
      this.stats.incSuspectsSent();
    } catch (CancelException e) {
      return;
    }

    if (failedRecipients != null && failedRecipients.size() > 0) {
      logger.trace("Unable to send suspect message to {}", failedRecipients);
    }
    logger.trace("Processing suspect message locally");
    processMessage(smm);
  }

  private static class ConnectTimeoutTask extends TimerTask implements ConnectionWatcher {

    final Timer scheduler;
    Socket socket;
    final long timeout;

    ConnectTimeoutTask(Timer scheduler, long timeout) {
      this.scheduler = scheduler;
      this.timeout = timeout;
    }

    @Override
    public void beforeConnect(Socket socket) {
      this.socket = socket;
      scheduler.schedule(this, timeout);
    }

    @Override
    public void afterConnect(Socket socket) {
      cancel();
    }

    @Override
    public void run() {
      try {
        if (socket != null) {
          socket.close();
        }
      } catch (IOException e) {
        // ignored - nothing useful to do here
      }
    }

  }

  public DMStats getStats() {
    return this.stats;
  }
}