+ * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + *
+ * http://www.apache.org/licenses/LICENSE-2.0 + *
+ * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.automq.stream.thirdparty.moe.cnkirito.kdio; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.channels.Channel; + +public interface DirectChannel extends Channel { + /** + * Writes from the src buffer into this channel at position. + * + * @param src + * The {@link ByteBuffer} to write from + * + * @param position + * The position within the file at which to start writing + * + * @return How many bytes were written from src into the file + * @throws IOException + */ + int write(ByteBuffer src, long position) throws IOException; + + /** + * Reads from this channel into the dst buffer from position. + * + * @param dst + * The {@link ByteBuffer} to read into + * + * @param position + * The position within the file at which to start reading + * + * @return How many bytes were placed into dst + * @throws IOException + */ + int read(ByteBuffer dst, long position) throws IOException; + + /** + * @return The file size for this channel + */ + long size(); + + /** + * @return true if this channel is read only, false otherwise + */ + boolean isReadOnly(); + + /** + * Truncates this file's length to fileLength. + * + * @param fileLength The length to which to truncate + * + * @return This UnsafeByteAlignedChannel + * + * @throws IOException + */ + DirectChannel truncate(long fileLength) throws IOException; + + /** + * @return The file descriptor for this channel + */ + int getFD(); +} diff --git a/s3stream/src/main/java/com/automq/stream/thirdparty/moe/cnkirito/kdio/DirectChannelImpl.java b/s3stream/src/main/java/com/automq/stream/thirdparty/moe/cnkirito/kdio/DirectChannelImpl.java new file mode 100755 index 000000000..7ba12863b --- /dev/null +++ b/s3stream/src/main/java/com/automq/stream/thirdparty/moe/cnkirito/kdio/DirectChannelImpl.java @@ -0,0 +1,128 @@ +/** + * Copyright 2019 xujingfeng (kirito.moe@foxmail.com) + *
+ * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + *
+ * http://www.apache.org/licenses/LICENSE-2.0 + *
+ * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.automq.stream.thirdparty.moe.cnkirito.kdio; + +import java.io.File; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.channels.ClosedChannelException; +import java.nio.channels.NonWritableChannelException; + +public class DirectChannelImpl implements DirectChannel { + private final DirectIOLib lib; + private final int fd; + private boolean isOpen; + private long fileLength; + private final boolean isReadOnly; + + private DirectChannelImpl(DirectIOLib lib, int fd, long fileLength, boolean readOnly) { + this.lib = lib; + this.fd = fd; + this.isOpen = true; + this.isReadOnly = readOnly; + this.fileLength = fileLength; + } + + public static DirectChannel getChannel(File file, boolean readOnly) throws IOException { + DirectIOLib lib = DirectIOLib.getLibForPath(file.toString()); + if (null == lib) { + throw new IOException("No DirectIOLib found for path " + file); + } + return getChannel(lib, file, readOnly); + } + + public static DirectChannel getChannel(DirectIOLib lib, File file, boolean readOnly) throws IOException { + int fd = lib.oDirectOpen(file.toString(), readOnly); + long length = file.length(); + return new DirectChannelImpl(lib, fd, length, readOnly); + } + + private void ensureOpen() throws ClosedChannelException { + if (!isOpen()) { + throw new ClosedChannelException(); + } + } + + private void ensureWritable() { + if (isReadOnly()) { + throw new NonWritableChannelException(); + } + } + + @Override + public int read(ByteBuffer dst, long position) throws IOException { + ensureOpen(); + return lib.pread(fd, dst, position); + } + + @Override + public int write(ByteBuffer src, long position) throws IOException { + ensureOpen(); + ensureWritable(); + assert src.position() == lib.blockStart(src.position()); + + int written = lib.pwrite(fd, src, position); + + // update file length if we wrote past it + fileLength = Math.max(position + written, fileLength); + return written; + } + + @Override + public DirectChannel truncate(final long length) throws IOException { + ensureOpen(); + ensureWritable(); + if (DirectIOLib.ftruncate(fd, length) < 0) { + throw new IOException("Error during truncate on descriptor " + fd + ": " + + DirectIOLib.getLastError()); + } + fileLength = length; + return this; + } + + @Override + public long size() { + return fileLength; + } + + @Override + public int getFD() { + return fd; + } + + + @Override + public boolean isOpen() { + return isOpen; + } + + @Override + public boolean isReadOnly() { + return isReadOnly; + } + + @Override + public void close() throws IOException { + if (!isOpen()) { + return; + } + isOpen = false; + if (lib.close(fd) < 0) { + throw new IOException("Error closing file with descriptor " + fd + ": " + + DirectIOLib.getLastError()); + } + } +} diff --git a/s3stream/src/main/java/com/automq/stream/thirdparty/moe/cnkirito/kdio/DirectIOLib.java b/s3stream/src/main/java/com/automq/stream/thirdparty/moe/cnkirito/kdio/DirectIOLib.java new file mode 100755 index 000000000..52f0d3cfa --- /dev/null +++ b/s3stream/src/main/java/com/automq/stream/thirdparty/moe/cnkirito/kdio/DirectIOLib.java @@ -0,0 +1,389 @@ +/** + * Copyright (C) 2014 Stephen Macke (smacke@cs.stanford.edu) + *
+ * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + *
+ * http://www.apache.org/licenses/LICENSE-2.0 + *
+ * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.automq.stream.thirdparty.moe.cnkirito.kdio; + +import com.sun.jna.Native; +import com.sun.jna.NativeLong; +import com.sun.jna.Platform; +import com.sun.jna.Pointer; +import com.sun.jna.ptr.PointerByReference; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import sun.nio.ch.DirectBuffer; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; + +/** + * Class containing native hooks and utility methods for performing direct I/O, using + * the Linux O_DIRECT flag. + *
+ * This class is initialized at class load time, by registering JNA hooks into native methods.
+ * It also calculates Linux kernel version-dependent alignment amount (in bytes) for use with the O_DIRECT flag,
+ * when given a string for a file or directory.
+ */
+public class DirectIOLib {
+ private static final Logger logger = LoggerFactory.getLogger(DirectIOLib.class);
+ public static boolean binit;
+ static final int PC_REC_XFER_ALIGN = 0x11;
+
+ static {
+ binit = false;
+ try {
+ if (!Platform.isLinux()) {
+ logger.warn("Not running Linux, jaydio support disabled");
+ } else { // now check to see if we have O_DIRECT...
+
+ final int linuxVersion = 0;
+ final int majorRev = 1;
+ final int minorRev = 2;
+
+ List
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.automq.stream.thirdparty.moe.cnkirito.kdio;
+
+import com.sun.jna.NativeLong;
+import com.sun.jna.Pointer;
+import com.sun.jna.ptr.PointerByReference;
+import jdk.internal.access.SharedSecrets;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+
+public class DirectIOUtils {
+ public static final ByteOrder NATIVE_BYTE_ORDER = ByteOrder.nativeOrder();
+
+ /**
+ * Allocate capacity bytes of native memory for use as a buffer, and
+ * return a {@link ByteBuffer} which gives an interface to this memory. The
+ * memory is allocated with
+ * {@link DirectIOLib#posix_memalign(PointerByReference, NativeLong, NativeLong) DirectIOLib#posix_memalign()}
+ * to ensure that the buffer can be used with O_DIRECT.
+ * *
+ *
+ * @param capacity The requested number of bytes to allocate
+ * @return A new JnaMemAlignedBuffer of capacity bytes aligned in native memory.
+ */
+ public static ByteBuffer allocateForDirectIO(DirectIOLib lib, int capacity) {
+ if (capacity % lib.blockSize() > 0) {
+ throw new IllegalArgumentException("Capacity (" + capacity + ") must be a multiple"
+ + "of the block size (" + lib.blockSize() + ")");
+ }
+ NativeLong blockSize = new NativeLong(lib.blockSize());
+ PointerByReference pointerToPointer = new PointerByReference();
+
+ // align memory for use with O_DIRECT
+ DirectIOLib.posix_memalign(pointerToPointer, blockSize, new NativeLong(capacity));
+ return wrapPointer(Pointer.nativeValue(pointerToPointer.getValue()), capacity);
+ }
+
+ /**
+ * @param ptr Pointer to wrap.
+ * @param len Memory location length.
+ * @return Byte buffer wrapping the given memory.
+ */
+ public static ByteBuffer wrapPointer(long ptr, int len) {
+ ByteBuffer buf = SharedSecrets.getJavaNioAccess().newDirectByteBuffer(ptr, len, null, null);
+
+ assert buf.isDirect();
+ return buf;
+ }
+}
diff --git a/s3stream/src/main/java/com/automq/stream/thirdparty/moe/cnkirito/kdio/DirectRandomAccessFile.java b/s3stream/src/main/java/com/automq/stream/thirdparty/moe/cnkirito/kdio/DirectRandomAccessFile.java
new file mode 100755
index 000000000..aabad015a
--- /dev/null
+++ b/s3stream/src/main/java/com/automq/stream/thirdparty/moe/cnkirito/kdio/DirectRandomAccessFile.java
@@ -0,0 +1,86 @@
+/**
+ * Copyright (C) 2014 Stephen Macke (smacke@cs.stanford.edu)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.automq.stream.thirdparty.moe.cnkirito.kdio;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.nio.ByteBuffer;
+
+/**
+ * Class to emulate the behavior of {@link RandomAccessFile}, but using direct I/O.
+ *
+ */
+public class DirectRandomAccessFile implements Closeable {
+
+ private final DirectChannel channel;
+
+
+ /**
+ * @param file The file to open
+ *
+ * @param mode Either "rw" or "r", depending on whether this file is read only
+ *
+ * @throws IOException
+ */
+ public DirectRandomAccessFile(File file, String mode)
+ throws IOException {
+
+ boolean readOnly = false;
+ if ("r".equals(mode)) {
+ readOnly = true;
+ } else if (!"rw".equals(mode)) {
+ throw new IllegalArgumentException("only r and rw modes supported");
+ }
+
+ if (readOnly && !file.isFile()) {
+ throw new FileNotFoundException("couldn't find file " + file);
+ }
+
+ this.channel = DirectChannelImpl.getChannel(file, readOnly);
+ }
+
+ @Override
+ public void close() throws IOException {
+ channel.close();
+ }
+
+
+ public int write(ByteBuffer src, long position) throws IOException {
+ return channel.write(src, position);
+ }
+
+ public int read(ByteBuffer dst, long position) throws IOException {
+ return channel.read(dst, position);
+ }
+
+ /**
+ * @return The current position in the file
+ */
+ public long getFilePointer() {
+ return channel.getFD();
+ }
+
+ /**
+ * @return The current length of the file
+ */
+ public long length() {
+ return channel.size();
+ }
+
+}
diff --git a/s3stream/src/main/java/com/automq/stream/thirdparty/moe/cnkirito/kdio/OpenFlags.java b/s3stream/src/main/java/com/automq/stream/thirdparty/moe/cnkirito/kdio/OpenFlags.java
new file mode 100755
index 000000000..08671a741
--- /dev/null
+++ b/s3stream/src/main/java/com/automq/stream/thirdparty/moe/cnkirito/kdio/OpenFlags.java
@@ -0,0 +1,32 @@
+/**
+ * Copyright (C) 2014 Stephen Macke (smacke@cs.stanford.edu)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.automq.stream.thirdparty.moe.cnkirito.kdio;
+
+/**
+ * Constants for {@link DirectIOLib#oDirectOpen(String, boolean)}.
+ */
+public final class OpenFlags {
+ public static final int O_RDONLY = 00;
+ public static final int O_WRONLY = 01;
+ public static final int O_RDWR = 02;
+ public static final int O_CREAT = 0100;
+ public static final int O_TRUNC = 01000;
+ public static final int O_DIRECT = 040000;
+ public static final int O_SYNC = 04000000;
+
+ private OpenFlags() {
+ }
+}