flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (FLINK-7406) Implement Netty receiver incoming pipeline for credit-based
Date Tue, 21 Nov 2017 14:50:01 GMT

    [ https://issues.apache.org/jira/browse/FLINK-7406?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16260820#comment-16260820
] 

ASF GitHub Bot commented on FLINK-7406:
---------------------------------------

Github user NicoK commented on a diff in the pull request:

    https://github.com/apache/flink/pull/4509#discussion_r152234362
  
    --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
---
    @@ -287,17 +290,22 @@ void notifyCreditAvailable() {
     	}
     
     	/**
    -	 * Exclusive buffer is recycled to this input channel directly and it may trigger notify
    -	 * credit to producer.
    +	 * Exclusive buffer is recycled to this input channel directly and it may trigger return
extra
    +	 * floating buffer and notify increased credit to the producer.
     	 *
     	 * @param segment The exclusive segment of this channel.
     	 */
     	@Override
     	public void recycle(MemorySegment segment) {
    -		synchronized (availableBuffers) {
    -			// Important: the isReleased check should be inside the synchronized block.
    -			// that way the segment can also be returned to global pool after added into
    -			// the available queue during releasing all resources.
    +		boolean floatingBufferRecycled = false;
    +
    +		synchronized (bufferQueue) {
    +			final int numRequiredBuffers = senderBacklog + initialCredit;
    +			checkState(bufferQueue.getAvailableBufferSize() <= numRequiredBuffers,
    +				"The number of available buffers " + bufferQueue.getAvailableBufferSize() + " should
not exceed " + numRequiredBuffers);
    +
    +			// Important: check the isReleased state inside synchronized block, so there is no
    +			// race condition when recycle and releaseAllResources running in parallel.
     			if (isReleased.get()) {
     				try {
     					inputGate.returnExclusiveSegments(Arrays.asList(segment));
    --- End diff --
    
    now that I see it, it should rather be `inputGate.returnExclusiveSegments(Collections.singletonList(segment));`


> Implement Netty receiver incoming pipeline for credit-based
> -----------------------------------------------------------
>
>                 Key: FLINK-7406
>                 URL: https://issues.apache.org/jira/browse/FLINK-7406
>             Project: Flink
>          Issue Type: Sub-task
>          Components: Network
>            Reporter: zhijiang
>            Assignee: zhijiang
>             Fix For: 1.5.0
>
>
> This is a part of work for credit-based network flow control.
> Currently {{PartitionRequestClientHandler}} receives and reads {{BufferResponse}} from
producer. It will request buffer from {{BufferPool}} for holding the message. If not got,
the message is staged temporarily and {{autoread}} for channel is set false.
> For credit-based mode, {{PartitionRequestClientHandler}} can always get buffer from {{RemoteInputChannel}}
for reading messages from producer.
> The related works are:
> * Add the backlog of producer in {{BufferResponse}} message structure
> * {{PartitionRequestClientHandler}} requests buffer from {{RemoteInputChannel}} directly
> * {{PartitionRequestClientHandler}} updates backlog for {{RemoteInputChannel}}, and it
may trigger requests of floating buffers from {{BufferPool}}



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

Mime
View raw message