/*
 * 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.internal.tcp;

import java.io.IOException;
import java.lang.ref.Reference;
import java.lang.ref.WeakReference;
import java.net.InetAddress;
import java.net.Socket;
import java.util.ArrayList;
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.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicReference;

import org.apache.logging.log4j.Logger;

import org.apache.geode.SystemFailure;
import org.apache.geode.alerting.internal.spi.AlertingAction;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.DistributedSystemDisconnectedException;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.distributed.internal.membership.MembershipManager;
import org.apache.geode.distributed.internal.membership.adapter.GMSMembershipManager;
import org.apache.geode.distributed.internal.membership.gms.api.Membership;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.SystemTimer;
import org.apache.geode.internal.logging.CoreLoggingExecutors;
import org.apache.geode.internal.net.BufferPool;
import org.apache.geode.internal.net.SocketCloser;
import org.apache.geode.logging.internal.executors.LoggingExecutors;
import org.apache.geode.logging.internal.log4j.api.LogService;

/**
 * <p>
 * ConnectionTable holds all of the Connection objects in a conduit. Connections represent a pipe
 * between two endpoints represented by generic DistributedMembers.
 * </p>
 *
 * @since GemFire 2.1
 */
public class ConnectionTable {
  private static final Logger logger = LogService.getLogger();

  /** warning when descriptor limit reached */
  @MakeNotStatic
  private static boolean ulimitWarningIssued;

  /**
   * true if the current thread wants non-shared resources
   */
  @MakeNotStatic
  private static ThreadLocal threadWantsOwnResources = new ThreadLocal();

  /**
   * Used for messages whose order must be preserved Only connections used for sending messages, and
   * receiving acks, will be put in this map.
   */
  protected final Map orderedConnectionMap = new ConcurrentHashMap();

  /**
   * ordered connections local to this thread. Note that accesses to the resulting map must be
   * synchronized because of static cleanup.
   */
  // ThreadLocal<Map>
  protected final ThreadLocal<Map> threadOrderedConnMap;

  /**
   * List of thread-owned ordered connection maps, for cleanup
   *
   * Accesses to the maps in this list need to be synchronized on their instance.
   */
  private final List threadConnMaps;

  /**
   * Timer to kill idle threads
   *
   * guarded.By this
   */
  private SystemTimer idleConnTimer;

  /**
   * Used to find connections owned by threads. The key is the same one used in
   * threadOrderedConnMap. The value is an ArrayList since we can have any number of connections
   * with the same key.
   */
  private ConcurrentMap threadConnectionMap;

  /**
   * Used for all non-ordered messages. Only connections used for sending messages, and receiving
   * acks, will be put in this map.
   */
  protected final Map unorderedConnectionMap = new ConcurrentHashMap();

  /**
   * Used for all accepted connections. These connections are read only; we never send messages,
   * except for acks; only receive.
   *
   * Consists of a list of Connection
   */
  private final List receivers = new ArrayList();

  /**
   * the conduit for this table
   */
  private final TCPConduit owner;

  private final BufferPool bufferPool;

  /**
   * true if this table is no longer in use
   */
  private volatile boolean closed = false;

  /**
   * Executor used by p2p reader and p2p handshaker threads.
   */
  private final Executor p2pReaderThreadPool;
  /**
   * Number of seconds to wait before timing out an unused p2p reader thread. Default is 120 (2
   * minutes).
   */
  private static final long READER_POOL_KEEP_ALIVE_TIME =
      Long.getLong("p2p.READER_POOL_KEEP_ALIVE_TIME", 120).longValue();

  private final SocketCloser socketCloser;

  /**
   * The most recent instance to be created
   *
   * TODO this assumes no more than one instance is created at a time?
   */
  @MakeNotStatic
  private static final AtomicReference lastInstance = new AtomicReference();

  /**
   * A set of sockets that are in the process of being connected
   */
  private Map connectingSockets = new HashMap();

  /**
   * Cause calling thread to share communication resources with other threads.
   */
  public static void threadWantsSharedResources() {
    threadWantsOwnResources.set(Boolean.FALSE);
  }

  /**
   * Cause calling thread to acquire exclusive access to communication resources. Exclusive access
   * may not be available in which case this call is ignored.
   */
  public static void threadWantsOwnResources() {
    threadWantsOwnResources.set(Boolean.TRUE);
  }

  /**
   * Returns true if calling thread owns its own communication resources.
   */
  boolean threadOwnsResources() {
    DistributionManager d = getDM();
    if (d != null) {
      return d.getSystem().threadOwnsResources() && !AlertingAction.isThreadAlerting();
    }
    return false;
  }

  public static Boolean getThreadOwnsResourcesRegistration() {
    return (Boolean) threadWantsOwnResources.get();
  }

  public TCPConduit getOwner() {
    return owner;
  }


  private ConnectionTable(TCPConduit conduit) throws IOException {
    this.owner = conduit;
    this.idleConnTimer = (this.owner.idleConnectionTimeout != 0)
        ? new SystemTimer(conduit.getDM().getSystem(), true) : null;
    this.threadOrderedConnMap = new ThreadLocal();
    this.threadConnMaps = new ArrayList();
    this.threadConnectionMap = new ConcurrentHashMap();
    this.p2pReaderThreadPool = createThreadPoolForIO(conduit.getDM().getSystem().isShareSockets());
    this.socketCloser = new SocketCloser();
    this.bufferPool = new BufferPool(owner.getStats());
  }

