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-9913) Improve output serialization only once in RecordWriter
Date Thu, 06 Sep 2018 13:10:00 GMT

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

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

pnowojski commented on a change in pull request #6417: [FLINK-9913][runtime] Improve output
serialization only once in RecordWriter
URL: https://github.com/apache/flink/pull/6417#discussion_r215615035
 
 

 ##########
 File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java
 ##########
 @@ -89,77 +88,101 @@ public RecordWriter(ResultPartitionWriter writer, ChannelSelector<T>
channelSele
 
 		this.numChannels = writer.getNumberOfSubpartitions();
 
-		/*
-		 * The runtime exposes a channel abstraction for the produced results
-		 * (see {@link ChannelSelector}). Every channel has an independent
-		 * serializer.
-		 */
-		this.serializers = new SpanningRecordSerializer[numChannels];
+		this.serializer = new SpanningRecordSerializer<T>();
 		this.bufferBuilders = new Optional[numChannels];
+		this.broadcastChannels = new int[numChannels];
 		for (int i = 0; i < numChannels; i++) {
-			serializers[i] = new SpanningRecordSerializer<T>();
+			broadcastChannels[i] = i;
 			bufferBuilders[i] = Optional.empty();
 		}
 	}
 
 	public void emit(T record) throws IOException, InterruptedException {
-		for (int targetChannel : channelSelector.selectChannels(record, numChannels)) {
-			sendToTarget(record, targetChannel);
-		}
+		emitToTargetChannels(record, channelSelector.selectChannels(record, numChannels));
 	}
 
 	/**
 	 * This is used to broadcast Streaming Watermarks in-band with records. This ignores
 	 * the {@link ChannelSelector}.
 	 */
 	public void broadcastEmit(T record) throws IOException, InterruptedException {
-		for (int targetChannel = 0; targetChannel < numChannels; targetChannel++) {
-			sendToTarget(record, targetChannel);
-		}
+		emitToTargetChannels(record, broadcastChannels);
 	}
 
 	/**
 	 * This is used to send LatencyMarks to a random target channel.
 	 */
 	public void randomEmit(T record) throws IOException, InterruptedException {
-		sendToTarget(record, rng.nextInt(numChannels));
+		serializer.serializeRecord(record);
+
+		if (copyToTargetBuffers(rng.nextInt(numChannels))) {
+			serializer.prune();
+		}
 	}
 
-	private void sendToTarget(T record, int targetChannel) throws IOException, InterruptedException
{
-		RecordSerializer<T> serializer = serializers[targetChannel];
+	private void emitToTargetChannels(T record, int[] targetChannels) throws IOException, InterruptedException
{
+		serializer.serializeRecord(record);
+
+		boolean pruneAfterCopying = false;
+		for (int channel : targetChannels) {
+			if (copyToTargetBuffers(channel)) {
+				pruneAfterCopying = true;
+			}
+		}
 
-		SerializationResult result = serializer.addRecord(record);
+		// Make sure we don't hold onto the large intermediate serialization buffer for too long
+		if (pruneAfterCopying) {
+			serializer.prune();
+		}
+	}
 
+	/**
+	 * Copies the intermediate serialization buffer to the BufferBuilder of the target channel,
also
+	 * checks to prune the intermediate buffer iif the target BufferBuilder is fulfilled and
the record
+	 * is full.
+	 *
+	 * @param targetChannel the target channel to get BufferBuilder
+	 * @return <tt>true</tt> if the intermediate serialization buffer should be
pruned
+	 */
+	private boolean copyToTargetBuffers(int targetChannel) throws IOException, InterruptedException
{
 
 Review comment:
   Third time I'm looking at this PR and third time I had to think for a minute what does
the this method. I'm always forgetting that `serializer` is a class field and that this method
copies from it.
   
   Maybe rename to `copyFromSerializerToTargetChannel`? Imo rename would allow us to drop
most of the java doc and simplify it to just:
    ```
   	/**
   	 * @param targetChannel
   	 * @return <tt>true</tt> if the intermediate serialization buffer should be
pruned
   	 */
   ```

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


> Improve output serialization only once in RecordWriter
> ------------------------------------------------------
>
>                 Key: FLINK-9913
>                 URL: https://issues.apache.org/jira/browse/FLINK-9913
>             Project: Flink
>          Issue Type: Improvement
>          Components: Network
>    Affects Versions: 1.6.0
>            Reporter: zhijiang
>            Assignee: zhijiang
>            Priority: Major
>              Labels: pull-request-available
>             Fix For: 1.7.0
>
>
> Currently the {{RecordWriter}} emits output into multi channels via {{ChannelSelector}} 
or broadcasts output to all channels directly. Each channel has a separate {{RecordSerializer}}
for serializing outputs, that means the output will be serialized as many times as the number
of selected channels.
> As we know, data serialization is a high cost operation, so we can get good benefits
by improving the serialization only once.
> I would suggest the following changes for realizing it.
>  # Only one {{RecordSerializer}} is created in {{RecordWriter}} for all the channels.
>  # The output is serialized into the intermediate data buffer only once for different
channels.
>  # The intermediate serialization results are copied into different {{BufferBuilder}}s
for different channels.
> An additional benefit by using a single serializer for all channels is that we get a
potentially significant reduction on heap space overhead from fewer intermediate serialization
buffers (only once we got over 5MiB, these buffers were pruned back to 128B!).



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Mime
View raw message