geode-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From bschucha...@apache.org
Subject [1/2] geode git commit: GEODE-2684 Connection & ConnectionTable cleanup
Date Wed, 05 Apr 2017 22:31:43 GMT
Repository: geode
Updated Branches:
  refs/heads/develop 39c72b204 -> 6b2b7b2f7


GEODE-2684 Connection & ConnectionTable cleanup

removed dead code and indirect access of TcpConduit through the
connection table.


Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/6b2b7b2f
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/6b2b7b2f
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/6b2b7b2f

Branch: refs/heads/develop
Commit: 6b2b7b2f7f3f63b8ae638e9afffa5edc0f763783
Parents: 391502a
Author: Bruce Schuchardt <bschuchardt@pivotal.io>
Authored: Wed Apr 5 15:13:01 2017 -0700
Committer: Bruce Schuchardt <bschuchardt@pivotal.io>
Committed: Wed Apr 5 15:31:33 2017 -0700

----------------------------------------------------------------------
 .../apache/geode/internal/tcp/Connection.java   | 188 ++++++-------------
 .../geode/internal/tcp/ConnectionTable.java     |  68 ++-----
 .../geode/internal/tcp/DirectReplySender.java   |   2 +-
 .../apache/geode/internal/tcp/MsgReader.java    |   2 +-
 .../apache/geode/internal/tcp/NIOMsgReader.java |   2 +-
 5 files changed, 82 insertions(+), 180 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/geode/blob/6b2b7b2f/geode-core/src/main/java/org/apache/geode/internal/tcp/Connection.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/tcp/Connection.java b/geode-core/src/main/java/org/apache/geode/internal/tcp/Connection.java
