You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by ds...@apache.org on 2015/09/23 01:39:35 UTC
[1/3] incubator-geode git commit: CacheClientProxy now always uses
the SocketCloser on its socket. It now uses an AtomicBoolean to prevent
scheduling it more than once.
Repository: incubator-geode
Updated Branches:
refs/heads/feature/GEODE-332 159d78167 -> 0bc967dbd
CacheClientProxy now always uses the SocketCloser on its socket.
It now uses an AtomicBoolean to prevent scheduling it more than once.
Project: http://git-wip-us.apache.org/repos/asf/incubator-geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-geode/commit/193dd386
Tree: http://git-wip-us.apache.org/repos/asf/incubator-geode/tree/193dd386
Diff: http://git-wip-us.apache.org/repos/asf/incubator-geode/diff/193dd386
Branch: refs/heads/feature/GEODE-332
Commit: 193dd3866010c794ee1b12e6a1f35fc3ea3f5270
Parents: 159d781
Author: Darrel Schneider <ds...@pivotal.io>
Authored: Tue Sep 22 16:09:24 2015 -0700
Committer: Darrel Schneider <ds...@pivotal.io>
Committed: Tue Sep 22 16:09:24 2015 -0700
----------------------------------------------------------------------
.../cache/tier/sockets/CacheClientProxy.java | 33 +++++++++++---------
1 file changed, 18 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/193dd386/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/CacheClientProxy.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/CacheClientProxy.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/CacheClientProxy.java
index 15ad73b..85c7493 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/CacheClientProxy.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/CacheClientProxy.java
@@ -63,7 +63,6 @@ import com.gemstone.gemfire.cache.query.internal.cq.InternalCqQuery;
import com.gemstone.gemfire.distributed.DistributedMember;
import com.gemstone.gemfire.distributed.internal.DistributionManager;
import com.gemstone.gemfire.distributed.internal.InternalDistributedSystem;
-import com.gemstone.gemfire.internal.SocketCloser;
import com.gemstone.gemfire.internal.SystemTimer;
import com.gemstone.gemfire.internal.SystemTimer.SystemTimerTask;
import com.gemstone.gemfire.internal.Version;
@@ -116,6 +115,8 @@ public class CacheClientProxy implements ClientSession {
* The socket between the server and the client
*/
protected Socket _socket;
+
+ private final AtomicBoolean _socketClosed = new AtomicBoolean();
/**
* A communication buffer used by each message we send to the client
@@ -960,10 +961,7 @@ public class CacheClientProxy implements ClientSession {
// to fix bug 37684
// 1. check to see if dispatcher is still alive
if (this._messageDispatcher.isAlive()) {
- if (this._socket != null && !this._socket.isClosed()) {
- this._cacheClientNotifier.getSocketCloser().asyncClose(this._socket, this._remoteHostAddress, null);
- getCacheClientNotifier().getAcceptorStats().decCurrentQueueConnections();
- }
+ closeSocket();
destroyRQ();
alreadyDestroyed = true;
this._messageDispatcher.interrupt();
@@ -996,21 +994,26 @@ public class CacheClientProxy implements ClientSession {
}
}
- private void closeTransientFields() {
- // Close the socket
- if (this._socket != null && !this._socket.isClosed()) {
- try {
- this._socket.close();
- getCacheClientNotifier().getAcceptorStats().decCurrentQueueConnections();
- } catch (IOException e) {/*ignore*/}
+ private void closeSocket() {
+ if (this._socketClosed.compareAndSet(false, true)) {
+ // Close the socket
+ this._cacheClientNotifier.getSocketCloser().asyncClose(this._socket, this._remoteHostAddress, null);
+ getCacheClientNotifier().getAcceptorStats().decCurrentQueueConnections();
}
+ }
+
+ private void closeTransientFields() {
+ closeSocket();
// Null out comm buffer, host address, ports and proxy id. All will be
// replaced when the client reconnects.
releaseCommBuffer();
- if (this._remoteHostAddress != null) {
- this._cacheClientNotifier.getSocketCloser().releaseResourcesForAddress(this._remoteHostAddress);
- this._remoteHostAddress = null;
+ {
+ String remoteHostAddress = this._remoteHostAddress;
+ if (remoteHostAddress != null) {
+ this._cacheClientNotifier.getSocketCloser().releaseResourcesForAddress(remoteHostAddress);
+ this._remoteHostAddress = null;
+ }
}
try {
this.cils[RegisterInterestTracker.interestListIndex].clearClientInterestList();
[2/3] incubator-geode git commit: The closer on CacheClientNotifier
now waits 50ms for the asyncClose request to complete. The unit tests now
configure a closer much like the one used by the notifier.
Posted by ds...@apache.org.
The closer on CacheClientNotifier now waits 50ms for the asyncClose request to complete.
The unit tests now configure a closer much like the one used by the notifier.
Project: http://git-wip-us.apache.org/repos/asf/incubator-geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-geode/commit/b317428c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-geode/tree/b317428c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-geode/diff/b317428c
Branch: refs/heads/feature/GEODE-332
Commit: b317428c35005f8c2827fe45d6e154c2b6d69669
Parents: 193dd38
Author: Darrel Schneider <ds...@pivotal.io>
Authored: Tue Sep 22 16:20:05 2015 -0700
Committer: Darrel Schneider <ds...@pivotal.io>
Committed: Tue Sep 22 16:20:05 2015 -0700
----------------------------------------------------------------------
.../java/com/gemstone/gemfire/internal/SocketCloser.java | 8 ++++++--
.../internal/cache/tier/sockets/CacheClientNotifier.java | 2 +-
.../com/gemstone/gemfire/internal/SocketCloserJUnitTest.java | 5 +++--
.../gemfire/internal/SocketCloserWithWaitJUnitTest.java | 3 ++-
4 files changed, 12 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/b317428c/gemfire-core/src/main/java/com/gemstone/gemfire/internal/SocketCloser.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/SocketCloser.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/SocketCloser.java
index be98db7..8468daa 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/SocketCloser.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/SocketCloser.java
@@ -56,8 +56,8 @@ public class SocketCloser {
public SocketCloser() {
this(ASYNC_CLOSE_POOL_KEEP_ALIVE_SECONDS, ASYNC_CLOSE_POOL_MAX_THREADS, ASYNC_CLOSE_WAIT_MILLISECONDS, TimeUnit.MILLISECONDS);
}
- public SocketCloser(int asyncClosePoolMaxThreads) {
- this(ASYNC_CLOSE_POOL_KEEP_ALIVE_SECONDS, asyncClosePoolMaxThreads, ASYNC_CLOSE_WAIT_MILLISECONDS, TimeUnit.MILLISECONDS);
+ public SocketCloser(int asyncClosePoolMaxThreads, long asyncCloseWaitMillis) {
+ this(ASYNC_CLOSE_POOL_KEEP_ALIVE_SECONDS, asyncClosePoolMaxThreads, asyncCloseWaitMillis, TimeUnit.MILLISECONDS);
}
public SocketCloser(long asyncClosePoolKeepAliveSeconds, int asyncClosePoolMaxThreads, long asyncCloseWaitTime, TimeUnit asyncCloseWaitUnits) {
this.asyncClosePoolKeepAliveSeconds = asyncClosePoolKeepAliveSeconds;
@@ -65,6 +65,10 @@ public class SocketCloser {
this.asyncCloseWaitTime = asyncCloseWaitTime;
this.asyncCloseWaitUnits = asyncCloseWaitUnits;
}
+
+ public int getMaxThreads() {
+ return this.asyncClosePoolMaxThreads;
+ }
private ThreadPoolExecutor getAsyncThreadExecutor(String address) {
synchronized (asyncCloseExecutors) {
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/b317428c/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/CacheClientNotifier.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/CacheClientNotifier.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/CacheClientNotifier.java
index d23e1d7..deddfd1 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/CacheClientNotifier.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/CacheClientNotifier.java
@@ -2123,7 +2123,7 @@ public class CacheClientNotifier {
// Set the Cache
this.setCache((GemFireCacheImpl)cache);
this.acceptorStats = acceptorStats;
- this.socketCloser = new SocketCloser(1); // we only need one thread per client
+ this.socketCloser = new SocketCloser(1, 50); // we only need one thread per client and wait 50ms for close
// Set the LogWriter
this.logWriter = (InternalLogWriter)cache.getLogger();
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/b317428c/gemfire-core/src/test/java/com/gemstone/gemfire/internal/SocketCloserJUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/test/java/com/gemstone/gemfire/internal/SocketCloserJUnitTest.java b/gemfire-core/src/test/java/com/gemstone/gemfire/internal/SocketCloserJUnitTest.java
index 681b2a7..0b66ec5 100644
--- a/gemfire-core/src/test/java/com/gemstone/gemfire/internal/SocketCloserJUnitTest.java
+++ b/gemfire-core/src/test/java/com/gemstone/gemfire/internal/SocketCloserJUnitTest.java
@@ -97,12 +97,13 @@ public class SocketCloserJUnitTest {
// Each thread pool (one for A and one for B) has a max of 8 threads.
// So verify that this many are currently waiting on cdl.
{
+ final int maxThreads = this.socketCloser.getMaxThreads();
WaitCriterion wc = new WaitCriterion() {
public boolean done() {
- return waitingToClose.get() == 2*8;
+ return waitingToClose.get() == 2*maxThreads;
}
public String description() {
- return "expected 16 waiters but found only " + waitingToClose.get();
+ return "expected " + 2*maxThreads + " waiters but found only " + waitingToClose.get();
}
};
DistributedTestCase.waitForCriterion(wc, 5000, 10, true);
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/b317428c/gemfire-core/src/test/java/com/gemstone/gemfire/internal/SocketCloserWithWaitJUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/test/java/com/gemstone/gemfire/internal/SocketCloserWithWaitJUnitTest.java b/gemfire-core/src/test/java/com/gemstone/gemfire/internal/SocketCloserWithWaitJUnitTest.java
index 5f3eb5b..9e63743 100644
--- a/gemfire-core/src/test/java/com/gemstone/gemfire/internal/SocketCloserWithWaitJUnitTest.java
+++ b/gemfire-core/src/test/java/com/gemstone/gemfire/internal/SocketCloserWithWaitJUnitTest.java
@@ -8,6 +8,7 @@ import com.gemstone.gemfire.test.junit.categories.UnitTest;
/**
* Tests SocketCloser with a wait time. The default SocketCloser does not wait.
+ * This test configures a closer much like the one used by CacheClientNotifier.
*/
@Category(UnitTest.class)
public class SocketCloserWithWaitJUnitTest extends SocketCloserJUnitTest {
@@ -15,7 +16,7 @@ public class SocketCloserWithWaitJUnitTest extends SocketCloserJUnitTest {
protected SocketCloser createSocketCloser() {
return new SocketCloser(
SocketCloser.ASYNC_CLOSE_POOL_KEEP_ALIVE_SECONDS,
- SocketCloser.ASYNC_CLOSE_POOL_MAX_THREADS,
+ 1, // max threads
1, TimeUnit.NANOSECONDS);
}
}
[3/3] incubator-geode git commit: Connection no longer passes a
Runnable to asyncClose. By doing this the Runnable keeps the Connection
instance alive. If you have a bunch of async closes queued up then this might
consume more memory than is desired. The
Posted by ds...@apache.org.
Connection no longer passes a Runnable to asyncClose.
By doing this the Runnable keeps the Connection instance alive.
If you have a bunch of async closes queued up then this might consume
more memory than is desired. The code that used to be in the Runnable
is now in a new prepareForAsyncClose method. This code will not block
for long periods of time like close can so it is ok to do it inline.
Project: http://git-wip-us.apache.org/repos/asf/incubator-geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-geode/commit/0bc967db
Tree: http://git-wip-us.apache.org/repos/asf/incubator-geode/tree/0bc967db
Diff: http://git-wip-us.apache.org/repos/asf/incubator-geode/diff/0bc967db
Branch: refs/heads/feature/GEODE-332
Commit: 0bc967dbda2b3d3b5d0958f33d5aec53c220f605
Parents: b317428
Author: Darrel Schneider <ds...@pivotal.io>
Authored: Tue Sep 22 16:32:51 2015 -0700
Committer: Darrel Schneider <ds...@pivotal.io>
Committed: Tue Sep 22 16:32:51 2015 -0700
----------------------------------------------------------------------
.../gemfire/internal/tcp/Connection.java | 28 +++++++++++---------
1 file changed, 15 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/0bc967db/gemfire-core/src/main/java/com/gemstone/gemfire/internal/tcp/Connection.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/tcp/Connection.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/tcp/Connection.java
index 6db01bd..88bca22 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/tcp/Connection.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/tcp/Connection.java
@@ -830,29 +830,31 @@ public class Connection implements Runnable {
// we do the close in a background thread because the operation may hang if
// there is a problem with the network. See bug #46659
- Runnable r = new Runnable() {
- public void run() {
- synchronized(stateLock) {
- if (readerThread != null && isRunning && !readerShuttingDown
- && (connectionState == STATE_READING
- || connectionState == STATE_READING_ACK)) {
- readerThread.interrupt();
- }
- }
- }
- };
+
// if simulating sickness, sockets must be closed in-line so that tests know
// that the vm is sick when the beSick operation completes
if (beingSick) {
- r.run();
+ prepareForAsyncClose();
}
else {
if (this.asyncCloseCalled.compareAndSet(false, true)) {
- this.owner.getSocketCloser().asyncClose(this.socket, String.valueOf(this.remoteAddr), r);
+ Socket s = this.socket;
+ if (s != null && !s.isClosed()) {
+ prepareForAsyncClose();
+ this.owner.getSocketCloser().asyncClose(s, String.valueOf(this.remoteAddr), null);
+ }
}
}
}
+ private void prepareForAsyncClose() {
+ synchronized(stateLock) {
+ if (readerThread != null && isRunning && !readerShuttingDown
+ && (connectionState == STATE_READING || connectionState == STATE_READING_ACK)) {
+ readerThread.interrupt();
+ }
+ }
+ }
private static final int CONNECT_HANDSHAKE_SIZE = 4096;