This is an automated email from the ASF dual-hosted git repository.
dlmarion pushed a commit to branch 2.1
in repository https://gitbox.apache.org/repos/asf/accumulo.git
The following commit(s) were added to refs/heads/2.1 by this push:
new 30da958f67 Reduce deadlock probability of ClientContext.close /
TabletLocator.getLocator (#6470)
30da958f67 is described below
commit 30da958f6722b21a4e3c8cdb3168c9a5365fa28a
Author: Dave Marion <[email protected]>
AuthorDate: Fri Jul 10 16:03:57 2026 -0400
Reduce deadlock probability of ClientContext.close /
TabletLocator.getLocator (#6470)
Looking at the non-deterministic failures of SessionBlockVerifyIT
I found a deadlock between the ClientContext.close and
TabletLocator.getLocator methods when there are unclosed
Scanners in the client. This change removes most of the
synchronization in ClientContext in favor of memoized
suppliers.
JStack found the deadlock at:
```
Java stack information for the threads listed above:
===================================================
"junit-timeout-thread-4":
at
org.apache.accumulo.core.clientImpl.TabletLocator.disable(TabletLocator.java:134)
- waiting to lock <0x000000042dbf08e8> (a java.lang.Class for
org.apache.accumulo.core.clientImpl.TabletLocator)
at
org.apache.accumulo.core.clientImpl.TabletLocator$1.disable(TabletLocator.java:245)
at
org.apache.accumulo.core.singletons.SingletonManager.disable(SingletonManager.java:106)
at
org.apache.accumulo.core.singletons.SingletonManager$$Lambda/0x00007fbd94325f88.accept(Unknown
Source)
at
java.util.ArrayList.forEach([email protected]/ArrayList.java:1596)
at
org.apache.accumulo.core.singletons.SingletonManager.transition(SingletonManager.java:196)
at
org.apache.accumulo.core.singletons.SingletonManager.releaseReservation(SingletonManager.java:145)
- locked <0x000000042dcc4b68> (a java.lang.Class for
org.apache.accumulo.core.singletons.SingletonManager)
at
org.apache.accumulo.core.singletons.SingletonReservation.close(SingletonReservation.java:50)
at
org.apache.accumulo.core.clientImpl.ClientContext.close(ClientContext.java:894)
- locked <0x000000042d857668> (a
org.apache.accumulo.core.clientImpl.ClientContext)
at
org.apache.accumulo.test.functional.SessionBlockVerifyIT.run(SessionBlockVerifyIT.java:179)
"pool-2-thread-5":
at
org.apache.accumulo.core.clientImpl.ClientContext.tableZooHelper(ClientContext.java:626)
- waiting to lock <0x000000042d857668> (a
org.apache.accumulo.core.clientImpl.ClientContext)
at
org.apache.accumulo.core.clientImpl.ClientContext.getTableState(ClientContext.java:671)
at
org.apache.accumulo.core.clientImpl.TabletLocator.getLocator(TabletLocator.java:148)
- locked <0x000000042dbf08e8> (a java.lang.Class for
org.apache.accumulo.core.clientImpl.TabletLocator)
at
org.apache.accumulo.core.clientImpl.ThriftScanner.scan(ThriftScanner.java:518)
at
org.apache.accumulo.core.clientImpl.ScannerIterator.readBatch(ScannerIterator.java:159)
- locked <0x000000042d863fb8> (a
org.apache.accumulo.core.clientImpl.ThriftScanner$ScanState)
at
org.apache.accumulo.core.clientImpl.ScannerIterator.getNextBatch(ScannerIterator.java:177)
at
org.apache.accumulo.core.clientImpl.ScannerIterator.hasNext(ScannerIterator.java:109)
at
org.apache.accumulo.test.functional.SessionBlockVerifyIT.lambda$0(SessionBlockVerifyIT.java:122)
at
org.apache.accumulo.test.functional.SessionBlockVerifyIT$$Lambda/0x00007fbd943acba8.call(Unknown
Source)
```
---
.../accumulo/core/clientImpl/ClientContext.java | 147 +++++++++------------
.../core/clientImpl/ThriftTransportPool.java | 2 +-
.../org/apache/accumulo/server/ServerContext.java | 7 +-
3 files changed, 65 insertions(+), 91 deletions(-)
diff --git
a/core/src/main/java/org/apache/accumulo/core/clientImpl/ClientContext.java
b/core/src/main/java/org/apache/accumulo/core/clientImpl/ClientContext.java
index d267d4b7d4..db1cd21ea3 100644
--- a/core/src/main/java/org/apache/accumulo/core/clientImpl/ClientContext.java
+++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/ClientContext.java
@@ -128,9 +128,9 @@ public class ClientContext implements AccumuloClient {
private final ZooReader zooReader;
private final ZooCache zooCache;
- private Credentials creds;
- private BatchWriterConfig batchWriterConfig;
- private ConditionalWriterConfig conditionalWriterConfig;
+ private Supplier<Credentials> creds;
+ private final Supplier<BatchWriterConfig> batchWriterConfig;
+ private final Supplier<ConditionalWriterConfig> conditionalWriterConfig;
private final AccumuloConfiguration serverConf;
private final Configuration hadoopConf;
@@ -141,21 +141,25 @@ public class ClientContext implements AccumuloClient {
private final Supplier<SslConnectionParams> sslSupplier;
private final Supplier<ScanServerSelector> scanServerSelectorSupplier;
private TCredentials rpcCreds;
- private ThriftTransportPool thriftTransportPool;
- private ZookeeperLockChecker zkLockChecker;
+ protected Supplier<ThriftTransportPool> thriftTransportPool;
+ private final Supplier<ZookeeperLockChecker> zkLockChecker;
+ private volatile boolean scannerReadAheadPoolCreated = false;
+ private volatile boolean cleanupThreadPoolCreated = false;
+ private volatile boolean thriftTransportPoolCreated = false;
private volatile boolean closed = false;
- private SecurityOperations secops = null;
+ private final Supplier<SecurityOperations> secops;
private final TableOperationsImpl tableops;
private final NamespaceOperations namespaceops;
- private InstanceOperations instanceops = null;
+ private final Supplier<InstanceOperations> instanceops;
@SuppressWarnings("deprecation")
private org.apache.accumulo.core.client.admin.ReplicationOperations
replicationops = null;
private final SingletonReservation singletonReservation;
private final Supplier<ThreadPools> clientThreadPools;
- private ThreadPoolExecutor cleanupThreadPool;
- private ThreadPoolExecutor scannerReadaheadPool;
+ private final Supplier<ThreadPoolExecutor> cleanupThreadPool;
+ private final Supplier<ThreadPoolExecutor> scannerReadaheadPool;
+ private final Supplier<TableZooHelper> tableZooHelper;
private void ensureOpen() {
if (closed) {
@@ -253,6 +257,21 @@ public class ClientContext implements AccumuloClient {
clientThreadPools = () ->
ThreadPools.getClientThreadPools(getConfiguration(), ueh);
}
}
+ scannerReadaheadPool = memoize(() -> clientThreadPools.get()
+
.getPoolBuilder(SCANNER_READ_AHEAD_POOL).numCoreThreads(0).numMaxThreads(Integer.MAX_VALUE)
+ .withTimeOut(3L, SECONDS).withQueue(new SynchronousQueue<>()).build());
+ cleanupThreadPool =
+ memoize(() ->
clientThreadPools.get().getPoolBuilder(CONDITIONAL_WRITER_CLEANUP_POOL)
+ .numCoreThreads(1).withTimeOut(3L, SECONDS).build());
+ creds = memoize(() -> new Credentials(info.getPrincipal(),
info.getAuthenticationToken()));
+ batchWriterConfig = memoize(() ->
getBatchWriterConfig(info.getProperties()));
+ conditionalWriterConfig = memoize(() ->
getConditionalWriterConfig(info.getProperties()));
+ tableZooHelper = memoize(() -> new TableZooHelper(this));
+ secops = memoize(() -> new SecurityOperationsImpl(this));
+ instanceops = memoize(() -> new InstanceOperationsImpl(this));
+ thriftTransportPool =
+ memoize(() ->
ThriftTransportPool.startNew(this::getTransportPoolMaxAgeMillis, false));
+ zkLockChecker = memoize(() -> new ZookeeperLockChecker(this));
}
public Ample getAmple() {
@@ -260,24 +279,16 @@ public class ClientContext implements AccumuloClient {
return new AmpleImpl(this);
}
- public synchronized Future<List<KeyValue>>
- submitScannerReadAheadTask(Callable<List<KeyValue>> c) {
+ public Future<List<KeyValue>>
submitScannerReadAheadTask(Callable<List<KeyValue>> c) {
ensureOpen();
- if (scannerReadaheadPool == null) {
- scannerReadaheadPool =
clientThreadPools.get().getPoolBuilder(SCANNER_READ_AHEAD_POOL)
- .numCoreThreads(0).numMaxThreads(Integer.MAX_VALUE).withTimeOut(3L,
SECONDS)
- .withQueue(new SynchronousQueue<>()).build();
- }
- return scannerReadaheadPool.submit(c);
+ scannerReadAheadPoolCreated = true;
+ return scannerReadaheadPool.get().submit(c);
}
- public synchronized void executeCleanupTask(Runnable r) {
+ public void executeCleanupTask(Runnable r) {
ensureOpen();
- if (cleanupThreadPool == null) {
- cleanupThreadPool =
clientThreadPools.get().getPoolBuilder(CONDITIONAL_WRITER_CLEANUP_POOL)
- .numCoreThreads(1).withTimeOut(3L, SECONDS).build();
- }
- this.cleanupThreadPool.execute(r);
+ cleanupThreadPoolCreated = true;
+ this.cleanupThreadPool.get().execute(r);
}
/**
@@ -291,12 +302,9 @@ public class ClientContext implements AccumuloClient {
/**
* Retrieve the credentials used to construct this context
*/
- public synchronized Credentials getCredentials() {
+ public Credentials getCredentials() {
ensureOpen();
- if (creds == null) {
- creds = new Credentials(info.getPrincipal(),
info.getAuthenticationToken());
- }
- return creds;
+ return creds.get();
}
public String getPrincipal() {
@@ -318,10 +326,10 @@ public class ClientContext implements AccumuloClient {
* Update the credentials in the current context after changing the current
user's password or
* other auth token
*/
- public synchronized void setCredentials(Credentials newCredentials) {
+ public void setCredentials(Credentials newCredentials) {
ensureOpen();
checkArgument(newCredentials != null, "newCredentials is null");
- creds = newCredentials;
+ creds = memoize(() -> newCredentials);
rpcCreds = null;
}
@@ -391,12 +399,9 @@ public class ClientContext implements AccumuloClient {
return batchWriterConfig;
}
- public synchronized BatchWriterConfig getBatchWriterConfig() {
+ public BatchWriterConfig getBatchWriterConfig() {
ensureOpen();
- if (batchWriterConfig == null) {
- batchWriterConfig = getBatchWriterConfig(info.getProperties());
- }
- return batchWriterConfig;
+ return batchWriterConfig.get();
}
/**
@@ -451,12 +456,9 @@ public class ClientContext implements AccumuloClient {
return conditionalWriterConfig;
}
- public synchronized ConditionalWriterConfig getConditionalWriterConfig() {
+ public ConditionalWriterConfig getConditionalWriterConfig() {
ensureOpen();
- if (conditionalWriterConfig == null) {
- conditionalWriterConfig =
getConditionalWriterConfig(info.getProperties());
- }
- return conditionalWriterConfig;
+ return conditionalWriterConfig.get();
}
/**
@@ -620,14 +622,9 @@ public class ClientContext implements AccumuloClient {
return zooCache;
}
- private TableZooHelper tableZooHelper;
-
- private synchronized TableZooHelper tableZooHelper() {
+ private TableZooHelper tableZooHelper() {
ensureOpen();
- if (tableZooHelper == null) {
- tableZooHelper = new TableZooHelper(this);
- }
- return tableZooHelper;
+ return tableZooHelper.get();
}
public TableId getTableId(String tableName) throws TableNotFoundException {
@@ -816,35 +813,27 @@ public class ClientContext implements AccumuloClient {
}
@Override
- public synchronized TableOperations tableOperations() {
+ public TableOperations tableOperations() {
ensureOpen();
return tableops;
}
@Override
- public synchronized NamespaceOperations namespaceOperations() {
+ public NamespaceOperations namespaceOperations() {
ensureOpen();
return namespaceops;
}
@Override
- public synchronized SecurityOperations securityOperations() {
+ public SecurityOperations securityOperations() {
ensureOpen();
- if (secops == null) {
- secops = new SecurityOperationsImpl(this);
- }
-
- return secops;
+ return secops.get();
}
@Override
- public synchronized InstanceOperations instanceOperations() {
+ public InstanceOperations instanceOperations() {
ensureOpen();
- if (instanceops == null) {
- instanceops = new InstanceOperationsImpl(this);
- }
-
- return instanceops;
+ return instanceops.get();
}
@Override
@@ -879,17 +868,15 @@ public class ClientContext implements AccumuloClient {
@Override
public synchronized void close() {
closed = true;
- if (thriftTransportPool != null) {
- thriftTransportPool.shutdown();
- }
- if (tableZooHelper != null) {
- tableZooHelper.close();
+ if (thriftTransportPoolCreated) {
+ thriftTransportPool.get().shutdown();
}
- if (scannerReadaheadPool != null) {
- scannerReadaheadPool.shutdownNow(); // abort all tasks, client is
shutting down
+ tableZooHelper.get().close();
+ if (scannerReadAheadPoolCreated) {
+ scannerReadaheadPool.get().shutdownNow(); // abort all tasks, client is
shutting down
}
- if (cleanupThreadPool != null) {
- cleanupThreadPool.shutdown(); // wait for shutdown tasks to execute
+ if (cleanupThreadPoolCreated) {
+ cleanupThreadPool.get().shutdown(); // wait for shutdown tasks to execute
}
singletonReservation.close();
}
@@ -1118,24 +1105,14 @@ public class ClientContext implements AccumuloClient {
return
ClientProperty.RPC_TRANSPORT_IDLE_TIMEOUT.getTimeInMillis(getProperties());
}
- public synchronized ThriftTransportPool getTransportPool() {
- return getTransportPoolImpl(false);
- }
-
- protected synchronized ThriftTransportPool getTransportPoolImpl(boolean
shouldHalt) {
+ public ThriftTransportPool getTransportPool() {
ensureOpen();
- if (thriftTransportPool == null) {
- thriftTransportPool =
- ThriftTransportPool.startNew(this::getTransportPoolMaxAgeMillis,
shouldHalt);
- }
- return thriftTransportPool;
+ thriftTransportPoolCreated = true;
+ return thriftTransportPool.get();
}
- public synchronized ZookeeperLockChecker getTServerLockChecker() {
+ public ZookeeperLockChecker getTServerLockChecker() {
ensureOpen();
- if (this.zkLockChecker == null) {
- this.zkLockChecker = new ZookeeperLockChecker(this);
- }
- return this.zkLockChecker;
+ return this.zkLockChecker.get();
}
}
diff --git
a/core/src/main/java/org/apache/accumulo/core/clientImpl/ThriftTransportPool.java
b/core/src/main/java/org/apache/accumulo/core/clientImpl/ThriftTransportPool.java
index 59a3b9d8c2..2f543eed31 100644
---
a/core/src/main/java/org/apache/accumulo/core/clientImpl/ThriftTransportPool.java
+++
b/core/src/main/java/org/apache/accumulo/core/clientImpl/ThriftTransportPool.java
@@ -117,7 +117,7 @@ public class ThriftTransportPool {
* and false if the thread is running within a client
* @return a new instance with its checker thread started to clean up idle
transports
*/
- static ThriftTransportPool startNew(LongSupplier maxAgeMillis, boolean
shouldHalt) {
+ public static ThriftTransportPool startNew(LongSupplier maxAgeMillis,
boolean shouldHalt) {
var pool = new ThriftTransportPool(maxAgeMillis, shouldHalt);
log.debug("Set thrift transport pool idle time to {}ms",
maxAgeMillis.getAsLong());
pool.checkThread.start();
diff --git
a/server/base/src/main/java/org/apache/accumulo/server/ServerContext.java
b/server/base/src/main/java/org/apache/accumulo/server/ServerContext.java
index 9f1a0f607d..12feff1371 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/ServerContext.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/ServerContext.java
@@ -132,6 +132,8 @@ public class ServerContext extends ClientContext {
memoize(() -> new AuditedSecurityOperation(this,
SecurityOperation.getAuthorizor(this),
SecurityOperation.getAuthenticator(this),
SecurityOperation.getPermHandler(this)));
metricsInfoSupplier = memoize(() -> new MetricsInfoImpl(this));
+ thriftTransportPool =
+ memoize(() ->
ThriftTransportPool.startNew(this::getTransportPoolMaxAgeMillis, true));
}
/**
@@ -451,11 +453,6 @@ public class ServerContext extends ClientContext {
return getClientTimeoutInMillis();
}
- @Override
- public synchronized ThriftTransportPool getTransportPool() {
- return getTransportPoolImpl(true);
- }
-
public AuditedSecurityOperation getSecurityOperation() {
return securityOperation.get();
}
