You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2023/08/02 13:29:21 UTC
[iotdb] 01/01: ..
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch fix-pipe-oom-with-pattern
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 4bb3784046f9406fefbf32e8b5bbb1c54c92ec20
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Wed Aug 2 21:29:00 2023 +0800
..
---
.../PipeRealtimeDataRegionHybridExtractor.java | 4 ++--
.../PipeRealtimeDataRegionLogExtractor.java | 2 +-
.../PipeRealtimeDataRegionTsFileExtractor.java | 2 +-
.../pipe/task/connection/BlockingPendingQueue.java | 18 +++++++++++----
.../pipe/task/connection/PipeEventCollector.java | 27 +++-------------------
5 files changed, 21 insertions(+), 32 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionHybridExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionHybridExtractor.java
index bbddc5d03b6..ffefb13daf0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionHybridExtractor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionHybridExtractor.java
@@ -81,7 +81,7 @@ public class PipeRealtimeDataRegionHybridExtractor extends PipeRealtimeDataRegio
}
if (!event.getTsFileEpoch().getState(this).equals(TsFileEpoch.State.USING_TSFILE)
- && !pendingQueue.offer(event)) {
+ && !pendingQueue.waitedOffer(event)) {
LOGGER.warn(
"extractTabletInsertion: pending queue of PipeRealtimeDataRegionHybridExtractor {} "
+ "has reached capacity, discard tablet event {}, current state {}",
@@ -101,7 +101,7 @@ public class PipeRealtimeDataRegionHybridExtractor extends PipeRealtimeDataRegio
state ->
state.equals(TsFileEpoch.State.EMPTY) ? TsFileEpoch.State.USING_TSFILE : state);
- if (!pendingQueue.offer(event)) {
+ if (!pendingQueue.waitedOffer(event)) {
LOGGER.warn(
"extractTsFileInsertion: pending queue of PipeRealtimeDataRegionHybridExtractor {} "
+ "has reached capacity, discard TsFile event {}, current state {}",
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionLogExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionLogExtractor.java
index 378af16857b..74ff0c533b1 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionLogExtractor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionLogExtractor.java
@@ -51,7 +51,7 @@ public class PipeRealtimeDataRegionLogExtractor extends PipeRealtimeDataRegionEx
return;
}
- if (!pendingQueue.offer(event)) {
+ if (!pendingQueue.waitedOffer(event)) {
LOGGER.warn(
"extract: pending queue of PipeRealtimeDataRegionLogExtractor {} "
+ "has reached capacity, discard tablet event {}, current state {}",
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionTsFileExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionTsFileExtractor.java
index ec351310c54..bab1ea2ec46 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionTsFileExtractor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionTsFileExtractor.java
@@ -51,7 +51,7 @@ public class PipeRealtimeDataRegionTsFileExtractor extends PipeRealtimeDataRegio
return;
}
- if (!pendingQueue.offer(event)) {
+ if (!pendingQueue.waitedOffer(event)) {
LOGGER.warn(
"extract: pending queue of PipeRealtimeDataRegionTsFileExtractor {} "
+ "has reached capacity, discard TsFile event {}, current state {}",
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/BlockingPendingQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/BlockingPendingQueue.java
index 2787adc4d0a..c32d54ea38b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/BlockingPendingQueue.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/BlockingPendingQueue.java
@@ -41,15 +41,25 @@ public abstract class BlockingPendingQueue<E extends Event> {
this.pendingQueue = pendingQueue;
}
- public boolean offer(E event) {
- boolean isAdded = false;
+ public boolean waitedOffer(E event) {
try {
- isAdded = pendingQueue.offer(event, MAX_BLOCKING_TIME_MS, TimeUnit.MILLISECONDS);
+ return pendingQueue.offer(event, MAX_BLOCKING_TIME_MS, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
LOGGER.info("pending queue offer is interrupted.", e);
Thread.currentThread().interrupt();
+ return false;
+ }
+ }
+
+ public boolean directOffer(E event) {
+ try {
+ pendingQueue.put(event);
+ return true;
+ } catch (InterruptedException e) {
+ LOGGER.info("pending queue offer is interrupted.", e);
+ Thread.currentThread().interrupt();
+ return false;
}
- return isAdded;
}
public E directPoll() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/PipeEventCollector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/PipeEventCollector.java
index 498fc5aecfa..99281ed399b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/PipeEventCollector.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/PipeEventCollector.java
@@ -22,25 +22,14 @@ package org.apache.iotdb.db.pipe.task.connection;
import org.apache.iotdb.db.pipe.event.EnrichedEvent;
import org.apache.iotdb.pipe.api.collector.EventCollector;
import org.apache.iotdb.pipe.api.event.Event;
-
-import java.util.LinkedList;
-import java.util.Queue;
+import org.apache.iotdb.pipe.api.exception.PipeException;
public class PipeEventCollector implements EventCollector {
private final BoundedBlockingPendingQueue<Event> pendingQueue;
- // buffer queue is used to store events that are not offered to pending queue
- // because the pending queue is full. when pending queue is full, pending queue
- // will notify tasks to stop extracting events, and buffer queue will be used to store
- // events before tasks are stopped. when pending queue is not full and tasks are
- // notified by the pending queue to start extracting events, buffer queue will be used to store
- // events before events in buffer queue are offered to pending queue.
- private final Queue<Event> bufferQueue;
-
public PipeEventCollector(BoundedBlockingPendingQueue<Event> pendingQueue) {
this.pendingQueue = pendingQueue;
- bufferQueue = new LinkedList<>();
}
@Override
@@ -49,18 +38,8 @@ public class PipeEventCollector implements EventCollector {
((EnrichedEvent) event).increaseReferenceCount(PipeEventCollector.class.getName());
}
- while (!bufferQueue.isEmpty()) {
- final Event bufferedEvent = bufferQueue.peek();
- if (pendingQueue.offer(bufferedEvent)) {
- bufferQueue.poll();
- } else {
- bufferQueue.offer(event);
- return;
- }
- }
-
- if (!pendingQueue.offer(event)) {
- bufferQueue.offer(event);
+ if (!pendingQueue.directOffer(event)) {
+ throw new PipeException("Unexpected error: interrupted while offering event to queue");
}
}
}