flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From NicoK <...@git.apache.org>
Subject [GitHub] flink pull request #4552: [FLINK-7456][network] Implement Netty sender incom...
Date Mon, 08 Jan 2018 08:02:34 GMT
Github user NicoK commented on a diff in the pull request:

    https://github.com/apache/flink/pull/4552#discussion_r157760366
  
    --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java
---
    @@ -41,30 +42,33 @@
     import java.io.IOException;
     import java.net.SocketAddress;
     import java.util.ArrayDeque;
    -import java.util.Queue;
     import java.util.concurrent.ConcurrentHashMap;
     import java.util.concurrent.ConcurrentMap;
     import java.util.concurrent.atomic.AtomicReference;
     
    +/**
    + * Channel handler to read the messages of buffer response or error response from the
    + * producer, to write and flush the unannounced credits for the producer.
    + */
     class PartitionRequestClientHandler extends ChannelInboundHandlerAdapter {
     
     	private static final Logger LOG = LoggerFactory.getLogger(PartitionRequestClientHandler.class);
     
    -	private final ConcurrentMap<InputChannelID, RemoteInputChannel> inputChannels
= new ConcurrentHashMap<InputChannelID, RemoteInputChannel>();
    -
    -	private final AtomicReference<Throwable> channelError = new AtomicReference<Throwable>();
    +	/** Channels, which already requested partitions from the producers. */
    +	private final ConcurrentMap<InputChannelID, RemoteInputChannel> inputChannels
= new ConcurrentHashMap<>();
     
    -	private final BufferListenerTask bufferListener = new BufferListenerTask();
    +	/** Channels, which will notify the producers about unannounced credit. */
    +	private final ArrayDeque<RemoteInputChannel> inputChannelsWithCredit = new ArrayDeque<>();
     
    -	private final Queue<Object> stagedMessages = new ArrayDeque<Object>();
    +	private final AtomicReference<Throwable> channelError = new AtomicReference<>();
     
    -	private final StagedMessagesHandlerTask stagedMessagesHandler = new StagedMessagesHandlerTask();
    +	private final ChannelFutureListener writeListener = new WriteAndFlushNextMessageIfPossibleListener();
     
     	/**
     	 * Set of cancelled partition requests. A request is cancelled iff an input channel
is cleared
     	 * while data is still coming in for this channel.
     	 */
    -	private final ConcurrentMap<InputChannelID, InputChannelID> cancelled = Maps.newConcurrentMap();
    +	private final ConcurrentMap<InputChannelID, InputChannelID> cancelled = new ConcurrentHashMap<>();
     
     	private volatile ChannelHandlerContext ctx;
    --- End diff --
    
    it looks like you missed to migrate some of the comments that were present in `CreditBasedClientHandler`
but are not present here, e.g. for `ctx`


---

Mime
View raw message