  private Executor createThreadPoolForIO(boolean conserveSockets) {
    if (conserveSockets) {
      return LoggingExecutors.newThreadOnEachExecute("SharedP2PReader");
    } else {
      return CoreLoggingExecutors.newThreadPoolWithSynchronousFeed("UnsharedP2PReader", 1,
          Integer.MAX_VALUE, READER_POOL_KEEP_ALIVE_TIME);
    }
  }

  /** conduit calls acceptConnection after an accept */
  protected void acceptConnection(Socket sock, PeerConnectionFactory peerConnectionFactory)
      throws IOException, ConnectionException, InterruptedException {
    InetAddress connAddress = sock.getInetAddress(); // for bug 44736
    boolean finishedConnecting = false;
    Connection connection = null;
    // boolean exceptionLogged = false;
    try {
      connection = peerConnectionFactory.createReceiver(this, sock);

      // check for shutdown (so it doesn't get missed in the finally block)
      this.owner.getCancelCriterion().checkCancelInProgress(null);
      finishedConnecting = true;
    } catch (IOException ex) {
      // check for shutdown...
      this.owner.getCancelCriterion().checkCancelInProgress(ex);
      logger.warn(String.format("Failed to accept connection from %s because: %s",
          new Object[] {(connAddress != null ? connAddress : "unavailable address"), ex}));
      throw ex;
    } catch (ConnectionException ex) {
      // check for shutdown...
      this.owner.getCancelCriterion().checkCancelInProgress(ex);
      logger.warn(String.format("Failed to accept connection from %s because: %s",
          new Object[] {(connAddress != null ? connAddress : "unavailable address"), ex}));
      throw ex;
    } finally {
      // note: no need to call incFailedAccept here because it will be done
      // in our caller.
      // no need to log error here since caller will log warning

      if (connection != null && !finishedConnecting) {
        // we must be throwing from checkCancelInProgress so close the connection
        closeCon("cancel after accept",
            connection);
        connection = null;
      }
    }

    if (connection != null) {
      synchronized (this.receivers) {
        this.owner.getStats().incReceivers();
        if (this.closed) {
          closeCon("Connection table no longer in use", connection);
          return;
        }
        // If connection.stopped is false, any connection cleanup thread will not yet have acquired
        // the receiver synchronization to remove the receiver. Therefore we can safely add it here.
        if (!(connection.isSocketClosed() || connection.isReceiverStopped())) {
          this.receivers.add(connection);
        }
      }
      if (logger.isDebugEnabled()) {
        logger.debug("Accepted {} myAddr={} theirAddr={}", connection, getConduit().getMemberId(),
            connection.remoteAddr);
      }
    }
  }


  /**
   * Process a newly created PendingConnection
   *
   * @param id DistributedMember on which the connection is created
   * @param sharedResource whether the connection is used by multiple threads
   * @param preserveOrder whether to preserve order
   * @param m map to add the connection to
   * @param pc the PendingConnection to process
   * @param startTime the ms clock start time for the operation
   * @param ackThreshold the ms ack-wait-threshold, or zero
   * @param ackSAThreshold the ms ack-severe_alert-threshold, or zero
   * @return the Connection, or null if someone else already created or closed it
   * @throws IOException if unable to connect
   */
  private Connection handleNewPendingConnection(DistributedMember id, boolean sharedResource,
      boolean preserveOrder, Map m, PendingConnection pc, long startTime, long ackThreshold,
      long ackSAThreshold) throws IOException, DistributedSystemDisconnectedException {
    // handle new pending connection
    Connection con = null;
    try {
      con = Connection.createSender(owner.getMembershipManager(), this, preserveOrder, id,
          sharedResource, startTime, ackThreshold, ackSAThreshold);
      this.owner.getStats().incSenders(sharedResource, preserveOrder);
    } finally {
      // our connection failed to notify anyone waiting for our pending con
      if (con == null) {
        this.owner.getStats().incFailedConnect();
        synchronized (m) {
          Object rmObj = m.remove(id);
          if (rmObj != pc && rmObj != null) {
            // put it back since it was not our pc
            m.put(id, rmObj);
          }
        }
        pc.notifyWaiters(null);
        // we must be throwing an exception
      }
    } // finally

    // Update our list of connections -- either the
    // orderedConnectionMap or unorderedConnectionMap
    //
    // Note that we added the entry _before_ we attempted the connect,
    // so it's possible something else got through in the mean time...
    synchronized (m) {
      Object e = m.get(id);
      if (e == pc) {
        m.put(id, con);
      } else if (e == null) {
        // someone closed our pending connection
        // so cleanup the connection we created
        con.requestClose(
            "pending connection cancelled");
        con = null;
      } else {
        if (e instanceof Connection) {
          Connection newCon = (Connection) e;
          if (!newCon.connected) {
            // Fix for bug 31590
            // someone closed our pending connect
            // so cleanup the connection we created
            if (con != null) {
              con.requestClose(
                  "pending connection closed");
              con = null;
            }
          } else {
            // This should not happen. It means that someone else
            // created the connection which should only happen if
            // our Connection was rejected.
            // Assert.assertTrue(false);
            // The above assertion was commented out to try the
            // following with bug 32680
            if (con != null) {
              con.requestClose("someone else created the connection");
            }
            con = newCon;
          }
        }
      }
    }
    pc.notifyWaiters(con);
    if (con != null && logger.isDebugEnabled()) {
      logger.debug("handleNewPendingConnection {} myAddr={} theirAddr={}", con,
          getConduit().getMemberId(), con.remoteAddr);
    }

    return con;
  }

