This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 98c4aa324ec Pipe: skip pipe CN heartbeat and sync when
`PipeTaskCoordinatorLock` is held by another thread (#11509)
98c4aa324ec is described below
commit 98c4aa324ecc256a750da58227d9e870bf798421
Author: V_Galaxy <[email protected]>
AuthorDate: Thu Nov 9 20:25:55 2023 +0800
Pipe: skip pipe CN heartbeat and sync when `PipeTaskCoordinatorLock` is
held by another thread (#11509)
When the DN processing logic is blocked during the create (start) pipe, due
to the timeout and retry mechanism, the write locks of CN's
`PipeTaskCoordinatorLock` and DN's `PipeMetaKeeper` will be held for a long
period of time, causing the timed `heartbeat` logic and `sync` logic to make no
progress at all during this period, which in turn leads to a large number of
related procedures and RPCs piling up. This PR partially sacrifices the
liveness of `heartbeat` and `sync` to avoid the [...]
---
Minimum Reproduction:
```sql
INSERT INTO root.db.d(time, m) values (1, 1);
create pipe a2b
with connector (
'connector'='iotdb-air-gap-connector',
'connector.ip'='192.0.2.0',
'connector.port'='12345',
'connector.air-gap.handshake-timeout-ms'='400000'
);
```
Here, the IP is a reserved address, so the socket connect process in
`iotdb-air-gap-connector` will continue to be blocked until timeout.
The client can observe the error message `Msg: 1107: Procedure execution
timed out.`. Meanwhile, CN will be unable to process additional pipe requests
due to the accumulation of procedures and RPCs.
---
Detailed Explanation for added skip logic:
For the `heartbeat` operation, if it is found that
`PipeTaskCoordinatorLock` is held by another thread before execution, according
to the calling relationship, there are no more than the following situations:
- create / start / stop / drop pipe, these procedures will call the
`pushSinglePipeMeta` rpc of DN in the `OPERATE_ON_DATA_NODES` step
(`org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv#pushSinglePipeMetaToDataNodes/dropSinglePipeOnDataNodes`),
these two steps will acquire the write lock of `PipeMetaKeeper` (DN) on the DN
side. Notice that `heartbeat` will also call the `pipeHeartbeat` rpc of DN, and
the read lock of `PipeMetaKeeper` will be acquired on the DN side. If [...]
- parse heartbeat, indicating that the previous round of `heartbeat` has
not been processed (impossible, guaranteed by `synchronized`?)
- auto restart / handle successful restart, indicating that `sync` is being
processed at this time, these two steps will acquire the write lock of
`PipeMetaKeeper` (CN) (No effect)
- `PipeMetaSyncProcedure` / `PipeHandleMetaChangeProcedure` /
`PipeHandleLeaderChangeProcedure`, these procedures will call the
`pushPipeMeta` rpc of DN, analysis of the same as above
In this way, when performing the `heartbeat` operation, if it is found that
`PipeTaskCoordinatorLock` is held by another thread, if the other thread is
blocked, the `heartbeat` will **most likely** be blocked, so just skip.
The same analysis can be performed for `sync` operation.
------
* fix: skip heartbeat and sync when PipeTaskCoordinatorLock is held by
another thread
* fix: skip the sync logic ahead of time
---
.../confignode/manager/pipe/runtime/PipeHeartbeatScheduler.java | 6 ++++++
.../iotdb/confignode/manager/pipe/runtime/PipeMetaSyncer.java | 8 +++++++-
.../iotdb/confignode/manager/pipe/task/PipeTaskCoordinator.java | 7 ++++++-
.../confignode/manager/pipe/task/PipeTaskCoordinatorLock.java | 4 ++++
4 files changed, 23 insertions(+), 2 deletions(-)
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeHeartbeatScheduler.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeHeartbeatScheduler.java
index 04ecb0fcf0c..02400e1a3da 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeHeartbeatScheduler.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeHeartbeatScheduler.java
@@ -82,6 +82,12 @@ public class PipeHeartbeatScheduler {
return;
}
+ if (configManager.getPipeManager().getPipeTaskCoordinator().isLocked()) {
+ LOGGER.warn(
+ "PipeTaskCoordinatorLock is held by another thread, skip this round
of heartbeat to avoid procedure and rpc accumulation as much as possible");
+ return;
+ }
+
final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
configManager.getNodeManager().getRegisteredDataNodeLocations();
final TPipeHeartbeatReq request = new
TPipeHeartbeatReq(System.currentTimeMillis());
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeMetaSyncer.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeMetaSyncer.java
index 82482aa8c41..c6047e3684f 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeMetaSyncer.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/runtime/PipeMetaSyncer.java
@@ -86,7 +86,13 @@ public class PipeMetaSyncer {
}
}
- private void sync() {
+ private synchronized void sync() {
+ if (configManager.getPipeManager().getPipeTaskCoordinator().isLocked()) {
+ LOGGER.warn(
+ "PipeTaskCoordinatorLock is held by another thread, skip this round
of sync to avoid procedure and rpc accumulation as much as possible");
+ return;
+ }
+
final ProcedureManager procedureManager =
configManager.getProcedureManager();
boolean somePipesNeedRestarting = false;
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/task/PipeTaskCoordinator.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/task/PipeTaskCoordinator.java
index 2e378bb9b7b..25d5911478a 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/task/PipeTaskCoordinator.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/task/PipeTaskCoordinator.java
@@ -66,9 +66,10 @@ public class PipeTaskCoordinator {
public AtomicReference<PipeTaskInfo> tryLock() {
if (pipeTaskCoordinatorLock.tryLock()) {
pipeTaskInfoHolder = new AtomicReference<>(pipeTaskInfo);
+ return pipeTaskInfoHolder;
}
- return pipeTaskInfoHolder;
+ return null;
}
/**
@@ -93,6 +94,10 @@ public class PipeTaskCoordinator {
}
}
+ public boolean isLocked() {
+ return pipeTaskCoordinatorLock.isLocked();
+ }
+
/** Caller should ensure that the method is called in the lock {@link
#tryLock()}. */
public TSStatus createPipe(TCreatePipeReq req) {
return configManager.getProcedureManager().createPipe(req);
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/task/PipeTaskCoordinatorLock.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/task/PipeTaskCoordinatorLock.java
index fff9e6e1029..fcaf98f2024 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/task/PipeTaskCoordinatorLock.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/task/PipeTaskCoordinatorLock.java
@@ -100,4 +100,8 @@ public class PipeTaskCoordinatorLock {
Thread.currentThread().getName());
}
}
+
+ boolean isLocked() {
+ return !deque.isEmpty();
+ }
}
