Caideyipi commented on code in PR #11532:
URL: https://github.com/apache/iotdb/pull/11532#discussion_r1392539754
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeMetaSyncer.java:
##########
@@ -87,9 +94,21 @@ public synchronized void start() {
}
private synchronized void sync() {
+ if (!configManager.getPipeManager().getPipeTaskCoordinator().hasAnyPipe()
Review Comment:
Now a big problem emerges that if we observe the task info as "empty", it
doesn't mean that it must be empty when we execute "metasync", because since
there is no lock to the procedure here, to extreme there might be a "create
pipe" slipped through the procedures between the check and the meta sync...
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeMetaSyncer.java:
##########
@@ -104,16 +123,42 @@ private synchronized void sync() {
pipeAutoRestartRoundCounter.set(0);
}
- final TSStatus status = procedureManager.pipeMetaSync();
-
- if (somePipesNeedRestarting
- && status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- handleSuccessfulRestartWithLock();
- procedureManager.pipeHandleMetaChange(true, false);
- }
+ final TSStatus metaSyncStatus = procedureManager.pipeMetaSync();
+
+ if (metaSyncStatus.getCode() ==
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ boolean canSkipNextSync = false;
+
+ if (somePipesNeedRestarting) {
+ final boolean isRestartSuccessful = handleSuccessfulRestartWithLock();
+ final TSStatus handleMetaChangeStatus =
+ procedureManager.pipeHandleMetaChangeWithBlock(true, false);
+ if
(!configManager.getPipeManager().getPipeTaskCoordinator().hasAnyPipe()
+ && isRestartSuccessful
+ && handleMetaChangeStatus.getCode() ==
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ canSkipNextSync = true;
+ }
+ } else {
+ if
(!configManager.getPipeManager().getPipeTaskCoordinator().hasAnyPipe()) {
+ canSkipNextSync = true;
+ }
+ }
- if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- procedureManager.pipeHandleMetaChange(true, true);
+ // When the sync operation succeeded while there was no pipe in CN, we
need to set the
+ // isLastEmptyPipeSyncSuccessful variable and record the version of
PipeTaskInfo at that time.
+ if (canSkipNextSync) {
+ LOGGER.info("Skip next round of sync if no pipe and no modification on
PipeTaskInfo.");
Review Comment:
Actually, it skips all rounds of syncs until one pipe is created, not just
next round
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeMetaSyncer.java:
##########
@@ -87,9 +94,21 @@ public synchronized void start() {
}
private synchronized void sync() {
+ if (!configManager.getPipeManager().getPipeTaskCoordinator().hasAnyPipe()
+ && isLastEmptyPipeSyncSuccessful
+ && lastEmptyPipeSyncWriteLockAcquiredCount
Review Comment:
And the pipe heartbeat won't produce "pipeTaskInfoLockCount" if it has
nothing to do with the update of the CN pipe task metas(*^_^*)
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeMetaSyncer.java:
##########
@@ -87,9 +94,21 @@ public synchronized void start() {
}
private synchronized void sync() {
+ if (!configManager.getPipeManager().getPipeTaskCoordinator().hasAnyPipe()
+ && isLastEmptyPipeSyncSuccessful
+ && lastEmptyPipeSyncWriteLockAcquiredCount
Review Comment:
I think the two things can be solved without extra lock, if we just check if
the nextCount == lastCount + 1 if we executed the metasync, or nextCount ==
lastCount if without it. If not equals, some other pipe procedures must
interfere here, then we must push the meta sync continually.
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeMetaSyncer.java:
##########
@@ -87,9 +94,21 @@ public synchronized void start() {
}
private synchronized void sync() {
+ if (!configManager.getPipeManager().getPipeTaskCoordinator().hasAnyPipe()
+ && isLastEmptyPipeSyncSuccessful
+ && lastEmptyPipeSyncWriteLockAcquiredCount
Review Comment:
Besides, an ABA might just happen between the two read/write of the
"lastLockCount", out of the "metasync" procedure... Also to extreme, it may
just happen between 2 arbitrary continuous lines here... Then “this count” ==
“last count” doesn't actually guarantee the absence of "ABA".
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeMetaSyncer.java:
##########
@@ -87,9 +94,21 @@ public synchronized void start() {
}
private synchronized void sync() {
+ if (!configManager.getPipeManager().getPipeTaskCoordinator().hasAnyPipe()
+ && isLastEmptyPipeSyncSuccessful
+ && lastEmptyPipeSyncWriteLockAcquiredCount
Review Comment:
Besides, an ABA might just happen between the two read/write of the
"lastLockCount", out of the "metasync" procedure... Also to extreme, it may
just happen between 2 arbitrary continuous lines here... Then “this count” ==
“last count” doesn't actually guarantee the absence of "ABA".
--
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]