  /**
   * unordered or conserve-sockets=true note that unordered connections are currently always shared
   *
   * @param id the DistributedMember on which we are creating a connection
   * @param scheduleTimeout whether unordered connection should time out
   * @param preserveOrder whether to preserve order
   * @param startTime the ms clock start time for the operation
   * @param ackTimeout the ms ack-wait-threshold, or zero
   * @param ackSATimeout the ms ack-severe-alert-threshold, or zero
   * @return the new Connection, or null if an error
   * @throws IOException if unable to create the connection
   */
  private Connection getSharedConnection(DistributedMember id, boolean scheduleTimeout,
      boolean preserveOrder, long startTime, long ackTimeout, long ackSATimeout)
      throws IOException, DistributedSystemDisconnectedException {
    Connection result = null;

    final Map m = preserveOrder ? this.orderedConnectionMap : this.unorderedConnectionMap;

    PendingConnection pc = null; // new connection, if needed
    Object mEntry = null; // existing connection (if we don't create a new one)

    // Look for pending connection
    synchronized (m) {
      mEntry = m.get(id);
      if (mEntry != null && (mEntry instanceof Connection)) {
        Connection existingCon = (Connection) mEntry;
        if (!existingCon.connected) {
          mEntry = null;
        }
      }
      if (mEntry == null) {
        pc = new PendingConnection(preserveOrder, id);
        m.put(id, pc);
      }
    } // synchronized

    if (pc != null) {
      if (logger.isDebugEnabled()) {
        logger.debug("created PendingConnection {}", pc);
      }
      result = handleNewPendingConnection(id, true /* fixes bug 43386 */, preserveOrder, m, pc,
          startTime, ackTimeout, ackSATimeout);
      if (!preserveOrder && scheduleTimeout) {
        scheduleIdleTimeout(result);
      }
    } else { // we have existing connection
      if (mEntry instanceof PendingConnection) {

        if (AlertingAction.isThreadAlerting()) {
          // do not change the text of this exception - it is looked for in exception handlers
          throw new IOException("Cannot form connection to alert listener " + id);
        }

        result = ((PendingConnection) mEntry).waitForConnect(this.owner.getMembershipManager(),
            startTime, ackTimeout, ackSATimeout);
        if (logger.isDebugEnabled()) {
          if (result != null) {
            logger.debug("getSharedConnection {} myAddr={} theirAddr={}", result,
                getConduit().getMemberId(), result.remoteAddr);
          } else {
            logger.debug("getSharedConnection: Connect failed");
          }
        }
      } else {
        result = (Connection) mEntry;
      }
    } // we have existing connection

    return result;
  }

  /**
   * Must be looking for an ordered connection that this thread owns
   *
   * @param id stub on which to create the connection
   * @param startTime the ms clock start time for the operation
   * @param ackTimeout the ms ack-wait-threshold, or zero
   * @param ackSATimeout the ms ack-severe-alert-threshold, or zero
   * @return the connection, or null if an error
   * @throws IOException if the connection could not be created
   */
  Connection getThreadOwnedConnection(DistributedMember id, long startTime, long ackTimeout,
      long ackSATimeout) throws IOException, DistributedSystemDisconnectedException {
    Connection result = null;

    // Look for result in the thread local
    Map m = (Map) this.threadOrderedConnMap.get();
    if (m == null) {
      // First time for this thread. Create thread local
      m = new HashMap();
      synchronized (this.threadConnMaps) {
        if (this.closed) {
          owner.getCancelCriterion().checkCancelInProgress(null);
          throw new DistributedSystemDisconnectedException(
              "Connection table is closed");
        }
        // check for stale references and remove them.
        for (Iterator it = this.threadConnMaps.iterator(); it.hasNext();) {
          Reference r = (Reference) it.next();
          if (r.get() == null) {
            it.remove();
          }
        } // for
        this.threadConnMaps.add(new WeakReference(m)); // ref added for bug 38011
      } // synchronized
      this.threadOrderedConnMap.set(m);
    } else {
      // Consult thread local.
      synchronized (m) {
        result = (Connection) m.get(id);
      }
      if (result != null && result.timedOut) {
        result = null;
      }
    }
    if (result != null)
      return result;

    // OK, we have to create a new connection.
    result = Connection.createSender(owner.getMembershipManager(), this, true /* preserveOrder */,
        id, false /* shared */, startTime, ackTimeout, ackSATimeout);
    if (logger.isDebugEnabled()) {
      logger.debug("ConnectionTable: created an ordered connection: {}", result);
    }
    this.owner.getStats().incSenders(false/* shared */, true /* preserveOrder */);

    // Update the list of connections owned by this thread....

    if (this.threadConnectionMap == null) {
      // This instance is being destroyed; fail the operation
      closeCon(
          "Connection table being destroyed",
          result);
      return null;
    }

    ArrayList al = (ArrayList) this.threadConnectionMap.get(id);
    if (al == null) {
      // First connection for this DistributedMember. Make sure list for this
      // stub is created if it isn't already there.
      al = new ArrayList();

      // Since it's a concurrent map, we just try to put it and then
      // return whichever we got.
      Object o = this.threadConnectionMap.putIfAbsent(id, al);
      if (o != null) {
        al = (ArrayList) o;
      }
    }

    // Add our Connection to the list
    synchronized (al) {
      al.add(result);
    }

    // Finally, add the connection to our thread local map.
    synchronized (m) {
      m.put(id, result);
    }

    scheduleIdleTimeout(result);
    return result;
  }

