vaijosh commented on code in PR #3164:
URL: https://github.com/apache/hugegraph/pull/3164#discussion_r4035327842
##########
hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandler.java:
##########
@@ -230,6 +230,14 @@ void lock(String path) throws InterruptedException,
void unlock(String path);
+ /**
+ * Non-blocking attempt to reserve the compactRange() window for partition
{@code id}.
+ * Returns false if a compaction is actively running for that partition
right now.
+ */
+ boolean tryLockCompactionRange(int id);
Review Comment:
Addressed.
##########
hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandlerImpl.java:
##########
@@ -1413,30 +1456,58 @@ public boolean dbCompaction(String graphName, int id,
String tableName) {
pathLock.putIfAbsent(path, new
AtomicInteger(compactionCanStart));
compactionState.putIfAbsent(id, new AtomicInteger(0));
log.info("Partition {} dbCompaction started", id);
- if (tableName.isEmpty()) {
- lock(path);
- setState(id, doing);
- log.info("Partition {}-{} got lock, dbCompaction
start", id, path);
- op.compactRange();
- setState(id, compactionDone);
- log.info("Partition {} dbCompaction end and start
to do snapshot", id);
- PartitionEngine pe =
HgStoreEngine.getInstance().getPartitionEngine(id);
- // find leader and send blankTask, after execution
- if (pe.isLeader()) {
- RaftClosure bc = (closure) -> {
- };
-
pe.addRaftTask(RaftOperation.create(RaftOperation.SYNC_BLANK_TASK),
- bc);
+ ReentrantLock rangeLock =
+ compactionRangeLock.computeIfAbsent(id,
+ k -> new
ReentrantLock());
+ boolean rangeLocked = false;
+ try {
+ rangeLocked =
rangeLock.tryLock(compactionRangeLockWaitMillis,
+
TimeUnit.MILLISECONDS);
+ if (rangeLocked) {
+ if (tableName.isEmpty()) {
+ lock(path);
+ setState(id, doing);
+ log.info("Partition {}-{} got lock,
dbCompaction start", id, path);
+ op.compactRange();
+ setState(id, compactionDone);
+ log.info("Partition {} dbCompaction end
and start to do snapshot", id);
+ PartitionEngine pe =
HgStoreEngine.getInstance().getPartitionEngine(id);
+ // find leader and send blankTask, after
execution
+ if (pe.isLeader()) {
+ RaftClosure bc = (closure) -> {
+ };
+
pe.addRaftTask(RaftOperation.create(RaftOperation.SYNC_BLANK_TASK),
+ bc);
+ } else {
+ HgCmdClient client =
HgStoreEngine.getInstance().getHgCmdClient();
+ BlankTaskRequest request = new
BlankTaskRequest();
+ request.setGraphName("");
+ request.setPartitionId(id);
+
client.tryInternalCallSyncWithRpc(request);
+ }
+ setAndNotifyState(id, compactionDone);
+ } else {
+ op.compactRange(tableName);
+ }
} else {
- HgCmdClient client =
HgStoreEngine.getInstance().getHgCmdClient();
- BlankTaskRequest request = new
BlankTaskRequest();
- request.setGraphName("");
- request.setPartitionId(id);
- client.tryInternalCallSyncWithRpc(request);
+ log.warn("Partition {} skip dbCompaction,
snapshot save " +
+ "still in progress after {}ms wait",
id,
+ compactionRangeLockWaitMillis);
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ log.warn("Partition {} dbCompaction interrupted
while waiting " +
+ "for snapshot range lock", id);
+ // Interrupted while waiting for the snapshot save
to release
+ // the range lock. The path lock was already
acquired and
+ // must be released here, otherwise later
compactions for this
+ // partition would block until the path lock
timeout.
+ unlock(path);
Review Comment:
Addressed
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
