ctubbsii commented on code in PR #6378:
URL: https://github.com/apache/accumulo/pull/6378#discussion_r3262338172
##########
server/manager/src/main/java/org/apache/accumulo/manager/Manager.java:
##########
@@ -201,8 +226,8 @@ public class Manager extends AbstractServer implements
LiveTServerSet.Listener,
Collections.synchronizedMap(new HashMap<>());
final Set<TServerInstance> serversToShutdown =
Collections.synchronizedSet(new HashSet<>());
final Migrations migrations = new Migrations();
+ private final MergeLocks mergeLocks = new MergeLocks();
Review Comment:
```suggestion
private final LoadingCache<TableId,ReentrantLock> mergeLocks =
Caffeine.newBuilder().weakValues()
.scheduler(Scheduler.systemScheduler()).build(k -> new
ReentrantLock());
```
My main concern with this is that the TableId in the keys will prevent
cleanup of the TableId.cache weak values. The scheduler is supposed to help
with cleanup of the keys whose values have been garbage collected, but it may
also be a good idea to use String (`tableId.canonical()`) for the key.
##########
server/manager/src/main/java/org/apache/accumulo/manager/Manager.java:
##########
@@ -496,15 +523,19 @@ public MergeInfo getMergeInfo(TableId tableId) {
log.warn("Unexpected error reading merge state", ex);
return new MergeInfo();
}
+ } finally {
+ l.unlock();
}
}
public void setMergeState(MergeInfo info, MergeState state)
throws KeeperException, InterruptedException {
ServerContext context = getContext();
- synchronized (mergeLock) {
- String path =
- getZooKeeperRoot() + Constants.ZTABLES + "/" +
info.getExtent().tableId() + "/merge";
+ final TableId tid = info.getExtent().tableId();
+ final ReentrantLock l = mergeLocks.getLock(tid);
Review Comment:
```suggestion
final ReentrantLock l = mergeLocks.get(tid);
```
##########
server/manager/src/main/java/org/apache/accumulo/manager/Manager.java:
##########
@@ -477,7 +502,9 @@ public TServerConnection getConnection(TServerInstance
server) {
public MergeInfo getMergeInfo(TableId tableId) {
ServerContext context = getContext();
- synchronized (mergeLock) {
+ final ReentrantLock l = mergeLocks.getLock(tableId);
Review Comment:
```suggestion
final ReentrantLock l = mergeLocks.get(tableId);
```
##########
server/manager/src/main/java/org/apache/accumulo/manager/Manager.java:
##########
@@ -519,16 +550,20 @@ public void setMergeState(MergeInfo info, MergeState
state)
state.equals(MergeState.STARTED) ? ZooUtil.NodeExistsPolicy.FAIL
: ZooUtil.NodeExistsPolicy.OVERWRITE);
}
- mergeLock.notifyAll();
+ } finally {
+ l.unlock();
}
nextEvent.event("Merge state of %s set to %s", info.getExtent(), state);
}
public void clearMergeState(TableId tableId) throws KeeperException,
InterruptedException {
- synchronized (mergeLock) {
+ final ReentrantLock l = mergeLocks.getLock(tableId);
Review Comment:
```suggestion
final ReentrantLock l = mergeLocks.get(tableId);
```
##########
server/manager/src/main/java/org/apache/accumulo/manager/Manager.java:
##########
@@ -1479,6 +1514,9 @@ boolean canSuspendTablets() {
throw new IllegalStateException("Exception updating manager lock", e);
}
+ ThreadPools.watchCriticalScheduledTask(
+
context.getScheduledExecutor().scheduleWithFixedDelay(mergeLocks::cleanup, 3,
3, HOURS));
+
Review Comment:
No need for this cleanup code with the weakValues cache's scheduler.
##########
server/manager/src/main/java/org/apache/accumulo/manager/Manager.java:
##########
@@ -172,6 +173,30 @@
public class Manager extends AbstractServer implements
LiveTServerSet.Listener, TableObserver,
CurrentState, HighlyAvailableService, ServerProcessService.Iface {
+ private final class MergeLocks {
+
+ private final Object lock = new Object();
+ private Map<TableId,ReentrantLock> lockStorage = new
HashMap<TableId,ReentrantLock>();
+
+ private ReentrantLock getLock(TableId tid) {
+ synchronized (lock) {
+ return lockStorage.computeIfAbsent(tid, k -> new ReentrantLock(true));
+ }
+ }
+
+ private void cleanup() {
+ synchronized (lock) {
+ Set<TableId> removals = new HashSet<>();
+ for (Entry<TableId,ReentrantLock> e : lockStorage.entrySet()) {
+ if (!getContext().tableNodeExists(e.getKey()) &&
!e.getValue().isLocked()) {
+ removals.add(e.getKey());
+ }
+ }
+ removals.forEach(lockStorage::remove);
+ }
+ }
+ }
+
Review Comment:
```suggestion
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
