| /* |
| * 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 static java.lang.Integer.MAX_VALUE; |
| import static java.util.concurrent.TimeUnit.SECONDS; |
| |
| import java.io.IOException; |
| 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.VisibleForTesting; |
| 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.api.MemberIdentifier; |
| import org.apache.geode.distributed.internal.membership.api.Membership; |
| import org.apache.geode.internal.Assert; |
| import org.apache.geode.internal.SystemTimer; |
| import org.apache.geode.internal.lang.utils.JavaWorkarounds; |
| 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; |
| |
| /** |
| * ConnectionTable holds all of the Connection objects in a conduit. Connections represent a pipe |
| * between two endpoints represented by generic DistributedMembers. |
| * |
| * @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 |
| */ |
| private static final ThreadLocal<Boolean> 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. |
| * Value may be {@link Connection} or {@link PendingConnection}. |
| */ |
| private final Map<DistributedMember, Object> orderedConnectionMap = new ConcurrentHashMap<>(); |
| |
| /** |
| * ordered connections local to this thread. |
| */ |
| static final ThreadLocal<Map<DistributedMember, Connection>> threadOrderedConnMap = |
| new ThreadLocal<>(); |
| |
| /** |
| * 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 final ConcurrentMap<DistributedMember, List<Connection>> threadConnectionMap; |
| |
| /** |
| * Used for all non-ordered messages. Only connections used for sending messages, and receiving |
| * acks, will be put in this map. |
| */ |
| private final Map<DistributedMember, Object> 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<Connection> 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; |
| |
| /** |
| * 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); |
| |
| private final SocketCloser socketCloser; |
| |
| /** |
| * The most recent instance to be created |
| * |
| * <p> |
| * TODO this assumes no more than one instance is created at a time? |
| */ |
| @MakeNotStatic |
| private static final AtomicReference<ConnectionTable> lastInstance = new AtomicReference<>(); |
| |
| /** |
| * A set of sockets that are in the process of being connected |
| */ |
| private final Map<Socket, ConnectingSocketInfo> 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. |
| */ |
| private boolean threadOwnsResources() { |
| DistributionManager d = getDM(); |
| if (d != null) { |
| return d.getSystem().threadOwnsResources() && !AlertingAction.isThreadAlerting(); |
| } |
| return false; |
| } |
| |
| public static Boolean getThreadOwnsResourcesRegistration() { |
| return threadWantsOwnResources.get(); |
| } |
| |
| public TCPConduit getOwner() { |
| return owner; |
| } |
| |
| public static ConnectionTable create(TCPConduit conduit) { |
| ConnectionTable ct = new ConnectionTable(conduit); |
| lastInstance.set(ct); |
| return ct; |
| } |
| |
| private ConnectionTable(TCPConduit conduit) { |
| owner = conduit; |
| idleConnTimer = owner.idleConnectionTimeout != 0 |
| ? new SystemTimer(conduit.getDM().getSystem()) : null; |
| threadConnectionMap = new ConcurrentHashMap<>(); |
| p2pReaderThreadPool = createThreadPoolForIO(conduit.getDM().getSystem().isShareSockets()); |
| socketCloser = new SocketCloser(); |
| bufferPool = conduit.getBufferPool(); |
| } |
| |
| private Executor createThreadPoolForIO(boolean conserveSockets) { |
| if (conserveSockets) { |
| return LoggingExecutors.newThreadOnEachExecute("SharedP2PReader"); |
| } |
| return CoreLoggingExecutors.newThreadPoolWithSynchronousFeed(1, MAX_VALUE, |
| READER_POOL_KEEP_ALIVE_TIME, SECONDS, "UnsharedP2PReader"); |
| } |
| |
| /** conduit calls acceptConnection after an accept */ |
| void acceptConnection(Socket sock, PeerConnectionFactory peerConnectionFactory) |
| throws IOException, ConnectionException { |
| InetAddress connAddress = sock.getInetAddress(); |
| boolean finishedConnecting = false; |
| Connection connection = null; |
| try { |
| connection = peerConnectionFactory.createReceiver(this, sock); |
| |
| // check for shutdown (so it doesn't get missed in the finally block) |
| owner.getCancelCriterion().checkCancelInProgress(null); |
| finishedConnecting = true; |
| } catch (ConnectionException | IOException ex) { |
| // check for shutdown... |
| owner.getCancelCriterion().checkCancelInProgress(ex); |
| logger.warn("Failed to accept connection from {} because: {}", |
| 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 (receivers) { |
| owner.getStats().incReceivers(); |
| if (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())) { |
| receivers.add(connection); |
| } |
| } |
| if (logger.isDebugEnabled()) { |
| logger.debug("Accepted {} myAddr={} theirAddr={}", connection, getConduit().getMemberId(), |
| connection.getRemoteAddress()); |
| } |
| } |
| } |
| |
| /** |
| * 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(InternalDistributedMember id, |
| boolean sharedResource, |
| boolean preserveOrder, Map<DistributedMember, Object> m, PendingConnection pc, long startTime, |
| long ackThreshold, |
| long ackSAThreshold) throws IOException, DistributedSystemDisconnectedException { |
| // handle new pending connection |
| Connection con = null; |
| try { |
| long senderCreateStartTime = owner.getStats().startSenderCreate(); |
| con = Connection.createSender(owner.getMembership(), this, preserveOrder, id, |
| sharedResource, startTime, ackThreshold, ackSAThreshold); |
| owner.getStats().incSenders(sharedResource, preserveOrder, senderCreateStartTime); |
| } finally { |
| // our connection failed to notify anyone waiting for our pending con |
| if (con == null) { |
| 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 |
| } |
| } |
| |
| // 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) { |
| // 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. |
| 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.getRemoteAddress()); |
| } |
| |
| 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(InternalDistributedMember id, boolean scheduleTimeout, |
| boolean preserveOrder, long startTime, long ackTimeout, long ackSATimeout) |
| throws IOException, DistributedSystemDisconnectedException { |
| |
| final Map<DistributedMember, Object> m = |
| preserveOrder ? orderedConnectionMap : unorderedConnectionMap; |
| |
| // new connection, if needed |
| PendingConnection pc = null; |
| |
| // existing connection (if we don't create a new one) |
| Object mEntry; |
| |
| // Look for pending connection |
| synchronized (m) { |
| mEntry = m.get(id); |
| if (mEntry instanceof Connection) { |
| Connection existingCon = (Connection) mEntry; |
| if (!existingCon.connected) { |
| mEntry = null; |
| } |
| } |
| if (mEntry == null) { |
| pc = new PendingConnection(preserveOrder, id); |
| m.put(id, pc); |
| } |
| } |
| |
| Connection result; |
| if (pc != null) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("created PendingConnection {}", pc); |
| } |
| result = handleNewPendingConnection(id, true, 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(owner.getMembership(), |
| startTime, ackTimeout, ackSATimeout); |
| if (logger.isDebugEnabled()) { |
| if (result != null) { |
| logger.debug("getSharedConnection {} myAddr={} theirAddr={}", result, |
| getConduit().getMemberId(), result.getRemoteAddress()); |
| } 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(InternalDistributedMember id, long startTime, long ackTimeout, |
| long ackSATimeout) throws IOException, DistributedSystemDisconnectedException { |
| Connection result; |
| |
| // Look for result in the thread local |
| Map<DistributedMember, Connection> m = threadOrderedConnMap.get(); |
| if (m == null) { |
| // First time for this thread. Create thread local |
| m = new HashMap<>(); |
| threadOrderedConnMap.set(m); |
| } else { |
| // No need to sync map since it is only referenced by ThreadLocal. |
| result = m.get(id); |
| if (result != null && !result.timedOut) { |
| return result; |
| } |
| } |
| |
| // OK, we have to create a new connection. |
| long senderCreateStartTime = owner.getStats().startSenderCreate(); |
| result = Connection.createSender(owner.getMembership(), this, true, id, false, startTime, |
| ackTimeout, ackSATimeout); |
| owner.getStats().incSenders(false, true, senderCreateStartTime); |
| if (logger.isDebugEnabled()) { |
| logger.debug("ConnectionTable: created an ordered connection: {}", result); |
| } |
| |
| // Update the list of connections owned by this thread.... |
| |
| final List<Connection> al = |
| JavaWorkarounds.computeIfAbsent(threadConnectionMap, id, k -> new ArrayList<>()); |
| |
| // Add our Connection to the list |
| synchronized (al) { |
| al.add(result); |
| } |
| |
| // Finally, add the connection to our thread local map. |
| // No need to sync map since it is only referenced by ThreadLocal. |
| m.put(id, result); |
| |
| scheduleIdleTimeout(result); |
| return result; |
| } |
| |
| /** schedule an idle-connection timeout task */ |
| private void scheduleIdleTimeout(Connection conn) { |
| if (conn == null) { |
| return; |
| } |
| // Set the idle timeout |
| if (owner.idleConnectionTimeout != 0) { |
| try { |
| synchronized (this) { |
| if (!closed) { |
| IdleConnTT task = new IdleConnTT(conn); |
| conn.setIdleTimeoutTask(task); |
| synchronized (task) { |
| if (!task.isCancelled()) { |
| getIdleConnTimer().scheduleAtFixedRate(task, owner.idleConnectionTimeout, |
| owner.idleConnectionTimeout); |
| } |
| } |
| } |
| } |
| } catch (IllegalStateException e) { |
| if (conn.isClosing()) { |
| // 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 IOException if the connection could not be created |
| */ |
| protected Connection get(InternalDistributedMember id, boolean preserveOrder, long startTime, |
| long ackTimeout, long ackSATimeout) |
| throws IOException, DistributedSystemDisconnectedException { |
| if (closed) { |
| owner.getCancelCriterion().checkCancelInProgress(null); |
| throw new DistributedSystemDisconnectedException("Connection table is closed"); |
| } |
| Connection result; |
| 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.getPreserveOrder() == preserveOrder); |
| } |
| return result; |
| } |
| |
| 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); |
| } |
| } |
| |
| TCPConduit getConduit() { |
| return owner; |
| } |
| |
| BufferPool getBufferPool() { |
| return bufferPool; |
| } |
| |
| public boolean isClosed() { |
| return 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); |
| } 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 |
| */ |
| synchronized SystemTimer getIdleConnTimer() { |
| if (closed) { |
| return null; |
| } |
| if (idleConnTimer == null) { |
| idleConnTimer = new SystemTimer(getDM().getSystem()); |
| } |
| return idleConnTimer; |
| } |
| |
| protected void close() { |
| if (closed) { |
| return; |
| } |
| closed = true; |
| synchronized (this) { |
| if (idleConnTimer != null) { |
| idleConnTimer.cancel(); |
| } |
| } |
| synchronized (orderedConnectionMap) { |
| for (Object o : orderedConnectionMap.values()) { |
| closeCon("Connection table being destroyed", o); |
| } |
| orderedConnectionMap.clear(); |
| } |
| synchronized (unorderedConnectionMap) { |
| for (Object o : unorderedConnectionMap.values()) { |
| closeCon("Connection table being destroyed", o); |
| } |
| unorderedConnectionMap.clear(); |
| } |
| if (threadConnectionMap != null) { |
| synchronized (threadConnectionMap) { |
| for (final List<Connection> connections : threadConnectionMap.values()) { |
| synchronized (connections) { |
| for (Object o : connections) { |
| closeCon("Connection table being destroyed", o); |
| } |
| } |
| } |
| threadConnectionMap.clear(); |
| } |
| } |
| Executor localExec = p2pReaderThreadPool; |
| if (localExec != null) { |
| if (localExec instanceof ExecutorService) { |
| ((ExecutorService) localExec).shutdown(); |
| } |
| } |
| closeReceivers(false); |
| |
| Map<DistributedMember, Connection> map = threadOrderedConnMap.get(); |
| if (map != null) { |
| // No need to synchronize map since it is only referenced by ThreadLocal. |
| map.clear(); |
| } |
| socketCloser.close(); |
| } |
| |
| public void executeCommand(Runnable runnable) { |
| Executor local = 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 |
| */ |
| void closeReceivers(boolean beingSick) { |
| synchronized (receivers) { |
| for (Iterator<Connection> it = receivers.iterator(); it.hasNext();) { |
| Connection con = it.next(); |
| if (!beingSick || con.getPreserveOrder()) { |
| closeCon("Connection table being destroyed", con, beingSick); |
| it.remove(); |
| } |
| } |
| // now close any sockets being formed |
| synchronized (connectingSockets) { |
| for (final Iterator<Map.Entry<Socket, ConnectingSocketInfo>> it = |
| connectingSockets.entrySet().iterator(); it.hasNext();) { |
| final Map.Entry<Socket, ?> entry = it.next(); |
| try { |
| entry.getKey().close(); |
| } catch (IOException e) { |
| // ignored - we're shutting down |
| } |
| it.remove(); |
| } |
| } |
| } |
| } |
| |
| void removeReceiver(Connection con) { |
| synchronized (receivers) { |
| receivers.remove(con); |
| } |
| } |
| |
| |
| /** |
| * remove an endpoint and notify the membership manager of the departure |
| */ |
| protected void removeEndpoint(DistributedMember stub, String reason) { |
| removeEndpoint(stub, reason, true); |
| } |
| |
| void removeEndpoint(DistributedMember memberID, String reason, boolean notifyDisconnect) { |
| if (closed) { |
| return; |
| } |
| boolean needsRemoval = false; |
| synchronized (orderedConnectionMap) { |
| if (orderedConnectionMap.get(memberID) != null) { |
| needsRemoval = true; |
| } |
| } |
| if (!needsRemoval) { |
| synchronized (unorderedConnectionMap) { |
| if (unorderedConnectionMap.get(memberID) != null) { |
| needsRemoval = true; |
| } |
| } |
| } |
| if (!needsRemoval) { |
| final ConcurrentMap<DistributedMember, List<Connection>> cm = threadConnectionMap; |
| if (cm != null) { |
| final List<Connection> al = cm.get(memberID); |
| needsRemoval = al != null && !al.isEmpty(); |
| } |
| } |
| |
| if (needsRemoval) { |
| InternalDistributedMember remoteAddress = null; |
| synchronized (orderedConnectionMap) { |
| Object c = orderedConnectionMap.remove(memberID); |
| if (c instanceof Connection) { |
| remoteAddress = ((Connection) c).getRemoteAddress(); |
| } |
| closeCon(reason, c); |
| } |
| synchronized (unorderedConnectionMap) { |
| Object c = unorderedConnectionMap.remove(memberID); |
| if (remoteAddress == null && c instanceof Connection) { |
| remoteAddress = ((Connection) c).getRemoteAddress(); |
| } |
| closeCon(reason, c); |
| } |
| |
| final ConcurrentMap<DistributedMember, List<Connection>> cm = threadConnectionMap; |
| if (cm != null) { |
| final List<Connection> al = cm.remove(memberID); |
| if (al != null) { |
| synchronized (al) { |
| for (Connection c : al) { |
| if (remoteAddress == null && c != null) { |
| remoteAddress = c.getRemoteAddress(); |
| } |
| closeCon(reason, c); |
| } |
| al.clear(); |
| } |
| } |
| } |
| |
| // close any sockets that are in the process of being connected |
| final Set<Socket> toRemove = new HashSet<>(); |
| synchronized (connectingSockets) { |
| for (final Iterator<Map.Entry<Socket, ConnectingSocketInfo>> it = |
| connectingSockets.entrySet().iterator(); it.hasNext();) { |
| final Map.Entry<Socket, ConnectingSocketInfo> entry = it.next(); |
| final ConnectingSocketInfo info = entry.getValue(); |
| if (info.peerAddress.equals(((MemberIdentifier) memberID).getInetAddress())) { |
| toRemove.add(entry.getKey()); |
| it.remove(); |
| } |
| } |
| } |
| |
| for (final Socket sock : toRemove) { |
| 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 |
| final Set<Connection> connectionsToClose = new HashSet<>(); |
| synchronized (receivers) { |
| for (final Iterator<Connection> it = receivers.iterator(); it.hasNext();) { |
| final Connection con = it.next(); |
| if (memberID.equals(con.getRemoteAddress())) { |
| it.remove(); |
| connectionsToClose.add(con); |
| } |
| } |
| } |
| for (final Connection con : connectionsToClose) { |
| closeCon(reason, con); |
| } |
| if (notifyDisconnect) { |
| if (owner.getDM().shutdownInProgress()) { |
| throw new DistributedSystemDisconnectedException("Shutdown in progress", |
| owner.getDM().getDistribution().getShutdownCause()); |
| } |
| } |
| |
| if (remoteAddress != null) { |
| socketCloser.releaseResourcesForAddress(remoteAddress.toString()); |
| } |
| } |
| } |
| |
| SocketCloser getSocketCloser() { |
| return socketCloser; |
| } |
| |
| /** check to see if there are still any receiver threads for the given end-point */ |
| boolean hasReceiversFor(DistributedMember endPoint) { |
| synchronized (receivers) { |
| for (Connection receiver : receivers) { |
| if (endPoint.equals(receiver.getRemoteAddress())) { |
| return true; |
| } |
| } |
| } |
| return false; |
| } |
| |
| private static void removeFromThreadConMap( |
| final ConcurrentMap<DistributedMember, List<Connection>> cm, final DistributedMember stub, |
| final Connection c) { |
| if (cm != null) { |
| List<Connection> al = cm.get(stub); |
| if (al != null) { |
| synchronized (al) { |
| al.remove(c); |
| } |
| } |
| } |
| } |
| |
| void removeThreadConnection(final DistributedMember stub, final Connection c) { |
| removeFromThreadConMap(threadConnectionMap, stub, c); |
| final Map<DistributedMember, Connection> m = threadOrderedConnMap.get(); |
| if (m != null) { |
| // No need to synchronize map since it is only referenced by ThreadLocal. |
| if (m.get(stub) == c) { |
| m.remove(stub); |
| } |
| } |
| } |
| |
| void removeSharedConnection(String reason, DistributedMember stub, boolean ordered, |
| Connection c) { |
| if (closed) { |
| return; |
| } |
| if (ordered) { |
| synchronized (orderedConnectionMap) { |
| if (orderedConnectionMap.get(stub) == c) { |
| closeCon(reason, orderedConnectionMap.remove(stub)); |
| } |
| } |
| } else { |
| synchronized (unorderedConnectionMap) { |
| if (unorderedConnectionMap.get(stub) == c) { |
| closeCon(reason, unorderedConnectionMap.remove(stub)); |
| } |
| } |
| } |
| } |
| |
| @VisibleForTesting |
| public static long getNumSenderSharedConnections() { |
| ConnectionTable ct = lastInstance.get(); |
| if (ct == null) { |
| return 0; |
| } |
| return (ct.getConduit().getStats().getSendersSU()); |
| } |
| |
| |
| |
| /** |
| * Clears lastInstance. Does not yet close underlying sockets, but probably not strictly |
| * necessary. |
| * |
| * @see SystemFailure#emergencyClose() |
| */ |
| public static void emergencyClose() { |
| ConnectionTable ct = lastInstance.get(); |
| if (ct == null) { |
| return; |
| } |
| lastInstance.set(null); |
| } |
| |
| void removeAndCloseThreadOwnedSockets() { |
| Map<DistributedMember, Connection> m = threadOrderedConnMap.get(); |
| if (m != null) { |
| // No need to synchronize map since it is only referenced by ThreadLocal. |
| for (final Iterator<Map.Entry<DistributedMember, Connection>> it = |
| m.entrySet().iterator(); it.hasNext();) { |
| final Map.Entry<DistributedMember, Connection> me = it.next(); |
| DistributedMember stub = me.getKey(); |
| Connection c = me.getValue(); |
| removeFromThreadConMap(threadConnectionMap, stub, c); |
| it.remove(); |
| closeCon("thread finalization", c); |
| } |
| } |
| } |
| |
| public static void releaseThreadsSockets() { |
| ConnectionTable ct = lastInstance.get(); |
| if (ct == null) { |
| return; |
| } |
| ct.removeAndCloseThreadOwnedSockets(); |
| } |
| |
| /** |
| * 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 |
| */ |
| void getThreadOwnedOrderedConnectionState(final DistributedMember member, |
| final Map<Long, Long> result) { |
| if (threadConnectionMap != null) { |
| List<Connection> al = threadConnectionMap.get(member); |
| if (al != null) { |
| synchronized (al) { |
| al = new ArrayList<>(al); |
| } |
| |
| for (final Connection connection : al) { |
| if (!connection.isSharedResource() && connection.getOriginatedHere() |
| && connection.getPreserveOrder()) { |
| result.put(connection.getUniqueId(), connection.getMessagesSent()); |
| } |
| } |
| } |
| } |
| } |
| |
| /** |
| * wait for the given incoming connections to receive at least the associated number of messages |
| */ |
| void waitForThreadOwnedOrderedConnectionState(final DistributedMember member, |
| final Map<Long, Long> connectionStates) |
| throws InterruptedException { |
| if (Thread.interrupted()) { |
| // wisest to do this before the synchronize below |
| throw new InterruptedException(); |
| } |
| List<Connection> r; |
| synchronized (receivers) { |
| r = new ArrayList<>(receivers); |
| } |
| for (Connection con : r) { |
| if (!con.stopped && !con.isClosing() && !con.getOriginatedHere() && con.getPreserveOrder() |
| && member.equals(con.getRemoteAddress())) { |
| final Long state = connectionStates.remove(con.getUniqueId()); |
| if (state != null) { |
| long count = state; |
| 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.isEmpty()) { |
| if (logger.isDebugEnabled()) { |
| StringBuilder sb = new StringBuilder(1000); |
| sb.append("These connections from "); |
| sb.append(member); |
| sb.append("could not be located during waitForThreadOwnedOrderedConnectionState: "); |
| for (Iterator<Map.Entry<Long, Long>> it = connectionStates.entrySet().iterator(); it |
| .hasNext();) { |
| Map.Entry<Long, Long> entry = it.next(); |
| sb.append(entry.getKey()).append('(').append(entry.getValue()).append(')'); |
| if (it.hasNext()) { |
| sb.append(','); |
| } |
| } |
| logger.debug(sb); |
| } |
| } |
| } |
| |
| protected DistributionManager getDM() { |
| return owner.getDM(); |
| } |
| |
| /** keep track of a socket that is trying to connect() for shutdown purposes */ |
| 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 */ |
| void removeConnectingSocket(Socket socket) { |
| synchronized (connectingSockets) { |
| connectingSockets.remove(socket); |
| } |
| } |
| |
| int getNumberOfReceivers() { |
| return receivers.size(); |
| } |
| |
| private class PendingConnection { |
| |
| /** |
| * true if this connection is still pending |
| */ |
| private boolean pending = true; |
| |
| /** |
| * the connection we are waiting on |
| */ |
| private Connection conn; |
| |
| /** |
| * whether the connection preserves message ordering |
| */ |
| private final boolean preserveOrder; |
| |
| /** |
| * the stub we are connecting to |
| */ |
| private final DistributedMember id; |
| |
| private final Thread connectingThread; |
| |
| private PendingConnection(boolean preserveOrder, DistributedMember id) { |
| this.preserveOrder = preserveOrder; |
| this.id = id; |
| connectingThread = Thread.currentThread(); |
| } |
| |
| /** |
| * Synchronously set the connection and notify waiters that we are ready. |
| * |
| * @param c the new connection |
| */ |
| private synchronized void notifyWaiters(Connection c) { |
| if (!pending) { |
| return; |
| } |
| |
| conn = c; |
| pending = false; |
| if (logger.isDebugEnabled()) { |
| logger.debug("Notifying waiters that pending {} connection to {} is ready; {}", |
| preserveOrder ? "ordered" : "unordered", id, 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 |
| */ |
| private synchronized Connection waitForConnect(Membership<InternalDistributedMember> mgr, |
| long startTime, long ackTimeout, |
| long ackSATimeout) { |
| if (connectingThread == Thread.currentThread()) { |
| throw new ReenteredConnectException("This thread is already trying to connect"); |
| } |
| |
| final Map<DistributedMember, Object> m = |
| preserveOrder ? orderedConnectionMap : unorderedConnectionMap; |
| |
| DistributedMember targetMember = null; |
| if (ackSATimeout > 0) { |
| targetMember = id; |
| } |
| |
| boolean suspected = false; |
| boolean severeAlertIssued = false; |
| for (int attempt = 0;;) { |
| if (!pending) { |
| break; |
| } |
| getConduit().getCancelCriterion().checkCancelInProgress(null); |
| |
| // wait a little bit... |
| boolean interrupted = Thread.interrupted(); |
| try { |
| wait(100); |
| } catch (InterruptedException e) { |
| interrupted = true; |
| getConduit().getCancelCriterion().checkCancelInProgress(e); |
| } finally { |
| if (interrupted) { |
| Thread.currentThread().interrupt(); |
| } |
| } |
| |
| if (!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 {} in over {} seconds", |
| id, ackTimeout / 1000); |
| mgr.suspectMember((InternalDistributedMember) targetMember, |
| "Unable to form a TCP/IP connection in a reasonable amount of time"); |
| suspected = true; |
| } |
| } |
| |
| Object e = m.get(id); |
| if (e == this) { |
| attempt += 1; |
| if (logger.isDebugEnabled() && attempt % 20 == 1) { |
| logger.debug("Waiting for pending connection to complete: {} connection to {}; {}", |
| preserveOrder ? "ordered" : "unordered", 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; |
| } |
| if (e instanceof Connection) { |
| notifyWaiters((Connection) e); |
| break; |
| } |
| // defer to the new instance |
| return ((PendingConnection) e).waitForConnect(mgr, startTime, ackTimeout, ackSATimeout); |
| |
| } |
| return conn; |
| } |
| |
| public String toString() { |
| return super.toString() + " created by " + connectingThread.getName(); |
| } |
| } |
| |
| private static class IdleConnTT extends SystemTimer.SystemTimerTask { |
| |
| private Connection connection; |
| |
| private IdleConnTT(Connection c) { |
| connection = c; |
| } |
| |
| @Override |
| public boolean cancel() { |
| Connection con = connection; |
| if (con != null) { |
| con.cleanUpOnIdleTaskCancel(); |
| } |
| connection = null; |
| return super.cancel(); |
| } |
| |
| @Override |
| public void run2() { |
| Connection con = connection; |
| if (con != null) { |
| if (con.checkForIdleTimeout()) { |
| cancel(); |
| } |
| } |
| } |
| } |
| |
| private static class ConnectingSocketInfo { |
| |
| private final InetAddress peerAddress; |
| |
| private ConnectingSocketInfo(InetAddress addr) { |
| peerAddress = addr; |
| } |
| } |
| } |
