Skip to content

Commit 4945fbc

Browse files
CaideyipiCaideyipi
authored
Pipe: Further reduce the pipe logs & Added configurations for tsFile segment lock (#16315) (#16327)
* logger-fix * add-ons * fix * fix * fix * warn-fix * further-fix * fix
1 parent 2b9fdb9 commit 4945fbc

9 files changed

Lines changed: 148 additions & 98 deletions

File tree

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -644,7 +644,7 @@ public TSStatus handleMetaChanges(final PipeHandleMetaChangePlan plan) {
644644
}
645645

646646
private TSStatus handleMetaChangesInternal(final PipeHandleMetaChangePlan plan) {
647-
LOGGER.info("Handling pipe meta changes ...");
647+
LOGGER.debug("Handling pipe meta changes ...");
648648

649649
pipeMetaKeeper.clear();
650650

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -104,12 +104,12 @@ protected AtomicReference<PipeTaskInfo> acquireLockInternal(
104104

105105
@Override
106106
protected ProcedureLockState acquireLock(ConfigNodeProcedureEnv configNodeProcedureEnv) {
107-
LOGGER.info("ProcedureId {} try to acquire pipe lock.", getProcId());
107+
LOGGER.debug("ProcedureId {} try to acquire pipe lock.", getProcId());
108108
pipeTaskInfo = acquireLockInternal(configNodeProcedureEnv);
109109
if (pipeTaskInfo == null) {
110110
LOGGER.warn("ProcedureId {} failed to acquire pipe lock.", getProcId());
111111
} else {
112-
LOGGER.info("ProcedureId {} acquired pipe lock.", getProcId());
112+
LOGGER.debug("ProcedureId {} acquired pipe lock.", getProcId());
113113
}
114114

115115
final ProcedureLockState procedureLockState = super.acquireLock(configNodeProcedureEnv);
@@ -120,7 +120,7 @@ protected ProcedureLockState acquireLock(ConfigNodeProcedureEnv configNodeProced
120120
"ProcedureId {}: LOCK_ACQUIRED. The following procedure should not be executed without pipe lock.",
121121
getProcId());
122122
} else {
123-
LOGGER.info(
123+
LOGGER.debug(
124124
"ProcedureId {}: LOCK_ACQUIRED. The following procedure should be executed with pipe lock.",
125125
getProcId());
126126
}
@@ -129,7 +129,7 @@ protected ProcedureLockState acquireLock(ConfigNodeProcedureEnv configNodeProced
129129
if (pipeTaskInfo == null) {
130130
LOGGER.warn("ProcedureId {}: LOCK_EVENT_WAIT. Without acquiring pipe lock.", getProcId());
131131
} else {
132-
LOGGER.info("ProcedureId {}: LOCK_EVENT_WAIT. Pipe lock will be released.", getProcId());
132+
LOGGER.debug("ProcedureId {}: LOCK_EVENT_WAIT. Pipe lock will be released.", getProcId());
133133
configNodeProcedureEnv
134134
.getConfigManager()
135135
.getPipeManager()
@@ -168,7 +168,7 @@ protected void releaseLock(ConfigNodeProcedureEnv configNodeProcedureEnv) {
168168
if (pipeTaskInfo == null) {
169169
LOGGER.warn("ProcedureId {} release lock. No need to release pipe lock.", getProcId());
170170
} else {
171-
LOGGER.info("ProcedureId {} release lock. Pipe lock will be released.", getProcId());
171+
LOGGER.debug("ProcedureId {} release lock. Pipe lock will be released.", getProcId());
172172
if (this instanceof PipeMetaSyncProcedure) {
173173
configNodeProcedureEnv
174174
.getConfigManager()

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedure.java‎

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -75,7 +75,7 @@ protected ProcedureLockState acquireLock(ConfigNodeProcedureEnv configNodeProced
7575
if (System.currentTimeMillis() - LAST_EXECUTION_TIME.get() < MIN_EXECUTION_INTERVAL_MS) {
7676
// Skip by setting the pipeTaskInfo to null
7777
pipeTaskInfo = null;
78-
LOGGER.info(
78+
LOGGER.debug(
7979
"PipeMetaSyncProcedure: acquireLock, skip the procedure due to the last execution time {}",
8080
LAST_EXECUTION_TIME.get());
8181
return ProcedureLockState.LOCK_ACQUIRED;
@@ -91,22 +91,22 @@ protected PipeTaskOperation getOperation() {
9191

9292
@Override
9393
public boolean executeFromValidateTask(ConfigNodeProcedureEnv env) {
94-
LOGGER.info("PipeMetaSyncProcedure: executeFromValidateTask");
94+
LOGGER.debug("PipeMetaSyncProcedure: executeFromValidateTask");
9595

9696
LAST_EXECUTION_TIME.set(System.currentTimeMillis());
9797
return true;
9898
}
9999

100100
@Override
101101
public void executeFromCalculateInfoForTask(ConfigNodeProcedureEnv env) {
102-
LOGGER.info("PipeMetaSyncProcedure: executeFromCalculateInfoForTask");
102+
LOGGER.debug("PipeMetaSyncProcedure: executeFromCalculateInfoForTask");
103103

104104
// Do nothing
105105
}
106106

107107
@Override
108108
public void executeFromWriteConfigNodeConsensus(ConfigNodeProcedureEnv env) {
109-
LOGGER.info("PipeMetaSyncProcedure: executeFromWriteConfigNodeConsensus");
109+
LOGGER.debug("PipeMetaSyncProcedure: executeFromWriteConfigNodeConsensus");
110110

111111
final List<PipeMeta> pipeMetaList = new ArrayList<>();
112112
for (final PipeMeta pipeMeta : pipeTaskInfo.get().getPipeMetaList()) {
@@ -132,7 +132,7 @@ public void executeFromWriteConfigNodeConsensus(ConfigNodeProcedureEnv env) {
132132
@Override
133133
public void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env)
134134
throws PipeException, IOException {
135-
LOGGER.info("PipeMetaSyncProcedure: executeFromOperateOnDataNodes");
135+
LOGGER.debug("PipeMetaSyncProcedure: executeFromOperateOnDataNodes");
136136

137137
Map<Integer, TPushPipeMetaResp> respMap = pushPipeMetaToDataNodes(env);
138138
if (pipeTaskInfo.get().recordDataNodePushPipeMetaExceptions(respMap)) {
@@ -145,28 +145,28 @@ public void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env)
145145

146146
@Override
147147
public void rollbackFromValidateTask(ConfigNodeProcedureEnv env) {
148-
LOGGER.info("PipeMetaSyncProcedure: rollbackFromValidateTask");
148+
LOGGER.debug("PipeMetaSyncProcedure: rollbackFromValidateTask");
149149

150150
// Do nothing
151151
}
152152

153153
@Override
154154
public void rollbackFromCalculateInfoForTask(ConfigNodeProcedureEnv env) {
155-
LOGGER.info("PipeMetaSyncProcedure: rollbackFromCalculateInfoForTask");
155+
LOGGER.debug("PipeMetaSyncProcedure: rollbackFromCalculateInfoForTask");
156156

157157
// Do nothing
158158
}
159159

160160
@Override
161161
public void rollbackFromWriteConfigNodeConsensus(ConfigNodeProcedureEnv env) {
162-
LOGGER.info("PipeMetaSyncProcedure: rollbackFromWriteConfigNodeConsensus");
162+
LOGGER.debug("PipeMetaSyncProcedure: rollbackFromWriteConfigNodeConsensus");
163163

164164
// Do nothing
165165
}
166166

167167
@Override
168168
public void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env) {
169-
LOGGER.info("PipeMetaSyncProcedure: rollbackFromOperateOnDataNodes");
169+
LOGGER.debug("PipeMetaSyncProcedure: rollbackFromOperateOnDataNodes");
170170

171171
// Do nothing
172172
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java‎

Lines changed: 11 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern;
2929
import org.apache.iotdb.commons.pipe.receiver.IoTDBFileReceiver;
3030
import org.apache.iotdb.commons.pipe.receiver.PipeReceiverStatusHandler;
31+
import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
3132
import org.apache.iotdb.commons.pipe.sink.payload.airgap.AirGapPseudoTPipeTransferRequest;
3233
import org.apache.iotdb.commons.pipe.sink.payload.thrift.common.PipeTransferSliceReqHandler;
3334
import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeRequestType;
@@ -376,7 +377,7 @@ public synchronized TPipeTransferResp receive(final TPipeTransferReq req) {
376377
} catch (final Exception e) {
377378
final String error =
378379
String.format("Exception %s encountered while handling request %s.", e.getMessage(), req);
379-
LOGGER.warn("Receiver id = {}: {}", receiverId.get(), error, e);
380+
PipeLogger.log(LOGGER::warn, e, "Receiver id = %s: %s", receiverId.get(), error);
380381
return new TPipeTransferResp(RpcUtils.getStatus(TSStatusCode.PIPE_ERROR, error));
381382
}
382383
}
@@ -533,7 +534,8 @@ private TPipeTransferResp handleTransferSchemaPlan(final PipeTransferPlanNodeReq
533534
final TSStatus status =
534535
((AlterLogicalViewNode) req.getPlanNode()).checkPermissionBeforeProcess(username);
535536
if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
536-
LOGGER.warn(
537+
PipeLogger.log(
538+
LOGGER::warn,
537539
"Receiver id = {}: Failed to check authority for statement {}, username = {}, response = {}.",
538540
receiverId.get(),
539541
StatementType.ALTER_LOGICAL_VIEW.name(),
@@ -681,15 +683,17 @@ private TSStatus executeStatementAndClassifyExceptions(
681683
|| result.getCode() == TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()) {
682684
return result;
683685
} else {
684-
LOGGER.warn(
686+
PipeLogger.log(
687+
LOGGER::warn,
685688
"Receiver id = {}: Failure status encountered while executing statement {}: {}",
686689
receiverId.get(),
687690
statement,
688691
result);
689692
return statement.accept(STATEMENT_STATUS_VISITOR, result);
690693
}
691694
} catch (final Exception e) {
692-
LOGGER.warn(
695+
PipeLogger.log(
696+
LOGGER::warn,
693697
"Receiver id = {}: Exception encountered while executing statement {}: ",
694698
receiverId.get(),
695699
statement,
@@ -733,8 +737,9 @@ private TSStatus executeStatement(final Statement statement) {
733737

734738
final TSStatus status = AuthorityChecker.checkAuthority(statement, clientSession);
735739
if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
736-
LOGGER.warn(
737-
"Receiver id = {}: Failed to check authority for statement {}, username = {}, response = {}.",
740+
PipeLogger.log(
741+
LOGGER::warn,
742+
"Receiver id = %s: Failed to check authority for statement %s, username = %s, response = %s.",
738743
receiverId.get(),
739744
statement.getType().name(),
740745
username,

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceSegmentLock.java‎

Lines changed: 22 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919

2020
package org.apache.iotdb.db.pipe.resource.tsfile;
2121

22+
import org.apache.iotdb.commons.pipe.config.PipeConfig;
2223
import org.apache.iotdb.db.storageengine.StorageEngine;
2324

2425
import org.slf4j.Logger;
@@ -31,25 +32,29 @@
3132
public class PipeTsFileResourceSegmentLock {
3233

3334
private static final Logger LOGGER = LoggerFactory.getLogger(PipeTsFileResourceSegmentLock.class);
34-
3535
private static final int SEGMENT_LOCK_MIN_SIZE = 32;
3636
private static final int SEGMENT_LOCK_MAX_SIZE = 128;
37-
3837
private volatile ReentrantLock[] locks;
3938

4039
private void initIfNecessary() {
4140
if (locks == null) {
4241
synchronized (this) {
42+
int lockSegmentSize = PipeConfig.getInstance().getPipeTsFileResourceSegmentLockNum();
4343
if (locks == null) {
44-
int lockSegmentSize = SEGMENT_LOCK_MIN_SIZE;
45-
try {
46-
lockSegmentSize = StorageEngine.getInstance().getAllDataRegionIds().size();
47-
} catch (final Exception e) {
48-
LOGGER.warn(
49-
"Cannot get data region ids, use default lock segment size: {}", lockSegmentSize);
44+
if (lockSegmentSize <= 0) {
45+
try {
46+
lockSegmentSize =
47+
Math.min(
48+
Math.max(
49+
StorageEngine.getInstance().getAllDataRegionIds().size(),
50+
SEGMENT_LOCK_MIN_SIZE),
51+
SEGMENT_LOCK_MAX_SIZE);
52+
} catch (final Exception e) {
53+
LOGGER.warn(
54+
"Cannot get data region ids, use default lock segment size: {}", lockSegmentSize);
55+
lockSegmentSize = SEGMENT_LOCK_MIN_SIZE;
56+
}
5057
}
51-
lockSegmentSize = Math.min(SEGMENT_LOCK_MAX_SIZE, lockSegmentSize);
52-
lockSegmentSize = Math.max(SEGMENT_LOCK_MIN_SIZE, lockSegmentSize);
5358

5459
final ReentrantLock[] tmpLocks = new ReentrantLock[lockSegmentSize];
5560
for (int i = 0; i < tmpLocks.length; i++) {
@@ -58,6 +63,13 @@ private void initIfNecessary() {
5863

5964
// publish this variable
6065
locks = tmpLocks;
66+
} else if (locks.length < lockSegmentSize) {
67+
final ReentrantLock[] tmpLocks = new ReentrantLock[lockSegmentSize];
68+
System.arraycopy(locks, 0, tmpLocks, 0, locks.length);
69+
for (int i = locks.length; i < lockSegmentSize; ++i) {
70+
tmpLocks[i] = new ReentrantLock();
71+
}
72+
locks = tmpLocks;
6173
}
6274
}
6375
}
@@ -74,39 +86,8 @@ public boolean tryLock(final File file, final long timeout, final TimeUnit timeU
7486
return locks[Math.abs(file.hashCode()) % locks.length].tryLock(timeout, timeUnit);
7587
}
7688

77-
public boolean tryLockAll(final long timeout, final TimeUnit timeUnit)
78-
throws InterruptedException {
79-
initIfNecessary();
80-
int alreadyLocked = 0;
81-
for (final ReentrantLock lock : locks) {
82-
if (lock.tryLock(timeout, timeUnit)) {
83-
alreadyLocked++;
84-
} else {
85-
break;
86-
}
87-
}
88-
89-
if (alreadyLocked == locks.length) {
90-
return true;
91-
} else {
92-
unlockUntil(alreadyLocked);
93-
return false;
94-
}
95-
}
96-
97-
private void unlockUntil(final int index) {
98-
for (int i = 0; i < index; i++) {
99-
locks[i].unlock();
100-
}
101-
}
102-
10389
public void unlock(final File file) {
10490
initIfNecessary();
10591
locks[Math.abs(file.hashCode()) % locks.length].unlock();
10692
}
107-
108-
public void unlockAll() {
109-
initIfNecessary();
110-
unlockUntil(locks.length);
111-
}
11293
}

‎iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java‎

Lines changed: 19 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -331,6 +331,7 @@ public class CommonConfig {
331331
private volatile double pipeThresholdAllocationStrategyFixedMemoryHighUsageThreshold = 0.8d;
332332
private volatile boolean pipeTransferTsFileSync = false;
333333
private volatile long pipeCheckAllSyncClientLiveTimeIntervalMs = 5 * 60 * 1000L; // 5 minutes
334+
private int pipeTsFileResourceSegmentLockNum = -1;
334335

335336
private long twoStageAggregateMaxCombinerLiveTimeInMs = 8 * 60 * 1000L; // 8 minutes
336337
private long twoStageAggregateDataRegionInfoCacheTimeInMs = 3 * 60 * 1000L; // 3 minutes
@@ -1926,14 +1927,28 @@ public long getPipeCheckAllSyncClientLiveTimeIntervalMs() {
19261927
}
19271928

19281929
public void setPipeCheckAllSyncClientLiveTimeIntervalMs(
1929-
long pipeCheckSyncAllClientLiveTimeIntervalMs) {
1930-
if (this.pipeCheckAllSyncClientLiveTimeIntervalMs == pipeCheckSyncAllClientLiveTimeIntervalMs) {
1930+
long pipeCheckAllSyncClientLiveTimeIntervalMs) {
1931+
if (this.pipeCheckAllSyncClientLiveTimeIntervalMs == pipeCheckAllSyncClientLiveTimeIntervalMs) {
19311932
return;
19321933
}
1933-
this.pipeCheckAllSyncClientLiveTimeIntervalMs = pipeCheckSyncAllClientLiveTimeIntervalMs;
1934+
this.pipeCheckAllSyncClientLiveTimeIntervalMs = pipeCheckAllSyncClientLiveTimeIntervalMs;
19341935
logger.info(
19351936
"pipeCheckSyncAllClientLiveTimeIntervalMs is set to {}",
1936-
pipeCheckSyncAllClientLiveTimeIntervalMs);
1937+
pipeCheckAllSyncClientLiveTimeIntervalMs);
1938+
}
1939+
1940+
public int getPipeTsFileResourceSegmentLockNum() {
1941+
return pipeTsFileResourceSegmentLockNum;
1942+
}
1943+
1944+
public void setPipeTsFileResourceSegmentLockNum(int pipeTsFileResourceSegmentLockNum) {
1945+
if (this.pipeTsFileResourceSegmentLockNum == pipeTsFileResourceSegmentLockNum) {
1946+
return;
1947+
}
1948+
this.pipeTsFileResourceSegmentLockNum = pipeTsFileResourceSegmentLockNum;
1949+
logger.info(
1950+
"pipeCheckSyncAllClientLiveTimeIntervalMs is set to {}",
1951+
pipeCheckAllSyncClientLiveTimeIntervalMs);
19371952
}
19381953

19391954
public double getPipeSendTsFileRateLimitBytesPerSecond() {

‎iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -293,6 +293,10 @@ public long getPipeCheckAllSyncClientLiveTimeIntervalMs() {
293293
return COMMON_CONFIG.getPipeCheckAllSyncClientLiveTimeIntervalMs();
294294
}
295295

296+
public int getPipeTsFileResourceSegmentLockNum() {
297+
return COMMON_CONFIG.getPipeTsFileResourceSegmentLockNum();
298+
}
299+
296300
/////////////////////////////// Meta Consistency ///////////////////////////////
297301

298302
public boolean isSeperatedPipeHeartbeatEnabled() {
@@ -525,6 +529,9 @@ public void printAllConfigs() {
525529
LOGGER.info(
526530
"PipeCheckAllSyncClientLiveTimeIntervalMs: {}",
527531
getPipeCheckAllSyncClientLiveTimeIntervalMs());
532+
LOGGER.info(
533+
"PipeCheckAllSyncClientLiveTimeIntervalMs: {}",
534+
getPipeCheckAllSyncClientLiveTimeIntervalMs());
528535

529536
LOGGER.info("PipeDynamicMemoryHistoryWeight: {}", getPipeDynamicMemoryHistoryWeight());
530537
LOGGER.info(

‎iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -498,6 +498,11 @@ public static void loadPipeInternalConfig(CommonConfig config, TrimProperties pr
498498
properties.getProperty(
499499
"pipe_check_all_sync_client_live_time_interval_ms",
500500
String.valueOf(config.getPipeCheckAllSyncClientLiveTimeIntervalMs()))));
501+
config.setPipeTsFileResourceSegmentLockNum(
502+
Integer.parseInt(
503+
properties.getProperty(
504+
"pipe_tsfile_resource_segment_lock_num",
505+
String.valueOf(config.getPipeTsFileResourceSegmentLockNum()))));
501506

502507
config.setPipeRemainingTimeCommitRateAutoSwitchSeconds(
503508
Long.parseLong(

0 commit comments

Comments
 (0)