  /** schedule an idle-connection timeout task */
  private void scheduleIdleTimeout(Connection conn) {
    if (conn == null) {
      // fix for bug 43529
      return;
    }
    // Set the idle timeout
    if (this.owner.idleConnectionTimeout != 0) {
      try {
        synchronized (this) {
          if (!this.closed) {
            IdleConnTT task = new IdleConnTT(conn);
            conn.setIdleTimeoutTask(task);
            this.getIdleConnTimer().scheduleAtFixedRate(task, this.owner.idleConnectionTimeout,
                this.owner.idleConnectionTimeout);
          }
        }
      } catch (IllegalStateException e) {
        if (conn.isClosing()) {
          // bug #45077 - connection is closed before we schedule the timeout task,
          // causing the task to be canceled
          return;
        }
        logger.debug("Got an illegal state exception: {}", e.getMessage(), e);
        // Unfortunately, cancelInProgress() is not set until *after*
        // the shutdown message has been sent, so we need to check the
        // "closeInProgress" bit instead.
        owner.getCancelCriterion().checkCancelInProgress(null);
        Throwable cause = owner.getShutdownCause();
        if (cause == null) {
          cause = e;
        }
        throw new DistributedSystemDisconnectedException(
            "The distributed system is shutting down",
            cause);
      }
    }
  }

  /**
   * Get a new connection
   *
   * @param id the DistributedMember on which to create the connection
   * @param preserveOrder whether order should be preserved
   * @param startTime the ms clock start time
   * @param ackTimeout the ms ack-wait-threshold, or zero
   * @param ackSATimeout the ms ack-severe-alert-threshold, or zero
   * @return the new Connection, or null if a problem
   * @throws java.io.IOException if the connection could not be created
   */
  protected Connection get(DistributedMember id, boolean preserveOrder, long startTime,
      long ackTimeout, long ackSATimeout)
      throws java.io.IOException, DistributedSystemDisconnectedException {
    if (this.closed) {
      this.owner.getCancelCriterion().checkCancelInProgress(null);
      throw new DistributedSystemDisconnectedException(
          "Connection table is closed");
    }
    Connection result = null;
    boolean threadOwnsResources = threadOwnsResources();
    if (!preserveOrder || !threadOwnsResources) {
      result = getSharedConnection(id, threadOwnsResources, preserveOrder, startTime, ackTimeout,
          ackSATimeout);
    } else {
      result = getThreadOwnedConnection(id, startTime, ackTimeout, ackSATimeout);
    }
    if (result != null) {
      Assert.assertTrue(result.preserveOrder == preserveOrder);
    }
    return result;
  }

  protected synchronized void fileDescriptorsExhausted() {
    if (!ulimitWarningIssued) {
      ulimitWarningIssued = true;
      logger.fatal(
          "This process is out of file descriptors.This will hamper communications and slow down the system.Any conserve-sockets setting is now being ignored.Please consider raising the descriptor limit.This alert is only issued once per process.");
      InternalDistributedSystem.getAnyInstance().setShareSockets(true);
      threadWantsOwnResources = new ThreadLocal();
    }
  }

  protected TCPConduit getConduit() {
    return owner;
  }

  public BufferPool getBufferPool() {
    return bufferPool;
  }

  public boolean isClosed() {
    return this.closed;
  }

  private static void closeCon(String reason, Object c) {
    closeCon(reason, c, false);
  }

  private static void closeCon(String reason, Object c, boolean beingSick) {
    if (c == null) {
      return;
    }
    if (c instanceof Connection) {
      ((Connection) c).closePartialConnect(reason, beingSick); // fix for bug 31666
    } else {
      ((PendingConnection) c).notifyWaiters(null);
    }
  }

  /**
   * returns the idle connection timer, or null if the connection table is closed. guarded by a sync
   * on the connection table
   */
  protected synchronized SystemTimer getIdleConnTimer() {
    if (this.closed) {
      return null;
    }
    if (this.idleConnTimer == null) {
      this.idleConnTimer = new SystemTimer(getDM().getSystem(), true);
    }
    return this.idleConnTimer;
  }

