Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/2753#discussion_r18810254
--- Diff:
network/common/src/main/java/org/apache/spark/network/client/SluiceClientHandler.java
---
@@ -0,0 +1,155 @@
+/*
+ * 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.spark.network.client;
+
+import java.net.SocketAddress;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import com.google.common.annotations.VisibleForTesting;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.spark.network.protocol.StreamChunkId;
+import org.apache.spark.network.protocol.response.ChunkFetchFailure;
+import org.apache.spark.network.protocol.response.ChunkFetchSuccess;
+import org.apache.spark.network.protocol.response.RpcFailure;
+import org.apache.spark.network.protocol.response.RpcResponse;
+import org.apache.spark.network.protocol.response.ServerResponse;
+
+/**
+ * Handler that processes server responses, in response to requests issued
from [[SluiceClient]].
+ * It works by tracking the list of outstanding requests (and their
callbacks).
+ *
+ * Concurrency: thread safe and can be called from multiple threads.
+ */
+public class SluiceClientHandler extends
SimpleChannelInboundHandler<ServerResponse> {
+ private final Logger logger =
LoggerFactory.getLogger(SluiceClientHandler.class);
+
+ private final Map<StreamChunkId, ChunkReceivedCallback>
outstandingFetches =
+ new ConcurrentHashMap<StreamChunkId, ChunkReceivedCallback>();
+
+ private final Map<Long, RpcResponseCallback> outstandingRpcs =
+ new ConcurrentHashMap<Long, RpcResponseCallback>();
+
+ public void addFetchRequest(StreamChunkId streamChunkId,
ChunkReceivedCallback callback) {
+ outstandingFetches.put(streamChunkId, callback);
+ }
+
+ public void removeFetchRequest(StreamChunkId streamChunkId) {
+ outstandingFetches.remove(streamChunkId);
+ }
+
+ public void addRpcRequest(long tag, RpcResponseCallback callback) {
+ outstandingRpcs.put(tag, callback);
+ }
+
+ public void removeRpcRequest(long tag) {
+ outstandingRpcs.remove(tag);
+ }
+
+ /**
+ * Fire the failure callback for all outstanding requests. This is
called when we have an
+ * uncaught exception or pre-mature connection termination.
+ */
+ private void failOutstandingRequests(Throwable cause) {
+ for (Map.Entry<StreamChunkId, ChunkReceivedCallback> entry :
outstandingFetches.entrySet()) {
+ entry.getValue().onFailure(entry.getKey().chunkIndex, cause);
+ }
+ // TODO(rxin): Maybe we need to synchronize the access? Otherwise we
could clear new requests
--- End diff --
maybe remove the todo and explain why this is ok instead.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
