You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by ja...@apache.org on 2014/06/25 21:07:56 UTC
[07/14] DRILL-1067: RPC layer improvements - Unfork Netty and upgrade
to 4.0.20.Final - Create support for EPOLL native impl on Linux (disable by
providing -Ddrill.exec.disable_linux_epoll=false) - Move BitData connection
to Multiplex with DataTunnel lev
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/memory/Accountor.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/memory/Accountor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/memory/Accountor.java
index 257f6fc..cd08367 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/memory/Accountor.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/memory/Accountor.java
@@ -17,6 +17,7 @@
*/
package org.apache.drill.exec.memory;
+import io.netty.buffer.AccountingByteBuf;
import io.netty.buffer.ByteBuf;
import java.util.Arrays;
@@ -57,6 +58,17 @@ public class Accountor {
}
}
+ public boolean transferTo(Accountor target, AccountingByteBuf buf, long size){
+ boolean withinLimit = target.forceAdditionalReservation(size);
+ release(buf, size);
+
+ if (ENABLE_ACCOUNTING) {
+ target.buffers.put(buf, new DebugStackTrace(buf.capacity(), Thread.currentThread().getStackTrace()));
+ }
+
+ return withinLimit;
+ }
+
public long getAvailable() {
if (parent != null) {
return Math.min(parent.getAvailable(), getCapacity() - getAllocation());
@@ -76,8 +88,12 @@ public class Accountor {
return remainder.get(size);
}
- public void forceAdditionalReservation(long size) {
- if(size > 0) remainder.forceGet(size);
+ public boolean forceAdditionalReservation(long size) {
+ if(size > 0){
+ return remainder.forceGet(size);
+ }else{
+ return true;
+ }
}
public void reserved(long expected, AccountingByteBuf buf){
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/memory/AtomicRemainder.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/memory/AtomicRemainder.java b/exec/java-exec/src/main/java/org/apache/drill/exec/memory/AtomicRemainder.java
index 1ae1e4c..0d54974 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/memory/AtomicRemainder.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/memory/AtomicRemainder.java
@@ -66,13 +66,14 @@ public class AtomicRemainder {
*
* @param size
*/
- public void forceGet(long size) {
- long newAvailableShared = availableShared.addAndGet(size);
-// if (DEBUG)
-// logger.info("Force get {}. a.s. {} a.p. {} hashcode: {}", size, availableShared, availablePrivate, hashCode(), new Exception());
-// assert newAvailableShared <= initShared;
- if (parent != null)
- parent.forceGet(size);
+ public boolean forceGet(long size) {
+ if(get(size)){
+ return true;
+ }else{
+ availableShared.addAndGet(size);
+ if (parent != null) parent.forceGet(size);
+ return false;
+ }
}
public boolean get(long size) {
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/memory/BufferAllocator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/memory/BufferAllocator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/memory/BufferAllocator.java
index 50f93fc..38fe4f1 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/memory/BufferAllocator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/memory/BufferAllocator.java
@@ -17,6 +17,7 @@
*/
package org.apache.drill.exec.memory;
+import io.netty.buffer.AccountingByteBuf;
import io.netty.buffer.ByteBufAllocator;
import java.io.Closeable;
@@ -32,7 +33,7 @@ public interface BufferAllocator extends Closeable {
/**
* Allocate a new or reused buffer of the provided size. Note that the buffer may technically be larger than the
* requested size for rounding purposes. However, the buffers capacity will be set to the configured size.
- *
+ *
* @param size
* The size in bytes.
* @return A new ByteBuf.
@@ -42,7 +43,7 @@ public interface BufferAllocator extends Closeable {
/**
* Allocate a new or reused buffer within provided range. Note that the buffer may technically be larger than the
* requested size for rounding purposes. However, the buffers capacity will be set to the configured size.
- *
+ *
* @param minSize The minimum size in bytes.
* @param maxSize The maximum size in bytes.
* @return A new ByteBuf.
@@ -54,6 +55,14 @@ public interface BufferAllocator extends Closeable {
public abstract BufferAllocator getChildAllocator(FragmentHandle handle, long initialReservation,
long maximumReservation) throws OutOfMemoryException;
+ /**
+ * Take over ownership of fragment accounting. Always takes over ownership.
+ * @param buf
+ * @return false if over allocation.
+ */
+ public boolean takeOwnership(AccountingByteBuf buf) ;
+
+
public PreAllocator getNewPreAllocator();
/**
@@ -71,7 +80,7 @@ public interface BufferAllocator extends Closeable {
*/
/**
- *
+ *
*/
/**
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/memory/TopLevelAllocator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/memory/TopLevelAllocator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/memory/TopLevelAllocator.java
index ecbf0b4..e785f0d 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/memory/TopLevelAllocator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/memory/TopLevelAllocator.java
@@ -17,10 +17,11 @@
*/
package org.apache.drill.exec.memory;
+import io.netty.buffer.AccountingByteBuf;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.PooledByteBufAllocatorL;
-import io.netty.buffer.PooledUnsafeDirectByteBufL;
import java.util.HashMap;
import java.util.HashSet;
@@ -37,7 +38,7 @@ public class TopLevelAllocator implements BufferAllocator {
private static final boolean ENABLE_ACCOUNTING = AssertionUtil.isAssertionsEnabled();
private final Set<ChildAllocator> children;
- private final PooledByteBufAllocatorL innerAllocator = PooledByteBufAllocatorL.DEFAULT;
+ private final PooledByteBufAllocator innerAllocator = PooledByteBufAllocatorL.DEFAULT;
private final Accountor acct;
private final boolean errorOnLeak;
@@ -63,10 +64,15 @@ public class TopLevelAllocator implements BufferAllocator {
);
}
+ @Override
+ public boolean takeOwnership(AccountingByteBuf buf) {
+ return buf.transferAccounting(acct);
+ }
+
public AccountingByteBuf buffer(int min, int max) {
if(!acct.reserve(min)) return null;
ByteBuf buffer = innerAllocator.directBuffer(min, max);
- AccountingByteBuf wrapped = new AccountingByteBuf(acct, (PooledUnsafeDirectByteBufL) buffer);
+ AccountingByteBuf wrapped = new AccountingByteBuf(acct, buffer);
acct.reserved(min, wrapped);
return wrapped;
}
@@ -124,6 +130,11 @@ public class TopLevelAllocator implements BufferAllocator {
}
@Override
+ public boolean takeOwnership(AccountingByteBuf buf) {
+ return buf.transferAccounting(childAcct);
+ }
+
+ @Override
public AccountingByteBuf buffer(int size, int max) {
if(!childAcct.reserve(size)){
logger.warn("Unable to allocate buffer of size {} due to memory limit. Current allocation: {}", size, getAllocatedMemory());
@@ -131,7 +142,7 @@ public class TopLevelAllocator implements BufferAllocator {
};
ByteBuf buffer = innerAllocator.directBuffer(size, max);
- AccountingByteBuf wrapped = new AccountingByteBuf(childAcct, (PooledUnsafeDirectByteBufL) buffer);
+ AccountingByteBuf wrapped = new AccountingByteBuf(childAcct, buffer);
childAcct.reserved(buffer.capacity(), wrapped);
return wrapped;
}
@@ -227,7 +238,7 @@ public class TopLevelAllocator implements BufferAllocator {
public AccountingByteBuf getAllocation(){
- AccountingByteBuf b = new AccountingByteBuf(acct, (PooledUnsafeDirectByteBufL) innerAllocator.buffer(bytes));
+ AccountingByteBuf b = new AccountingByteBuf(acct, innerAllocator.buffer(bytes));
acct.reserved(bytes, b);
return b;
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/proto/helper/QueryIdHelper.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/proto/helper/QueryIdHelper.java b/exec/java-exec/src/main/java/org/apache/drill/exec/proto/helper/QueryIdHelper.java
index 749b3d5..3bbe231 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/proto/helper/QueryIdHelper.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/proto/helper/QueryIdHelper.java
@@ -19,14 +19,19 @@
package org.apache.drill.exec.proto.helper;
import java.util.UUID;
+
+import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
import org.apache.drill.exec.proto.UserBitShared.QueryId;
/* Helper class around the QueryId protobuf */
public class QueryIdHelper {
- /* Generate a UUID from the two parts of the queryid */
- public static String getQueryId(QueryId queryId)
- {
- return (new UUID(queryId.getPart1(), queryId.getPart2())).toString();
- }
+ /* Generate a UUID from the two parts of the queryid */
+ public static String getQueryId(QueryId queryId) {
+ return (new UUID(queryId.getPart1(), queryId.getPart2())).toString();
+ }
+
+ public static String getQueryIdentifier(FragmentHandle h) {
+ return getQueryId(h.getQueryId()) + ":" + h.getMajorFragmentId() + ":" + h.getMinorFragmentId();
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/record/RawFragmentBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/RawFragmentBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/RawFragmentBatch.java
index a7d3f4b..39eae07 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/record/RawFragmentBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/RawFragmentBatch.java
@@ -21,6 +21,8 @@ import io.netty.buffer.ByteBuf;
import org.apache.drill.exec.proto.BitData.FragmentRecordBatch;
import org.apache.drill.exec.rpc.RemoteConnection;
+import org.apache.drill.exec.rpc.ResponseSender;
+import org.apache.drill.exec.rpc.data.DataRpcConfig;
public class RawFragmentBatch {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(RawFragmentBatch.class);
@@ -28,12 +30,14 @@ public class RawFragmentBatch {
final RemoteConnection connection;
final FragmentRecordBatch header;
final ByteBuf body;
+ final ResponseSender sender;
- public RawFragmentBatch(RemoteConnection connection, FragmentRecordBatch header, ByteBuf body) {
+ public RawFragmentBatch(RemoteConnection connection, FragmentRecordBatch header, ByteBuf body, ResponseSender sender) {
super();
this.header = header;
this.body = body;
this.connection = connection;
+ this.sender = sender;
if(body != null) body.retain();
}
@@ -49,7 +53,7 @@ public class RawFragmentBatch {
public String toString() {
return "RawFragmentBatch [header=" + header + ", body=" + body + "]";
}
-
+
public void release(){
if(body != null) body.release();
}
@@ -58,5 +62,12 @@ public class RawFragmentBatch {
return connection;
}
-
+ public ResponseSender getSender() {
+ return sender;
+ }
+
+ public void sendOk(){
+ sender.send(DataRpcConfig.OK);
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/BasicClient.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/BasicClient.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/BasicClient.java
index 562fa90..0b7fd0f 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/BasicClient.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/BasicClient.java
@@ -28,6 +28,7 @@ import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
+import io.netty.channel.epoll.EpollChannelOption;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.GenericFutureListener;
@@ -59,8 +60,10 @@ public abstract class BasicClient<T extends EnumLite, R extends RemoteConnection
b = new Bootstrap() //
.group(eventLoopGroup) //
- .channel(NioSocketChannel.class) //
+ .channel(TransportCheck.getClientSocketChannel()) //
.option(ChannelOption.ALLOCATOR, alloc) //
+ .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 30*1000)
+ .option(ChannelOption.SO_REUSEADDR, true)
.option(ChannelOption.SO_RCVBUF, 1 << 17) //
.option(ChannelOption.SO_SNDBUF, 1 << 17) //
.option(ChannelOption.TCP_NODELAY, true)
@@ -82,9 +85,11 @@ public abstract class BasicClient<T extends EnumLite, R extends RemoteConnection
);
connect = true;
}
- }) //
+ }); //
- ;
+ if(TransportCheck.SUPPORTS_EPOLL){
+ b.option(EpollChannelOption.SO_REUSEPORT, true); //
+ }
}
public abstract ProtobufLengthDecoder getDecoder(BufferAllocator allocator);
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/BasicServer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/BasicServer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/BasicServer.java
index 599f0fd..82d37db 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/BasicServer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/BasicServer.java
@@ -25,10 +25,11 @@ import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
+import io.netty.channel.epoll.EpollChannelOption;
+import io.netty.channel.epoll.EpollServerSocketChannel;
+import io.netty.channel.socket.ServerSocketChannel;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
-import io.netty.handler.logging.LogLevel;
-import io.netty.handler.logging.LoggingHandler;
import java.io.IOException;
import java.net.BindException;
@@ -56,11 +57,12 @@ public abstract class BasicServer<T extends EnumLite, C extends RemoteConnection
public BasicServer(RpcConfig rpcMapping, ByteBufAllocator alloc, EventLoopGroup eventLoopGroup) {
super(rpcMapping);
this.eventLoopGroup = eventLoopGroup;
-
b = new ServerBootstrap() //
- .channel(NioServerSocketChannel.class) //
- .option(ChannelOption.SO_BACKLOG, 100) //
+ .channel(TransportCheck.getServerSocketChannel()) //
+ .option(ChannelOption.SO_BACKLOG, 1000) //
+ .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 30*1000)
.option(ChannelOption.TCP_NODELAY, true)
+ .option(ChannelOption.SO_REUSEADDR, true)
.option(ChannelOption.SO_RCVBUF, 1 << 17) //
.option(ChannelOption.SO_SNDBUF, 1 << 17) //
.group(eventLoopGroup) //
@@ -84,6 +86,10 @@ public abstract class BasicServer<T extends EnumLite, C extends RemoteConnection
// logger.debug("Server connection initialization completed.");
}
});
+
+ if(TransportCheck.SUPPORTS_EPOLL){
+ b.option(EpollChannelOption.SO_REUSEPORT, true); //
+ }
}
public OutOfMemoryHandler getOutOfMemoryHandler() {
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/ResponseSender.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/ResponseSender.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/ResponseSender.java
new file mode 100644
index 0000000..6dc9ae1
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/ResponseSender.java
@@ -0,0 +1,24 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.exec.rpc;
+
+
+public interface ResponseSender {
+ public void send(Response r);
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcBus.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcBus.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcBus.java
index 30101b2..05cca5a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcBus.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcBus.java
@@ -18,6 +18,7 @@
package org.apache.drill.exec.rpc;
import com.google.common.base.Stopwatch;
+
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufInputStream;
import io.netty.channel.Channel;
@@ -45,7 +46,7 @@ import com.google.protobuf.Parser;
/**
* The Rpc Bus deals with incoming and outgoing communication and is used on both the server and the client side of a
* system.
- *
+ *
* @param <T>
*/
public abstract class RpcBus<T extends EnumLite, C extends RemoteConnection> implements Closeable {
@@ -55,6 +56,10 @@ public abstract class RpcBus<T extends EnumLite, C extends RemoteConnection> imp
protected abstract MessageLite getResponseDefaultInstance(int rpcType) throws RpcException;
+ protected void handle(C connection, int rpcType, ByteBuf pBody, ByteBuf dBody, ResponseSender sender) throws RpcException{
+ sender.send(handle(connection, rpcType, pBody, dBody));
+ }
+
protected abstract Response handle(C connection, int rpcType, ByteBuf pBody, ByteBuf dBody) throws RpcException;
public abstract boolean isClient();
@@ -70,19 +75,19 @@ public abstract class RpcBus<T extends EnumLite, C extends RemoteConnection> imp
DrillRpcFutureImpl<RECEIVE> rpcFuture = new DrillRpcFutureImpl<RECEIVE>();
this.send(rpcFuture, connection, rpcType, protobufBody, clazz, dataBodies);
return rpcFuture;
- }
-
+ }
+
public <SEND extends MessageLite, RECEIVE extends MessageLite> void send(RpcOutcomeListener<RECEIVE> listener, C connection, T rpcType,
SEND protobufBody, Class<RECEIVE> clazz, ByteBuf... dataBodies) {
send(listener, connection, rpcType, protobufBody, clazz, false, dataBodies);
}
-
+
public <SEND extends MessageLite, RECEIVE extends MessageLite> void send(RpcOutcomeListener<RECEIVE> listener, C connection, T rpcType,
SEND protobufBody, Class<RECEIVE> clazz, boolean allowInEventLoop, ByteBuf... dataBodies) {
-
+
if(!allowInEventLoop){
if(connection.inEventLoop()) throw new IllegalStateException("You attempted to send while inside the rpc event thread. This isn't allowed because sending will block if the channel is backed up.");
-
+
if(!connection.blockOnNotWritable(listener)) return;
}
@@ -136,6 +141,27 @@ public abstract class RpcBus<T extends EnumLite, C extends RemoteConnection> imp
return new ChannelClosedHandler();
}
+ private class ResponseSenderImpl implements ResponseSender {
+
+ RemoteConnection connection;
+ int coordinationId;
+
+ public ResponseSenderImpl(RemoteConnection connection, int coordinationId) {
+ super();
+ this.connection = connection;
+ this.coordinationId = coordinationId;
+ }
+
+ public void send(Response r){
+ assert rpcConfig.checkResponseSend(r.rpcType, r.pBody.getClass());
+ OutboundRpcMessage outMessage = new OutboundRpcMessage(RpcMode.RESPONSE, r.rpcType, coordinationId,
+ r.pBody, r.dBodies);
+ if (RpcConstants.EXTRA_DEBUGGING) logger.debug("Adding message to outbound buffer. {}", outMessage);
+ connection.getChannel().writeAndFlush(outMessage);
+ }
+
+ }
+
protected class InboundHandler extends MessageToMessageDecoder<InboundRpcMessage> {
private final C connection;
@@ -151,13 +177,9 @@ public abstract class RpcBus<T extends EnumLite, C extends RemoteConnection> imp
switch (msg.mode) {
case REQUEST: {
// handle message and ack.
- Response r = handle(connection, msg.rpcType, msg.pBody, msg.dBody);
+ ResponseSender sender = new ResponseSenderImpl(connection, msg.coordinationId);
+ handle(connection, msg.rpcType, msg.pBody, msg.dBody, sender);
msg.release(); // we release our ownership. Handle could have taken over ownership.
- assert rpcConfig.checkResponseSend(r.rpcType, r.pBody.getClass());
- OutboundRpcMessage outMessage = new OutboundRpcMessage(RpcMode.RESPONSE, r.rpcType, msg.coordinationId,
- r.pBody, r.dBodies);
- if (RpcConstants.EXTRA_DEBUGGING) logger.debug("Adding message to outbound buffer. {}", outMessage);
- ctx.writeAndFlush(outMessage);
break;
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/TransportCheck.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/TransportCheck.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/TransportCheck.java
new file mode 100644
index 0000000..b326a1f
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/TransportCheck.java
@@ -0,0 +1,78 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.exec.rpc;
+
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.epoll.EpollEventLoopGroup;
+import io.netty.channel.epoll.EpollServerSocketChannel;
+import io.netty.channel.epoll.EpollSocketChannel;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.ServerSocketChannel;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.util.internal.SystemPropertyUtil;
+
+import java.util.Locale;
+
+/**
+ * TransportCheck decides whether or not to use the native EPOLL mechanism for communication.
+ */
+@SuppressWarnings("unused")
+public class TransportCheck {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TransportCheck.class);
+
+ public static final boolean SUPPORTS_EPOLL;
+
+ static{
+
+ String name = SystemPropertyUtil.get("os.name").toLowerCase(Locale.US).trim();
+
+ if (!name.startsWith("linux") || SystemPropertyUtil.getBoolean("drill.exec.disable-linux-epoll", false) //
+// /* disable epoll */ || true //
+ ) {
+ SUPPORTS_EPOLL = false;
+ }else{
+ SUPPORTS_EPOLL = true;
+ }
+ }
+
+ public static Class<? extends ServerSocketChannel> getServerSocketChannel(){
+ if(SUPPORTS_EPOLL){
+ return EpollServerSocketChannel.class;
+ }else{
+ return NioServerSocketChannel.class;
+ }
+ }
+
+ public static Class<? extends SocketChannel> getClientSocketChannel(){
+ if(SUPPORTS_EPOLL){
+ return EpollSocketChannel.class;
+ }else{
+ return NioSocketChannel.class;
+ }
+ }
+
+ public static EventLoopGroup createEventLoopGroup(int nThreads, String prefix) {
+ if(SUPPORTS_EPOLL){
+ return new EpollEventLoopGroup(nThreads, new NamedThreadFactory(prefix));
+ }else{
+ return new NioEventLoopGroup(nThreads, new NamedThreadFactory(prefix));
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/BitServerConnection.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/BitServerConnection.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/BitServerConnection.java
index 58fa403..4583f88 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/BitServerConnection.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/BitServerConnection.java
@@ -17,110 +17,24 @@
*/
package org.apache.drill.exec.rpc.data;
-import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.Channel;
-import org.apache.drill.exec.memory.AccountingByteBuf;
import org.apache.drill.exec.memory.BufferAllocator;
-import org.apache.drill.exec.memory.OutOfMemoryException;
-import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
import org.apache.drill.exec.rpc.RemoteConnection;
import org.apache.drill.exec.work.fragment.FragmentManager;
public class BitServerConnection extends RemoteConnection{
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(BitServerConnection.class);
- private AllocatorProxy proxy = new AllocatorProxy();
- private volatile FragmentManager manager;
-
- public BitServerConnection(Channel channel, BufferAllocator initialAllocator) {
+ private final BufferAllocator allocator;
+ public BitServerConnection(Channel channel, BufferAllocator allocator) {
super(channel);
- proxy.setAllocator(initialAllocator);
- }
-
- void setManager(FragmentManager manager){
- this.manager = manager;
- if (manager != null) { // Do this check for TestBitRpc test
- this.proxy.setAllocator(manager.getFragmentContext().getAllocator());
- manager.addConnection(this);
- }
+ this.allocator = allocator;
}
@Override
public BufferAllocator getAllocator() {
- if(manager != null){
- return manager.getFragmentContext().getAllocator();
- }
- return proxy;
- }
-
- public FragmentManager getFragmentManager(){
- return manager;
+ return allocator;
}
- final static String ERROR_MESSAGE = "Attempted to access AllocatorProxy";
-
- private static class AllocatorProxy implements BufferAllocator {
- private BufferAllocator allocator;
-
- public void setAllocator(BufferAllocator allocator) {
- this.allocator = allocator;
- }
-
- @Override
- public AccountingByteBuf buffer(int size) {
- if (allocator == null) {
- throw new UnsupportedOperationException(ERROR_MESSAGE);
- }
- return allocator.buffer(size);
- }
-
- @Override
- public AccountingByteBuf buffer(int minSize, int maxSize) {
- if (allocator == null) {
- throw new UnsupportedOperationException(ERROR_MESSAGE);
- }
- return allocator.buffer(minSize, maxSize);
- }
-
- @Override
- public ByteBufAllocator getUnderlyingAllocator() {
- if (allocator == null) {
- throw new UnsupportedOperationException(ERROR_MESSAGE);
- }
- return allocator.getUnderlyingAllocator();
- }
-
- @Override
- public BufferAllocator getChildAllocator(FragmentHandle handle, long initialReservation, long maximumReservation) throws OutOfMemoryException {
- if (allocator == null) {
- throw new UnsupportedOperationException(ERROR_MESSAGE);
- }
- return allocator.getChildAllocator(handle, initialReservation, maximumReservation);
- }
-
- @Override
- public PreAllocator getNewPreAllocator() {
- if (allocator == null) {
- throw new UnsupportedOperationException(ERROR_MESSAGE);
- }
- return allocator.getNewPreAllocator();
- }
-
- @Override
- public void close() {
- if (allocator != null) {
- allocator.close();
- }
- }
-
- @Override
- public long getAllocatedMemory() {
- if (allocator == null) {
- throw new UnsupportedOperationException(ERROR_MESSAGE);
- }
- return allocator.getAllocatedMemory();
- }
- }
-
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataConnectionCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataConnectionCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataConnectionCreator.java
index 9c2ef5b..197996d 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataConnectionCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataConnectionCreator.java
@@ -18,6 +18,7 @@
package org.apache.drill.exec.rpc.data;
import java.io.Closeable;
+import java.util.concurrent.ConcurrentMap;
import org.apache.drill.exec.exception.DrillbitStartupException;
import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
@@ -25,10 +26,11 @@ import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
import org.apache.drill.exec.rpc.control.WorkEventBus;
import org.apache.drill.exec.server.BootStrapContext;
+import com.google.common.collect.Maps;
import com.google.common.io.Closeables;
/**
- * Manages a connection pool for each endpoint.
+ * Manages a connection for each endpoint.
*/
public class DataConnectionCreator implements Closeable {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DataConnectionCreator.class);
@@ -38,6 +40,7 @@ public class DataConnectionCreator implements Closeable {
private final WorkEventBus workBus;
private final DataResponseHandler dataHandler;
private final boolean allowPortHunting;
+ private ConcurrentMap<DrillbitEndpoint, DataConnectionManager> connectionManager = Maps.newConcurrentMap();
public DataConnectionCreator(BootStrapContext context, WorkEventBus workBus, DataResponseHandler dataHandler, boolean allowPortHunting) {
super();
@@ -55,7 +58,12 @@ public class DataConnectionCreator implements Closeable {
}
public DataTunnel getTunnel(DrillbitEndpoint endpoint, FragmentHandle handle) {
- return new DataTunnel(new DataConnectionManager(handle, endpoint, context));
+ DataConnectionManager newManager = new DataConnectionManager(handle, endpoint, context);
+ DataConnectionManager oldManager = connectionManager.putIfAbsent(endpoint, newManager);
+ if(oldManager != null){
+ newManager = oldManager;
+ }
+ return new DataTunnel(newManager);
}
public void close() {
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataResponseHandler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataResponseHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataResponseHandler.java
index aaff504..f1c94d3 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataResponseHandler.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataResponseHandler.java
@@ -22,10 +22,14 @@ import io.netty.buffer.ByteBuf;
import org.apache.drill.exec.proto.BitData.FragmentRecordBatch;
import org.apache.drill.exec.rpc.RemoteConnection;
import org.apache.drill.exec.rpc.Response;
+import org.apache.drill.exec.rpc.ResponseSender;
import org.apache.drill.exec.rpc.RpcException;
import org.apache.drill.exec.work.fragment.FragmentManager;
public interface DataResponseHandler {
- public Response handle(RemoteConnection connection, FragmentManager manager, FragmentRecordBatch fragmentBatch, ByteBuf data) throws RpcException;
+ public void handle(RemoteConnection connection, FragmentManager manager, FragmentRecordBatch fragmentBatch,
+ ByteBuf data, ResponseSender responder) throws RpcException;
+
+ public void informOutOfMemory();
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataResponseHandlerImpl.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataResponseHandlerImpl.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataResponseHandlerImpl.java
index b5b41a3..5d71c90 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataResponseHandlerImpl.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataResponseHandlerImpl.java
@@ -26,6 +26,7 @@ import org.apache.drill.exec.record.RawFragmentBatch;
import org.apache.drill.exec.rpc.Acks;
import org.apache.drill.exec.rpc.RemoteConnection;
import org.apache.drill.exec.rpc.Response;
+import org.apache.drill.exec.rpc.ResponseSender;
import org.apache.drill.exec.rpc.RpcException;
import org.apache.drill.exec.work.WorkManager.WorkerBee;
import org.apache.drill.exec.work.fragment.FragmentManager;
@@ -40,10 +41,17 @@ public class DataResponseHandlerImpl implements DataResponseHandler{
this.bee = bee;
}
- public Response handle(RemoteConnection connection, FragmentManager manager, FragmentRecordBatch fragmentBatch, ByteBuf data) throws RpcException {
+
+ @Override
+ public void informOutOfMemory() {
+ logger.error("Out of memory outside any particular fragment.");
+ }
+
+
+ public void handle(RemoteConnection connection, FragmentManager manager, FragmentRecordBatch fragmentBatch, ByteBuf data, ResponseSender sender) throws RpcException {
// logger.debug("Fragment Batch received {}", fragmentBatch);
try {
- boolean canRun = manager.handle(new RawFragmentBatch(connection, fragmentBatch, data));
+ boolean canRun = manager.handle(new RawFragmentBatch(connection, fragmentBatch, data, sender));
if (canRun) {
// logger.debug("Arriving batch means local batch can run, starting local batch.");
// if we've reached the canRun threshold, we'll proceed. This expects handler.handle() to only return a single
@@ -56,9 +64,8 @@ public class DataResponseHandlerImpl implements DataResponseHandler{
bee.getContext().getWorkBus().removeFragmentManager(manager.getHandle());
}
- return DataRpcConfig.OK;
} catch (FragmentSetupException e) {
logger.error("Failure while attempting to setup new fragment.", e);
- return new Response(RpcType.ACK, Acks.FAIL);
+ sender.send(new Response(RpcType.ACK, Acks.FAIL));
}
}}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataServer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataServer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataServer.java
index ad57956..a449032 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataServer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataServer.java
@@ -17,6 +17,7 @@
*/
package org.apache.drill.exec.rpc.data;
+import io.netty.buffer.AccountingByteBuf;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
@@ -28,15 +29,19 @@ import org.apache.drill.exec.proto.BitData.BitClientHandshake;
import org.apache.drill.exec.proto.BitData.BitServerHandshake;
import org.apache.drill.exec.proto.BitData.FragmentRecordBatch;
import org.apache.drill.exec.proto.BitData.RpcType;
+import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
import org.apache.drill.exec.proto.UserBitShared.RpcChannel;
-import org.apache.drill.exec.record.RawFragmentBatch;
+import org.apache.drill.exec.proto.helper.QueryIdHelper;
+import org.apache.drill.exec.rpc.Acks;
import org.apache.drill.exec.rpc.BasicServer;
import org.apache.drill.exec.rpc.OutOfMemoryHandler;
import org.apache.drill.exec.rpc.ProtobufLengthDecoder;
import org.apache.drill.exec.rpc.Response;
+import org.apache.drill.exec.rpc.ResponseSender;
import org.apache.drill.exec.rpc.RpcException;
import org.apache.drill.exec.rpc.control.WorkEventBus;
import org.apache.drill.exec.server.BootStrapContext;
+import org.apache.drill.exec.work.fragment.FragmentManager;
import com.google.protobuf.MessageLite;
@@ -47,7 +52,6 @@ public class DataServer extends BasicServer<RpcType, BitServerConnection> {
private final BootStrapContext context;
private final WorkEventBus workBus;
private final DataResponseHandler dataHandler;
- private BitServerConnection connection;
public DataServer(BootStrapContext context, WorkEventBus workBus, DataResponseHandler dataHandler) {
super(DataRpcConfig.MAPPING, context.getAllocator().getUnderlyingAllocator(), context.getBitLoopGroup());
@@ -69,7 +73,7 @@ public class DataServer extends BasicServer<RpcType, BitServerConnection> {
@Override
public BitServerConnection initRemoteConnection(Channel channel) {
- return connection = new BitServerConnection(channel, context.getAllocator());
+ return new BitServerConnection(channel, context.getAllocator());
}
@Override
@@ -86,20 +90,35 @@ public class DataServer extends BasicServer<RpcType, BitServerConnection> {
throw new RpcException(String.format("Invalid NodeMode. Expected BIT_DATA but received %s.",
inbound.getChannel()));
- connection.setManager(workBus.getOrCreateFragmentManager(inbound.getHandle()));
return BitServerHandshake.newBuilder().setRpcVersion(DataRpcConfig.RPC_VERSION).build();
}
};
}
+ private final static FragmentRecordBatch OOM_FRAGMENT = FragmentRecordBatch.newBuilder().setIsOutOfMemory(true).build();
@Override
- protected Response handle(BitServerConnection connection, int rpcType, ByteBuf pBody, ByteBuf body) throws RpcException {
+ protected void handle(BitServerConnection connection, int rpcType, ByteBuf pBody, ByteBuf body, ResponseSender sender) throws RpcException {
assert rpcType == RpcType.REQ_RECORD_BATCH_VALUE;
FragmentRecordBatch fragmentBatch = get(pBody, FragmentRecordBatch.PARSER);
- return dataHandler.handle(connection, connection.getFragmentManager(), fragmentBatch, body);
+ FragmentHandle handle = fragmentBatch.getHandle();
+
+ try {
+ FragmentManager manager = workBus.getOrCreateFragmentManager(fragmentBatch.getHandle());
+ BufferAllocator allocator = manager.getFragmentContext().getAllocator();
+ if(body != null){
+ if(!allocator.takeOwnership((AccountingByteBuf) body.unwrap())){
+ dataHandler.handle(connection, manager, OOM_FRAGMENT, null, null);
+ }
+ }
+ dataHandler.handle(connection, manager, fragmentBatch, body, sender);
+
+ } catch (FragmentSetupException e) {
+ logger.error("Failure while getting fragment manager. {}", QueryIdHelper.getQueryIdentifier(handle), e);
+ sender.send(new Response(RpcType.ACK, Acks.FAIL));
+ }
}
private class ProxyCloseHandler implements GenericFutureListener<ChannelFuture> {
@@ -111,14 +130,6 @@ public class DataServer extends BasicServer<RpcType, BitServerConnection> {
this.handler = handler;
}
- public GenericFutureListener<ChannelFuture> getHandler() {
- return handler;
- }
-
- public void setHandler(GenericFutureListener<ChannelFuture> handler) {
- this.handler = handler;
- }
-
@Override
public void operationComplete(ChannelFuture future) throws Exception {
handler.operationComplete(future);
@@ -126,20 +137,14 @@ public class DataServer extends BasicServer<RpcType, BitServerConnection> {
}
- private final static FragmentRecordBatch OOM_FRAGMENT = FragmentRecordBatch.newBuilder().setIsOutOfMemory(true).build();
+
@Override
public OutOfMemoryHandler getOutOfMemoryHandler() {
return new OutOfMemoryHandler() {
@Override
public void handle() {
- try {
- logger.debug("Setting autoRead false");
- connection.getFragmentManager().setAutoRead(false);
- connection.getFragmentManager().handle(new RawFragmentBatch(connection, OOM_FRAGMENT, null));
- } catch (FragmentSetupException e) {
- throw new RuntimeException();
- }
+ dataHandler.informOutOfMemory();
}
};
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataTunnel.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataTunnel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataTunnel.java
index 3c2b9e3..546f967 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataTunnel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/data/DataTunnel.java
@@ -18,6 +18,9 @@
package org.apache.drill.exec.rpc.data;
import io.netty.buffer.ByteBuf;
+
+import java.util.concurrent.Semaphore;
+
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.proto.BitData.RpcType;
import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
@@ -25,13 +28,16 @@ import org.apache.drill.exec.record.FragmentWritableBatch;
import org.apache.drill.exec.rpc.DrillRpcFuture;
import org.apache.drill.exec.rpc.FutureBitCommand;
import org.apache.drill.exec.rpc.ListeningCommand;
+import org.apache.drill.exec.rpc.RpcException;
import org.apache.drill.exec.rpc.RpcOutcomeListener;
+import org.apache.drill.exec.rpc.RpcConnectionHandler.FailureType;
public class DataTunnel {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DataTunnel.class);
private final DataConnectionManager manager;
+ private final Semaphore sendingSemaphore = new Semaphore(3);
public DataTunnel(DataConnectionManager manager) {
this.manager = manager;
@@ -39,18 +45,49 @@ public class DataTunnel {
public void sendRecordBatch(RpcOutcomeListener<Ack> outcomeListener, FragmentWritableBatch batch) {
SendBatchAsyncListen b = new SendBatchAsyncListen(outcomeListener, batch);
- manager.runCommand(b);
+ try{
+ sendingSemaphore.acquire();
+ manager.runCommand(b);
+ }catch(InterruptedException e){
+ outcomeListener.failed(new RpcException("Interrupted while trying to get sending semaphore.", e));
+ }
}
public DrillRpcFuture<Ack> sendRecordBatch(FragmentContext context, FragmentWritableBatch batch) {
SendBatchAsyncFuture b = new SendBatchAsyncFuture(batch, context);
- manager.runCommand(b);
+ try{
+ sendingSemaphore.acquire();
+ manager.runCommand(b);
+ }catch(InterruptedException e){
+ b.connectionFailed(FailureType.CONNECTION, new RpcException("Interrupted while trying to get sending semaphore.", e));
+ }
return b.getFuture();
}
-
- private static class SendBatchAsyncListen extends ListeningCommand<Ack, DataClientConnection> {
+ private class ThrottlingOutcomeListener implements RpcOutcomeListener<Ack>{
+ RpcOutcomeListener<Ack> inner;
+
+ public ThrottlingOutcomeListener(RpcOutcomeListener<Ack> inner) {
+ super();
+ this.inner = inner;
+ }
+
+ @Override
+ public void failed(RpcException ex) {
+ sendingSemaphore.release();
+ inner.failed(ex);
+ }
+
+ @Override
+ public void success(Ack value, ByteBuf buffer) {
+ sendingSemaphore.release();
+ inner.success(value, buffer);
+ }
+
+ }
+
+ private class SendBatchAsyncListen extends ListeningCommand<Ack, DataClientConnection> {
final FragmentWritableBatch batch;
public SendBatchAsyncListen(RpcOutcomeListener<Ack> listener, FragmentWritableBatch batch) {
@@ -60,7 +97,7 @@ public class DataTunnel {
@Override
public void doRpcCall(RpcOutcomeListener<Ack> outcomeListener, DataClientConnection connection) {
- connection.send(outcomeListener, RpcType.REQ_RECORD_BATCH, batch.getHeader(), Ack.class, batch.getBuffers());
+ connection.send(new ThrottlingOutcomeListener(outcomeListener), RpcType.REQ_RECORD_BATCH, batch.getHeader(), Ack.class, batch.getBuffers());
}
@Override
@@ -77,7 +114,7 @@ public class DataTunnel {
}
}
- private static class SendBatchAsyncFuture extends FutureBitCommand<Ack, DataClientConnection> {
+ private class SendBatchAsyncFuture extends FutureBitCommand<Ack, DataClientConnection> {
final FragmentWritableBatch batch;
final FragmentContext context;
@@ -89,7 +126,7 @@ public class DataTunnel {
@Override
public void doRpcCall(RpcOutcomeListener<Ack> outcomeListener, DataClientConnection connection) {
- connection.send(outcomeListener, RpcType.REQ_RECORD_BATCH, batch.getHeader(), Ack.class, batch.getBuffers());
+ connection.send(new ThrottlingOutcomeListener(outcomeListener), RpcType.REQ_RECORD_BATCH, batch.getHeader(), Ack.class, batch.getBuffers());
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/server/BootStrapContext.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/BootStrapContext.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/BootStrapContext.java
index ddfb42c..41d13a0 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/BootStrapContext.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/BootStrapContext.java
@@ -17,6 +17,8 @@
*/
package org.apache.drill.exec.server;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import java.io.Closeable;
@@ -27,6 +29,7 @@ import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.memory.TopLevelAllocator;
import org.apache.drill.exec.metrics.DrillMetrics;
import org.apache.drill.exec.rpc.NamedThreadFactory;
+import org.apache.drill.exec.rpc.TransportCheck;
import com.codahale.metrics.MetricRegistry;
@@ -34,16 +37,16 @@ public class BootStrapContext implements Closeable{
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(BootStrapContext.class);
private final DrillConfig config;
- private final NioEventLoopGroup loop;
- private final NioEventLoopGroup loop2;
+ private final EventLoopGroup loop;
+ private final EventLoopGroup loop2;
private final MetricRegistry metrics;
private final BufferAllocator allocator;
public BootStrapContext(DrillConfig config) {
super();
this.config = config;
- this.loop = new NioEventLoopGroup(config.getInt(ExecConstants.BIT_SERVER_RPC_THREADS), new NamedThreadFactory("BitServer-"));
- this.loop2 = new NioEventLoopGroup(config.getInt(ExecConstants.BIT_SERVER_RPC_THREADS), new NamedThreadFactory("BitClient-"));
+ this.loop = TransportCheck.createEventLoopGroup(config.getInt(ExecConstants.BIT_SERVER_RPC_THREADS), "BitServer-");
+ this.loop2 = TransportCheck.createEventLoopGroup(config.getInt(ExecConstants.BIT_SERVER_RPC_THREADS), "BitClient-");
this.metrics = DrillMetrics.getInstance();
this.allocator = new TopLevelAllocator(config);
}
@@ -52,11 +55,11 @@ public class BootStrapContext implements Closeable{
return config;
}
- public NioEventLoopGroup getBitLoopGroup() {
+ public EventLoopGroup getBitLoopGroup() {
return loop;
}
- public NioEventLoopGroup getBitClientLoopGroup() {
+ public EventLoopGroup getBitClientLoopGroup() {
return loop2;
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/server/DrillbitContext.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/DrillbitContext.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/DrillbitContext.java
index 556724e..eed0126 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/DrillbitContext.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/DrillbitContext.java
@@ -17,7 +17,7 @@
*/
package org.apache.drill.exec.server;
-import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.EventLoopGroup;
import java.util.Collection;
@@ -117,7 +117,7 @@ public class DrillbitContext {
return this.storagePlugins;
}
- public NioEventLoopGroup getBitLoopGroup(){
+ public EventLoopGroup getBitLoopGroup(){
return context.getBitLoopGroup();
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/vector/BitVector.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/vector/BitVector.java b/exec/java-exec/src/main/java/org/apache/drill/exec/vector/BitVector.java
index 73f97fe..01fcfe0 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/vector/BitVector.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/vector/BitVector.java
@@ -17,12 +17,12 @@
*/
package org.apache.drill.exec.vector;
+import io.netty.buffer.AccountingByteBuf;
import io.netty.buffer.ByteBuf;
import org.apache.drill.common.expression.FieldReference;
import org.apache.drill.exec.expr.holders.BitHolder;
import org.apache.drill.exec.expr.holders.NullableBitHolder;
-import org.apache.drill.exec.memory.AccountingByteBuf;
import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.memory.OutOfMemoryRuntimeException;
import org.apache.drill.exec.proto.UserBitShared.SerializedField;
@@ -366,11 +366,7 @@ public final class BitVector extends BaseDataValueVector implements FixedWidthVe
} else if (allocationMonitor > 0) {
allocationMonitor = 0;
}
- data.writerIndex(idx);
- if (data instanceof AccountingByteBuf) {
- data.capacity(idx);
- data.writerIndex(idx);
- }
+ VectorTrimmer.trim(data, idx);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/vector/VectorTrimmer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/vector/VectorTrimmer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/vector/VectorTrimmer.java
new file mode 100644
index 0000000..e946afe
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/vector/VectorTrimmer.java
@@ -0,0 +1,33 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.exec.vector;
+
+import io.netty.buffer.AccountingByteBuf;
+import io.netty.buffer.ByteBuf;
+
+public class VectorTrimmer {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(VectorTrimmer.class);
+
+ public static void trim(ByteBuf data, int idx) {
+ data.writerIndex(idx);
+ if (data instanceof AccountingByteBuf) {
+ // data.capacity(idx);
+ data.writerIndex(idx);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/AbstractDataCollector.java
----------------------------------------------------------------------
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 a67f06b..23794d9 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
@@ -23,25 +23,23 @@ import java.lang.reflect.InvocationTargetException;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerArray;
-import java.util.concurrent.atomic.AtomicReferenceArray;
import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.base.Receiver;
import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
import org.apache.drill.exec.record.RawFragmentBatch;
-import org.apache.drill.exec.rpc.RemoteConnection;
+import org.apache.drill.exec.rpc.ResponseSender;
import com.google.common.base.Preconditions;
-public abstract class AbstractDataCollector implements DataCollector, ReadController{
+public abstract class AbstractDataCollector implements DataCollector{
private final List<DrillbitEndpoint> incoming;
private final int oppositeMajorFragmentId;
private final AtomicIntegerArray remainders;
private final AtomicInteger remainingRequired;
protected final RawBatchBuffer[] buffers;
- private final AtomicReferenceArray<RemoteConnection> connections;
private final AtomicInteger parentAccounter;
private final AtomicInteger finishedStreams = new AtomicInteger();
@@ -52,15 +50,14 @@ public abstract class AbstractDataCollector implements DataCollector, ReadContro
this.parentAccounter = parentAccounter;
this.incoming = receiver.getProvidingEndpoints();
- this.connections = new AtomicReferenceArray<>(incoming.size());
this.remainders = new AtomicIntegerArray(incoming.size());
this.oppositeMajorFragmentId = receiver.getOppositeMajorFragmentId();
this.buffers = new RawBatchBuffer[minInputsRequired];
try {
String bufferClassName = context.getConfig().getString(ExecConstants.INCOMING_BUFFER_IMPL);
- Constructor<?> bufferConstructor = Class.forName(bufferClassName).getConstructor(FragmentContext.class, ReadController.class, int.class);
+ Constructor<?> bufferConstructor = Class.forName(bufferClassName).getConstructor(FragmentContext.class, int.class);
for(int i = 0; i < buffers.length; i++) {
- buffers[i] = (RawBatchBuffer) bufferConstructor.newInstance(context, this, incoming.size());
+ buffers[i] = (RawBatchBuffer) bufferConstructor.newInstance(context, incoming.size());
}
} catch (InstantiationException | IllegalAccessException | InvocationTargetException |
NoSuchMethodException | ClassNotFoundException e) {
@@ -82,17 +79,6 @@ public abstract class AbstractDataCollector implements DataCollector, ReadContro
}
- public void setAutoRead(boolean enabled){
- for(int i = 0; i < connections.length(); i++){
- setAutoRead(i, enabled);
- }
- }
-
- public void setAutoRead(int minorFragmentId, boolean enabled){
- RemoteConnection c = connections.get(minorFragmentId);
- if(c != null) c.setAutoRead(enabled);
- }
-
public abstract void streamFinished(int minorFragmentId);
public boolean batchArrived(int minorFragmentId, RawFragmentBatch batch) throws IOException {
@@ -104,10 +90,6 @@ public abstract class AbstractDataCollector implements DataCollector, ReadContro
}
}
- // add the connection to the connection list if this is the first time we're seeing it.
- // TODO: move this to a better location (e.g. connection setup).
- if(connections.compareAndSet(minorFragmentId, null, batch.getConnection()));
-
// check to see if we have enough fragments reporting to proceed.
boolean decremented = false;
if (remainders.compareAndSet(minorFragmentId, 0, 1)) {
@@ -138,11 +120,6 @@ public abstract class AbstractDataCollector implements DataCollector, ReadContro
@Override
public void close() {
- for (int i = 0; i < connections.length(); i++) {
- if (connections.get(i) != null) {
- connections.get(i).close();
- };
- }
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/DataCollector.java
----------------------------------------------------------------------
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 dc016be..42d8eae 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
@@ -20,6 +20,7 @@ package org.apache.drill.exec.work.batch;
import java.io.IOException;
import org.apache.drill.exec.record.RawFragmentBatch;
+import org.apache.drill.exec.rpc.ResponseSender;
interface DataCollector extends AutoCloseable {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DataCollector.class);
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/IncomingBuffers.java
----------------------------------------------------------------------
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 3b97934..af5e9c7 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
@@ -28,6 +28,7 @@ import org.apache.drill.exec.physical.base.AbstractPhysicalVisitor;
import org.apache.drill.exec.physical.base.PhysicalOperator;
import org.apache.drill.exec.physical.base.Receiver;
import org.apache.drill.exec.record.RawFragmentBatch;
+import org.apache.drill.exec.rpc.ResponseSender;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
@@ -81,10 +82,12 @@ public class IncomingBuffers implements AutoCloseable {
DataCollector fSet = fragCounts.get(sendMajorFragmentId);
if (fSet == null) throw new FragmentSetupException(String.format("We received a major fragment id that we were not expecting. The id was %d. %s", sendMajorFragmentId, Arrays.toString(fragCounts.values().toArray())));
try {
- boolean decremented = fSet.batchArrived(batch.getHeader().getSendingMinorFragmentId(), batch);
+ synchronized(this){
+ boolean decremented = fSet.batchArrived(batch.getHeader().getSendingMinorFragmentId(), batch);
- // we should only return true if remaining required has been decremented and is currently equal to zero.
- return decremented && remainingRequired.get() == 0;
+ // we should only return true if remaining required has been decremented and is currently equal to zero.
+ return decremented && remainingRequired.get() == 0;
+ }
} catch (IOException e) {
throw new FragmentSetupException(e);
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/RawBatchBuffer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/RawBatchBuffer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/RawBatchBuffer.java
index 23443fa..bc8a1dc 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/RawBatchBuffer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/RawBatchBuffer.java
@@ -21,7 +21,7 @@ import java.io.IOException;
import org.apache.drill.exec.record.RawFragmentBatch;
import org.apache.drill.exec.record.RawFragmentBatchProvider;
-import org.apache.drill.exec.rpc.RemoteConnection;
+import org.apache.drill.exec.rpc.ResponseSender;
/**
* A batch buffer is responsible for queuing incoming batches until a consumer is ready to receive them. It will also
@@ -36,6 +36,7 @@ public interface RawBatchBuffer extends RawFragmentBatchProvider {
* @param batch
* Batch to enqueue
* @throws IOException
+ * @returns Whether response should be returned.
*/
public void enqueue(RawFragmentBatch batch) throws IOException;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/ReadController.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/ReadController.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/ReadController.java
deleted file mode 100644
index d23eb4c..0000000
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/ReadController.java
+++ /dev/null
@@ -1,23 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.drill.exec.work.batch;
-
-public interface ReadController {
- public void setAutoRead(boolean enabled);
- public void setAutoRead(int minorFragmentId, boolean enabled);
-}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/ResponseSenderQueue.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/ResponseSenderQueue.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/ResponseSenderQueue.java
new file mode 100644
index 0000000..5a9316e
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/ResponseSenderQueue.java
@@ -0,0 +1,45 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.exec.work.batch;
+
+import java.util.Queue;
+
+import org.apache.drill.exec.rpc.ResponseSender;
+import org.apache.drill.exec.rpc.data.DataRpcConfig;
+
+import com.google.common.collect.Queues;
+
+public class ResponseSenderQueue {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ResponseSenderQueue.class);
+
+ private Queue<ResponseSender> q = Queues.newConcurrentLinkedQueue();
+
+ public void enqueueResponse(ResponseSender sender){
+ q.add(sender);
+ }
+
+ public void flushResponses(){
+ while(!q.isEmpty()){
+ ResponseSender s = q.poll();
+ if(s != null){
+ s.send(DataRpcConfig.OK);
+ }
+ }
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/SpoolingRawBatchBuffer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/SpoolingRawBatchBuffer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/SpoolingRawBatchBuffer.java
index 9ee4d0f..76d642c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/SpoolingRawBatchBuffer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/SpoolingRawBatchBuffer.java
@@ -37,7 +37,9 @@ import org.apache.drill.exec.proto.ExecProtos;
import org.apache.drill.exec.proto.helper.QueryIdHelper;
import org.apache.drill.exec.record.RawFragmentBatch;
import org.apache.drill.exec.rpc.RemoteConnection;
+import org.apache.drill.exec.rpc.ResponseSender;
import org.apache.drill.exec.rpc.data.BitServerConnection;
+import org.apache.drill.exec.rpc.data.DataRpcConfig;
import org.apache.drill.exec.store.LocalSyncableFileSystem;
import org.apache.drill.exec.work.fragment.FragmentManager;
import org.apache.hadoop.conf.Configuration;
@@ -77,7 +79,7 @@ public class SpoolingRawBatchBuffer implements RawBatchBuffer {
private boolean closed = false;
private FragmentManager fragmentManager;
- public SpoolingRawBatchBuffer(FragmentContext context, ReadController readController, int fragmentCount) throws IOException, OutOfMemoryException {
+ public SpoolingRawBatchBuffer(FragmentContext context, int fragmentCount) throws IOException, OutOfMemoryException {
this.context = context;
this.allocator = context.getNewChildAllocator(ALLOCATOR_INITIAL_RESERVATION, ALLOCATOR_MAX_RESERVATION);
this.threshold = context.getConfig().getLong(ExecConstants.SPOOLING_BUFFER_MEMORY);
@@ -99,7 +101,8 @@ public class SpoolingRawBatchBuffer implements RawBatchBuffer {
public synchronized void enqueue(RawFragmentBatch batch) throws IOException {
if (batch.getHeader().getIsOutOfMemory()) {
if (fragmentManager == null) {
- fragmentManager = ((BitServerConnection) batch.getConnection()).getFragmentManager();
+ throw new UnsupportedOperationException("Need to fix.");
+// fragmentManager = ((BitServerConnection) batch.getConnection()).getFragmentManager();
}
// fragmentManager.setAutoRead(false);
// logger.debug("Setting autoRead false");
@@ -109,6 +112,7 @@ public class SpoolingRawBatchBuffer implements RawBatchBuffer {
} else {
logger.debug("ignoring duplicate OOM message");
}
+ batch.sendOk();
return;
}
RawFragmentBatchWrapper wrapper;
@@ -270,7 +274,7 @@ public class SpoolingRawBatchBuffer implements RawBatchBuffer {
BitData.FragmentRecordBatch header = BitData.FragmentRecordBatch.parseDelimitedFrom(stream);
ByteBuf buf = allocator.buffer(bodyLength);
buf.writeBytes(stream, bodyLength);
- batch = new RawFragmentBatch(null, header, buf);
+ batch = new RawFragmentBatch(null, header, buf, null);
buf.release();
available = true;
latch.countDown();
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/UnlimitedRawBatchBuffer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/UnlimitedRawBatchBuffer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/UnlimitedRawBatchBuffer.java
index d14e50c..11e20c7 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/UnlimitedRawBatchBuffer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/UnlimitedRawBatchBuffer.java
@@ -23,6 +23,8 @@ import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.record.RawFragmentBatch;
+import org.apache.drill.exec.rpc.ResponseSender;
+import org.apache.drill.exec.rpc.data.DataRpcConfig;
import com.google.common.collect.Queues;
@@ -35,32 +37,20 @@ public class UnlimitedRawBatchBuffer implements RawBatchBuffer{
private final int startlimit;
private final AtomicBoolean overlimit = new AtomicBoolean(false);
private final AtomicBoolean outOfMemory = new AtomicBoolean(false);
- private final ReadController readController;
- private final boolean multiFragment;
+ private final ResponseSenderQueue readController = new ResponseSenderQueue();
- public UnlimitedRawBatchBuffer(FragmentContext context, ReadController readController, int fragmentCount) {
+ public UnlimitedRawBatchBuffer(FragmentContext context, int fragmentCount) {
int bufferSizePerSocket = context.getConfig().getInt(ExecConstants.INCOMING_BUFFER_SIZE);
- this.multiFragment = fragmentCount > 1;
- this.readController = readController;
this.softlimit = bufferSizePerSocket * fragmentCount;
this.startlimit = Math.max(softlimit/2, 1);
this.buffer = Queues.newLinkedBlockingDeque();
}
- private void setRead(int minorFragmentId, boolean setting){
- if(multiFragment){
- readController.setAutoRead(setting);
- }else{
- readController.setAutoRead(minorFragmentId, setting);
- }
- }
-
@Override
public void enqueue(RawFragmentBatch batch) {
if (batch.getHeader().getIsOutOfMemory()) {
logger.debug("Setting autoread false");
- readController.setAutoRead(false);
if (!outOfMemory.get() && !buffer.peekFirst().getHeader().getIsOutOfMemory()) {
buffer.addFirst(batch);
}
@@ -70,7 +60,9 @@ public class UnlimitedRawBatchBuffer implements RawBatchBuffer{
buffer.add(batch);
if(buffer.size() == softlimit){
overlimit.set(true);
- setRead(batch.getHeader().getSendingMinorFragmentId(), false);
+ readController.enqueueResponse(batch.getSender());
+ }else{
+ batch.sendOk();
}
}
@@ -98,7 +90,7 @@ public class UnlimitedRawBatchBuffer implements RawBatchBuffer{
if (outOfMemory.get() && buffer.size() < 10) {
logger.debug("Setting autoread true");
outOfMemory.set(false);
- readController.setAutoRead(true);
+ readController.flushResponses();
}
RawFragmentBatch b = null;
@@ -116,7 +108,6 @@ public class UnlimitedRawBatchBuffer implements RawBatchBuffer{
if (b != null && b.getHeader().getIsOutOfMemory()) {
outOfMemory.set(true);
- readController.setAutoRead(false);
return b;
}
@@ -125,8 +116,7 @@ public class UnlimitedRawBatchBuffer implements RawBatchBuffer{
if(!finished && overlimit.get()){
if(buffer.size() == startlimit){
overlimit.set(false);
- setRead(b.getHeader().getSendingMinorFragmentId(), true);
- readController.setAutoRead(true);
+ readController.flushResponses();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/FragmentManager.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/FragmentManager.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/FragmentManager.java
index 0a4b235..68849ff 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/FragmentManager.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/FragmentManager.java
@@ -22,6 +22,7 @@ import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
import org.apache.drill.exec.record.RawFragmentBatch;
import org.apache.drill.exec.rpc.RemoteConnection;
+import org.apache.drill.exec.rpc.ResponseSender;
/**
* The Fragment Manager is responsible managing incoming data and executing a fragment. Once enough data and resources
@@ -31,7 +32,7 @@ public interface FragmentManager {
/**
* Handle the next incoming record batch.
- *
+ *
* @param batch
* @return True if the fragment has enough incoming data to be able to be run.
* @throws FragmentSetupException
@@ -40,7 +41,7 @@ public interface FragmentManager {
/**
* Get the fragment runner for this incoming fragment. Note, this can only be requested once.
- *
+ *
* @return
*/
public abstract FragmentExecutor getRunnable();
@@ -50,7 +51,7 @@ public interface FragmentManager {
public boolean isWaiting();
public abstract FragmentHandle getHandle();
-
+
public abstract FragmentContext getFragmentContext();
public abstract void addConnection(RemoteConnection connection);
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/NonRootFragmentManager.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/NonRootFragmentManager.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/NonRootFragmentManager.java
index 84fb806..51bf81c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/NonRootFragmentManager.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/NonRootFragmentManager.java
@@ -34,6 +34,7 @@ import org.apache.drill.exec.proto.BitControl.PlanFragment;
import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
import org.apache.drill.exec.record.RawFragmentBatch;
import org.apache.drill.exec.rpc.RemoteConnection;
+import org.apache.drill.exec.rpc.ResponseSender;
import org.apache.drill.exec.server.DrillbitContext;
import org.apache.drill.exec.work.batch.IncomingBuffers;
@@ -48,7 +49,6 @@ public class NonRootFragmentManager implements FragmentManager {
private volatile FragmentExecutor runner;
private volatile boolean cancel = false;
private final FragmentContext context;
- private final PhysicalPlanReader reader;
private List<RemoteConnection> connections = new CopyOnWriteArrayList<>();
public NonRootFragmentManager(PlanFragment fragment, DrillbitContext context) throws FragmentSetupException{
@@ -59,7 +59,6 @@ public class NonRootFragmentManager implements FragmentManager {
this.buffers = new IncomingBuffers(root, this.context);
this.context.setBuffers(buffers);
this.runnerListener = new NonRootStatusReporter(this.context, context.getController().getTunnel(fragment.getForeman()));
- this.reader = context.getPlanReader();
}catch(ExecutionSetupException | IOException e){
throw new FragmentSetupException("Failure while decoding fragment.", e);
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/RootFragmentManager.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/RootFragmentManager.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/RootFragmentManager.java
index c763d55..4b9ee52 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/RootFragmentManager.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/fragment/RootFragmentManager.java
@@ -17,16 +17,17 @@
*/
package org.apache.drill.exec.work.fragment;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
import org.apache.drill.exec.exception.FragmentSetupException;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
import org.apache.drill.exec.record.RawFragmentBatch;
import org.apache.drill.exec.rpc.RemoteConnection;
+import org.apache.drill.exec.rpc.ResponseSender;
import org.apache.drill.exec.work.batch.IncomingBuffers;
-import java.util.List;
-import java.util.concurrent.CopyOnWriteArrayList;
-
public class RootFragmentManager implements FragmentManager{
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(RootFragmentManager.class);
@@ -35,7 +36,7 @@ public class RootFragmentManager implements FragmentManager{
private final FragmentHandle handle;
private volatile boolean cancel = false;
private List<RemoteConnection> connections = new CopyOnWriteArrayList<>();
-
+
public RootFragmentManager(FragmentHandle handle, IncomingBuffers buffers, FragmentExecutor runner) {
super();
this.handle = handle;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/java-exec/src/test/java/org/apache/drill/exec/server/TestBitRpc.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/server/TestBitRpc.java b/exec/java-exec/src/test/java/org/apache/drill/exec/server/TestBitRpc.java
index f579448..347cfb0 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/server/TestBitRpc.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/server/TestBitRpc.java
@@ -25,6 +25,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import mockit.Injectable;
+import mockit.NonStrictExpectations;
import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.common.expression.ExpressionPosition;
@@ -34,8 +35,8 @@ import org.apache.drill.common.types.Types;
import org.apache.drill.exec.ExecTest;
import org.apache.drill.exec.expr.TypeHelper;
import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.proto.BitData.FragmentRecordBatch;
-import org.apache.drill.exec.proto.BitData.RpcType;
import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
@@ -43,14 +44,14 @@ import org.apache.drill.exec.proto.UserBitShared.QueryId;
import org.apache.drill.exec.record.FragmentWritableBatch;
import org.apache.drill.exec.record.MaterializedField;
import org.apache.drill.exec.record.WritableBatch;
-import org.apache.drill.exec.rpc.Acks;
import org.apache.drill.exec.rpc.RemoteConnection;
-import org.apache.drill.exec.rpc.Response;
+import org.apache.drill.exec.rpc.ResponseSender;
import org.apache.drill.exec.rpc.RpcException;
import org.apache.drill.exec.rpc.RpcOutcomeListener;
import org.apache.drill.exec.rpc.control.WorkEventBus;
import org.apache.drill.exec.rpc.data.DataConnectionManager;
import org.apache.drill.exec.rpc.data.DataResponseHandler;
+import org.apache.drill.exec.rpc.data.DataRpcConfig;
import org.apache.drill.exec.rpc.data.DataServer;
import org.apache.drill.exec.rpc.data.DataTunnel;
import org.apache.drill.exec.vector.Float8Vector;
@@ -66,14 +67,26 @@ public class TestBitRpc extends ExecTest {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestBitRpc.class);
@Test
- public void testConnectionBackpressure(@Injectable WorkerBee bee, @Injectable WorkEventBus workBus) throws Exception {
- int port = 1234;
- BootStrapContext c = new BootStrapContext(DrillConfig.create());
+ public void testConnectionBackpressure(@Injectable WorkerBee bee, @Injectable final WorkEventBus workBus, @Injectable final FragmentManager fman, @Injectable final FragmentContext fcon) throws Exception {
+
+ final BootStrapContext c = new BootStrapContext(DrillConfig.create());
BootStrapContext c2 = new BootStrapContext(DrillConfig.create());
+
+ new NonStrictExpectations(){{
+ workBus.getOrCreateFragmentManager((FragmentHandle) any); result = fman;
+ workBus.getFragmentManager( (FragmentHandle) any); result = fman;
+ fman.getFragmentContext(); result = fcon;
+ fcon.getAllocator(); result = c.getAllocator();
+ }};
+
+
+ int port = 1234;
+
DataResponseHandler drp = new BitComTestHandler();
DataServer server = new DataServer(c, workBus, drp);
+
port = server.bind(port, false);
DrillbitEndpoint ep = DrillbitEndpoint.newBuilder().setAddress("localhost").setDataPort(port).build();
DataConnectionManager manager = new DataConnectionManager(FragmentHandle.getDefaultInstance(), ep, c2);
@@ -140,7 +153,7 @@ public class TestBitRpc extends ExecTest {
int v = 0;
@Override
- public Response handle(RemoteConnection connection, FragmentManager manager, FragmentRecordBatch fragmentBatch, ByteBuf data)
+ public void handle(RemoteConnection connection, FragmentManager manager, FragmentRecordBatch fragmentBatch, ByteBuf data, ResponseSender sender)
throws RpcException {
// System.out.println("Received.");
try {
@@ -153,7 +166,11 @@ public class TestBitRpc extends ExecTest {
// TODO Auto-generated catch block
e.printStackTrace();
}
- return new Response(RpcType.ACK, Acks.OK);
+ sender.send(DataRpcConfig.OK);
+ }
+
+ @Override
+ public void informOutOfMemory() {
}
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/exec/pom.xml
----------------------------------------------------------------------
diff --git a/exec/pom.xml b/exec/pom.xml
index 908d3bc..7248597 100644
--- a/exec/pom.xml
+++ b/exec/pom.xml
@@ -31,9 +31,7 @@
<dependencies>
</dependencies>
<modules>
- <module>bufferl</module>
<module>java-exec</module>
<module>jdbc</module>
- <!-- <module>jdbc-all</module> -->
</modules>
</project>
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/af29c255/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index f43962a..7dbdef0 100644
--- a/pom.xml
+++ b/pom.xml
@@ -395,7 +395,7 @@
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-handler</artifactId>
- <version>4.0.7.Final</version>
+ <version>4.0.20.Final</version>
</dependency>
<dependency>