  protected void close() {
    /*
     * NOMUX if (inputMuxManager != null) { inputMuxManager.stop(); }
     */
    if (this.closed) {
      return;
    }
    this.closed = true;
    synchronized (this) {
      if (this.idleConnTimer != null) {
        this.idleConnTimer.cancel();
      }
    }
    synchronized (this.orderedConnectionMap) {
      for (Iterator it = this.orderedConnectionMap.values().iterator(); it.hasNext();) {
        closeCon(
            "Connection table being destroyed",
            it.next());
      }
      this.orderedConnectionMap.clear();
    }
    synchronized (this.unorderedConnectionMap) {
      for (Iterator it = this.unorderedConnectionMap.values().iterator(); it.hasNext();) {
        closeCon(
            "Connection table being destroyed",
            it.next());
      }
      this.unorderedConnectionMap.clear();
    }
    if (this.threadConnectionMap != null) {
      this.threadConnectionMap = null;
    }
    if (this.threadConnMaps != null) {
      synchronized (this.threadConnMaps) {
        for (Iterator it = this.threadConnMaps.iterator(); it.hasNext();) {
          Reference r = (Reference) it.next();
          Map m = (Map) r.get();
          if (m != null) {
            synchronized (m) {
              for (Iterator mit = m.values().iterator(); mit.hasNext();) {
                closeCon("Connection table being destroyed", mit.next());
              }
            }
          }
        }
        this.threadConnMaps.clear();
      }
    }
    {
      Executor localExec = this.p2pReaderThreadPool;
      if (localExec != null) {
        if (localExec instanceof ExecutorService) {
          ((ExecutorService) localExec).shutdown();
        }
      }
    }
    closeReceivers(false);

    Map m = (Map) this.threadOrderedConnMap.get();
    if (m != null) {
      synchronized (m) {
        m.clear();
      }
    }
    this.socketCloser.close();
  }

  public void executeCommand(Runnable runnable) {
    Executor local = this.p2pReaderThreadPool;
    if (local != null) {
      local.execute(runnable);
    }
  }

  /**
   * Close all receiving threads. This is used during shutdown and is also used by a test hook that
   * makes us deaf to incoming messages.
   *
   * @param beingSick a test hook to simulate a sick process
   */
  protected void closeReceivers(boolean beingSick) {
    synchronized (this.receivers) {
      for (Iterator it = this.receivers.iterator(); it.hasNext();) {
        Connection con = (Connection) it.next();
        if (!beingSick || con.preserveOrder) {
          closeCon(
              "Connection table being destroyed",
              con, beingSick);
          it.remove();
        }
      }
      // now close any sockets being formed
      synchronized (connectingSockets) {
        for (Iterator it = connectingSockets.entrySet().iterator(); it.hasNext();) {
          Map.Entry entry = (Map.Entry) it.next();
          // ConnectingSocketInfo info = (ConnectingSocketInfo)entry.getValue();
          try {
            ((Socket) entry.getKey()).close();
          } catch (IOException e) {
            // ignored - we're shutting down
          }
          it.remove();
        }
      }
    }
  }


  protected void removeReceiver(Object con) {
    synchronized (this.receivers) {
      this.receivers.remove(con);
    }
  }

  /**
   * Return true if our owner already knows that this endpoint is departing
   */
  protected boolean isEndpointShuttingDown(DistributedMember id) {
    return giveUpOnMember(owner.getDM().getMembershipManager(), id);
  }

  protected boolean giveUpOnMember(MembershipManager mgr, DistributedMember remoteAddr) {
    return !mgr.memberExists(remoteAddr) || mgr.isShunned(remoteAddr) || mgr.shutdownInProgress();
  }

  /** remove an endpoint and notify the membership manager of the departure */
  protected void removeEndpoint(DistributedMember stub, String reason) {
    removeEndpoint(stub, reason, true);
  }