index a0af245..c57a0ba 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/tcp/Connection.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/tcp/Connection.java
@@ -72,8 +72,6 @@ public class Connection implements Runnable {
   public final static int CHUNKED_MSG_TYPE = 0x4d; // a chunk of one logical msg
   public final static int END_CHUNKED_MSG_TYPE = 0x4e; // last in a series of chunks
   public final static int DIRECT_ACK_BIT = 0x20;
-  // We no longer support early ack
-  // public final static int EARLY_ACK_BIT = 0x10;
 
   public static final int MSG_HEADER_SIZE_OFFSET = 0;
   public static final int MSG_HEADER_TYPE_OFFSET = 4;
@@ -95,7 +93,9 @@ public class Connection implements Runnable {
       "member unexpectedly shut down shared, unordered connection";
 
   /** the table holding this connection */
-  final ConnectionTable owner;
+  private final ConnectionTable owner;
+
+  private final TCPConduit conduit;
 
   /**
    * Set to false once run() is terminating. Using this instead of Thread.isAlive as the
reader
@@ -113,15 +113,6 @@ public class Connection implements Runnable {
   /** The idle timeout timer task for this connection */
   private SystemTimerTask idleTask;
 
-  /**
-   * Returns the depth of unshared reader threads from this thread to the original
-   * non-reader-thread. E.g., ServerConnection -> reader(domino=1) -> reader(domino=2)
->
-   * reader(domino=3)
-   */
-  public static int getDominoCount() {
-    return dominoCount.get().intValue();
-  }
-
   private final static ThreadLocal isReaderThread = new ThreadLocal();
 
   public final static void makeReaderThread() {
@@ -129,7 +120,7 @@ public class Connection implements Runnable {
     makeReaderThread(true);
   }
 
-  private final static void makeReaderThread(boolean v) {
+  private static void makeReaderThread(boolean v) {
     isReaderThread.set(v);
   }
 
@@ -150,7 +141,7 @@ public class Connection implements Runnable {
     if (connectTimeoutStr != null) {
       P2P_CONNECT_TIMEOUT = Integer.parseInt(connectTimeoutStr);
     } else {
-      P2P_CONNECT_TIMEOUT = 6 * this.owner.owner.getDM().getConfig().getMemberTimeout();
+      P2P_CONNECT_TIMEOUT = 6 * this.conduit.getDM().getConfig().getMemberTimeout();
     }
     IS_P2P_CONNECT_TIMEOUT_INITIALIZED = true;
     return P2P_CONNECT_TIMEOUT;
@@ -367,20 +358,18 @@ public class Connection implements Runnable {
   /** the buffer used for NIO message receipt */
   ByteBuffer nioInputBuffer;
 
-  /** the position of the next message's content */
-  // int nioMessageStart;
-
   /** the length of the next message to be dispatched */
   int nioMessageLength;
-  // byte nioMessageVersion;
 
   /** the type of message being received */
   byte nioMessageType;
 
   /** used to lock access to destreamer data */
   private final Object destreamerLock = new Object();
+
   /** caches a msg destreamer that is currently not being used */
   MsgDestreamer idleMsgDestreamer;
+
   /**
    * used to map a msgId to a MsgDestreamer which are used for destreaming chunked messages
using
    * nio
@@ -409,8 +398,6 @@ public class Connection implements Runnable {
   private int sendBufferSize = -1;
   private int recvBufferSize = -1;
 
-  private ReplySender replySender;
-
   private void setSendBufferSize(Socket sock) {
     setSendBufferSize(sock, this.owner.getConduit().tcpBufferSize);
   }
@@ -541,6 +528,7 @@ public class Connection implements Runnable {
       throw new IllegalArgumentException(
           LocalizedStrings.Connection_NULL_CONNECTIONTABLE.toLocalizedString());
     }
+    this.conduit = t.getConduit();
     this.isReceiver = true;
     this.owner = t;
     this.socket = socket;
@@ -628,7 +616,7 @@ public class Connection implements Runnable {
     bytes[MSG_HEADER_SIZE_OFFSET + 2] = (byte) ((msglen / 0x100) & 0xff);
     bytes[MSG_HEADER_SIZE_OFFSET + 3] = (byte) (msglen & 0xff);
     bytes[MSG_HEADER_TYPE_OFFSET] = (byte) NORMAL_MSG_TYPE; // message type
-    bytes[MSG_HEADER_ID_OFFSET] = (byte) ((MsgIdGenerator.NO_MSG_ID / 0x100) & 0xff);
+    bytes[MSG_HEADER_ID_OFFSET] = (byte) ((MsgIdGenerator.NO_MSG_ID >> 8) & 0xff);
     bytes[MSG_HEADER_ID_OFFSET + 1] = (byte) (MsgIdGenerator.NO_MSG_ID & 0xff);
     bytes[MSG_HEADER_BYTES] = REPLY_CODE_OK;
     int allocSize = bytes.length;
@@ -707,19 +695,16 @@ public class Connection implements Runnable {
       my_okHandshakeBytes = okHandshakeBytes;
     }
     if (useNIO()) {
+      assert my_okHandshakeBuf != null;
       synchronized (my_okHandshakeBuf) {
         my_okHandshakeBuf.position(0);
         nioWriteFully(getSocket().getChannel(), my_okHandshakeBuf, false, null);
       }
     } else {
       synchronized (outLock) {
-        try {
-          // this.writerThread = Thread.currentThread();
-          this.output.write(my_okHandshakeBytes, 0, my_okHandshakeBytes.length);
-          this.output.flush();
-        } finally {
-          // this.writerThread = null;
-        }
+        assert my_okHandshakeBytes != null;
+        this.output.write(my_okHandshakeBytes, 0, my_okHandshakeBytes.length);
+        this.output.flush();
       }
     }
   }
@@ -832,7 +817,7 @@ public class Connection implements Runnable {
   /**
    * asynchronously close this connection
    * 
-   * @param beingSick
+   * @param beingSick test hook to simulate sickness in communications & membership
    */
   private void asyncClose(boolean beingSick) {
     // note: remoteAddr may be null if this is a receiver that hasn't finished its handshake
@@ -890,8 +875,7 @@ public class Connection implements Runnable {
 
     InternalDistributedMember myAddr = this.owner.getConduit().getMemberId();
     final MsgOutputStream connectHandshake = new MsgOutputStream(CONNECT_HANDSHAKE_SIZE);
-    // connectHandshake.reset();
-    /**
+    /*
      * Note a byte of zero is always written because old products serialized a member id
with always
      * sends an ip address. My reading of the ip-address specs indicated that the first byte
of a
      * valid address would never be 0.
@@ -925,8 +909,6 @@ public class Connection implements Runnable {
   private void handshakeStream() throws IOException {
     waitForAddressCompletion();
 
-    // this.output = new BufferedOutputStream(getSocket().getOutputStream(),
-    // owner.getConduit().bufferSize);
     this.output = getSocket().getOutputStream();
     ByteArrayOutputStream baos = new ByteArrayOutputStream(CONNECT_HANDSHAKE_SIZE);
     DataOutputStream os = new DataOutputStream(baos);
@@ -961,17 +943,12 @@ public class Connection implements Runnable {
     lenbytes[MSG_HEADER_SIZE_OFFSET + 2] = (byte) ((len / 0x100) & 0xff);
     lenbytes[MSG_HEADER_SIZE_OFFSET + 3] = (byte) (len & 0xff);
     lenbytes[MSG_HEADER_TYPE_OFFSET] = (byte) NORMAL_MSG_TYPE;
-    lenbytes[MSG_HEADER_ID_OFFSET] = (byte) ((MsgIdGenerator.NO_MSG_ID / 0x100) & 0xff);
+    lenbytes[MSG_HEADER_ID_OFFSET] = (byte) ((MsgIdGenerator.NO_MSG_ID >> 8) &
0xff);
     lenbytes[MSG_HEADER_ID_OFFSET + 1] = (byte) (MsgIdGenerator.NO_MSG_ID & 0xff);
     synchronized (outLock) {
-      try {
-        // this.writerThread = Thread.currentThread();
-        this.output.write(lenbytes, 0, lenbytes.length);
-        this.output.write(msg, 0, msg.length);
-        this.output.flush();
-      } finally {
-        // this.writerThread = null;
-      }
+      this.output.write(lenbytes, 0, lenbytes.length);
+      this.output.write(msg, 0, msg.length);
+      this.output.flush();
     }
   }
 
@@ -1091,7 +1068,7 @@ public class Connection implements Runnable {
         // create connection
         try {
           conn = null;
-          conn = new Connection(mgr, t, preserveOrder, remoteAddr, sharedResource);
+          conn = new Connection(t, preserveOrder, remoteAddr, sharedResource);
         } catch (javax.net.ssl.SSLHandshakeException se) {
           // no need to retry if certificates were rejected
           throw se;
@@ -1206,7 +1183,7 @@ public class Connection implements Runnable {
 
   private void setRemoteAddr(DistributedMember m) {
     this.remoteAddr = this.owner.getDM().getCanonicalId(m);
-    MembershipManager mgr = this.owner.owner.getMembershipManager();
+    MembershipManager mgr = this.conduit.getMembershipManager();
     mgr.addSurpriseMember(m);
   }
 
@@ -1214,9 +1191,8 @@ public class Connection implements Runnable {
    * creates a new connection to a remote server. We are initiating this connection; the
other side
    * must accept us We will almost always send messages; small acks are received.
    */
-  private Connection(MembershipManager mgr, ConnectionTable t, boolean preserveOrder,
-      DistributedMember remoteID, boolean sharedResource)
-      throws IOException, DistributedSystemDisconnectedException {
+  private Connection(ConnectionTable t, boolean preserveOrder, DistributedMember remoteID,
+      boolean sharedResource) throws IOException, DistributedSystemDisconnectedException
{
 
     // initialize a socket upfront. So that the
     InternalDistributedMember remoteAddr = (InternalDistributedMember) remoteID;
@@ -1224,6 +1200,7 @@ public class Connection implements Runnable {
       throw new IllegalArgumentException(
           LocalizedStrings.Connection_CONNECTIONTABLE_IS_NULL.toLocalizedString());
     }
+    this.conduit = t.getConduit();
     this.isReceiver = false;
     this.owner = t;
     this.sharedResource = sharedResource;
@@ -1248,7 +1225,7 @@ public class Connection implements Runnable {
 
         channel.socket().setKeepAlive(SocketCreator.ENABLE_TCP_KEEP_ALIVE);
 
-        /**
+        /*
          * If conserve-sockets is false, the socket can be used for receiving responses,
so set the
          * receive buffer accordingly.
          */
@@ -1261,7 +1238,7 @@ public class Connection implements Runnable {
         setSendBufferSize(channel.socket());
         channel.configureBlocking(true);
 
-        int connectTime = getP2PConnectTimeout();;
+        int connectTime = getP2PConnectTimeout();
 
         try {
           channel.socket().connect(addr, connectTime);
@@ -1276,7 +1253,7 @@ public class Connection implements Runnable {
             Thread.currentThread().interrupt();
           }
           throw c;
-        } catch (CancelledKeyException e) {
+        } catch (CancelledKeyException | ClosedSelectorException e) {
           // bug #44469: for some reason NIO throws this runtime exception
           // instead of an IOException on timeouts
           ConnectException c = new ConnectException(
@@ -1284,14 +1261,6 @@ public class Connection implements Runnable {
                   .toLocalizedString(new Object[] {connectTime}));
           c.initCause(e);
           throw c;
-        } catch (ClosedSelectorException e) {
-          // bug #44808: for some reason JRockit NIO thorws this runtime exception
-          // instead of an IOException on timeouts
-          ConnectException c = new ConnectException(
-              LocalizedStrings.Connection_ATTEMPT_TO_CONNECT_TIMED_OUT_AFTER_0_MILLISECONDS
-                  .toLocalizedString(new Object[] {connectTime}));
-          c.initCause(e);
-          throw c;
         }
       } finally {
         this.owner.removeConnectingSocket(channel.socket());
@@ -1309,7 +1278,6 @@ public class Connection implements Runnable {
         setSocketBufferSize(this.socket, false, socketBufferSize, true);
         setSendBufferSize(this.socket);
       } else {
-        // socket = new Socket(remoteAddr.getInetAddress(), remoteAddr.getPort());
         Socket s = new Socket();
         this.socket = s;
         s.setTcpNoDelay(true);
@@ -1335,13 +1303,12 @@ public class Connection implements Runnable {
    * must not be doing it correctly.
    */
   private static final boolean BATCH_SENDS = Boolean.getBoolean("p2p.batchSends");
-  protected static final int BATCH_BUFFER_SIZE =
+  private static final int BATCH_BUFFER_SIZE =
       Integer.getInteger("p2p.batchBufferSize", 1024 * 1024).intValue();
-  protected static final int BATCH_FLUSH_MS =
-      Integer.getInteger("p2p.batchFlushTime", 50).intValue();
-  protected Object batchLock;
-  protected ByteBuffer fillBatchBuffer;
-  protected ByteBuffer sendBatchBuffer;
+  private static final int BATCH_FLUSH_MS = Integer.getInteger("p2p.batchFlushTime", 50).intValue();
+  private Object batchLock;
+  private ByteBuffer fillBatchBuffer;
+  private ByteBuffer sendBatchBuffer;
   private BatchBufferFlusher batchFlusher;
 
   private void createBatchSendBuffer() {
@@ -1446,13 +1413,7 @@ public class Connection implements Runnable {
                   SocketChannel channel = getSocket().getChannel();
                   nioWriteFully(channel, sendBatchBuffer, false, null);
                   sendBatchBuffer.clear();
-                } catch (IOException ex) {
-                  logger.fatal(LocalizedMessage.create(
-                      LocalizedStrings.Connection_EXCEPTION_FLUSHING_BATCH_SEND_BUFFER_0,
ex));
-                  readerShuttingDown = true;
-                  requestClose(LocalizedStrings.Connection_EXCEPTION_FLUSHING_BATCH_SEND_BUFFER_0
-                      .toLocalizedString(ex));
-                } catch (ConnectionException ex) {
+                } catch (IOException | ConnectionException ex) {
                   logger.fatal(LocalizedMessage.create(
                       LocalizedStrings.Connection_EXCEPTION_FLUSHING_BATCH_SEND_BUFFER_0,
ex));
                   readerShuttingDown = true;
@@ -1526,13 +1487,6 @@ public class Connection implements Runnable {
     return this.closing.get();
   }
 
-  /**
-   * Used to close a connection that has not yet been registered with the distribution manager.
-   */
-  void closePartialConnect(String reason) {
-    close(reason, false, false, false, false);
-  }
-
   void closePartialConnect(String reason, boolean beingSick) {
     close(reason, false, false, beingSick, false);
   }
@@ -1619,9 +1573,9 @@ public class Connection implements Runnable {
       // if network partition detection is enabled or this is an admin vm
       // we can't wait for the reader thread when running in an IBM JRE. See
       // bug 41889
-      if (this.owner.owner.config.getEnableNetworkPartitionDetection()
-          || this.owner.owner.getMemberId().getVmKind() == DistributionManager.ADMIN_ONLY_DM_TYPE
-          || this.owner.owner.getMemberId().getVmKind() == DistributionManager.LOCATOR_DM_TYPE)
{
+      if (this.conduit.config.getEnableNetworkPartitionDetection()
+          || this.conduit.getMemberId().getVmKind() == DistributionManager.ADMIN_ONLY_DM_TYPE
+          || this.conduit.getMemberId().getVmKind() == DistributionManager.LOCATOR_DM_TYPE)
{
         isIBM = "IBM Corporation".equals(System.getProperty("java.vm.vendor"));
       }
       {
@@ -1673,6 +1627,7 @@ public class Connection implements Runnable {
               }
             }
           } else {
+            // noinspection ConstantConditions
             this.owner.removeSharedConnection(reason, this.remoteAddr, this.preserveOrder,
this);
           }
         } else if (!this.isReceiver) {
@@ -1735,7 +1690,7 @@ public class Connection implements Runnable {
       initiateSuspicionIfSharedUnordered();
       if (this.isReceiver) {
         if (!this.sharedResource) {
-          this.owner.owner.stats.incThreadOwnedReceivers(-1L, dominoCount.get());
+          this.conduit.stats.incThreadOwnedReceivers(-1L, dominoCount.get());
         }
         asyncClose(false);
         this.owner.removeAndCloseThreadOwnedSockets();
@@ -1759,7 +1714,7 @@ public class Connection implements Runnable {
   }
 
   private String p2pReaderName() {
-    StringBuffer sb = new StringBuffer(64);
+    StringBuilder sb = new StringBuilder(64);
     if (this.isReceiver) {
       sb.append("P2P message reader@");
     } else {
@@ -1973,8 +1928,8 @@ public class Connection implements Runnable {
     }
 
     msg = msg.toLowerCase();
-    return (msg.indexOf("forcibly closed") >= 0) || (msg.indexOf("reset by peer") >=
0)
-        || (msg.indexOf("connection reset") >= 0);
+    return (msg.contains("forcibly closed")) || (msg.contains("reset by peer"))
+        || (msg.contains("connection reset"));
   }
 
   private static boolean validMsgType(int msgType) {
@@ -2012,7 +1967,7 @@ public class Connection implements Runnable {
           this.idleMsgDestreamer = null;
         } else {
           result = new MsgDestreamer(this.owner.getConduit().stats,
-              this.owner.owner.getCancelCriterion(), v);
+              this.conduit.getCancelCriterion(), v);
         }
         result.setName(p2pReaderName() + " msgId=" + msgId);
         this.destreamerMap.put(key, result);
@@ -2103,7 +2058,7 @@ public class Connection implements Runnable {
         /* byte msgHdrVersion = */ calcHdrVersion(len);
         len = calcMsgByteSize(len);
         int msgType = lenbytes[MSG_HEADER_TYPE_OFFSET];
-        short msgId = (short) ((lenbytes[MSG_HEADER_ID_OFFSET] & 0xff * 0x100)
+        short msgId = (short) (((lenbytes[MSG_HEADER_ID_OFFSET] & 0xff) << 8)
             + (lenbytes[MSG_HEADER_ID_OFFSET + 1] & 0xff));
         boolean myDirectAck = (msgType & DIRECT_ACK_BIT) != 0;
         if (myDirectAck) {
@@ -2384,7 +2339,7 @@ public class Connection implements Runnable {
                   // logger.fine("thread-owned receiver with domino count of " + dominoNumber
+ "
                   // will prefer shared sockets");
                 }
-                this.owner.owner.stats.incThreadOwnedReceivers(1L, dominoNumber);
+                this.conduit.stats.incThreadOwnedReceivers(1L, dominoNumber);
               }
 
               if (logger.isDebugEnabled()) {
@@ -2541,11 +2496,6 @@ public class Connection implements Runnable {
         }
         bytesSoFar += bytesThisTime;
       } catch (InterruptedIOException io) {
-        // try { Thread.sleep(10); }
-        // catch (InterruptedException ie) {
-        // Thread.currentThread().interrupt();
-        // }
-
         // Current thread has been interrupted. Regard it similar to an EOF
         this.readerShuttingDown = true;
         try {
@@ -2582,7 +2532,7 @@ public class Connection implements Runnable {
     final boolean origSocketInUse = this.socketInUse;
     byte originalState = -1;
     synchronized (stateLock) {
-      originalState = this.connectionState;;
+      originalState = this.connectionState;
       this.connectionState = STATE_SENDING;
     }
     this.socketInUse = true;
@@ -2597,13 +2547,8 @@ public class Connection implements Runnable {
         } else {
           byte[] bytesToWrite = getBytesToWrite(buffer);
           synchronized (outLock) {
-            try {
-              // this.writerThread = Thread.currentThread();
-              this.output.write(bytesToWrite);
-              this.output.flush();
-            } finally {
-              // this.writerThread = null;
-            }
+            this.output.write(bytesToWrite);
+            this.output.flush();
           }
         }
       }
@@ -2763,15 +2708,7 @@ public class Connection implements Runnable {
     return bytesToWrite;
   }
 
-  // private String socketInfo() {
-  // return (" socket: " + getSocket().getLocalAddress() + ":" + getSocket().getLocalPort()
+ " -> "
-  // +
-  // getSocket().getInetAddress() + ":" + getSocket().getPort() + " connection = " +
-  // System.identityHashCode(this));
-  //
-  // }
-
-  private final boolean addToQueue(ByteBuffer buffer, DistributionMessage msg, boolean force)
+  private boolean addToQueue(ByteBuffer buffer, DistributionMessage msg, boolean force)
       throws ConnectionException {
     final DMStats stats = this.owner.getConduit().stats;
     long start = DistributionStats.getStatTime();
@@ -2891,7 +2828,7 @@ public class Connection implements Runnable {
    * 
    * @throws ConnectionException if the conduit has stopped
    */
-  private final boolean handleBlockedWrite(ByteBuffer buffer, DistributionMessage msg)
+  private boolean handleBlockedWrite(ByteBuffer buffer, DistributionMessage msg)
       throws ConnectionException {
     if (!addToQueue(buffer, msg, true)) {
       return false;
@@ -2931,7 +2868,7 @@ public class Connection implements Runnable {
     this.pusherThread.start();
   }
 
-  private final ByteBuffer takeFromOutgoingQueue() throws InterruptedException {
+  private ByteBuffer takeFromOutgoingQueue() throws InterruptedException {
     ByteBuffer result = null;
     final DMStats stats = this.owner.getConduit().stats;
     long start = DistributionStats.getStatTime();
@@ -3152,7 +3089,7 @@ public class Connection implements Runnable {
    * Return false if socket writes to be done async/nonblocking Return true if socket writes
to be
    * done sync/blocking
    */
-  private final boolean useSyncWrites(boolean forceAsync) {
+  private boolean useSyncWrites(boolean forceAsync) {
     if (forceAsync) {
       return false;
     }
@@ -3185,7 +3122,7 @@ public class Connection implements Runnable {
 
   static private final int MAX_WAIT_TIME = (1 << 5); // ms (must be a power of 2)
 
-  private final void writeAsync(SocketChannel channel, ByteBuffer buffer, boolean forceAsync,
+  private void writeAsync(SocketChannel channel, ByteBuffer buffer, boolean forceAsync,
       DistributionMessage p_msg, final DMStats stats) throws IOException {
     DistributionMessage msg = p_msg;
     // async/non-blocking
@@ -3301,7 +3238,7 @@ public class Connection implements Runnable {
                 if (msToWait <= 0) {
                   Thread.yield();
                 } else {
-                  boolean interrupted = Thread.interrupted();;
+                  boolean interrupted = Thread.interrupted();
                   try {
                     Thread.sleep(msToWait);
                   } catch (InterruptedException ex) {
@@ -3401,10 +3338,10 @@ public class Connection implements Runnable {
   /**
    * stateLock is used to synchronize state changes.
    */
-  protected Object stateLock = new Object();
+  private final Object stateLock = new Object();
 
   /** for timeout processing, this is the current state of the connection */
-  protected byte connectionState = STATE_IDLE;
+  private byte connectionState = STATE_IDLE;
 
   /* ~~~~~~~~~~~~~ connection states ~~~~~~~~~~~~~~~ */
   /** the connection is idle, but may be in use */
@@ -3420,16 +3357,11 @@ public class Connection implements Runnable {
   /** the connection is in use and is reading a message */
   protected static final byte STATE_READING = 5;
 
-  protected static final String[] STATE_NAMES =
-      new String[] {"idle", "sending", "post_sending", "reading_ack", "received_ack", "reading"};
   /* ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ */
 
   /** set to true if we exceeded the ack-wait-threshold waiting for a response */
   protected volatile boolean ackTimedOut;
 
-  private static int ACK_SIZE = 1;
-  private static byte ACK_BYTE = 37;
-
   /**
    * @param msToWait number of milliseconds to wait for an ack. If 0 then wait forever.
    * @param msInterval interval between checks
@@ -3899,7 +3831,7 @@ public class Connection implements Runnable {
                     // } else {
                     // ConnectionTable.threadWantsSharedResources();
                   }
-                  this.owner.owner.stats.incThreadOwnedReceivers(1L, dominoNumber);
+                  this.conduit.stats.incThreadOwnedReceivers(1L, dominoNumber);
                   // Because this thread is not shared resource, it will be used for direct
                   // ack. Direct ack messages can be large. This call will resize the send
                   // buffer.
@@ -4039,6 +3971,10 @@ public class Connection implements Runnable {
     }
   }
 
+  protected TCPConduit getConduit() {
+    return this.conduit;
+  }
+
   protected Socket getSocket() throws SocketException {
     // fix for bug 37286
     Socket result = this.socket;
@@ -4177,7 +4113,7 @@ public class Connection implements Runnable {
   boolean nioChecked;
   boolean useNIO;
 
-  private final boolean useNIO() {
+  private boolean useNIO() {
     if (TCPConduit.useSSL) {
       return false;
     }
@@ -4193,7 +4129,7 @@ public class Connection implements Runnable {
     if (this.socket != null && (this.socket.getInetAddress() instanceof Inet6Address))
{
       String os = System.getProperty("os.name");
       if (os != null) {
-        if (os.indexOf("Windows") != -1) {
+        if (os.contains("Windows")) {
           this.useNIO = false;
         }
       }

http://git-wip-us.apache.org/repos/asf/geode/blob/6b2b7b2f/geode-core/src/main/java/org/apache/geode/internal/tcp/ConnectionTable.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/tcp/ConnectionTable.java b/geode-core/src/main/java/org/apache/geode/internal/tcp/ConnectionTable.java
index 08a9009..c55af82 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/tcp/ConnectionTable.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/tcp/ConnectionTable.java
@@ -64,17 +64,9 @@ import org.apache.geode.internal.logging.log4j.LocalizedMessage;
  * 
  * @since GemFire 2.1
  */
-/*
- * Note: We no longer use InputMultiplexer If InputMux is reinstated then the manager needs
to be
- * initialized and all lines that have a NOMUX preface should be uncommented
- * 
- */
 public class ConnectionTable {
   private static final Logger logger = LogService.getLogger();
 
-  /** a random number generator for secondary connection selection */
-  // static java.util.Random random = new java.util.Random();
-
   /** warning when descriptor limit reached */
   private static boolean ulimitWarningIssued;
 
@@ -82,6 +74,7 @@ public class ConnectionTable {
    * true if the current thread wants non-shared resources
    */
   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.
@@ -132,9 +125,7 @@ public class ConnectionTable {
   /**
    * the conduit for this table
    */
-  protected final TCPConduit owner;
-  // ARB: temp making this protected to provide access to Connection.
-  // private final TCPConduit owner;
+  private final TCPConduit owner;
 
   /**
    * true if this table is no longer in use
@@ -205,17 +196,10 @@ public class ConnectionTable {
     return (Boolean) threadWantsOwnResources.get();
   }
 
-  // public static void setThreadOwnsResourcesRegistration(
-  // Boolean newValue) {
-  // threadWantsOwnResources.set(newValue);
-  // }
-  // private Map connections = new HashMap();
-  /* NOMUX: private InputMuxManager inputMuxManager; */
-  // private int lowWater;
-  // private int highWater;
+  public TCPConduit getOwner() {
+    return owner;
+  }
 
-  // private static boolean TRACK_SERVER_CONNECTIONS =
-  // System.getProperty("p2p.bidirectional", "true").equals("true");
 
   private ConnectionTable(TCPConduit c) throws IOException {
     this.owner = c;
@@ -226,10 +210,6 @@ public class ConnectionTable {
     this.threadConnectionMap = new ConcurrentHashMap();
     this.p2pReaderThreadPool = createThreadPoolForIO(c.getDM().getSystem().isShareSockets());
     this.socketCloser = new SocketCloser();
-    /*
-     * NOMUX: if (TCPConduit.useNIO) { inputMuxManager = new InputMuxManager(this);
-     * inputMuxManager.start(c.logger); }
-     */
   }
 
   private Executor createThreadPoolForIO(boolean conserveSockets) {
@@ -306,7 +286,6 @@ public class ConnectionTable {
       }
     }
 
-    // Stub id = conn.getRemoteId();
     if (conn != null) {
       synchronized (this.receivers) {
         this.owner.stats.incReceivers();
@@ -322,22 +301,9 @@ public class ConnectionTable {
             conn.remoteAddr);
       }
     }
-    // cleanupHighWater();
   }
 
 
-  // /** returns the connection associated with the given key, or null if
-  // no such connection exists */
-  // protected Connection basicGet(Serializable id) {
-  // synchronized (this.orderedConnectionMap) {
-  // return (Connection) this.orderedConnectionMap.get(id);
-  // }
-  // }
-
-  // protected Connection get(Serializable id) throws java.io.IOException {
-  // return get(id, false);
-  // }
-
 
   /**
    * Process a newly created PendingConnection
@@ -432,10 +398,10 @@ public class ConnectionTable {
   }
 
   /**
-   * unordered or conserve-sockets note that unordered connections are currently always shared
+   * 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 threadOwnsResources whether unordered conn is owned by the current thread
+   * @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
@@ -444,9 +410,9 @@ public class ConnectionTable {
    * @throws IOException if unable to create the connection
    * @throws DistributedSystemDisconnectedException
    */
-  private Connection getUnorderedOrConserveSockets(DistributedMember id,
-      boolean threadOwnsResources, boolean preserveOrder, long startTime, long ackTimeout,
-      long ackSATimeout) throws IOException, DistributedSystemDisconnectedException {
+  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;
@@ -472,7 +438,7 @@ public class ConnectionTable {
     if (pc != null) {
       result = handleNewPendingConnection(id, true /* fixes bug 43386 */, preserveOrder,
m, pc,
           startTime, ackTimeout, ackSATimeout);
-      if (!preserveOrder && threadOwnsResources) {
+      if (!preserveOrder && scheduleTimeout) {
         scheduleIdleTimeout(result);
       }
     } else { // we have existing connection
@@ -487,10 +453,10 @@ public class ConnectionTable {
             startTime, ackTimeout, ackSATimeout);
         if (logger.isDebugEnabled()) {
           if (result != null) {
-            logger.debug("getUnorderedOrConserveSockets {} myAddr={} theirAddr={}", result,
+            logger.debug("getSharedConnection {} myAddr={} theirAddr={}", result,
                 getConduit().getMemberId(), result.remoteAddr);
           } else {
-            logger.debug("getUnorderedOrConserveSockets: Connect failed");
+            logger.debug("getSharedConnection: Connect failed");
           }
         }
       } else {
@@ -512,7 +478,7 @@ public class ConnectionTable {
    * @throws IOException if the connection could not be created
    * @throws DistributedSystemDisconnectedException
    */
-  Connection getOrderedAndOwned(DistributedMember id, long startTime, long ackTimeout,
+  Connection getThreadOwnedConnection(DistributedMember id, long startTime, long ackTimeout,
       long ackSATimeout) throws IOException, DistributedSystemDisconnectedException {
     Connection result = null;
 
@@ -658,10 +624,10 @@ public class ConnectionTable {
     Connection result = null;
     boolean threadOwnsResources = threadOwnsResources();
     if (!preserveOrder || !threadOwnsResources) {
-      result = getUnorderedOrConserveSockets(id, threadOwnsResources, preserveOrder, startTime,
-          ackTimeout, ackSATimeout);
+      result = getSharedConnection(id, threadOwnsResources, preserveOrder, startTime, ackTimeout,
+          ackSATimeout);
     } else {
-      result = getOrderedAndOwned(id, startTime, ackTimeout, ackSATimeout);
+      result = getThreadOwnedConnection(id, startTime, ackTimeout, ackSATimeout);
     }
     if (result != null) {
       Assert.assertTrue(result.preserveOrder == preserveOrder);

http://git-wip-us.apache.org/repos/asf/geode/blob/6b2b7b2f/geode-core/src/main/java/org/apache/geode/internal/tcp/DirectReplySender.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/tcp/DirectReplySender.java
b/geode-core/src/main/java/org/apache/geode/internal/tcp/DirectReplySender.java
index 3872ee9..bf06953 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/tcp/DirectReplySender.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/tcp/DirectReplySender.java
@@ -56,7 +56,7 @@ class DirectReplySender implements ReplySender {
     // mutates the list when it has exceptions.
 
     // fix for bug #42199 - cancellation check
-    this.conn.owner.getDM().getCancelCriterion().checkCancelInProgress(null);
+    this.conn.getConduit().getDM().getCancelCriterion().checkCancelInProgress(null);
 
     if (logger.isTraceEnabled(LogMarker.DM)) {
       logger.trace(LogMarker.DM, "Sending a direct reply {} to {}", msg, conn.getRemoteAddress());

http://git-wip-us.apache.org/repos/asf/geode/blob/6b2b7b2f/geode-core/src/main/java/org/apache/geode/internal/tcp/MsgReader.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/tcp/MsgReader.java b/geode-core/src/main/java/org/apache/geode/internal/tcp/MsgReader.java
index fc56271..be1f533 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/tcp/MsgReader.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/tcp/MsgReader.java
@@ -94,7 +94,7 @@ public abstract class MsgReader {
   public abstract ByteBuffer readAtLeast(int bytes) throws IOException;
 
   protected DMStats getStats() {
-    return conn.owner.getConduit().stats;
+    return conn.getConduit().stats;
   }
 
   public static class Header {

http://git-wip-us.apache.org/repos/asf/geode/blob/6b2b7b2f/geode-core/src/main/java/org/apache/geode/internal/tcp/NIOMsgReader.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/tcp/NIOMsgReader.java b/geode-core/src/main/java/org/apache/geode/internal/tcp/NIOMsgReader.java
index 50f5fae..a4e35a4 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/tcp/NIOMsgReader.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/tcp/NIOMsgReader.java
@@ -79,7 +79,7 @@ public class NIOMsgReader extends MsgReader {
     if (nioInputBuffer == null) {
       int allocSize = conn.getReceiveBufferSize();
       if (allocSize == -1) {
-        allocSize = conn.owner.getConduit().tcpBufferSize;
+        allocSize = conn.getConduit().tcpBufferSize;
       }
       if (allocSize > bufferSize) {
         bufferSize = allocSize;


Mime
View raw message