You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@drill.apache.org by ja...@apache.org on 2013/09/04 13:09:58 UTC
[28/52] [abbrv] [partial] DRILL-165: Reorganize directories (moves
only)
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcException.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcException.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcException.java
new file mode 100644
index 0000000..500f959
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcException.java
@@ -0,0 +1,58 @@
+/*******************************************************************************
+ * 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 java.util.concurrent.ExecutionException;
+
+import org.apache.drill.common.exceptions.DrillIOException;
+
+/**
+ * Parent class for all rpc exceptions.
+ */
+public class RpcException extends DrillIOException{
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(RpcException.class);
+
+ public RpcException() {
+ super();
+ }
+
+ public RpcException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public RpcException(String message) {
+ super(message);
+ }
+
+ public RpcException(Throwable cause) {
+ super(cause);
+ }
+
+ public static RpcException mapException(Throwable t){
+ while(t instanceof ExecutionException) t = ((ExecutionException)t).getCause();
+ if(t instanceof RpcException) return ((RpcException) t);
+ return new RpcException(t);
+ }
+
+ public static RpcException mapException(String message, Throwable t){
+ while(t instanceof ExecutionException) t = ((ExecutionException)t).getCause();
+ return new RpcException(message, t);
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcExceptionHandler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcExceptionHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcExceptionHandler.java
new file mode 100644
index 0000000..0123cad
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcExceptionHandler.java
@@ -0,0 +1,47 @@
+/*******************************************************************************
+ * 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.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+
+public class RpcExceptionHandler implements ChannelHandler{
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(RpcExceptionHandler.class);
+
+ public RpcExceptionHandler(){
+ }
+
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+ if(!ctx.channel().isOpen()) return;
+ logger.error("Exception in pipeline. Closing channel between local " + ctx.channel().localAddress() + " and remote " + ctx.channel().remoteAddress(), cause);
+ ctx.close();
+ }
+
+
+ @Override
+ public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
+ }
+
+
+ @Override
+ public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcMessage.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcMessage.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcMessage.java
new file mode 100644
index 0000000..08ea150
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcMessage.java
@@ -0,0 +1,40 @@
+/*******************************************************************************
+ * 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.buffer.ByteBuf;
+
+import org.apache.drill.exec.proto.GeneralRPCProtos.RpcMode;
+
+public abstract class RpcMessage {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(RpcMessage.class);
+
+ public RpcMode mode;
+ public int rpcType;
+ public int coordinationId;
+
+ public RpcMessage(RpcMode mode, int rpcType, int coordinationId) {
+ this.mode = mode;
+ this.rpcType = rpcType;
+ this.coordinationId = coordinationId;
+ }
+
+ public abstract int getBodySize();
+ abstract void release();
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcOutcome.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcOutcome.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcOutcome.java
new file mode 100644
index 0000000..112924d
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcOutcome.java
@@ -0,0 +1,28 @@
+/*******************************************************************************
+ * 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.buffer.ByteBuf;
+
+public interface RpcOutcome<T> {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(RpcOutcome.class);
+
+ public void set(Object value, ByteBuf buffer);
+ public void setException(Throwable t);
+ public Class<T> getOutcomeType();
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcOutcomeListener.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcOutcomeListener.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcOutcomeListener.java
new file mode 100644
index 0000000..7633316
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/RpcOutcomeListener.java
@@ -0,0 +1,29 @@
+/*******************************************************************************
+ * 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.buffer.ByteBuf;
+
+public interface RpcOutcomeListener<V> {
+
+ public void failed(RpcException ex);
+ public void success(V value, ByteBuf buffer);
+
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/ZeroCopyProtobufLengthDecoder.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/ZeroCopyProtobufLengthDecoder.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/ZeroCopyProtobufLengthDecoder.java
new file mode 100644
index 0000000..6186056
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/ZeroCopyProtobufLengthDecoder.java
@@ -0,0 +1,89 @@
+/*******************************************************************************
+ * 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.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.ByteToMessageDecoder;
+import io.netty.handler.codec.CorruptedFrameException;
+
+import java.util.List;
+
+import com.google.protobuf.CodedInputStream;
+
+/**
+ * Modified version of ProtobufVarint32FrameDecoder that avoids bytebuf copy.
+ */
+public class ZeroCopyProtobufLengthDecoder extends ByteToMessageDecoder {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ZeroCopyProtobufLengthDecoder.class);
+
+ @Override
+ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
+
+ if (!ctx.channel().isOpen()) {
+ if (in.readableBytes() > 0)
+ logger.info("Channel is closed, discarding remaining {} byte(s) in buffer.", in.readableBytes());
+ in.skipBytes(in.readableBytes());
+ return;
+ }
+
+ in.markReaderIndex();
+ final byte[] buf = new byte[5];
+ for (int i = 0; i < buf.length; i++) {
+ if (!in.isReadable()) {
+ in.resetReaderIndex();
+ return;
+ }
+
+ buf[i] = in.readByte();
+ if (buf[i] >= 0) {
+
+ int length = CodedInputStream.newInstance(buf, 0, i + 1).readRawVarint32();
+
+ if (length < 0) {
+ throw new CorruptedFrameException("negative length: " + length);
+ }
+ if (length == 0) {
+ throw new CorruptedFrameException("Received a message of length 0.");
+ }
+
+ if (in.readableBytes() < length) {
+ in.resetReaderIndex();
+ return;
+ } else {
+ // need to make buffer copy, otherwise netty will try to refill this buffer if we move the readerIndex forward...
+ // TODO: Can we avoid this copy?
+ ByteBuf outBuf = in.copy(in.readerIndex(), length);
+ in.skipBytes(length);
+
+ if (RpcConstants.EXTRA_DEBUGGING)
+ logger.debug(String.format(
+ "ReaderIndex is %d after length header of %d bytes and frame body of length %d bytes.",
+ in.readerIndex(), i + 1, length));
+ out.add(outBuf);
+ return;
+ }
+ }
+ }
+
+ // Couldn't find the byte whose MSB is off.
+ throw new CorruptedFrameException("length wider than 32-bit");
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/AvailabilityListener.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/AvailabilityListener.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/AvailabilityListener.java
new file mode 100644
index 0000000..ecaf8d3
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/AvailabilityListener.java
@@ -0,0 +1,22 @@
+/*******************************************************************************
+ * 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.bit;
+
+public interface AvailabilityListener {
+ public void isAvailable(BitConnection connection);
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitClient.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitClient.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitClient.java
new file mode 100644
index 0000000..82a6aa6
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitClient.java
@@ -0,0 +1,99 @@
+/*******************************************************************************
+ * 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.bit;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.util.concurrent.GenericFutureListener;
+
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.proto.ExecProtos.BitHandshake;
+import org.apache.drill.exec.proto.ExecProtos.RpcType;
+import org.apache.drill.exec.rpc.BasicClient;
+import org.apache.drill.exec.rpc.Response;
+import org.apache.drill.exec.rpc.RpcBus;
+import org.apache.drill.exec.rpc.RpcConnectionHandler;
+import org.apache.drill.exec.rpc.RpcException;
+import org.apache.drill.exec.rpc.bit.BitConnectionManager.CloseHandlerCreator;
+import org.apache.drill.exec.server.BootStrapContext;
+import org.apache.drill.exec.work.batch.BitComHandler;
+
+import com.google.protobuf.MessageLite;
+
+public class BitClient extends BasicClient<RpcType, BitConnection, BitHandshake, BitHandshake>{
+
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(BitClient.class);
+
+ private final BitComHandler handler;
+ private final DrillbitEndpoint remoteEndpoint;
+ private volatile BitConnection connection;
+ private final ListenerPool listeners;
+ private final CloseHandlerCreator closeHandlerFactory;
+ private final DrillbitEndpoint localIdentity;
+
+ public BitClient(DrillbitEndpoint remoteEndpoint, DrillbitEndpoint localEndpoint, BitComHandler handler, BootStrapContext context, CloseHandlerCreator closeHandlerFactory, ListenerPool listeners) {
+ super(BitRpcConfig.MAPPING, context.getAllocator().getUnderlyingAllocator(), context.getBitLoopGroup(), RpcType.HANDSHAKE, BitHandshake.class, BitHandshake.PARSER);
+ this.localIdentity = localEndpoint;
+ this.remoteEndpoint = remoteEndpoint;
+ this.handler = handler;
+ this.listeners = listeners;
+ this.closeHandlerFactory = closeHandlerFactory;
+ }
+
+ public void connect(RpcConnectionHandler<BitConnection> connectionHandler) {
+ connectAsClient(connectionHandler, BitHandshake.newBuilder().setRpcVersion(BitRpcConfig.RPC_VERSION).setEndpoint(localIdentity).build(), remoteEndpoint.getAddress(), remoteEndpoint.getBitPort());
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public BitConnection initRemoteConnection(Channel channel) {
+ this.connection = new BitConnection(channel, (RpcBus<RpcType, BitConnection>) (RpcBus<?, ?>) this, listeners);
+ return connection;
+ }
+
+ @Override
+ protected GenericFutureListener<ChannelFuture> getCloseHandler(BitConnection clientConnection) {
+ return closeHandlerFactory.getHandler(clientConnection, super.getCloseHandler(clientConnection));
+ }
+
+ @Override
+ public MessageLite getResponseDefaultInstance(int rpcType) throws RpcException {
+ return BitComDefaultInstanceHandler.getResponseDefaultInstance(rpcType);
+ }
+
+ @Override
+ protected Response handle(BitConnection connection, int rpcType, ByteBuf pBody, ByteBuf dBody) throws RpcException {
+ return handler.handle(connection, rpcType, pBody, dBody);
+ }
+
+ @Override
+ protected void validateHandshake(BitHandshake handshake) throws RpcException {
+ if(handshake.getRpcVersion() != BitRpcConfig.RPC_VERSION) throw new RpcException(String.format("Invalid rpc version. Expected %d, actual %d.", handshake.getRpcVersion(), BitRpcConfig.RPC_VERSION));
+ }
+
+ @Override
+ protected void finalizeConnection(BitHandshake handshake, BitConnection connection) {
+ connection.setEndpoint(handshake.getEndpoint());
+ }
+
+ public BitConnection getConnection(){
+ return this.connection;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitCom.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitCom.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitCom.java
new file mode 100644
index 0000000..f7f508e
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitCom.java
@@ -0,0 +1,56 @@
+/*******************************************************************************
+ * 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.bit;
+
+import java.io.Closeable;
+
+import org.apache.drill.exec.exception.DrillbitStartupException;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.work.fragment.IncomingFragmentHandler;
+
+/**
+ * Service that allows one Drillbit to communicate with another. Internally manages whether each particular bit is a
+ * server or a client depending on who initially made the connection. If no connection exists, BitCom is responsible for
+ * making a connection. BitCom should automatically straight route local BitCommunication rather than connecting to its
+ * self.
+ */
+public interface BitCom extends Closeable {
+
+ /**
+ * Get a Bit to Bit communication tunnel. If the BitCom doesn't have a tunnel attached to the node already, it will
+ * start creating one. This create the connection asynchronously.
+ *
+ * @param node
+ * @return
+ */
+ public BitTunnel getTunnel(DrillbitEndpoint node) ;
+
+ public DrillbitEndpoint start(DrillbitEndpoint partialEndpoint) throws InterruptedException, DrillbitStartupException;
+
+ /**
+ * Register an incoming batch handler for a local foreman.
+ * @param handler
+ */
+ public void registerIncomingBatchHandler(IncomingFragmentHandler handler);
+
+ /**
+ * Get ListenerPool
+ * @return
+ */
+ public ListenerPool getListeners();
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitComDefaultInstanceHandler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitComDefaultInstanceHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitComDefaultInstanceHandler.java
new file mode 100644
index 0000000..e1d4902
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitComDefaultInstanceHandler.java
@@ -0,0 +1,51 @@
+/*******************************************************************************
+ * 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.bit;
+
+import org.apache.drill.exec.proto.ExecProtos.BitHandshake;
+import org.apache.drill.exec.proto.ExecProtos.BitStatus;
+import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
+import org.apache.drill.exec.proto.ExecProtos.FragmentStatus;
+import org.apache.drill.exec.proto.ExecProtos.RpcType;
+import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
+import org.apache.drill.exec.rpc.RpcException;
+
+import com.google.protobuf.MessageLite;
+
+public class BitComDefaultInstanceHandler {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(BitComDefaultInstanceHandler.class);
+
+
+ public static MessageLite getResponseDefaultInstance(int rpcType) throws RpcException {
+ switch (rpcType) {
+ case RpcType.ACK_VALUE:
+ return Ack.getDefaultInstance();
+ case RpcType.HANDSHAKE_VALUE:
+ return BitHandshake.getDefaultInstance();
+ case RpcType.RESP_FRAGMENT_HANDLE_VALUE:
+ return FragmentHandle.getDefaultInstance();
+ case RpcType.RESP_FRAGMENT_STATUS_VALUE:
+ return FragmentStatus.getDefaultInstance();
+ case RpcType.RESP_BIT_STATUS_VALUE:
+ return BitStatus.getDefaultInstance();
+
+ default:
+ throw new UnsupportedOperationException();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitComImpl.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitComImpl.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitComImpl.java
new file mode 100644
index 0000000..d1cadc7
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitComImpl.java
@@ -0,0 +1,85 @@
+/*******************************************************************************
+ * 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.bit;
+
+import java.util.concurrent.ConcurrentMap;
+
+import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.exec.exception.DrillbitStartupException;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.server.BootStrapContext;
+import org.apache.drill.exec.work.batch.BitComHandler;
+import org.apache.drill.exec.work.fragment.IncomingFragmentHandler;
+
+import com.google.common.collect.Maps;
+import com.google.common.io.Closeables;
+
+/**
+ * Manages communication tunnels between nodes.
+ */
+public class BitComImpl implements BitCom {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(BitComImpl.class);
+
+ private final ListenerPool listeners;
+ private volatile BitServer server;
+ private final BitComHandler handler;
+ private final BootStrapContext context;
+ private final ConnectionManagerRegistry connectionRegistry;
+
+ public BitComImpl(BootStrapContext context, BitComHandler handler) {
+ super();
+ this.handler = handler;
+ this.context = context;
+ this.listeners = new ListenerPool(8);
+ this.connectionRegistry = new ConnectionManagerRegistry(handler, context, listeners);
+ }
+
+ @Override
+ public DrillbitEndpoint start(DrillbitEndpoint partialEndpoint) throws InterruptedException, DrillbitStartupException {
+ server = new BitServer(handler, context, connectionRegistry, listeners);
+ int port = context.getConfig().getInt(ExecConstants.INITIAL_BIT_PORT);
+ port = server.bind(port);
+ DrillbitEndpoint completeEndpoint = partialEndpoint.toBuilder().setBitPort(port).build();
+ connectionRegistry.setEndpoint(completeEndpoint);
+ return completeEndpoint;
+ }
+
+
+
+ public ListenerPool getListeners() {
+ return listeners;
+ }
+
+ @Override
+ public BitTunnel getTunnel(DrillbitEndpoint endpoint) {
+ return new BitTunnel(endpoint, connectionRegistry.getConnectionManager(endpoint));
+ }
+
+ @Override
+ public void registerIncomingBatchHandler(IncomingFragmentHandler handler) {
+ this.handler.registerIncomingFragmentHandler(handler);
+ }
+
+ public void close() {
+ Closeables.closeQuietly(server);
+ for (BitConnectionManager bt : connectionRegistry) {
+ bt.close();
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitCommand.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitCommand.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitCommand.java
new file mode 100644
index 0000000..692c63e
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitCommand.java
@@ -0,0 +1,28 @@
+/*******************************************************************************
+ * 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.bit;
+
+import org.apache.drill.exec.rpc.RpcConnectionHandler;
+
+import com.google.protobuf.MessageLite;
+
+public interface BitCommand<T extends MessageLite> extends RpcConnectionHandler<BitConnection>{
+
+ public abstract void connectionAvailable(BitConnection connection);
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitConnection.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitConnection.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitConnection.java
new file mode 100644
index 0000000..f85ea74
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitConnection.java
@@ -0,0 +1,117 @@
+/*******************************************************************************
+ * 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.bit;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.util.concurrent.GenericFutureListener;
+
+import java.util.UUID;
+import java.util.concurrent.ConcurrentMap;
+
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
+import org.apache.drill.exec.proto.ExecProtos.FragmentStatus;
+import org.apache.drill.exec.proto.ExecProtos.PlanFragment;
+import org.apache.drill.exec.proto.ExecProtos.RpcType;
+import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
+import org.apache.drill.exec.record.FragmentWritableBatch;
+import org.apache.drill.exec.rpc.DrillRpcFuture;
+import org.apache.drill.exec.rpc.RemoteConnection;
+import org.apache.drill.exec.rpc.RpcBus;
+import org.apache.drill.exec.rpc.RpcConnectionHandler;
+import org.apache.drill.exec.rpc.RpcOutcomeListener;
+
+import com.google.common.base.Preconditions;
+import com.google.common.io.Closeables;
+import com.google.protobuf.MessageLite;
+
+public class BitConnection extends RemoteConnection{
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(BitConnection.class);
+
+ private final RpcBus<RpcType, BitConnection> bus;
+ private final ListenerPool listeners;
+ private volatile DrillbitEndpoint endpoint;
+ private volatile boolean active = false;
+ private final UUID id;
+
+ public BitConnection(Channel channel, RpcBus<RpcType, BitConnection> bus, ListenerPool listeners){
+ super(channel);
+ this.bus = bus;
+ // we use a local listener pool unless a global one is provided.
+ this.listeners = listeners != null ? listeners : new ListenerPool(2);
+ this.id = UUID.randomUUID();
+ }
+
+ void setEndpoint(DrillbitEndpoint endpoint){
+ assert this.endpoint == null : "Endpoint should only be set once (only in the case in incoming server requests).";
+ this.endpoint = endpoint;
+ active = true;
+ }
+
+ protected DrillbitEndpoint getEndpoint() {
+ return endpoint;
+ }
+
+ public ListenerPool getListenerPool(){
+ return listeners;
+ }
+
+
+ public <SEND extends MessageLite, RECEIVE extends MessageLite> void send(RpcOutcomeListener<RECEIVE> outcomeListener, RpcType rpcType,
+ SEND protobufBody, Class<RECEIVE> clazz, ByteBuf... dataBodies){
+ bus.send(outcomeListener, this, rpcType, protobufBody, clazz, dataBodies);
+ }
+
+
+ public void disable(){
+ active = false;
+ }
+
+ public boolean isActive(){
+ return active;
+ }
+
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result = prime * result + ((id == null) ? 0 : id.hashCode());
+ return result;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) return true;
+ if (obj == null) return false;
+ if (getClass() != obj.getClass()) return false;
+ BitConnection other = (BitConnection) obj;
+ if (id == null) {
+ if (other.id != null) return false;
+ } else if (!id.equals(other.id)) return false;
+ return true;
+ }
+
+
+
+ public void shutdownIfClient(){
+ if(bus.isClient()) Closeables.closeQuietly(bus);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitConnectionManager.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitConnectionManager.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitConnectionManager.java
new file mode 100644
index 0000000..d99bb22
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitConnectionManager.java
@@ -0,0 +1,189 @@
+/*******************************************************************************
+ * 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.bit;
+
+import io.netty.channel.ChannelFuture;
+import io.netty.util.concurrent.GenericFutureListener;
+
+import java.io.Closeable;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.rpc.RpcConnectionHandler;
+import org.apache.drill.exec.rpc.RpcException;
+import org.apache.drill.exec.server.BootStrapContext;
+import org.apache.drill.exec.work.batch.BitComHandler;
+
+import com.google.protobuf.MessageLite;
+
+/**
+ * Manager all connections between two particular bits.
+ */
+public class BitConnectionManager implements Closeable{
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(BitConnectionManager.class);
+
+ private final DrillbitEndpoint endpoint;
+ private final AtomicReference<BitConnection> connectionHolder;
+ private final BitComHandler handler;
+ private final BootStrapContext context;
+ private final ListenerPool listenerPool;
+ private final DrillbitEndpoint localIdentity;
+
+ public BitConnectionManager(DrillbitEndpoint remoteEndpoint, DrillbitEndpoint localIdentity, BitComHandler handler, BootStrapContext context, ListenerPool listenerPool) {
+ assert remoteEndpoint != null : "Endpoint cannot be null.";
+ assert remoteEndpoint.getAddress() != null && !remoteEndpoint.getAddress().isEmpty(): "Endpoint address cannot be null.";
+ assert remoteEndpoint.getBitPort() > 0 : String.format("Bit Port must be set to a port between 1 and 65k. Was set to %d.", remoteEndpoint.getBitPort());
+
+ this.connectionHolder = new AtomicReference<BitConnection>();
+ this.endpoint = remoteEndpoint;
+ this.localIdentity = localIdentity;
+ this.handler = handler;
+ this.context = context;
+ this.listenerPool = listenerPool;
+ }
+
+ public <R extends MessageLite> BitCommand<R> runCommand(BitCommand<R> cmd){
+ logger.debug("Running command {}", cmd);
+ BitConnection connection = connectionHolder.get();
+ if(connection != null){
+ if(connection.isActive()){
+ cmd.connectionAvailable(connection);
+ return cmd;
+ }else{
+ // remove the old connection. (don't worry if we fail since someone else should have done it.
+ connectionHolder.compareAndSet(connection, null);
+ }
+ }
+
+ /** We've arrived here without a connection, let's make sure only one of us makes a connection. (fyi, another endpoint could create a reverse connection **/
+ synchronized(this){
+ connection = connectionHolder.get();
+ if(connection != null){
+ cmd.connectionAvailable(connection);
+ }else{
+ BitClient client = new BitClient(endpoint, localIdentity, handler, context, new CloseHandlerCreator(), listenerPool);
+
+ client.connect(new ConnectionListeningDecorator(cmd, !endpoint.equals(localIdentity)));
+ }
+ return cmd;
+
+ }
+ }
+
+ CloseHandlerCreator getCloseHandlerCreator(){
+ return new CloseHandlerCreator();
+ }
+
+ /** Factory for close handlers **/
+ class CloseHandlerCreator{
+ public GenericFutureListener<ChannelFuture> getHandler(BitConnection connection, GenericFutureListener<ChannelFuture> parent){
+ return new CloseHandler(connection, parent);
+ }
+ }
+
+
+
+ /**
+ * Listens for connection closes and clears connection holder.
+ */
+ private class CloseHandler implements GenericFutureListener<ChannelFuture>{
+ private BitConnection connection;
+ private GenericFutureListener<ChannelFuture> parent;
+
+ public CloseHandler(BitConnection connection, GenericFutureListener<ChannelFuture> parent) {
+ super();
+ this.connection = connection;
+ this.parent = parent;
+ }
+
+ @Override
+ public void operationComplete(ChannelFuture future) throws Exception {
+ connectionHolder.compareAndSet(connection, null);
+ parent.operationComplete(future);
+ }
+
+ }
+
+ /**
+ * Decorate a connection creation so that we capture a success and keep it available for future requests. If we have raced and another is already available... we return that one and close things down on this one.
+ */
+ private class ConnectionListeningDecorator implements RpcConnectionHandler<BitConnection>{
+
+ private final RpcConnectionHandler<BitConnection> delegate;
+ private final boolean closeOnDupe;
+
+ public ConnectionListeningDecorator(RpcConnectionHandler<BitConnection> delegate, boolean closeOnDupe) {
+ this.delegate = delegate;
+ this.closeOnDupe = closeOnDupe;
+ }
+
+ @Override
+ public void connectionSucceeded(BitConnection incoming) {
+ BitConnection connection = connectionHolder.get();
+ while(true){
+ boolean setted = connectionHolder.compareAndSet(null, incoming);
+ if(setted){
+ connection = incoming;
+ break;
+ }
+ connection = connectionHolder.get();
+ if(connection != null) break;
+ }
+
+
+ if(connection == incoming){
+ delegate.connectionSucceeded(connection);
+ }else{
+
+ if(closeOnDupe){
+ // close the incoming because another channel was created in the mean time (unless this is a self connection).
+ logger.debug("Closing incoming connection because a connection was already set.");
+ incoming.getChannel().close();
+ }
+ delegate.connectionSucceeded(connection);
+ }
+ }
+
+ @Override
+ public void connectionFailed(org.apache.drill.exec.rpc.RpcConnectionHandler.FailureType type, Throwable t) {
+ delegate.connectionFailed(type, t);
+ }
+
+ }
+
+ public DrillbitEndpoint getEndpoint() {
+ return endpoint;
+ }
+
+ public void addServerConnection(BitConnection connection){
+ // if the connection holder is not set, set it to this incoming connection.
+ logger.debug("Setting server connection.");
+ this.connectionHolder.compareAndSet(null, connection);
+ }
+
+ @Override
+ public void close() {
+ BitConnection c = connectionHolder.getAndSet(null);
+ if(c != null){
+ c.getChannel().close();
+ }
+ }
+
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitRpcConfig.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitRpcConfig.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitRpcConfig.java
new file mode 100644
index 0000000..32fd4f9
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitRpcConfig.java
@@ -0,0 +1,46 @@
+/*******************************************************************************
+ * 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.bit;
+
+
+import org.apache.drill.exec.proto.ExecProtos.BitHandshake;
+import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
+import org.apache.drill.exec.proto.ExecProtos.FragmentRecordBatch;
+import org.apache.drill.exec.proto.ExecProtos.FragmentStatus;
+import org.apache.drill.exec.proto.ExecProtos.PlanFragment;
+import org.apache.drill.exec.proto.ExecProtos.RpcType;
+import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
+import org.apache.drill.exec.rpc.Acks;
+import org.apache.drill.exec.rpc.Response;
+import org.apache.drill.exec.rpc.RpcConfig;
+
+public class BitRpcConfig {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(BitRpcConfig.class);
+
+ public static RpcConfig MAPPING = RpcConfig.newBuilder("BIT-RPC-MAPPING") //
+ .add(RpcType.HANDSHAKE, BitHandshake.class, RpcType.HANDSHAKE, BitHandshake.class)
+ .add(RpcType.REQ_INIATILIZE_FRAGMENT, PlanFragment.class, RpcType.ACK, Ack.class)
+ .add(RpcType.REQ_CANCEL_FRAGMENT, FragmentHandle.class, RpcType.ACK, Ack.class)
+ .add(RpcType.REQ_FRAGMENT_STATUS, FragmentStatus.class, RpcType.ACK, Ack.class)
+ .add(RpcType.REQ_RECORD_BATCH, FragmentRecordBatch.class, RpcType.ACK, Ack.class)
+ .build();
+
+ public static int RPC_VERSION = 2;
+
+ public static final Response OK = new Response(RpcType.ACK, Acks.OK);
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitServer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitServer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitServer.java
new file mode 100644
index 0000000..d4665a8
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitServer.java
@@ -0,0 +1,127 @@
+/*******************************************************************************
+ * 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.bit;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.util.concurrent.GenericFutureListener;
+
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.proto.ExecProtos.BitHandshake;
+import org.apache.drill.exec.proto.ExecProtos.RpcType;
+import org.apache.drill.exec.rpc.BasicServer;
+import org.apache.drill.exec.rpc.Response;
+import org.apache.drill.exec.rpc.RpcException;
+import org.apache.drill.exec.rpc.bit.BitConnectionManager.CloseHandlerCreator;
+import org.apache.drill.exec.server.BootStrapContext;
+import org.apache.drill.exec.work.batch.BitComHandler;
+
+import com.google.protobuf.MessageLite;
+
+public class BitServer extends BasicServer<RpcType, BitConnection>{
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(BitServer.class);
+
+ private final BitComHandler handler;
+ private final ListenerPool listeners;
+ private final ConnectionManagerRegistry connectionRegistry;
+ private volatile ProxyCloseHandler proxyCloseHandler;
+
+ public BitServer(BitComHandler handler, BootStrapContext context, ConnectionManagerRegistry connectionRegistry, ListenerPool listeners) {
+ super(BitRpcConfig.MAPPING, context.getAllocator().getUnderlyingAllocator(), context.getBitLoopGroup());
+ this.handler = handler;
+ this.connectionRegistry = connectionRegistry;
+ this.listeners = listeners;
+ }
+
+ @Override
+ public MessageLite getResponseDefaultInstance(int rpcType) throws RpcException {
+ return BitComDefaultInstanceHandler.getResponseDefaultInstance(rpcType);
+ }
+
+ @Override
+ protected Response handle(BitConnection connection, int rpcType, ByteBuf pBody, ByteBuf dBody) throws RpcException {
+ return handler.handle(connection, rpcType, pBody, dBody);
+ }
+
+ @Override
+ protected GenericFutureListener<ChannelFuture> getCloseHandler(BitConnection connection) {
+ this.proxyCloseHandler = new ProxyCloseHandler(super.getCloseHandler(connection));
+ return proxyCloseHandler;
+ }
+
+ @Override
+ public BitConnection initRemoteConnection(Channel channel) {
+ return new BitConnection(channel, this, listeners);
+ }
+
+
+ @Override
+ protected ServerHandshakeHandler<BitHandshake> getHandshakeHandler(final BitConnection connection) {
+ return new ServerHandshakeHandler<BitHandshake>(RpcType.HANDSHAKE, BitHandshake.PARSER){
+
+ @Override
+ public MessageLite getHandshakeResponse(BitHandshake inbound) throws Exception {
+// logger.debug("Handling handshake from other bit. {}", inbound);
+ if(inbound.getRpcVersion() != BitRpcConfig.RPC_VERSION) throw new RpcException(String.format("Invalid rpc version. Expected %d, actual %d.", inbound.getRpcVersion(), BitRpcConfig.RPC_VERSION));
+ if(!inbound.hasEndpoint() || inbound.getEndpoint().getAddress().isEmpty() || inbound.getEndpoint().getBitPort() < 1) throw new RpcException(String.format("RPC didn't provide valid counter endpoint information. Received %s.", inbound.getEndpoint()));
+ connection.setEndpoint(inbound.getEndpoint());
+
+ // add the
+ BitConnectionManager manager = connectionRegistry.getConnectionManager(inbound.getEndpoint());
+
+ // update the close handler.
+ proxyCloseHandler.setHandler(manager.getCloseHandlerCreator().getHandler(connection, proxyCloseHandler.getHandler()));
+
+ // add to the connection manager.
+ manager.addServerConnection(connection);
+
+ return BitHandshake.newBuilder().setRpcVersion(BitRpcConfig.RPC_VERSION).build();
+ }
+
+ };
+ }
+
+
+ private class ProxyCloseHandler implements GenericFutureListener<ChannelFuture> {
+
+ private volatile GenericFutureListener<ChannelFuture> handler;
+
+ public ProxyCloseHandler(GenericFutureListener<ChannelFuture> handler) {
+ super();
+ 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);
+ }
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitTunnel.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitTunnel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitTunnel.java
new file mode 100644
index 0000000..83b7959
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/BitTunnel.java
@@ -0,0 +1,137 @@
+/*******************************************************************************
+ * 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.bit;
+
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
+import org.apache.drill.exec.proto.ExecProtos.FragmentStatus;
+import org.apache.drill.exec.proto.ExecProtos.PlanFragment;
+import org.apache.drill.exec.proto.ExecProtos.RpcType;
+import org.apache.drill.exec.proto.GeneralRPCProtos.Ack;
+import org.apache.drill.exec.record.FragmentWritableBatch;
+import org.apache.drill.exec.rpc.DrillRpcFuture;
+import org.apache.drill.exec.rpc.RpcOutcomeListener;
+
+
+public class BitTunnel {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(BitTunnel.class);
+
+ private final BitConnectionManager manager;
+ private final DrillbitEndpoint endpoint;
+
+ public BitTunnel(DrillbitEndpoint endpoint, BitConnectionManager manager) {
+ this.manager = manager;
+ this.endpoint = endpoint;
+ }
+
+ public DrillbitEndpoint getEndpoint(){
+ return manager.getEndpoint();
+ }
+
+ public void sendRecordBatch(RpcOutcomeListener<Ack> outcomeListener, FragmentContext context, FragmentWritableBatch batch) {
+ SendBatch b = new SendBatch(outcomeListener, batch, context);
+ manager.runCommand(b);
+ }
+
+ public void sendFragment(RpcOutcomeListener<Ack> outcomeListener, PlanFragment fragment){
+ SendFragment b = new SendFragment(outcomeListener, fragment);
+ manager.runCommand(b);
+ }
+
+ public DrillRpcFuture<Ack> cancelFragment(FragmentHandle handle){
+ CancelFragment b = new CancelFragment(handle);
+ manager.runCommand(b);
+ return b.getFuture();
+ }
+
+ public DrillRpcFuture<Ack> sendFragmentStatus(FragmentStatus status){
+ SendFragmentStatus b = new SendFragmentStatus(status);
+ manager.runCommand(b);
+ return b.getFuture();
+ }
+
+ public static class SendBatch extends ListeningBitCommand<Ack> {
+ final FragmentWritableBatch batch;
+ final FragmentContext context;
+
+ public SendBatch(RpcOutcomeListener<Ack> listener, FragmentWritableBatch batch, FragmentContext context) {
+ super(listener);
+ this.batch = batch;
+ this.context = context;
+ }
+
+ @Override
+ public void doRpcCall(RpcOutcomeListener<Ack> outcomeListener, BitConnection connection) {
+ connection.send(outcomeListener, RpcType.REQ_RECORD_BATCH, batch.getHeader(), Ack.class, batch.getBuffers());
+ }
+
+ @Override
+ public String toString() {
+ return "SendBatch [batch.header=" + batch.getHeader() + "]";
+ }
+
+
+ }
+
+ public static class SendFragmentStatus extends FutureBitCommand<Ack> {
+ final FragmentStatus status;
+
+ public SendFragmentStatus(FragmentStatus status) {
+ super();
+ this.status = status;
+ }
+
+ @Override
+ public void doRpcCall(RpcOutcomeListener<Ack> outcomeListener, BitConnection connection) {
+ connection.send(outcomeListener, RpcType.REQ_FRAGMENT_STATUS, status, Ack.class);
+ }
+
+ }
+
+ public static class CancelFragment extends FutureBitCommand<Ack> {
+ final FragmentHandle handle;
+
+ public CancelFragment(FragmentHandle handle) {
+ super();
+ this.handle = handle;
+ }
+
+ @Override
+ public void doRpcCall(RpcOutcomeListener<Ack> outcomeListener, BitConnection connection) {
+ connection.send(outcomeListener, RpcType.REQ_CANCEL_FRAGMENT, handle, Ack.class);
+ }
+
+ }
+
+ public static class SendFragment extends ListeningBitCommand<Ack> {
+ final PlanFragment fragment;
+
+ public SendFragment(RpcOutcomeListener<Ack> listener, PlanFragment fragment) {
+ super(listener);
+ this.fragment = fragment;
+ }
+
+ @Override
+ public void doRpcCall(RpcOutcomeListener<Ack> outcomeListener, BitConnection connection) {
+ connection.send(outcomeListener, RpcType.REQ_INIATILIZE_FRAGMENT, fragment, Ack.class);
+ }
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/ConnectionManagerRegistry.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/ConnectionManagerRegistry.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/ConnectionManagerRegistry.java
new file mode 100644
index 0000000..8afbc33
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/ConnectionManagerRegistry.java
@@ -0,0 +1,73 @@
+/*******************************************************************************
+ * 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.bit;
+
+import io.netty.channel.Channel;
+
+import java.util.Iterator;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.rpc.RpcException;
+import org.apache.drill.exec.server.BootStrapContext;
+import org.apache.drill.exec.work.batch.BitComHandler;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Maps;
+import com.google.common.util.concurrent.CheckedFuture;
+
+public class ConnectionManagerRegistry implements Iterable<BitConnectionManager>{
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ConnectionManagerRegistry.class);
+
+ private final ConcurrentMap<DrillbitEndpoint, BitConnectionManager> registry = Maps.newConcurrentMap();
+
+ private final BitComHandler handler;
+ private final BootStrapContext context;
+ private final ListenerPool listenerPool;
+ private volatile DrillbitEndpoint localEndpoint;
+
+ public ConnectionManagerRegistry(BitComHandler handler, BootStrapContext context, ListenerPool listenerPool) {
+ super();
+ this.handler = handler;
+ this.context = context;
+ this.listenerPool = listenerPool;
+ }
+
+ public BitConnectionManager getConnectionManager(DrillbitEndpoint endpoint){
+ assert localEndpoint != null : "DrillbitEndpoint must be set before a connection manager can be retrieved";
+ BitConnectionManager m = registry.get(endpoint);
+ if(m == null){
+ m = new BitConnectionManager(endpoint, localEndpoint, handler, context, listenerPool);
+ BitConnectionManager m2 = registry.putIfAbsent(endpoint, m);
+ if(m2 != null) m = m2;
+ }
+
+ return m;
+ }
+
+ @Override
+ public Iterator<BitConnectionManager> iterator() {
+ return registry.values().iterator();
+ }
+
+ public void setEndpoint(DrillbitEndpoint endpoint){
+ this.localEndpoint = endpoint;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/FutureBitCommand.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/FutureBitCommand.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/FutureBitCommand.java
new file mode 100644
index 0000000..7914fc2
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/FutureBitCommand.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.bit;
+
+import io.netty.buffer.ByteBuf;
+
+import org.apache.drill.exec.rpc.DrillRpcFuture;
+import org.apache.drill.exec.rpc.RpcCheckedFuture;
+import org.apache.drill.exec.rpc.RpcException;
+import org.apache.drill.exec.rpc.RpcOutcomeListener;
+
+import com.google.common.util.concurrent.SettableFuture;
+import com.google.protobuf.MessageLite;
+
+public abstract class FutureBitCommand<T extends MessageLite> implements BitCommand<T> {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(FutureBitCommand.class);
+
+ protected final SettableFuture<T> settableFuture;
+ private final RpcCheckedFuture<T> parentFuture;
+
+ public FutureBitCommand() {
+ this.settableFuture = SettableFuture.create();
+ this.parentFuture = new RpcCheckedFuture<T>(settableFuture);
+ }
+
+ public abstract void doRpcCall(RpcOutcomeListener<T> outcomeListener, BitConnection connection);
+
+ @Override
+ public void connectionAvailable(BitConnection connection) {
+
+ doRpcCall(new DeferredRpcOutcome(), connection);
+ }
+
+ @Override
+ public void connectionSucceeded(BitConnection connection) {
+ connectionAvailable(connection);
+ }
+
+ private class DeferredRpcOutcome implements RpcOutcomeListener<T> {
+
+ @Override
+ public void failed(RpcException ex) {
+ settableFuture.setException(ex);
+ }
+
+ @Override
+ public void success(T value, ByteBuf buf) {
+ settableFuture.set(value);
+ }
+
+ }
+
+ public DrillRpcFuture<T> getFuture() {
+ return parentFuture;
+ }
+
+ @Override
+ public void connectionFailed(FailureType type, Throwable t) {
+ settableFuture.setException(RpcException.mapException(
+ String.format("Command failed while establishing connection. Failure type %s.", type), t));
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/ListenerPool.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/ListenerPool.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/ListenerPool.java
new file mode 100644
index 0000000..84dba85
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/ListenerPool.java
@@ -0,0 +1,59 @@
+/*******************************************************************************
+ * 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.bit;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
+import org.apache.drill.exec.proto.ExecProtos.FragmentStatus;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.rpc.RpcException;
+import org.apache.drill.exec.work.foreman.FragmentStatusListener;
+
+public class ListenerPool {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ListenerPool.class);
+
+ private final ConcurrentMap<QueryId, FragmentStatusListener> listeners;
+
+ public ListenerPool(int par){
+ listeners = new ConcurrentHashMap<QueryId, FragmentStatusListener>(16, 0.75f, par);
+ }
+
+ public void removeFragmentStatusListener(FragmentHandle handle) throws RpcException{
+ logger.debug("Removing framgent status listener for handle {}.", handle);
+ listeners.remove(handle);
+ }
+
+ public void addFragmentStatusListener(FragmentHandle handle, FragmentStatusListener listener) throws RpcException{
+ logger.debug("Adding framgent status listener for handle {}.", handle);
+ FragmentStatusListener old = listeners.putIfAbsent(handle.getQueryId(), listener);
+ if(old != null) throw new RpcException("Failure. The provided handle already exists in the listener pool. You need to remove one listener before adding another.");
+ }
+
+ public void status(FragmentStatus status){
+ FragmentStatusListener l = listeners.get(status.getHandle().getQueryId());
+ if(l == null){
+
+ logger.error("A fragment message arrived but there was no registered listener for that message for handle {}.", status.getHandle());
+ return;
+ }else{
+ l.statusUpdate(status);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/ListeningBitCommand.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/ListeningBitCommand.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/ListeningBitCommand.java
new file mode 100644
index 0000000..b322105
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/ListeningBitCommand.java
@@ -0,0 +1,70 @@
+/*******************************************************************************
+ * 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.bit;
+
+import io.netty.buffer.ByteBuf;
+
+import org.apache.drill.exec.rpc.RpcException;
+import org.apache.drill.exec.rpc.RpcOutcomeListener;
+
+import com.google.protobuf.MessageLite;
+
+public abstract class ListeningBitCommand<T extends MessageLite> implements BitCommand<T> {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ListeningBitCommand.class);
+
+ private final RpcOutcomeListener<T> listener;
+
+ public ListeningBitCommand(RpcOutcomeListener<T> listener) {
+ this.listener = listener;
+ }
+
+ public abstract void doRpcCall(RpcOutcomeListener<T> outcomeListener, BitConnection connection);
+
+ @Override
+ public void connectionAvailable(BitConnection connection) {
+
+ doRpcCall(new DeferredRpcOutcome(), connection);
+ }
+
+ @Override
+ public void connectionSucceeded(BitConnection connection) {
+ connectionAvailable(connection);
+ }
+
+ private class DeferredRpcOutcome implements RpcOutcomeListener<T> {
+
+ @Override
+ public void failed(RpcException ex) {
+ listener.failed(ex);
+ }
+
+ @Override
+ public void success(T value, ByteBuf buf) {
+ listener.success(value, buf);
+ }
+
+ }
+
+
+ @Override
+ public void connectionFailed(FailureType type, Throwable t) {
+ listener.failed(RpcException.mapException(
+ String.format("Command failed while establishing connection. Failure type %s.", type), t));
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/SendProgress.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/SendProgress.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/SendProgress.java
new file mode 100644
index 0000000..8ede9e6
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/bit/SendProgress.java
@@ -0,0 +1,22 @@
+/*******************************************************************************
+ * 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.bit;
+
+public class SendProgress {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(SendProgress.class);
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultBatch.java
new file mode 100644
index 0000000..c3aad22
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultBatch.java
@@ -0,0 +1,61 @@
+/*******************************************************************************
+ * 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.user;
+
+import io.netty.buffer.ByteBuf;
+
+import org.apache.drill.exec.proto.UserProtos.QueryResult;
+
+public class QueryResultBatch {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(QueryResultBatch.class);
+
+ private final QueryResult header;
+ private final ByteBuf data;
+
+ public QueryResultBatch(QueryResult header, ByteBuf data) {
+// logger.debug("New Result Batch with header {} and data {}", header, data);
+ this.header = header;
+ this.data = data;
+ if(this.data != null) data.retain();
+ }
+
+ public QueryResult getHeader() {
+ return header;
+ }
+
+ public ByteBuf getData() {
+ return data;
+ }
+
+
+ public boolean hasData(){
+ return data != null;
+ }
+
+ public void release(){
+ if(data != null) data.release();
+ }
+
+ @Override
+ public String toString() {
+ return "QueryResultBatch [header=" + header + ", data=" + data + "]";
+ }
+
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/a593d908/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultHandler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultHandler.java
new file mode 100644
index 0000000..50f8c5a
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/QueryResultHandler.java
@@ -0,0 +1,193 @@
+/*******************************************************************************
+ * 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.user;
+
+import io.netty.buffer.ByteBuf;
+
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ConcurrentMap;
+
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.proto.UserProtos.QueryResult;
+import org.apache.drill.exec.proto.UserProtos.QueryResult.QueryState;
+import org.apache.drill.exec.rpc.BaseRpcOutcomeListener;
+import org.apache.drill.exec.rpc.RpcBus;
+import org.apache.drill.exec.rpc.RpcException;
+import org.apache.drill.exec.rpc.RpcOutcomeListener;
+
+import com.google.common.collect.Maps;
+import com.google.common.collect.Queues;
+
+/**
+ * Encapsulates the future management of query submissions. This entails a potential race condition. Normal ordering is:
+ * 1. Submit query to be executed. 2. Receive QueryHandle for buffer management 3. Start receiving results batches for
+ * query.
+ *
+ * However, 3 could potentially occur before 2. As such, we need to handle this case and then do a switcheroo.
+ *
+ */
+public class QueryResultHandler {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(QueryResultHandler.class);
+
+ private ConcurrentMap<QueryId, UserResultsListener> resultsListener = Maps.newConcurrentMap();
+
+
+ public RpcOutcomeListener<QueryId> getWrappedListener(UserResultsListener listener){
+ return new SubmissionListener(listener);
+ }
+
+ public void batchArrived(ByteBuf pBody, ByteBuf dBody) throws RpcException {
+ final QueryResult result = RpcBus.get(pBody, QueryResult.PARSER);
+ final QueryResultBatch batch = new QueryResultBatch(result, dBody);
+ UserResultsListener l = resultsListener.get(result.getQueryId());
+
+ boolean failed = batch.getHeader().getQueryState() == QueryState.FAILED;
+ // logger.debug("For QueryId [{}], retrieved result listener {}", result.getQueryId(), l);
+ if (l == null) {
+ BufferingListener bl = new BufferingListener();
+ l = resultsListener.putIfAbsent(result.getQueryId(), bl);
+ // if we had a succesful insert, use that reference. Otherwise, just throw away the new bufering listener.
+ if (l == null) l = bl;
+ if (result.getQueryId().toString().equals("")) {
+ failAll();
+ }
+ }
+
+ if(failed){
+ l.submissionFailed(new RpcException("Remote failure while running query." + batch.getHeader().getErrorList()));
+ resultsListener.remove(result.getQueryId(), l);
+ }else{
+ l.resultArrived(batch);
+ }
+
+ if (
+ (failed || result.getIsLastChunk())
+ &&
+ (!(l instanceof BufferingListener) || ((BufferingListener)l).output != null)
+ ) {
+ resultsListener.remove(result.getQueryId(), l);
+ }
+
+
+ }
+
+ private void failAll() {
+ for (UserResultsListener l : resultsListener.values()) {
+ l.submissionFailed(new RpcException("Received result without QueryId"));
+ }
+ }
+
+
+
+ private class BufferingListener implements UserResultsListener {
+
+ private ConcurrentLinkedQueue<QueryResultBatch> results = Queues.newConcurrentLinkedQueue();
+ private volatile boolean finished = false;
+ private volatile RpcException ex;
+ private volatile UserResultsListener output;
+
+ public boolean transferTo(UserResultsListener l) {
+ synchronized (this) {
+ output = l;
+ boolean last = false;
+ for (QueryResultBatch r : results) {
+ l.resultArrived(r);
+ last = r.getHeader().getIsLastChunk();
+ }
+ if(ex != null){
+ l.submissionFailed(ex);
+ return true;
+ }
+ return last;
+ }
+ }
+
+
+ @Override
+ public void resultArrived(QueryResultBatch result) {
+ if(result.getHeader().getIsLastChunk()) finished = true;
+
+ synchronized (this) {
+ if (output == null) {
+ this.results.add(result);
+ } else {
+ output.resultArrived(result);
+ }
+ }
+ }
+
+ @Override
+ public void submissionFailed(RpcException ex) {
+ finished = true;
+ synchronized (this) {
+ if (output == null){
+ this.ex = ex;
+ } else{
+ output.submissionFailed(ex);
+ }
+ }
+ }
+
+ public boolean isFinished(){
+ return finished;
+ }
+
+ }
+
+ private class SubmissionListener extends BaseRpcOutcomeListener<QueryId> {
+ private UserResultsListener listener;
+
+ public SubmissionListener(UserResultsListener listener) {
+ super();
+ this.listener = listener;
+ }
+
+ @Override
+ public void failed(RpcException ex) {
+ listener.submissionFailed(ex);
+ }
+
+ @Override
+ public void success(QueryId queryId, ByteBuf buf) {
+ logger.debug("Received QueryId {} succesfully. Adding listener {}", queryId, listener);
+ UserResultsListener oldListener = resultsListener.putIfAbsent(queryId, listener);
+
+ // we need to deal with the situation where we already received results by the time we got the query id back. In
+ // that case, we'll need to transfer the buffering listener over, grabbing a lock against reception of additional
+ // results during the transition
+ if (oldListener != null) {
+ logger.debug("Unable to place user results listener, buffering listener was already in place.");
+ if (oldListener instanceof BufferingListener) {
+ resultsListener.remove(oldListener);
+ boolean all = ((BufferingListener) oldListener).transferTo(this.listener);
+ // simply remove the buffering listener if we already have the last response.
+ if (all) {
+ resultsListener.remove(oldListener);
+ } else {
+ boolean replaced = resultsListener.replace(queryId, oldListener, listener);
+ if (!replaced) throw new IllegalStateException();
+ }
+ } else {
+ throw new IllegalStateException("Trying to replace a non-buffering User Results listener.");
+ }
+ }
+
+ }
+
+ }
+}