  protected void removeEndpoint(DistributedMember memberID, String reason,
      boolean notifyDisconnect) {
    if (this.closed) {
      return;
    }
    boolean needsRemoval = false;
    synchronized (this.orderedConnectionMap) {
      if (this.orderedConnectionMap.get(memberID) != null)
        needsRemoval = true;
    }
    if (!needsRemoval) {
      synchronized (this.unorderedConnectionMap) {
        if (this.unorderedConnectionMap.get(memberID) != null)
          needsRemoval = true;
      }
    }
    if (!needsRemoval) {
      ConcurrentMap cm = this.threadConnectionMap;
      if (cm != null) {
        ArrayList al = (ArrayList) cm.get(memberID);
        needsRemoval = al != null && al.size() > 0;
      }
    }

    if (needsRemoval) {
      InternalDistributedMember remoteAddress = null;
      synchronized (this.orderedConnectionMap) {
        Object c = this.orderedConnectionMap.remove(memberID);
        if (c instanceof Connection) {
          remoteAddress = ((Connection) c).getRemoteAddress();
        }
        closeCon(reason, c);
      }
      synchronized (this.unorderedConnectionMap) {
        Object c = this.unorderedConnectionMap.remove(memberID);
        if (remoteAddress == null && (c instanceof Connection)) {
          remoteAddress = ((Connection) c).getRemoteAddress();
        }
        closeCon(reason, c);
      }

      {
        ConcurrentMap cm = this.threadConnectionMap;
        if (cm != null) {
          ArrayList al = (ArrayList) cm.remove(memberID);
          if (al != null) {
            synchronized (al) {
              for (Iterator it = al.iterator(); it.hasNext();) {
                Object c = it.next();
                if (remoteAddress == null && (c instanceof Connection)) {
                  remoteAddress = ((Connection) c).getRemoteAddress();
                }
                closeCon(reason, c);
              }
              al.clear();
            }
          }
        }
      }

      // close any sockets that are in the process of being connected
      Set toRemove = new HashSet();
      synchronized (connectingSockets) {
        for (Iterator it = connectingSockets.entrySet().iterator(); it.hasNext();) {
          Map.Entry entry = (Map.Entry) it.next();
          ConnectingSocketInfo info = (ConnectingSocketInfo) entry.getValue();
          if (info.peerAddress.equals(((InternalDistributedMember) memberID).getInetAddress())) {
            toRemove.add(entry.getKey());
            it.remove();
          }
        }
      }
      for (Iterator it = toRemove.iterator(); it.hasNext();) {
        Socket sock = (Socket) it.next();
        try {
          sock.close();
        } catch (IOException e) {
          if (logger.isDebugEnabled()) {
            logger.debug("caught exception while trying to close connecting socket for {}",
                memberID, e);
          }
        }
      }

      // close any receivers
      // avoid deadlock when a NIC has failed by closing connections outside
      // of the receivers sync (bug 38731)
      toRemove.clear();
      synchronized (this.receivers) {
        for (Iterator it = receivers.iterator(); it.hasNext();) {
          Connection con = (Connection) it.next();
          if (memberID.equals(con.getRemoteAddress())) {
            it.remove();
            toRemove.add(con);
          }
        }
      }
      for (Iterator it = toRemove.iterator(); it.hasNext();) {
        Connection con = (Connection) it.next();
        closeCon(reason, con);
      }
      if (notifyDisconnect) {
        // Before the removal of TCPConduit Stub addresses this used
        // to call MembershipManager.getMemberForStub, which checked
        // for a shutdown in progress and threw this exception:
        if (owner.getDM().shutdownInProgress()) {
          throw new DistributedSystemDisconnectedException("Shutdown in progress",
              owner.getDM().getMembershipManager().getShutdownCause());
        }
      }

      if (remoteAddress != null) {
        this.socketCloser.releaseResourcesForAddress(remoteAddress.toString());
      }
    }
  }

  SocketCloser getSocketCloser() {
    return this.socketCloser;
  }

  /** check to see if there are still any receiver threads for the given end-point */
  protected boolean hasReceiversFor(DistributedMember endPoint) {
    synchronized (this.receivers) {
      for (Iterator it = receivers.iterator(); it.hasNext();) {
        Connection con = (Connection) it.next();
        if (endPoint.equals(con.getRemoteAddress())) {
          return true;
        }
      }
    }
    return false;
  }

  private static void removeFromThreadConMap(ConcurrentMap cm, DistributedMember stub,
      Connection c) {
    if (cm != null) {
      ArrayList al = (ArrayList) cm.get(stub);
      if (al != null) {
        synchronized (al) {
          al.remove(c);
        }
      }
    }
  }

  protected void removeThreadConnection(DistributedMember stub, Connection c) {
    /*
     * if (this.closed) { return; }
     */
    removeFromThreadConMap(this.threadConnectionMap, stub, c);
    Map m = (Map) this.threadOrderedConnMap.get();
    if (m != null) {
      // Static cleanup thread might intervene, so we MUST synchronize
      synchronized (m) {
        if (m.get(stub) == c) {
          m.remove(stub);
        }
      } // synchronized
    } // m != null
  }

  void removeSharedConnection(String reason, DistributedMember stub, boolean ordered,
      Connection c) {
    if (this.closed) {
      return;
    }
    if (ordered) {
      synchronized (this.orderedConnectionMap) {
        if (this.orderedConnectionMap.get(stub) == c) {
          closeCon(reason, this.orderedConnectionMap.remove(stub));
        }
      }
    } else {
      synchronized (this.unorderedConnectionMap) {
        if (this.unorderedConnectionMap.get(stub) == c) {
          closeCon(reason, this.unorderedConnectionMap.remove(stub));
        }
      }
    }
  }

  /**
   * Just ensure that this class gets loaded.
   *
   * @see SystemFailure#loadEmergencyClasses()
   */
  public static void loadEmergencyClasses() {
    // don't go any further, Frodo!
  }

  /**
   * Clears lastInstance. Does not yet close underlying sockets, but probably not strictly
   * necessary.
   *
   * @see SystemFailure#emergencyClose()
   */
  public static void emergencyClose() {
    ConnectionTable ct = (ConnectionTable) lastInstance.get();
    if (ct == null) {
      return;
    }
    lastInstance.set(null);
  }

  public void removeAndCloseThreadOwnedSockets() {
    Map m = (Map) this.threadOrderedConnMap.get();
    if (m != null) {
      // Static cleanup may intervene; we MUST synchronize.
      synchronized (m) {
        Iterator it = m.entrySet().iterator();
        while (it.hasNext()) {
          Map.Entry me = (Map.Entry) it.next();
          DistributedMember stub = (DistributedMember) me.getKey();
          Connection c = (Connection) me.getValue();
          removeFromThreadConMap(this.threadConnectionMap, stub, c);
          it.remove();
          closeCon("thread finalization", c);
        } // while
      } // synchronized m
    }
  }

