This is an automated email from the ASF dual-hosted git repository.
jackietien 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 360e356cb3 fix the issue related to LocalSinkHandle/LocalSourceHandle
(#6296)
360e356cb3 is described below
commit 360e356cb34ee64c194a9656dee993cae849e559
Author: Zhang.Jinrui <[email protected]>
AuthorDate: Wed Jun 15 21:11:08 2022 +0800
fix the issue related to LocalSinkHandle/LocalSourceHandle (#6296)
---
.../execution/datatransfer/LocalSinkHandle.java | 30 ++++++++++++++--------
.../execution/datatransfer/LocalSourceHandle.java | 21 +++++++++------
.../execution/datatransfer/SharedTsBlockQueue.java | 28 +++++++++++++++++++-
3 files changed, 60 insertions(+), 19 deletions(-)
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/execution/datatransfer/LocalSinkHandle.java
b/server/src/main/java/org/apache/iotdb/db/mpp/execution/datatransfer/LocalSinkHandle.java
index d1a8ac33f1..fa59e7c98e 100644
---
a/server/src/main/java/org/apache/iotdb/db/mpp/execution/datatransfer/LocalSinkHandle.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/execution/datatransfer/LocalSinkHandle.java
@@ -57,6 +57,7 @@ public class LocalSinkHandle implements ISinkHandle {
this.localFragmentInstanceId = Validate.notNull(localFragmentInstanceId);
this.sinkHandleListener = Validate.notNull(sinkHandleListener);
this.queue = Validate.notNull(queue);
+ this.queue.setSinkHandle(this);
}
@Override
@@ -87,6 +88,14 @@ public class LocalSinkHandle implements ISinkHandle {
return queue.hasNoMoreTsBlocks() && queue.isEmpty();
}
+ public void checkAndInvokeOnFinished() {
+ if (isFinished()) {
+ synchronized (this) {
+ sinkHandleListener.onFinish(this);
+ }
+ }
+ }
+
@Override
public synchronized void send(List<TsBlock> tsBlocks) {
Validate.notNull(tsBlocks, "tsBlocks is null");
@@ -99,6 +108,7 @@ public class LocalSinkHandle implements ISinkHandle {
if (queue.hasNoMoreTsBlocks()) {
return;
}
+ logger.info("send TsBlocks. Size: {}", tsBlocks.size());
for (TsBlock tsBlock : tsBlocks) {
blocked = queue.add(tsBlock);
}
@@ -110,17 +120,17 @@ public class LocalSinkHandle implements ISinkHandle {
}
@Override
- public synchronized void setNoMoreTsBlocks() {
- logger.info("Set no-more-tsblocks.");
- if (aborted) {
- return;
- }
- queue.setNoMoreTsBlocks(true);
- sinkHandleListener.onEndOfBlocks(this);
- if (isFinished()) {
- sinkHandleListener.onFinish(this);
+ public void setNoMoreTsBlocks() {
+ synchronized (this) {
+ logger.info("set noMoreTsBlocks.");
+ if (aborted) {
+ return;
+ }
+ queue.setNoMoreTsBlocks(true);
+ sinkHandleListener.onEndOfBlocks(this);
}
- logger.info("No-more-tsblocks has been set.");
+ checkAndInvokeOnFinished();
+ logger.info("noMoreTsBlocks has been set.");
}
@Override
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/execution/datatransfer/LocalSourceHandle.java
b/server/src/main/java/org/apache/iotdb/db/mpp/execution/datatransfer/LocalSourceHandle.java
index 14a0ebc664..ac91afacda 100644
---
a/server/src/main/java/org/apache/iotdb/db/mpp/execution/datatransfer/LocalSourceHandle.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/execution/datatransfer/LocalSourceHandle.java
@@ -26,16 +26,11 @@ import org.apache.iotdb.tsfile.read.common.block.TsBlock;
import com.google.common.util.concurrent.ListenableFuture;
import io.airlift.concurrent.SetThreadName;
import org.apache.commons.lang3.Validate;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import static
com.google.common.util.concurrent.Futures.nonCancellationPropagating;
import static
org.apache.iotdb.db.mpp.execution.datatransfer.DataBlockManager.createFullIdFrom;
public class LocalSourceHandle implements ISourceHandle {
-
- private static final Logger logger =
LoggerFactory.getLogger(LocalSourceHandle.class);
-
private final TFragmentInstanceId remoteFragmentInstanceId;
private final TFragmentInstanceId localFragmentInstanceId;
private final String localPlanNodeId;
@@ -55,6 +50,7 @@ public class LocalSourceHandle implements ISourceHandle {
this.localFragmentInstanceId = Validate.notNull(localFragmentInstanceId);
this.localPlanNodeId = Validate.notNull(localPlanNodeId);
this.queue = Validate.notNull(queue);
+ this.queue.setSourceHandle(this);
this.sourceHandleListener = Validate.notNull(sourceHandleListener);
this.threadName =
createFullIdFrom(localFragmentInstanceId, localPlanNodeId + "." +
"SourceHandle");
@@ -88,9 +84,7 @@ public class LocalSourceHandle implements ISourceHandle {
synchronized (this) {
tsBlock = queue.remove();
}
- if (isFinished()) {
- sourceHandleListener.onFinished(this);
- }
+ checkAndInvokeOnFinished();
return tsBlock;
}
}
@@ -100,6 +94,17 @@ public class LocalSourceHandle implements ISourceHandle {
return queue.hasNoMoreTsBlocks() && queue.isEmpty();
}
+ public void checkAndInvokeOnFinished() {
+ if (isFinished()) {
+ // Putting synchronized here rather than marking in method is to avoid
deadlock.
+ // There are two locks need to invoke this method. One is lock of
SharedTsBlockQueue,
+ // the other is lock of LocalSourceHandle.
+ synchronized (this) {
+ sourceHandleListener.onFinished(this);
+ }
+ }
+ }
+
@Override
public ListenableFuture<Void> isBlocked() {
if (aborted) {
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/execution/datatransfer/SharedTsBlockQueue.java
b/server/src/main/java/org/apache/iotdb/db/mpp/execution/datatransfer/SharedTsBlockQueue.java
index 3d991b1aa9..61b94f3ecd 100644
---
a/server/src/main/java/org/apache/iotdb/db/mpp/execution/datatransfer/SharedTsBlockQueue.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/execution/datatransfer/SharedTsBlockQueue.java
@@ -26,6 +26,8 @@ import org.apache.iotdb.tsfile.read.common.block.TsBlock;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import org.apache.commons.lang3.Validate;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import javax.annotation.concurrent.GuardedBy;
@@ -34,6 +36,8 @@ import java.util.Queue;
public class SharedTsBlockQueue {
+ private static final Logger logger =
LoggerFactory.getLogger(SharedTsBlockQueue.class);
+
private final TFragmentInstanceId localFragmentInstanceId;
private final LocalMemoryManager localMemoryManager;
@@ -55,6 +59,9 @@ public class SharedTsBlockQueue {
@GuardedBy("this")
private boolean destroyed = false;
+ private LocalSourceHandle sourceHandle;
+ private LocalSinkHandle sinkHandle;
+
public SharedTsBlockQueue(
TFragmentInstanceId fragmentInstanceId, LocalMemoryManager
localMemoryManager) {
this.localFragmentInstanceId =
@@ -79,12 +86,26 @@ public class SharedTsBlockQueue {
return queue.isEmpty();
}
+ public void setSinkHandle(LocalSinkHandle sinkHandle) {
+ this.sinkHandle = sinkHandle;
+ }
+
+ public void setSourceHandle(LocalSourceHandle sourceHandle) {
+ this.sourceHandle = sourceHandle;
+ }
+
/** Notify no more tsblocks will be added to the queue. */
public synchronized void setNoMoreTsBlocks(boolean noMoreTsBlocks) {
if (destroyed) {
throw new IllegalStateException("queue has been destroyed");
}
this.noMoreTsBlocks = noMoreTsBlocks;
+ if (!blocked.isDone()) {
+ blocked.set(null);
+ }
+ if (this.sourceHandle != null) {
+ this.sourceHandle.checkAndInvokeOnFinished();
+ }
}
/**
@@ -96,6 +117,11 @@ public class SharedTsBlockQueue {
throw new IllegalStateException("queue has been destroyed");
}
TsBlock tsBlock = queue.remove();
+ // Every time LocalSourceHandle consumes a TsBlock, it needs to send the
event to
+ // corresponding LocalSinkHandle.
+ if (sinkHandle != null) {
+ sinkHandle.checkAndInvokeOnFinished();
+ }
localMemoryManager
.getQueryPool()
.free(localFragmentInstanceId.getQueryId(),
tsBlock.getRetainedSizeInBytes());
@@ -115,7 +141,7 @@ public class SharedTsBlockQueue {
throw new IllegalStateException("queue has been destroyed");
}
- Validate.notNull(tsBlock, "tsblock cannot be null");
+ Validate.notNull(tsBlock, "TsBlock cannot be null");
Validate.isTrue(blockedOnMemory == null || blockedOnMemory.isDone(),
"queue is full");
blockedOnMemory =
localMemoryManager
