GEODE-8947: Use waiting thread pool only in limited senario. (#6038)
* GEODE-8947: Use waiting thread pool only in limited senario.
* Only when processing transactional message and when conserve-sockets set to true case, a
separate thread in waiting pool will be used to process the message.
* This is to addresses performance issue if there won't be deadlock.
(cherry picked from commit b98cf4a3d1fb5e5d64c2d298827d937183ab2b79)
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tx/RemoteOperationMessage.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tx/RemoteOperationMessage.java
index ba49d73..a4cbef5 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tx/RemoteOperationMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tx/RemoteOperationMessage.java
@@ -14,6 +14,7 @@
*/
package org.apache.geode.internal.cache.tx;
+
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
@@ -176,7 +177,23 @@
sendReply(getSender(), this.processorId, dm, replyException, null, 0);
return;
}
- dm.getExecutors().getWaitingThreadPool().execute(() -> doRemoteOperation(dm, cache));
+
+ if (dm.getSystem().threadOwnsResources()) {
+ // reply inline if thread owns socket.
+ doRemoteOperation(dm, cache);
+ return;
+ }
+
+ if (isTransactional()) {
+ dm.getExecutors().getWaitingThreadPool().execute(() -> doRemoteOperation(dm, cache));
+ } else {
+ // reply inline for non-transactional case.
+ doRemoteOperation(dm, cache);
+ }
+ }
+
+ boolean isTransactional() {
+ return getTXUniqId() != TXManagerImpl.NOTX && canParticipateInTransaction();
}
void doRemoteOperation(ClusterDistributionManager dm, InternalCache cache) {
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/tx/RemoteOperationMessageTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/tx/RemoteOperationMessageTest.java
index e10c2f0..aa175d3 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/tx/RemoteOperationMessageTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/tx/RemoteOperationMessageTest.java
@@ -21,6 +21,7 @@
import static org.mockito.Mockito.anyInt;
import static org.mockito.Mockito.anyLong;
import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.mock;
@@ -302,6 +303,52 @@
.hasMessageContaining("system failure");
}
+
+ @Test
+ public void processInvokesDoRemoteOperationIfThreadOwnsResources() {
+ when(system.threadOwnsResources()).thenReturn(true);
+ doNothing().when(msg).doRemoteOperation(dm, cache);
+
+ msg.process(dm);
+
+ verify(msg).doRemoteOperation(dm, cache);
+ verify(msg, never()).isTransactional();
+ }
+
+ @Test
+ public void processInvokesDoRemoteOperationIfThreadDoesNotOwnResourcesAndNotTransactional() {
+ when(system.threadOwnsResources()).thenReturn(false);
+ doReturn(false).when(msg).isTransactional();
+ doNothing().when(msg).doRemoteOperation(dm, cache);
+
+ msg.process(dm);
+
+ verify(msg).doRemoteOperation(dm, cache);
+ verify(msg).isTransactional();
+ }
+
+ @Test
+ public void isTransactionalReturnsFalseIfTXUniqueIdIsNOTX() {
+ assertThat(msg.getTXUniqId()).isEqualTo(TXManagerImpl.NOTX);
+ assertThat(msg.isTransactional()).isFalse();
+ }
+
+ @Test
+ public void isTransactionalReturnsFalseIfCannotParticipateInTransaction() {
+ doReturn(1).when(msg).getTXUniqId();
+ doReturn(false).when(msg).canParticipateInTransaction();
+
+ assertThat(msg.isTransactional()).isFalse();
+ }
+
+ @Test
+ public void isTransactionalReturnsTrueIfHasTXUniqueIdAndCanParticipateInTransaction() {
+ doReturn(1).when(msg).getTXUniqId();
+
+ assertThat(msg.canParticipateInTransaction()).isTrue();
+ assertThat(msg.isTransactional()).isTrue();
+ }
+
private static class TestableRemoteOperationMessage extends RemoteOperationMessage {
private boolean operationOnRegionResult = true;