  public static void releaseThreadsSockets() {
    ConnectionTable ct = (ConnectionTable) lastInstance.get();
    if (ct == null) {
      return;
    }
    ct.removeAndCloseThreadOwnedSockets();
    // lastInstance = null;
  }

  /**
   * records the current outgoing message count on all thread-owned ordered connections. This does
   * not synchronize or stop new connections from being formed or new messages from being sent
   *
   * @since GemFire 5.1
   */
  protected void getThreadOwnedOrderedConnectionState(DistributedMember member, Map result) {

    ConcurrentMap cm = this.threadConnectionMap;
    if (cm != null) {
      ArrayList al = (ArrayList) cm.get(member);
      if (al != null) {
        synchronized (al) {
          al = new ArrayList(al);
        }

        for (Iterator it = al.iterator(); it.hasNext();) {
          Connection conn = (Connection) it.next();
          if (!conn.isSharedResource() && conn.getOriginatedHere() && conn.getPreserveOrder()) {
            result.put(Long.valueOf(conn.getUniqueId()), Long.valueOf(conn.getMessagesSent()));
          }
        }
      }
    }
  }

  /**
   * wait for the given incoming connections to receive at least the associated number of messages
   */
  protected void waitForThreadOwnedOrderedConnectionState(DistributedMember member,
      Map connectionStates) throws InterruptedException {
    if (Thread.interrupted())
      throw new InterruptedException(); // wisest to do this before the synchronize below
    List r = null;
    synchronized (receivers) {
      r = new ArrayList(receivers);
    }
    for (Iterator it = r.iterator(); it.hasNext();) {
      Connection con = (Connection) it.next();
      if (!con.stopped && !con.isClosing() && !con.getOriginatedHere() && con.getPreserveOrder()
          && member.equals(con.getRemoteAddress())) {
        Long state = (Long) connectionStates.remove(Long.valueOf(con.getUniqueId()));
        if (state != null) {
          long count = state.longValue();
          while (!con.stopped && !con.isClosing() && con.getMessagesReceived() < count) {
            if (logger.isDebugEnabled()) {
              logger.debug("Waiting for connection {}/{} currently={} need={}",
                  con.getRemoteAddress(), con.getUniqueId(), con.getMessagesReceived(), count);
            }
            Thread.sleep(100);
          }
        }
      }
    }
    if (connectionStates.size() > 0) {
      if (logger.isDebugEnabled()) {
        StringBuffer sb = new StringBuffer(1000);
        sb.append("These connections from ");
        sb.append(member);
        sb.append("could not be located during waitForThreadOwnedOrderedConnectionState: ");
        for (Iterator it = connectionStates.entrySet().iterator(); it.hasNext();) {
          Map.Entry entry = (Map.Entry) it.next();
          sb.append(entry.getKey()).append('(').append(entry.getValue()).append(')');
          if (it.hasNext()) {
            sb.append(',');
          }
        }
        logger.debug(sb);
      }
    }
  }

  protected DistributionManager getDM() {
    return this.owner.getDM();
  }

  // public boolean isShuttingDown() {
  // return this.owner.isShuttingDown();
  // }

  // protected void cleanupHighWater() {
  // cleanup(highWater);
  // }

  // protected void cleanupLowWater() {
  // cleanup(lowWater);
  // }

  // private void cleanup(int maxConnections) {
  /*
   * if (maxConnections == 0 || maxConnections >= connections.size()) { return; } while
   * (connections.size() > maxConnections) { Connection oldest = null; synchronized(connections) {
   * for (Iterator iter = connections.values().iterator(); iter.hasNext(); ) { Connection c =
   * (Connection)iter.next(); if (oldest == null || c.getTimeStamp() < oldest.getTimeStamp()) {
   * oldest = c; } } } // sanity check - don't close anything fresher than 10 seconds or // we'll
   * start thrashing if (oldest.getTimeStamp() > (System.currentTimeMillis() - 10000)) { if
   * (owner.lowWaterConnectionCount > 0) { owner.lowWaterConnectionCount += 10; } if
   * (owner.highWaterConnectionCount > 0) { owner.highWaterConnectionCount += 10; } new Object[] {
   * owner.lowWaterConnectionCount, owner.highWaterConnectionCount }); break; } if (oldest != null)
   * { oldest.close(); } }
   */
  // }

