You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by bs...@apache.org on 2020/03/23 15:28:08 UTC
[geode] branch feature/GEODE-6008 updated: work in progress
This is an automated email from the ASF dual-hosted git repository.
bschuchardt pushed a commit to branch feature/GEODE-6008
in repository https://gitbox.apache.org/repos/asf/geode.git
The following commit(s) were added to refs/heads/feature/GEODE-6008 by this push:
new fe92b05 work in progress
fe92b05 is described below
commit fe92b05021f87a6708970b9754f2f48b11916c0a
Author: Bruce Schuchardt <bs...@pivotal.io>
AuthorDate: Mon Mar 23 08:27:04 2020 -0700
work in progress
removed old test. Changed NioSslEngine to throw IOException instead
of IllegalStateException. There are handlers for the latter in reader
threads but not for the former.
---
.../geode/internal/tcp/CloseConnectionTest.java | 13 ++++
.../internal/tcp/TCPConduitBareBonesDUnitTest.java | 76 ----------------------
.../apache/geode/internal/net/NioSslEngine.java | 4 +-
.../org/apache/geode/internal/tcp/Connection.java | 17 +++--
.../geode/internal/net/NioSslEngineTest.java | 8 ++-
5 files changed, 33 insertions(+), 85 deletions(-)
diff --git a/geode-core/src/distributedTest/java/org/apache/geode/internal/tcp/CloseConnectionTest.java b/geode-core/src/distributedTest/java/org/apache/geode/internal/tcp/CloseConnectionTest.java
index 154e908..abe8ea9 100644
--- a/geode-core/src/distributedTest/java/org/apache/geode/internal/tcp/CloseConnectionTest.java
+++ b/geode-core/src/distributedTest/java/org/apache/geode/internal/tcp/CloseConnectionTest.java
@@ -22,6 +22,8 @@ import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DistributionImpl;
+import org.apache.geode.distributed.internal.InternalDistributedSystem;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.cache.CacheTestCase;
@@ -63,6 +65,17 @@ public class CloseConnectionTest extends CacheTestCase {
ConnectionTable conTable = getConnectionTable();
assertThat(conTable.getNumberOfReceivers()).isEqualTo(2);
});
+
+ // Make sure the Sender connection has a reader thread
+ vm1.invoke(() -> {
+ ConnectionTable conTable = getConnectionTable();
+ InternalDistributedSystem distributedSystem = getSystem();
+ InternalDistributedMember otherMember = distributedSystem.getDistributionManager()
+ .getOtherNormalDistributionManagerIds().iterator().next();
+ Connection connection = conTable.getConduit().getConnection(otherMember, true, false,
+ System.currentTimeMillis(), 15000, 0);
+ assertThat(connection.hasResidualReaderThread()).isTrue();
+ });
}
private ConnectionTable getConnectionTable() {
diff --git a/geode-core/src/distributedTest/java/org/apache/geode/internal/tcp/TCPConduitBareBonesDUnitTest.java b/geode-core/src/distributedTest/java/org/apache/geode/internal/tcp/TCPConduitBareBonesDUnitTest.java
deleted file mode 100644
index 77eb02e..0000000
--- a/geode-core/src/distributedTest/java/org/apache/geode/internal/tcp/TCPConduitBareBonesDUnitTest.java
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
- * 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 org.assertj.core.api.Assertions.assertThat;
-
-import java.util.Properties;
-
-import org.junit.Rule;
-import org.junit.Test;
-
-import org.apache.geode.distributed.ConfigurationProperties;
-import org.apache.geode.distributed.internal.DistributionConfig;
-import org.apache.geode.distributed.internal.DistributionConfigImpl;
-import org.apache.geode.distributed.internal.DistributionImpl;
-import org.apache.geode.distributed.internal.InternalDistributedSystem;
-import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
-import org.apache.geode.test.dunit.DistributedTestUtils;
-import org.apache.geode.test.dunit.rules.DistributedRule;
-
-public class TCPConduitBareBonesDUnitTest {
-
- @Rule
- public DistributedRule distributedRule = new DistributedRule(0);
-
- @Test
- public void testShortTermSharedConnections() throws Exception {
- // create and close many shared, ordered connections. Also see CloseConnectionTest,
- // which ensures that the outgoing socket in a shared connection is properly closed
- Properties properties = new Properties();
- properties.put(ConfigurationProperties.LOG_LEVEL, "fine");
- properties.put(ConfigurationProperties.LOCATORS, DistributedTestUtils.getLocators());
- DistributionConfig configuration = new DistributionConfigImpl(properties);
- InternalDistributedSystem distributedSystem =
- (InternalDistributedSystem) InternalDistributedSystem.connect(configuration.toProperties());
- TCPConduit conduit =
- ((DistributionImpl) distributedSystem.getDistributionManager().getDistribution())
- .getDirectChannel().getConduit();
-
- InternalDistributedMember memberID =
- distributedSystem.getDistributionManager().getDistribution().getView().getCreator();
- ConnectionTable.threadWantsSharedResources();
-
- try {
- for (int i = 0; i < 1000; i++) {
- long buffersSize =
- distributedSystem.getDistributionManager().getStats().getSenderBufferSize(true);
- // connect to the dunit locator. The first iteration will get the Connection that was formed
- // during startup, but we'll close it and start creating new ones.
- Connection connection =
- conduit.getConnection(memberID, true, false, System.currentTimeMillis(), 15000, 0);
- System.out.println("Test iteration " + i + ": " + connection);
- connection.requestClose("for test");
- // the connection should be stopped at this point
- assertThat(connection.isReceiverStopped()).isTrue();
- // make sure there are no double releases of ByteBuffers
- assertThat(distributedSystem.getDistributionManager().getStats().getSenderBufferSize(true))
- .isEqualTo(buffersSize);
- }
- } finally {
- distributedSystem.disconnect();
- }
- }
-}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/net/NioSslEngine.java b/geode-core/src/main/java/org/apache/geode/internal/net/NioSslEngine.java
index 9d67594..2d55fa3 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/net/NioSslEngine.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/net/NioSslEngine.java
@@ -212,9 +212,9 @@ public class NioSslEngine implements NioFilter {
return bufferPool.expandWriteBufferIfNeeded(type, existing, desiredCapacity);
}
- void checkClosed() {
+ void checkClosed() throws IOException {
if (closed) {
- throw new IllegalStateException("NioSslEngine has been closed");
+ throw new IOException("NioSslEngine has been closed");
}
}
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 893d42d..e568f5a 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
@@ -474,7 +474,7 @@ public class Connection implements Runnable {
* a Reader thread for an shared Connection will remain around in order to
* ensure that the socket is properly closed.
*/
- private volatile boolean isResidualReaderThread;
+ private volatile boolean hasResidualReaderThread;
/**
* creates a "reader" connection that we accepted (it was initiated by an explicit connect being
@@ -1332,7 +1332,7 @@ public class Connection implements Runnable {
}
// make sure our socket is closed
asyncClose(false);
- if (!isReceiver && !isResidualReaderThread) {
+ if (!isReceiver && !hasResidualReaderThread()) {
// receivers release the input buffer when exiting run(). Senders use the
// inputBuffer for reading direct-reply responses
releaseInputBuffer();
@@ -1634,7 +1634,7 @@ public class Connection implements Runnable {
} else {
// not exiting and not a Reader spawned from a ServerSocket.accept(), so
// let's set some state noting that this is happening
- isResidualReaderThread = true;
+ hasResidualReaderThread = true;
}
}
@@ -1690,7 +1690,7 @@ public class Connection implements Runnable {
}
}
} finally {
- isResidualReaderThread = false;
+ hasResidualReaderThread = false;
if (!handshakeHasBeenRead || (sharedResource && !asyncMode)) {
synchronized (stateLock) {
connectionState = STATE_IDLE;
@@ -3251,6 +3251,15 @@ public class Connection implements Runnable {
}
/**
+ * A shared sender connection will leave a reader thread around to ensure that the
+ * socket is properly closed at this end. When that is the case isResidualReaderThread
+ * will return true.
+ */
+ public boolean hasResidualReaderThread() {
+ return hasResidualReaderThread;
+ }
+
+ /**
* answers whether this connection is used for ordered message delivery
*/
boolean getPreserveOrder() {
diff --git a/geode-core/src/test/java/org/apache/geode/internal/net/NioSslEngineTest.java b/geode-core/src/test/java/org/apache/geode/internal/net/NioSslEngineTest.java
index 1d5e4a6..ef16a21 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/net/NioSslEngineTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/net/NioSslEngineTest.java
@@ -34,6 +34,7 @@ import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import java.io.IOException;
import java.net.Socket;
import java.net.SocketException;
import java.nio.ByteBuffer;
@@ -196,11 +197,11 @@ public class NioSslEngineTest {
@Test
- public void checkClosed() {
+ public void checkClosed() throws Exception {
nioSslEngine.checkClosed();
}
- @Test(expected = IllegalStateException.class)
+ @Test(expected = IOException.class)
public void checkClosedThrows() throws Exception {
when(mockEngine.wrap(any(ByteBuffer.class), any(ByteBuffer.class))).thenReturn(
new SSLEngineResult(CLOSED, FINISHED, 0, 100));
@@ -340,7 +341,8 @@ public class NioSslEngineTest {
when(mockEngine.wrap(any(ByteBuffer.class), any(ByteBuffer.class))).thenReturn(
new SSLEngineResult(CLOSED, FINISHED, 0, 0));
nioSslEngine.close(mockChannel);
- assertThatThrownBy(() -> nioSslEngine.checkClosed()).isInstanceOf(IllegalStateException.class);
+ assertThatThrownBy(() -> nioSslEngine.checkClosed()).isInstanceOf(IOException.class)
+ .hasMessageContaining("NioSslEngine has been closed");
nioSslEngine.close(mockChannel);
}