[FLINK-5059] only serialise events once in RecordWriter#broadcastEvent #2805

NicoK · 2016-11-14T21:05:29Z

Only serialise events once in RecordWriter#broadcastEvent.

While adapting this, also clean up some related APIs which is now unused or used similar patterns.

zentol · 2016-11-14T21:45:47Z

Please adjust the PR title to the following format: [FLINK-5059] <issue title/description>, this allows the mirroring of comments made here to JIRA.

zentol · 2016-11-14T21:50:41Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java

 				final CancelCheckpointMarker message = new CancelCheckpointMarker(checkpointMetaData.getCheckpointId());
 				for (ResultPartitionWriter output : getEnvironment().getAllWriters()) {
-					output.writeEventToAllChannels(message);
+					final Buffer eventBuffer = EventSerializer.toBuffer(message);


Why not add a method to the ResultPartitionWriter that encapsulates this behavior? The IterationHeadTask now contains an exact copy of this code.

That was the case before and I could have adapted ResultPartitionWriter#writeEventToAllChannels() accordingly. The question is, however, whether we want ResultPartitionWriter to be aware of the difference between events and buffers or offer a cleaner API that is based on buffers only...

in any case the entire try/finally block could be moved into the ResultPartitionWriter, correct?

yes, that's right - Let's go with writeBufferToAllChannels(Buffer) instead.

uce

The changes look good to me in general. It's a very low hanging fruit to only serialize the event buffers once since they don't come out of the buffer pools anyways. +1 to merge this after the comments are addressed.

uce · 2016-12-04T13:29:39Z

...aming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamMockEnvironment.java

 	}

+	private <T> void addBufferToOutputList(RecordDeserializer<DeserializationDelegate<T>> recordDeserializer,
+									   NonReusingDeserializationDelegate<T> delegate, Buffer buffer,


In test scope, this is not a problem, but in main scope the style guidelines (and check style plugin) do not allow leading whitespace. Personally, I don't think it's good practice to differentiate between tests and main code. You might consider removing the leading space here, too (for example one argument per line).

uce · 2016-12-04T13:30:10Z

...aming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamMockEnvironment.java

 		}
 	}