  /*
   * public void dumpConnectionTable() { Iterator iter = connectionMap.keySet().iterator(); while
   * (iter.hasNext()) { Object key = iter.next(); Object val = connectionMap.get(key); } }
   */
  private /* static */ class PendingConnection {
    /**
     * true if this connection is still pending
     */
    private boolean pending = true;

    /**
     * the connection we are waiting on
     */
    private Connection conn = null;

    /**
     * whether the connection preserves message ordering
     */
    private final boolean preserveOrder;

    /**
     * the stub we are connecting to
     */
    private final DistributedMember id;

    private final Thread connectingThread;

    public PendingConnection(boolean preserveOrder, DistributedMember id) {
      this.preserveOrder = preserveOrder;
      this.id = id;
      this.connectingThread = Thread.currentThread();
    }

    /**
     * Synchronously set the connection and notify waiters that we are ready.
     *
     * @param c the new connection
     */
    public synchronized void notifyWaiters(Connection c) {
      if (!this.pending)
        return; // already done.

      this.conn = c;
      this.pending = false;
      if (logger.isDebugEnabled()) {
        logger.debug("Notifying waiters that pending {} connection to {} is ready; {}",
            ((this.preserveOrder) ? "ordered" : "unordered"), this.id, this);
      }
      this.notifyAll();
    }

    /**
     * Wait for a connection
     *
     * @param mgr the membership manager that can instigate suspect processing if necessary
     * @param startTime the ms clock start time for the operation
     * @param ackTimeout the ms ack-wait-threshold, or zero
     * @param ackSATimeout the ms ack-severe-alert-threshold, or zero
     * @return the new connection
     */
    public synchronized Connection waitForConnect(Membership mgr, long startTime,
        long ackTimeout, long ackSATimeout) throws IOException {
      if (connectingThread == Thread.currentThread()) {
        throw new ReenteredConnectException("This thread is already trying to connect");
      }

      final Map m = this.preserveOrder ? orderedConnectionMap : unorderedConnectionMap;

      boolean severeAlertIssued = false;
      boolean suspected = false;
      DistributedMember targetMember = null;
      if (ackSATimeout > 0) {
        targetMember = this.id;
      }

      int attempt = 0;
      for (;;) {
        if (!this.pending) {
          break;
        }
        getConduit().getCancelCriterion().checkCancelInProgress(null);

        // wait a little bit...
        boolean interrupted = Thread.interrupted();
        try {
          this.wait(100); // spurious wakeup ok
        } catch (InterruptedException ignore) {
          interrupted = true;
          getConduit().getCancelCriterion().checkCancelInProgress(ignore);
        } finally {
          if (interrupted) {
            Thread.currentThread().interrupt();
          }
        }

        if (!this.pending)
          break;

        // Still pending...
        long now = System.currentTimeMillis();
        if (!severeAlertIssued && ackSATimeout > 0 && startTime + ackTimeout < now) {
          if (startTime + ackTimeout + ackSATimeout < now) {
            if (targetMember != null) {
              logger.fatal("Unable to form a TCP/IP connection to {} in over {} seconds",
                  targetMember, (ackSATimeout + ackTimeout) / 1000);
            }
            severeAlertIssued = true;
          } else if (!suspected) {
            logger.warn("Unable to form a TCP/IP connection to %s in over %s seconds",
                this.id, (ackTimeout) / 1000);
            ((GMSMembershipManager) mgr).suspectMember((InternalDistributedMember) targetMember,
                "Unable to form a TCP/IP connection in a reasonable amount of time");
            suspected = true;
          }
        }

        Object e;
        // synchronized (m) {
        e = m.get(this.id);
        // }
        if (e == this) {
          attempt += 1;
          if (logger.isDebugEnabled() && (attempt % 20 == 1)) {
            logger.debug("Waiting for pending connection to complete: {} connection to {}; {}",
                ((this.preserveOrder) ? "ordered" : "unordered"), this.id, this);
          }
          continue;
        }

        // Odd state change. Process and exit.
        if (logger.isDebugEnabled()) {
          logger.debug("Pending connection changed to {} unexpectedly", e);
        }

        if (e == null) {
          // We were removed
          notifyWaiters(null);
          break;
        } else if (e instanceof Connection) {
          notifyWaiters((Connection) e);
          break;
        } else {
          // defer to the new instance
          return ((PendingConnection) e).waitForConnect(mgr, startTime, ackTimeout, ackSATimeout);
        }

      } // for
      return this.conn;

    }

    public String toString() {
      return super.toString() + " created by " + connectingThread.getName();
    }
  }


  private static class IdleConnTT extends SystemTimer.SystemTimerTask {

    private Connection c;

    IdleConnTT(Connection c) {
      this.c = c;
    }

    @Override
    public boolean cancel() {
      Connection con = this.c;
      if (con != null) {
        con.cleanUpOnIdleTaskCancel();
      }
      this.c = null;
      return super.cancel();
    }

    @Override
    public void run2() {
      Connection con = this.c;
      if (con != null) {
        if (con.checkForIdleTimeout()) {
          cancel();
        }
      }
    }
  }

  public static ConnectionTable create(TCPConduit conduit) throws IOException {
    ConnectionTable ct = new ConnectionTable(conduit);
    lastInstance.set(ct);
    return ct;
  }

  /** keep track of a socket that is trying to connect() for shutdown purposes */
  public void addConnectingSocket(Socket socket, InetAddress addr) {
    synchronized (connectingSockets) {
      connectingSockets.put(socket, new ConnectingSocketInfo(addr));
    }
  }

  /** remove a socket from the tracked set. It should be connected at this point */
  public void removeConnectingSocket(Socket socket) {
    synchronized (connectingSockets) {
      connectingSockets.remove(socket);
    }
  }


  private static class ConnectingSocketInfo {
    InetAddress peerAddress;
    Thread connectingThread;

    public ConnectingSocketInfo(InetAddress addr) {
      this.peerAddress = addr;
      this.connectingThread = Thread.currentThread();
    }
  }

  public int getNumberOfReceivers() {
    return receivers.size();
  }
}