@Test public void testConcurrentOnBufferAndRelease() throws Exception { testConcurrentReleaseAndSomething(8192, (inputChannel, buffer, j) -> { inputChannel.onBuffer(buffer, j, -1); return null; }); }
public void onBuffer(Buffer buffer, int sequenceNumber, int backlog) throws IOException { boolean recycleBuffer = true; try { final boolean wasEmpty; synchronized (receivedBuffers) { if (isReleased.get()) { return; } if (expectedSequenceNumber != sequenceNumber) { onError(new BufferReorderingException(expectedSequenceNumber, sequenceNumber)); return; } wasEmpty = receivedBuffers.isEmpty(); receivedBuffers.add(buffer); recycleBuffer = false; } ++expectedSequenceNumber; if (wasEmpty) { notifyChannelNonEmpty(); } if (backlog >= 0) { onSenderBacklog(backlog); } } finally { if (recycleBuffer) { buffer.recycleBuffer(); } } }
RemoteInputChannel extends InputChannel implements BufferRecycler, BufferListener { public void onBuffer(Buffer buffer, int sequenceNumber, int backlog) throws IOException { boolean recycleBuffer = true; try { final boolean wasEmpty; synchronized (receivedBuffers) { if (isReleased.get()) { return; } if (expectedSequenceNumber != sequenceNumber) { onError(new BufferReorderingException(expectedSequenceNumber, sequenceNumber)); return; } wasEmpty = receivedBuffers.isEmpty(); receivedBuffers.add(buffer); recycleBuffer = false; } ++expectedSequenceNumber; if (wasEmpty) { notifyChannelNonEmpty(); } if (backlog >= 0) { onSenderBacklog(backlog); } } finally { if (recycleBuffer) { buffer.recycleBuffer(); } } } }
RemoteInputChannel extends InputChannel implements BufferRecycler, BufferListener { public void onBuffer(Buffer buffer, int sequenceNumber, int backlog) throws IOException { boolean recycleBuffer = true; try { final boolean wasEmpty; synchronized (receivedBuffers) { if (isReleased.get()) { return; } if (expectedSequenceNumber != sequenceNumber) { onError(new BufferReorderingException(expectedSequenceNumber, sequenceNumber)); return; } wasEmpty = receivedBuffers.isEmpty(); receivedBuffers.add(buffer); recycleBuffer = false; } ++expectedSequenceNumber; if (wasEmpty) { notifyChannelNonEmpty(); } if (backlog >= 0) { onSenderBacklog(backlog); } } finally { if (recycleBuffer) { buffer.recycleBuffer(); } } } RemoteInputChannel(
		SingleInputGate inputGate,
		int channelIndex,
		ResultPartitionID partitionId,
		ConnectionID connectionId,
		ConnectionManager connectionManager,
		int initialBackOff,
		int maxBackoff,
		InputChannelMetrics metrics,
		@Nonnull MemorySegmentProvider memorySegmentProvider); }
RemoteInputChannel extends InputChannel implements BufferRecycler, BufferListener { public void onBuffer(Buffer buffer, int sequenceNumber, int backlog) throws IOException { boolean recycleBuffer = true; try { final boolean wasEmpty; synchronized (receivedBuffers) { if (isReleased.get()) { return; } if (expectedSequenceNumber != sequenceNumber) { onError(new BufferReorderingException(expectedSequenceNumber, sequenceNumber)); return; } wasEmpty = receivedBuffers.isEmpty(); receivedBuffers.add(buffer); recycleBuffer = false; } ++expectedSequenceNumber; if (wasEmpty) { notifyChannelNonEmpty(); } if (backlog >= 0) { onSenderBacklog(backlog); } } finally { if (recycleBuffer) { buffer.recycleBuffer(); } } } RemoteInputChannel(
		SingleInputGate inputGate,
		int channelIndex,
		ResultPartitionID partitionId,
		ConnectionID connectionId,
		ConnectionManager connectionManager,
		int initialBackOff,
		int maxBackoff,
		InputChannelMetrics metrics,
		@Nonnull MemorySegmentProvider memorySegmentProvider); @VisibleForTesting @Override void requestSubpartition(int subpartitionIndex); @Override boolean isReleased(); @Override String toString(); @Override void recycle(MemorySegment segment); int getNumberOfAvailableBuffers(); int getNumberOfRequiredBuffers(); int getSenderBacklog(); @Override NotificationResult notifyBufferAvailable(Buffer buffer); @Override void notifyBufferDestroyed(); int getUnannouncedCredit(); int getAndResetUnannouncedCredit(); int getNumberOfQueuedBuffers(); @Override int unsynchronizedGetNumberOfQueuedBuffers(); int unsynchronizedGetExclusiveBuffersUsed(); int unsynchronizedGetFloatingBuffersAvailable(); InputChannelID getInputChannelId(); int getInitialCredit(); BufferProvider getBufferProvider(); @Nullable Buffer requestBuffer(); void onBuffer(Buffer buffer, int sequenceNumber, int backlog); void onEmptyBuffer(int sequenceNumber, int backlog); void onFailedPartitionRequest(); void onError(Throwable cause); }
RemoteInputChannel extends InputChannel implements BufferRecycler, BufferListener { public void onBuffer(Buffer buffer, int sequenceNumber, int backlog) throws IOException { boolean recycleBuffer = true; try { final boolean wasEmpty; synchronized (receivedBuffers) { if (isReleased.get()) { return; } if (expectedSequenceNumber != sequenceNumber) { onError(new BufferReorderingException(expectedSequenceNumber, sequenceNumber)); return; } wasEmpty = receivedBuffers.isEmpty(); receivedBuffers.add(buffer); recycleBuffer = false; } ++expectedSequenceNumber; if (wasEmpty) { notifyChannelNonEmpty(); } if (backlog >= 0) { onSenderBacklog(backlog); } } finally { if (recycleBuffer) { buffer.recycleBuffer(); } } } RemoteInputChannel(
		SingleInputGate inputGate,
		int channelIndex,
		ResultPartitionID partitionId,
		ConnectionID connectionId,
		ConnectionManager connectionManager,
		int initialBackOff,
		int maxBackoff,
		InputChannelMetrics metrics,
		@Nonnull MemorySegmentProvider memorySegmentProvider); @VisibleForTesting @Override void requestSubpartition(int subpartitionIndex); @Override boolean isReleased(); @Override String toString(); @Override void recycle(MemorySegment segment); int getNumberOfAvailableBuffers(); int getNumberOfRequiredBuffers(); int getSenderBacklog(); @Override NotificationResult notifyBufferAvailable(Buffer buffer); @Override void notifyBufferDestroyed(); int getUnannouncedCredit(); int getAndResetUnannouncedCredit(); int getNumberOfQueuedBuffers(); @Override int unsynchronizedGetNumberOfQueuedBuffers(); int unsynchronizedGetExclusiveBuffersUsed(); int unsynchronizedGetFloatingBuffersAvailable(); InputChannelID getInputChannelId(); int getInitialCredit(); BufferProvider getBufferProvider(); @Nullable Buffer requestBuffer(); void onBuffer(Buffer buffer, int sequenceNumber, int backlog); void onEmptyBuffer(int sequenceNumber, int backlog); void onFailedPartitionRequest(); void onError(Throwable cause);  }