You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by za...@apache.org on 2024/10/15 10:05:46 UTC
(flink) branch master updated: [FLINK-36526][state/forst] Optimize the overhead of writing with direct buffer (#25508)
This is an automated email from the ASF dual-hosted git repository.
zakelly pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new f7a532a66c3 [FLINK-36526][state/forst] Optimize the overhead of writing with direct buffer (#25508)
f7a532a66c3 is described below
commit f7a532a66c3816ab67b8eb5b25ae2b4233e98715
Author: Zakelly <za...@gmail.com>
AuthorDate: Tue Oct 15 18:05:38 2024 +0800
[FLINK-36526][state/forst] Optimize the overhead of writing with direct buffer (#25508)
---
.../org/apache/flink/core/memory/MemoryUtils.java | 2 +-
.../fs/ByteBufferWritableFSDataOutputStream.java | 27 ++++++++++++++++++++++
2 files changed, 28 insertions(+), 1 deletion(-)
diff --git a/flink-core/src/main/java/org/apache/flink/core/memory/MemoryUtils.java b/flink-core/src/main/java/org/apache/flink/core/memory/MemoryUtils.java
index bb8904a39f2..2155dc12159 100644
--- a/flink-core/src/main/java/org/apache/flink/core/memory/MemoryUtils.java
+++ b/flink-core/src/main/java/org/apache/flink/core/memory/MemoryUtils.java
@@ -154,7 +154,7 @@ public class MemoryUtils {
* @param buffer {@link ByteBuffer} which wraps the native memory address to get
* @return native memory address wrapped by the given {@link ByteBuffer}
*/
- static long getByteBufferAddress(ByteBuffer buffer) {
+ public static long getByteBufferAddress(ByteBuffer buffer) {
Preconditions.checkNotNull(buffer, "buffer is null");
Preconditions.checkArgument(
buffer.isDirect(), "Can't get address of a non-direct ByteBuffer.");
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/ByteBufferWritableFSDataOutputStream.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/ByteBufferWritableFSDataOutputStream.java
index 5d8fc2cd86d..aa7b975a02b 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/ByteBufferWritableFSDataOutputStream.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/ByteBufferWritableFSDataOutputStream.java
@@ -20,6 +20,7 @@ package org.apache.flink.state.forst.fs;
import org.apache.flink.annotation.Experimental;
import org.apache.flink.core.fs.FSDataOutputStream;
+import org.apache.flink.core.memory.MemoryUtils;
import java.io.IOException;
import java.nio.ByteBuffer;
@@ -34,6 +35,16 @@ import java.nio.ByteBuffer;
@Experimental
public class ByteBufferWritableFSDataOutputStream extends FSDataOutputStream {
+ /** The unsafe handle for transparent memory copied (heap / off-heap). */
+ @SuppressWarnings("restriction")
+ private static final sun.misc.Unsafe UNSAFE = MemoryUtils.UNSAFE;
+
+ /** The beginning of the byte array contents, relative to the byte array object. */
+ @SuppressWarnings("restriction")
+ private static final long BYTE_ARRAY_BASE_OFFSET = UNSAFE.arrayBaseOffset(byte[].class);
+
+ private static final int SEGMENT_BUFFER_SIZE = 16 * 1024;
+
private final FSDataOutputStream originalOutputStream;
public ByteBufferWritableFSDataOutputStream(FSDataOutputStream originalOutputStream) {
@@ -58,6 +69,22 @@ public class ByteBufferWritableFSDataOutputStream extends FSDataOutputStream {
}
if (bb.hasArray()) {
write(bb.array(), bb.arrayOffset() + bb.position(), bb.remaining());
+ } else if (bb.isDirect()) {
+ int len = bb.remaining();
+ int segment = Math.min(len, SEGMENT_BUFFER_SIZE);
+ byte[] bytes = new byte[segment];
+ for (int i = 0; i < len; ) {
+ int copy = Math.min(segment, bb.remaining());
+ UNSAFE.copyMemory(
+ null,
+ MemoryUtils.getByteBufferAddress(bb) + bb.position(),
+ bytes,
+ BYTE_ARRAY_BASE_OFFSET,
+ copy);
+ originalOutputStream.write(bytes, 0, copy);
+ bb.position(bb.position() + copy);
+ i += copy;
+ }
} else {
int len = bb.remaining();
for (int i = 0; i < len; i++) {