You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by ar...@apache.org on 2018/05/19 20:37:46 UTC
[drill] 01/04: DRILL-6348: Received batches are now owned by the
receive operators instead of the parent
This is an automated email from the ASF dual-hosted git repository.
arina pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/drill.git
commit dc1db98f613f041e40c2c704862051ee2a51734a
Author: Salim Achouche <sa...@gmail.com>
AuthorDate: Sun Apr 22 18:02:35 2018 -0700
DRILL-6348: Received batches are now owned by the receive operators instead of the parent
closes #1237
---
.../org/apache/drill/exec/ops/FragmentContextImpl.java | 5 ++++-
.../exec/physical/impl/MergingReceiverCreator.java | 2 +-
.../physical/impl/mergereceiver/MergingRecordBatch.java | 3 +++
.../impl/unorderedreceiver/UnorderedReceiverBatch.java | 3 +++
.../unorderedreceiver/UnorderedReceiverCreator.java | 2 +-
.../drill/exec/work/batch/AbstractDataCollector.java | 17 +++++++++++++++++
.../org/apache/drill/exec/work/batch/DataCollector.java | 14 +++++++++++++-
.../apache/drill/exec/work/batch/IncomingBuffers.java | 10 +++++++---
8 files changed, 49 insertions(+), 7 deletions(-)
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/FragmentContextImpl.java b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/FragmentContextImpl.java
index b192850..503ebdd 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/FragmentContextImpl.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/FragmentContextImpl.java
@@ -449,13 +449,16 @@ public class FragmentContextImpl extends BaseFragmentContext implements Executor
public void close() {
waitForSendComplete();
+ // Close the buffers before closing the operators; this is needed as buffer ownership
+ // is attached to the receive operators.
+ suppressingClose(buffers);
+
// close operator context
for (OperatorContextImpl opContext : contexts) {
suppressingClose(opContext);
}
suppressingClose(bufferManager);
- suppressingClose(buffers);
suppressingClose(allocator);
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/MergingReceiverCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/MergingReceiverCreator.java
index 0ef84b9..66a0cc2 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/MergingReceiverCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/MergingReceiverCreator.java
@@ -42,7 +42,7 @@ public class MergingReceiverCreator implements BatchCreator<MergingReceiverPOP>
IncomingBuffers bufHolder = context.getBuffers();
assert bufHolder != null : "IncomingBuffers must be defined for any place a receiver is declared.";
- RawBatchBuffer[] buffers = bufHolder.getBuffers(receiver.getOppositeMajorFragmentId());
+ RawBatchBuffer[] buffers = bufHolder.getCollector(receiver.getOppositeMajorFragmentId()).getBuffers();
return new MergingRecordBatch(context, receiver, buffers);
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/mergereceiver/MergingRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/mergereceiver/MergingRecordBatch.java
index 7e5ff21..9087757 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/mergereceiver/MergingRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/mergereceiver/MergingRecordBatch.java
@@ -136,6 +136,9 @@ public class MergingRecordBatch extends AbstractRecordBatch<MergingReceiverPOP>
this.config = config;
this.inputCounts = new long[config.getNumSenders()];
this.outputCounts = new long[config.getNumSenders()];
+
+ // Register this operator's buffer allocator so that incoming buffers are owned by this allocator
+ context.getBuffers().getCollector(config.getOppositeMajorFragmentId()).setAllocator(oContext.getAllocator());
}
@SuppressWarnings("resource")
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverBatch.java
index 433e0c8..424a733 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverBatch.java
@@ -86,6 +86,9 @@ public class UnorderedReceiverBatch implements CloseableRecordBatch {
this.stats = oContext.getStats();
this.stats.setLongStat(Metric.NUM_SENDERS, config.getNumSenders());
this.config = config;
+
+ // Register this operator's buffer allocator so that incoming buffers are owned by this allocator
+ context.getBuffers().getCollector(config.getOppositeMajorFragmentId()).setAllocator(oContext.getAllocator());
}
@Override
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverCreator.java
index 01a4588..3dcdfc4 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unorderedreceiver/UnorderedReceiverCreator.java
@@ -37,7 +37,7 @@ public class UnorderedReceiverCreator implements BatchCreator<UnorderedReceiver>
IncomingBuffers bufHolder = context.getBuffers();
assert bufHolder != null : "IncomingBuffers must be defined for any place a receiver is declared.";
- RawBatchBuffer[] buffers = bufHolder.getBuffers(receiver.getOppositeMajorFragmentId());
+ RawBatchBuffer[] buffers = bufHolder.getCollector(receiver.getOppositeMajorFragmentId()).getBuffers();
assert buffers.length == 1;
RawBatchBuffer buffer = buffers[0];
return new UnorderedReceiverBatch(context, buffer, receiver);
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/AbstractDataCollector.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/AbstractDataCollector.java
index b6b4183..bb3a5a2 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/AbstractDataCollector.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/AbstractDataCollector.java
@@ -22,6 +22,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerArray;
import org.apache.drill.common.AutoCloseables;
+import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.proto.BitControl.Collector;
import org.apache.drill.exec.record.RawFragmentBatch;
@@ -37,6 +38,8 @@ public abstract class AbstractDataCollector implements DataCollector {
private final int incomingStreams;
protected final RawBatchBuffer[] buffers;
protected final ArrayWrappedIntIntMap fragmentMap;
+ /** Allocator which owns incoming batches */
+ protected BufferAllocator ownerAllocator;
/**
* @param parentAccounter
@@ -53,6 +56,7 @@ public abstract class AbstractDataCollector implements DataCollector {
this.parentAccounter = parentAccounter;
this.remainders = new AtomicIntegerArray(incomingStreams);
this.oppositeMajorFragmentId = collector.getOppositeMajorFragmentId();
+ this.ownerAllocator = context.getAllocator();
// Create fragmentId to index that is within the range [0, incoming.size()-1]
// We use this mapping to find objects belonging to the fragment in buffers and remainders arrays.
fragmentMap = new ArrayWrappedIntIntMap();
@@ -116,4 +120,17 @@ public abstract class AbstractDataCollector implements DataCollector {
AutoCloseables.close(buffers);
}
+ /** {@inheritDoc} */
+ @Override
+ public BufferAllocator getAllocator() {
+ return this.ownerAllocator;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void setAllocator(BufferAllocator allocator) {
+ Preconditions.checkArgument(allocator != null, "buffer allocator cannot be null");
+ this.ownerAllocator = allocator;
+ }
+
}
\ No newline at end of file
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/DataCollector.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/DataCollector.java
index 026fc81..fa74677 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/DataCollector.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/DataCollector.java
@@ -19,13 +19,25 @@ package org.apache.drill.exec.work.batch;
import java.io.IOException;
+import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.record.RawFragmentBatch;
-interface DataCollector extends AutoCloseable {
+public interface DataCollector extends AutoCloseable {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DataCollector.class);
public boolean batchArrived(int minorFragmentId, RawFragmentBatch batch) throws IOException ;
public int getOppositeMajorFragmentId();
public RawBatchBuffer[] getBuffers();
public int getTotalIncomingFragments();
public void close() throws Exception;
+ /**
+ * Enables caller (e.g., receiver) to attach its buffer allocator to this Data Collector in order
+ * to claim ownership of incoming batches; by default, the fragment allocator owns these batches.
+ *
+ * @param allocator operator buffer allocator
+ */
+ void setAllocator(BufferAllocator allocator);
+ /**
+ * @return allocator
+ */
+ BufferAllocator getAllocator();
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/IncomingBuffers.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/IncomingBuffers.java
index 876c8b5..2d1b4f2 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/IncomingBuffers.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/IncomingBuffers.java
@@ -27,6 +27,7 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.drill.common.AutoCloseables;
import org.apache.drill.common.concurrent.AutoCloseableLock;
import org.apache.drill.exec.exception.FragmentSetupException;
+import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.proto.BitControl.Collector;
import org.apache.drill.exec.proto.BitControl.PlanFragment;
@@ -103,8 +104,11 @@ public class IncomingBuffers implements AutoCloseable {
Arrays.toString(collectorMap.values().toArray())));
}
+ // Use the Data Collector's buffer allocator if set, otherwise the fragment's one
+ BufferAllocator ownerAllocator = collector.getAllocator();
+
synchronized (collector) {
- final RawFragmentBatch newRawFragmentBatch = incomingBatch.newRawFragmentBatch(context.getAllocator());
+ final RawFragmentBatch newRawFragmentBatch = incomingBatch.newRawFragmentBatch(ownerAllocator);
boolean decrementedToZero = collector
.batchArrived(incomingBatch.getHeader().getSendingMinorFragmentId(), newRawFragmentBatch);
newRawFragmentBatch.release();
@@ -125,8 +129,8 @@ public class IncomingBuffers implements AutoCloseable {
return rem;
}
- public RawBatchBuffer[] getBuffers(int senderMajorFragmentId) {
- return collectorMap.get(senderMajorFragmentId).getBuffers();
+ public DataCollector getCollector(int senderMajorFragmentId) {
+ return collectorMap.get(senderMajorFragmentId);
}
public boolean isDone() {
--
To stop receiving notification emails like this one, please contact
arina@apache.org.