+	private <T> void addBufferToOutputList(RecordDeserializer<DeserializationDelegate<T>> recordDeserializer,


Not that it was commented before, but you might want to add a high level comment about what happens here.

uce · 2016-12-04T14:00:14Z

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java

-				if (buffer != null) {
-					writeAndClearBuffer(buffer, targetChannel, serializer);
+					// retain the buffer so that it can be recycled by each channel of targetPartition
+					eventBuffer.retain();


It would be good to add a special RecordWriterTest that ensure that the reference counting logic works.

@Test public void testBroadcastEventBufferReferenceCounting() throws Exception { Buffer buffer = EventSerializer.toBuffer(EndOfPartitionEvent.INSTANCE); // Partial mocking of static method... PowerMockito.stub(PowerMockito.method(EventSerializer.class, "toBuffer")).toReturn(buffer); @SuppressWarnings("unchecked") ArrayDeque<BufferOrEvent>[] queues = new ArrayDeque[] { new ArrayDeque(), new ArrayDeque() }; ResultPartitionWriter partition = createCollectingPartitionWriter(queues, new TestInfiniteBufferProvider()); RecordWriter<?> writer = new RecordWriter<>(partition); writer.broadcastEvent(EndOfPartitionEvent.INSTANCE); // Verify added to all queues assertEquals(1, queues[0].size()); assertEquals(1, queues[1].size()); assertTrue(buffer.isRecycled()); }

You have to adjust the createCollectingPartitionWriter to correctly recycle event buffers and replace the PrepareForTest class annotation with @PrepareForTest({ResultPartitionWriter.class, EventSerializer.class}).

Not changed in this PR, but to work correctly this relies on the ResultPartition to recycle the buffer if the add calls fails. It might make sense to add a special test (to ResultPartitionTest or RecordWriterTest) where we ensure that this actually happens to guard against future behaviour changes in ResultPartition. A possible better behaviour would be to let the RecordWriter recycle it if an Exception occurs. This should be addressed in a different PR though.

done, thanks

I opened [FLINK-5277] for the ResultPartition#add test: https://issues.apache.org/jira/browse/FLINK-5277

uce · 2016-12-04T14:00:49Z

...time/src/main/java/org/apache/flink/runtime/io/network/api/writer/ResultPartitionWriter.java

+		try {
+			for (int targetChannel = 0; targetChannel < partition.getNumberOfSubpartitions(); targetChannel++) {
+				// retain the buffer so that it can be recycled by each channel of targetPartition
+				eventBuffer.retain();


Similar comments as in the RecordWriter apply here.

uce · 2016-12-04T14:02:32Z

...k-runtime/src/test/java/org/apache/flink/runtime/io/network/api/writer/RecordWriterTest.java

+				} else {
+					// is event:
+					AbstractEvent event = EventSerializer.fromBuffer(buffer, getClass().getClassLoader());
+					Integer targetChannel = (Integer) invocationOnMock.getArguments()[1];


Was missing before, too, but we should recycle the buffer after creating the event.

…Event

…p and RecordWriter#sendEndOfSuperstep

…annels with writeBufferToAllChannels * also apply the "serialise once for all channels/sub-partitions" pattern

uce · 2016-12-12T11:41:03Z

Thanks for addressing the comments, Nico. Going to merge this with the next batch.

…#broadcastEvent This closes apache#2805.

zentol requested changes Nov 14, 2016

View reviewed changes

NicoK changed the title ~~Flink 5059~~ [FLINK-5059] only serialise events once in RecordWriter#broadcastEvent Nov 15, 2016

NicoK force-pushed the FLINK-5059 branch 3 times, most recently from 5df291d to a96c169 Compare November 15, 2016 09:31

uce reviewed Dec 4, 2016

View reviewed changes

Nico Kruber added 5 commits December 7, 2016 15:29

[FLINK-5059] only serialise events once during RecordWriter#broadcast…

313adbf

…Event

[FLINK-5059] cleanup: remove ResultPartitionWriter#writeEvent()

25007c8

[FLINK-5059] cleanup: remove ResultPartitionWriter#writeEndOfSuperste…

a14f00c

…p and RecordWriter#sendEndOfSuperstep

[FLINK-5059] cleanup: replace ResultPartitionWriter#writeEventToAllCh…

60b6bcb

…annels with writeBufferToAllChannels * also apply the "serialise once for all channels/sub-partitions" pattern

[FLINK-5059] address PR comments by @uce

3bff116

NicoK force-pushed the FLINK-5059 branch from a96c169 to 3bff116 Compare December 7, 2016 14:38

uce pushed a commit to uce/flink that referenced this pull request Dec 12, 2016

[FLINK-5059] [network] Only serialise events once during RecordWriter…

7616b09

…#broadcastEvent This closes apache#2805.

uce pushed a commit to uce/flink that referenced this pull request Jan 9, 2017

[FLINK-5059] [network] Only serialise events once during RecordWriter…

afbd673

…#broadcastEvent This closes apache#2805.

asfgit closed this in 9cff8c9 Jan 9, 2017

liuyuzhong7 pushed a commit to liuyuzhong7/flink that referenced this pull request Jan 17, 2017

[FLINK-5059] [network] Only serialise events once during RecordWriter…

71a3b07

…#broadcastEvent This closes apache#2805.

joseprupi pushed a commit to joseprupi/flink that referenced this pull request Feb 12, 2017

[FLINK-5059] [network] Only serialise events once during RecordWriter…

606b58a

…#broadcastEvent This closes apache#2805.

rmetzger added the component=Runtime/Network label Mar 14, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[FLINK-5059] only serialise events once in RecordWriter#broadcastEvent #2805

[FLINK-5059] only serialise events once in RecordWriter#broadcastEvent #2805

Uh oh!

NicoK commented Nov 14, 2016

Uh oh!

zentol commented Nov 14, 2016 •

edited

Loading

Uh oh!

zentol Nov 14, 2016

Uh oh!

NicoK Nov 15, 2016

Uh oh!

zentol Nov 15, 2016

Uh oh!

NicoK Nov 15, 2016

Uh oh!

uce left a comment

Uh oh!

uce Dec 4, 2016

Uh oh!

uce Dec 4, 2016

Uh oh!

uce Dec 4, 2016

Uh oh!

NicoK Dec 7, 2016

Uh oh!

uce Dec 4, 2016

Uh oh!

uce Dec 4, 2016

Uh oh!

uce commented Dec 12, 2016

Uh oh!

Uh oh!

[FLINK-5059] only serialise events once in RecordWriter#broadcastEvent #2805

[FLINK-5059] only serialise events once in RecordWriter#broadcastEvent #2805

Uh oh!

Conversation

NicoK commented Nov 14, 2016

Uh oh!

zentol commented Nov 14, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

uce left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

uce commented Dec 12, 2016

Uh oh!

Uh oh!

zentol commented Nov 14, 2016 •

edited

Loading