@Test public void testConcurrentNotifyBufferAvailableAndRelease() throws Exception { testConcurrentReleaseAndSomething(1024, (inputChannel, buffer, j) -> inputChannel.notifyBufferAvailable(buffer) ); }
@Override public NotificationResult notifyBufferAvailable(Buffer buffer) { NotificationResult notificationResult = NotificationResult.BUFFER_NOT_USED; try { synchronized (bufferQueue) { checkState(isWaitingForFloatingBuffers, "This channel should be waiting for floating buffers."); if (isReleased.get() || bufferQueue.getAvailableBufferSize() >= numRequiredBuffers) { isWaitingForFloatingBuffers = false; return notificationResult; } bufferQueue.addFloatingBuffer(buffer); if (bufferQueue.getAvailableBufferSize() == numRequiredBuffers) { isWaitingForFloatingBuffers = false; notificationResult = NotificationResult.BUFFER_USED_NO_NEED_MORE; } else { notificationResult = NotificationResult.BUFFER_USED_NEED_MORE; } } if (unannouncedCredit.getAndAdd(1) == 0) { notifyCreditAvailable(); } } catch (Throwable t) { setError(t); } return notificationResult; }
RemoteInputChannel extends InputChannel implements BufferRecycler, BufferListener { @Override public NotificationResult notifyBufferAvailable(Buffer buffer) { NotificationResult notificationResult = NotificationResult.BUFFER_NOT_USED; try { synchronized (bufferQueue) { checkState(isWaitingForFloatingBuffers, "This channel should be waiting for floating buffers."); if (isReleased.get() || bufferQueue.getAvailableBufferSize() >= numRequiredBuffers) { isWaitingForFloatingBuffers = false; return notificationResult; } bufferQueue.addFloatingBuffer(buffer); if (bufferQueue.getAvailableBufferSize() == numRequiredBuffers) { isWaitingForFloatingBuffers = false; notificationResult = NotificationResult.BUFFER_USED_NO_NEED_MORE; } else { notificationResult = NotificationResult.BUFFER_USED_NEED_MORE; } } if (unannouncedCredit.getAndAdd(1) == 0) { notifyCreditAvailable(); } } catch (Throwable t) { setError(t); } return notificationResult; } }
RemoteInputChannel extends InputChannel implements BufferRecycler, BufferListener { @Override public NotificationResult notifyBufferAvailable(Buffer buffer) { NotificationResult notificationResult = NotificationResult.BUFFER_NOT_USED; try { synchronized (bufferQueue) { checkState(isWaitingForFloatingBuffers, "This channel should be waiting for floating buffers."); if (isReleased.get() || bufferQueue.getAvailableBufferSize() >= numRequiredBuffers) { isWaitingForFloatingBuffers = false; return notificationResult; } bufferQueue.addFloatingBuffer(buffer); if (bufferQueue.getAvailableBufferSize() == numRequiredBuffers) { isWaitingForFloatingBuffers = false; notificationResult = NotificationResult.BUFFER_USED_NO_NEED_MORE; } else { notificationResult = NotificationResult.BUFFER_USED_NEED_MORE; } } if (unannouncedCredit.getAndAdd(1) == 0) { notifyCreditAvailable(); } } catch (Throwable t) { setError(t); } return notificationResult; } RemoteInputChannel(
		SingleInputGate inputGate,
		int channelIndex,
		ResultPartitionID partitionId,
		ConnectionID connectionId,
		ConnectionManager connectionManager,
		TaskIOMetricGroup metrics);  RemoteInputChannel(
		SingleInputGate inputGate,
		int channelIndex,
		ResultPartitionID partitionId,
		ConnectionID connectionId,
		ConnectionManager connectionManager,
		int initialBackOff,
		int maxBackoff,
		TaskIOMetricGroup metrics); }
RemoteInputChannel extends InputChannel implements BufferRecycler, BufferListener { @Override public NotificationResult notifyBufferAvailable(Buffer buffer) { NotificationResult notificationResult = NotificationResult.BUFFER_NOT_USED; try { synchronized (bufferQueue) { checkState(isWaitingForFloatingBuffers, "This channel should be waiting for floating buffers."); if (isReleased.get() || bufferQueue.getAvailableBufferSize() >= numRequiredBuffers) { isWaitingForFloatingBuffers = false; return notificationResult; } bufferQueue.addFloatingBuffer(buffer); if (bufferQueue.getAvailableBufferSize() == numRequiredBuffers) { isWaitingForFloatingBuffers = false; notificationResult = NotificationResult.BUFFER_USED_NO_NEED_MORE; } else { notificationResult = NotificationResult.BUFFER_USED_NEED_MORE; } } if (unannouncedCredit.getAndAdd(1) == 0) { notifyCreditAvailable(); } } catch (Throwable t) { setError(t); } return notificationResult; } RemoteInputChannel(
		SingleInputGate inputGate,
		int channelIndex,
		ResultPartitionID partitionId,
		ConnectionID connectionId,
		ConnectionManager connectionManager,
		TaskIOMetricGroup metrics);  RemoteInputChannel(
		SingleInputGate inputGate,
		int channelIndex,
		ResultPartitionID partitionId,
		ConnectionID connectionId,
		ConnectionManager connectionManager,
		int initialBackOff,
		int maxBackoff,
		TaskIOMetricGroup metrics); @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(); int unsynchronizedGetNumberOfQueuedBuffers(); 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 { @Override public NotificationResult notifyBufferAvailable(Buffer buffer) { NotificationResult notificationResult = NotificationResult.BUFFER_NOT_USED; try { synchronized (bufferQueue) { checkState(isWaitingForFloatingBuffers, "This channel should be waiting for floating buffers."); if (isReleased.get() || bufferQueue.getAvailableBufferSize() >= numRequiredBuffers) { isWaitingForFloatingBuffers = false; return notificationResult; } bufferQueue.addFloatingBuffer(buffer); if (bufferQueue.getAvailableBufferSize() == numRequiredBuffers) { isWaitingForFloatingBuffers = false; notificationResult = NotificationResult.BUFFER_USED_NO_NEED_MORE; } else { notificationResult = NotificationResult.BUFFER_USED_NEED_MORE; } } if (unannouncedCredit.getAndAdd(1) == 0) { notifyCreditAvailable(); } } catch (Throwable t) { setError(t); } return notificationResult; } RemoteInputChannel(
		SingleInputGate inputGate,
		int channelIndex,
		ResultPartitionID partitionId,
		ConnectionID connectionId,
		ConnectionManager connectionManager,
		TaskIOMetricGroup metrics);  RemoteInputChannel(
		SingleInputGate inputGate,
		int channelIndex,
		ResultPartitionID partitionId,
		ConnectionID connectionId,
		ConnectionManager connectionManager,
		int initialBackOff,
		int maxBackoff,
		TaskIOMetricGroup metrics); @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(); int unsynchronizedGetNumberOfQueuedBuffers(); 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);  }