You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@bookkeeper.apache.org by GitBox <gi...@apache.org> on 2022/05/09 02:15:46 UTC
[GitHub] [bookkeeper] hangc0276 opened a new pull request, #3263: BP-47 (task6): Direct I/O entrylogger support
hangc0276 opened a new pull request, #3263:
URL: https://github.com/apache/bookkeeper/pull/3263
### Motivation
Task 6 of BP-47, Direct I/O entrylogger support.
This is a standalone module for direct I/O entry logger support.
### Modification
The implementation uses JNI to do direct I/O to files via posix syscalls. Fallocate is used if running on linux, otherwise this is skipped (at the cost of more filesystem operates during writing).
There are two calls to write, writeAt and writeDelimited. I expect writeAt to be used for the entrylog headers, which entries will go through writeDelimited. In both cases, the calls may return before the syscalls occur. #flush() needs to be called to ensure things are actually written.
The entry log format isn't much changed from what is used by the existing entrylogger. The biggest difference is the padding. Direct I/O must write in aligned blocked. The size of the alignment varies by machine configuration, but 4K is a safe bet on most. As it is unlikely that entry data will land exactly on the alignment boundary, we need to add padding to writes. The existing entry logger has been changed to take this padding into account. When read as a signed int/long/byte the padding will aways parse to a negative value, which distinguishes it from valid entry data (the entry size will always be positive) and also from preallocated space (which is always 0).
Another difference in the format is that the header is now 4K rather than 1K. Again, this is to allow aligned rights. No changes are necessary to allow the existing entry logger to deal with the header change, as we create a dummy entry in the extra header space that the existing entry logger already knows to ignore.
### Others
The commit is made by @ivankelly . This is the sub task of https://github.com/apache/bookkeeper/pull/2932 , which pushed out by @mauricebarnum. However, this PR contains too many changes and the number of code lines reaches 8K+, and it is hard to review. According to your suggestion https://github.com/apache/bookkeeper/pull/2932#issuecomment-1072882245, and after communicate with @mauricebarnum, we are planing to divide the PR into 6 PRs, Please refer to https://github.com/apache/bookkeeper/issues/2943#issuecomment-1086446251.
However, @mauricebarnum doesn't have enough time to deal with those issues, and we desperately need this feature. After communicated with @mauricebarnum and @merlimat , I help to work on divide the PRs and push them out. We are hoping to contain this feature in BookKeeper 4.16.0
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [bookkeeper] hangc0276 commented on a diff in pull request #3263: BP-47 (task6): Direct I/O entrylogger support
Posted by GitBox <gi...@apache.org>.
hangc0276 commented on code in PR #3263:
URL: https://github.com/apache/bookkeeper/pull/3263#discussion_r895162615
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/EntryLogIdsImpl.java:
##########
@@ -0,0 +1,157 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.bookkeeper.bookie.storage;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.apache.bookkeeper.bookie.LedgerDirsManager;
+import org.apache.bookkeeper.bookie.storage.directentrylogger.Events;
+import org.apache.bookkeeper.slogger.Slogger;
+
+/**
+ * EntryLogIdsImpl.
+ */
+public class EntryLogIdsImpl implements EntryLogIds {
+ public static final Pattern FILE_PATTERN = Pattern.compile("^([0-9a-fA-F]+)\\.log$");
+ public static final Pattern COMPACTED_FILE_PATTERN =
+ Pattern.compile("^([0-9a-fA-F]+)\\.log\\.([0-9a-fA-F]+)\\.compacted$");
+
+ private final LedgerDirsManager ledgerDirsManager;
+ private final Slogger slog;
+ private int nextId;
+ private int maxId;
+
+ public EntryLogIdsImpl(LedgerDirsManager ledgerDirsManager,
+ Slogger slog) throws IOException {
+ this.ledgerDirsManager = ledgerDirsManager;
+ this.slog = slog;
+ findLargestGap();
+ }
+
+ @Override
+ public int nextId() throws IOException {
+ while (true) {
+ synchronized (this) {
+ int current = nextId;
+ nextId++;
+ if (nextId == maxId) {
+ findLargestGap();
+ } else {
+ return current;
+ }
+ }
+ }
+ }
+
+ private void findLargestGap() throws IOException {
+ long start = System.nanoTime();
+ List<Integer> currentIds = new ArrayList<Integer>();
+
+ for (File ledgerDir : ledgerDirsManager.getAllLedgerDirs()) {
+ currentIds.addAll(logIdsInDirectory(ledgerDir));
+ currentIds.addAll(compactedLogIdsInDirectory(ledgerDir));
+ }
+
+ int[] gap = findLargestGap(currentIds);
+ nextId = gap[0];
+ maxId = gap[1];
+ slog.kv("dirs", ledgerDirsManager.getAllLedgerDirs())
+ .kv("nextId", nextId)
+ .kv("maxId", maxId)
+ .kv("durationMs", TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start))
+ .info(Events.ENTRYLOG_IDS_CANDIDATES_SELECTED);
+ }
+
+ /**
+ * O(nlogn) algorithm to find largest contiguous gap between
+ * integers in a passed list. n should be relatively small.
+ * Entry logs should be about 1GB in size, so even if the node
+ * stores a PB, there should be only 1000000 entry logs.
+ */
+ static int[] findLargestGap(List<Integer> currentIds) {
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [bookkeeper] hangc0276 commented on a diff in pull request #3263: BP-47 (task6): Direct I/O entrylogger support
Posted by GitBox <gi...@apache.org>.
hangc0276 commented on code in PR #3263:
URL: https://github.com/apache/bookkeeper/pull/3263#discussion_r868739844
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/DefaultEntryLogger.java:
##########
@@ -1086,7 +1087,9 @@ EntryLogMetadata extractEntryLogMetadataFromIndex(long entryLogId) throws IOExce
bc.read(sizeBuffer.get(), offset);
int ledgersMapSize = sizeBuffer.get().readInt();
-
+ if (ledgersMapSize <= 0) {
Review Comment:
If there are not ledgers in ledgerMap, we just skip the following reading.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [bookkeeper] hangc0276 commented on a diff in pull request #3263: BP-47 (task6): Direct I/O entrylogger support
Posted by GitBox <gi...@apache.org>.
hangc0276 commented on code in PR #3263:
URL: https://github.com/apache/bookkeeper/pull/3263#discussion_r895143384
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/directentrylogger/DirectWriter.java:
##########
@@ -0,0 +1,322 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.bookkeeper.bookie.storage.directentrylogger;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.bookkeeper.common.util.ExceptionMessageHelper.exMsg;
+
+import io.netty.buffer.ByteBuf;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+
+import org.apache.bookkeeper.common.util.nativeio.NativeIO;
+import org.apache.bookkeeper.common.util.nativeio.NativeIOException;
+import org.apache.bookkeeper.slogger.Slogger;
+import org.apache.commons.lang3.SystemUtils;
+
+class DirectWriter implements LogWriter {
+ final NativeIO nativeIO;
+ final int fd;
+ final int id;
+ final String filename;
+ final BufferPool bufferPool;
+ final ExecutorService writeExecutor;
+ final Object bufferLock = new Object();
+ final List<Future<?>> outstandingWrites = new ArrayList<Future<?>>();
+ Buffer nativeBuffer;
+ long offset;
+ private volatile boolean useFallocate = true;
+
+ DirectWriter(int id,
+ String filename,
+ long maxFileSize,
+ ExecutorService writeExecutor,
+ BufferPool bufferPool,
+ NativeIO nativeIO, Slogger slog) throws IOException {
+ checkArgument(maxFileSize > 0, "Max file size (%d) must be positive");
+ this.id = id;
+ this.filename = filename;
+ this.writeExecutor = writeExecutor;
+ this.nativeIO = nativeIO;
+
+ offset = 0;
+
+ try {
+ fd = nativeIO.open(filename,
+ NativeIO.O_CREAT | NativeIO.O_WRONLY | NativeIO.O_DIRECT,
+ 00755);
+ checkState(fd >= 0, "Open should have thrown exception, fd is invalid : %d", fd);
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg(ne.getMessage()).kv("file", filename)
+ .kv("errno", ne.getErrno()).toString(), ne);
+ }
+
+ if (useFallocate) {
+ if (!SystemUtils.IS_OS_LINUX) {
+ useFallocate = false;
+ slog.warn(Events.FALLOCATE_NOT_AVAILABLE);
+ } else {
+ try {
+ int ret = nativeIO.fallocate(fd, NativeIO.FALLOC_FL_ZERO_RANGE, 0, maxFileSize);
+ checkState(ret == 0, "Exception should have been thrown on non-zero ret: %d", ret);
+ } catch (NativeIOException ex) {
+ // fallocate(2) is not supported on all filesystems. Since this is an optimization, disable
+ // subsequent usage instead of failing the operation.
+ useFallocate = false;
+ slog.kv("message", ex.getMessage())
+ .kv("file", filename)
+ .kv("errno", ex.getErrno())
+ .warn(Events.FALLOCATE_NOT_AVAILABLE);
+ }
+ }
+ }
+
+ this.bufferPool = bufferPool;
+ this.nativeBuffer = bufferPool.acquire();
+ }
+
+ @Override
+ public int logId() {
+ return id;
+ }
+
+ @Override
+ public void writeAt(long offset, ByteBuf buf) throws IOException {
+ checkArgument(Buffer.isAligned(offset),
+ "Offset to writeAt must be aligned to %d: %d is not", Buffer.ALIGNMENT, offset);
+ checkArgument(Buffer.isAligned(buf.readableBytes()),
+ "Buffer must write multiple of alignment bytes (%d), %d is not",
+ Buffer.ALIGNMENT, buf.readableBytes());
+ Buffer tmpBuffer = bufferPool.acquire();
+ int bytesToWrite = buf.readableBytes();
+ tmpBuffer.reset();
+ tmpBuffer.writeByteBuf(buf);
+ Future<?> f = writeExecutor.submit(() -> {
+ try {
+ int ret = nativeIO.pwrite(fd, tmpBuffer.pointer(), bytesToWrite, offset);
+ if (ret != bytesToWrite) {
+ throw new IOException(exMsg("Incomplete write")
+ .kv("filename", filename)
+ .kv("writeSize", bytesToWrite)
+ .kv("bytesWritten", ret)
+ .kv("offset", offset).toString());
+ }
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg("Write error")
+ .kv("filename", filename)
+ .kv("writeSize", bytesToWrite)
+ .kv("errno", ne.getErrno())
+ .kv("offset", offset).toString());
+ } finally {
+ bufferPool.release(tmpBuffer);
+ }
+ return null;
+ });
+ addOutstandingWrite(f);
+ }
+
+ @Override
+ public int writeDelimited(ByteBuf buf) throws IOException {
+ synchronized (bufferLock) {
+ if (!nativeBuffer.hasSpace(serializedSize(buf))) {
+ flushBuffer();
+ }
+
+ int readable = buf.readableBytes();
+ long bufferPosition = position() + Integer.BYTES;
+ if (bufferPosition > Integer.MAX_VALUE) {
+ throw new IOException(exMsg("Cannot write past max int")
+ .kv("filename", filename)
+ .kv("writeSize", readable)
+ .kv("position", bufferPosition)
+ .toString());
+ }
+ nativeBuffer.writeInt(readable);
+ nativeBuffer.writeByteBuf(buf);
+ return (int) bufferPosition;
+ }
+ }
+
+ @Override
+ public void position(long offset) throws IOException {
+ synchronized (bufferLock) {
+ if (nativeBuffer != null && nativeBuffer.position() > 0) {
+ flushBuffer();
+ }
+ if ((offset % Buffer.ALIGNMENT) != 0) {
+ throw new IOException(exMsg("offset must be multiple of alignment")
+ .kv("offset", offset)
+ .kv("alignment", Buffer.ALIGNMENT)
+ .toString());
+ }
+ this.offset = offset;
+ }
+ }
+
+ @Override
+ public long position() {
+ synchronized (bufferLock) {
+ return this.offset + (nativeBuffer != null ? nativeBuffer.position() : 0);
+ }
+ }
+
+ @Override
+ public void flush() throws IOException {
+ flushBuffer();
+
+ waitForOutstandingWrites();
+
+ try {
+ int ret = nativeIO.fsync(fd);
+ checkState(ret == 0, "Fsync should throw exception on non-zero return (%d)", ret);
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg(ne.getMessage())
+ .kv("file", filename)
+ .kv("errno", ne.getErrno()).toString());
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ synchronized (bufferLock) {
+ if (nativeBuffer != null && nativeBuffer.position() > 0) {
+ flushBuffer();
+ }
+ }
+
+ try {
+ int ret = nativeIO.close(fd);
+ checkState(ret == 0, "Close should throw exception on non-zero return (%d)", ret);
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg(ne.getMessage())
+ .kv("file", filename)
+ .kv("errno", ne.getErrno()).toString());
+ }
+ synchronized (bufferLock) {
+ bufferPool.release(nativeBuffer);
+ nativeBuffer = null;
+ }
+ }
+
+ private void addOutstandingWrite(Future<?> toAdd) throws IOException {
+ synchronized (outstandingWrites) {
+ outstandingWrites.add(toAdd);
+
+ Iterator<Future<?>> iter = outstandingWrites.iterator();
+ while (iter.hasNext()) { // clear out completed futures
+ Future<?> f = iter.next();
+ if (f.isDone()) {
+ waitForFuture(f);
+ iter.remove();
+ } else {
+ break;
+ }
+ }
+ }
+ }
+
+ private void waitForOutstandingWrites() throws IOException {
+ synchronized (outstandingWrites) {
+ Iterator<Future<?>> iter = outstandingWrites.iterator();
+ while (iter.hasNext()) { // clear out completed futures
+ Future<?> f = iter.next();
+ waitForFuture(f);
+ iter.remove();
+ }
+ }
+ }
+
+ private void waitForFuture(Future<?> f) throws IOException {
+ try {
+ f.get();
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ throw new IOException(ie);
+ } catch (Throwable t) {
+ if (t.getCause() instanceof IOException) {
+ throw (IOException) t.getCause();
+ } else {
+ throw new IOException(t);
+ }
+ }
+ }
+
+ private void flushBuffer() throws IOException {
+ synchronized (bufferLock) {
+ if (this.nativeBuffer != null) {
+ int bytesToWrite = this.nativeBuffer.padToAlignment();
+ if (bytesToWrite == 0) {
+ return;
+ }
+
+ Buffer bufferToFlush = this.nativeBuffer;
+ this.nativeBuffer = null;
+
+ long offsetToWrite = offset;
+ offset += bytesToWrite;
+
+ Future<?> f = writeExecutor.submit(() -> {
+ try {
+ if (bytesToWrite <= 0) {
+ return null;
+ }
+ int ret = nativeIO.pwrite(fd, bufferToFlush.pointer(), bytesToWrite, offsetToWrite);
+ if (ret != bytesToWrite) {
+ throw new IOException(exMsg("Incomplete write")
+ .kv("filename", filename)
+ .kv("pointer", bufferToFlush.pointer())
+ .kv("offset", offsetToWrite)
+ .kv("writeSize", bytesToWrite)
+ .kv("bytesWritten", ret).toString());
+ }
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg(ne.getMessage())
+ .kv("filename", filename)
+ .kv("offset", offsetToWrite)
+ .kv("writeSize", bytesToWrite)
+ .kv("pointer", bufferToFlush.pointer())
+ .kv("errno", ne.getErrno()).toString(), ne);
+ } finally {
+ bufferPool.release(bufferToFlush);
+ }
+ return null;
Review Comment:
The max line length is 112, so no need to refactor.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [bookkeeper] hangc0276 commented on pull request #3263: BP-47 (task6): Direct I/O entrylogger support
Posted by GitBox <gi...@apache.org>.
hangc0276 commented on PR #3263:
URL: https://github.com/apache/bookkeeper/pull/3263#issuecomment-1153142010
> > The biggest difference is the padding. Direct I/O must write in aligned blocked.
> > Another difference in the format is that the header is now 4K rather than 1K. Again, this is to allow aligned rights. No changes are necessary to allow the existing entry logger to deal with the header change, as we create a dummy entry in the extra header space that the existing entry logger already knows to ignore.
>
> I think this calls to addition of backwards-compat tests, upgrade/downgrade scenario (data written with these changes can be read by previous version of BK and vice versa)
>
> https://github.com/apache/bookkeeper/tree/master/tests/backward-compat
Ok, I will add the backwards-compact tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [bookkeeper] merlimat merged pull request #3263: BP-47 (task6): Direct I/O entrylogger support
Posted by GitBox <gi...@apache.org>.
merlimat merged PR #3263:
URL: https://github.com/apache/bookkeeper/pull/3263
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [bookkeeper] dlg99 commented on a diff in pull request #3263: BP-47 (task6): Direct I/O entrylogger support
Posted by GitBox <gi...@apache.org>.
dlg99 commented on code in PR #3263:
URL: https://github.com/apache/bookkeeper/pull/3263#discussion_r902965307
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/directentrylogger/DirectWriter.java:
##########
@@ -0,0 +1,322 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.bookkeeper.bookie.storage.directentrylogger;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.bookkeeper.common.util.ExceptionMessageHelper.exMsg;
+
+import io.netty.buffer.ByteBuf;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+
+import org.apache.bookkeeper.common.util.nativeio.NativeIO;
+import org.apache.bookkeeper.common.util.nativeio.NativeIOException;
+import org.apache.bookkeeper.slogger.Slogger;
+import org.apache.commons.lang3.SystemUtils;
+
+class DirectWriter implements LogWriter {
+ final NativeIO nativeIO;
+ final int fd;
+ final int id;
+ final String filename;
+ final BufferPool bufferPool;
+ final ExecutorService writeExecutor;
+ final Object bufferLock = new Object();
+ final List<Future<?>> outstandingWrites = new ArrayList<Future<?>>();
+ Buffer nativeBuffer;
+ long offset;
+ private volatile boolean useFallocate = true;
+
+ DirectWriter(int id,
+ String filename,
+ long maxFileSize,
+ ExecutorService writeExecutor,
+ BufferPool bufferPool,
+ NativeIO nativeIO, Slogger slog) throws IOException {
+ checkArgument(maxFileSize > 0, "Max file size (%d) must be positive");
+ this.id = id;
+ this.filename = filename;
+ this.writeExecutor = writeExecutor;
+ this.nativeIO = nativeIO;
+
+ offset = 0;
+
+ try {
+ fd = nativeIO.open(filename,
+ NativeIO.O_CREAT | NativeIO.O_WRONLY | NativeIO.O_DIRECT,
+ 00755);
+ checkState(fd >= 0, "Open should have thrown exception, fd is invalid : %d", fd);
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg(ne.getMessage()).kv("file", filename)
+ .kv("errno", ne.getErrno()).toString(), ne);
+ }
+
+ if (useFallocate) {
+ if (!SystemUtils.IS_OS_LINUX) {
+ useFallocate = false;
+ slog.warn(Events.FALLOCATE_NOT_AVAILABLE);
+ } else {
+ try {
+ int ret = nativeIO.fallocate(fd, NativeIO.FALLOC_FL_ZERO_RANGE, 0, maxFileSize);
+ checkState(ret == 0, "Exception should have been thrown on non-zero ret: %d", ret);
+ } catch (NativeIOException ex) {
+ // fallocate(2) is not supported on all filesystems. Since this is an optimization, disable
+ // subsequent usage instead of failing the operation.
+ useFallocate = false;
+ slog.kv("message", ex.getMessage())
+ .kv("file", filename)
+ .kv("errno", ex.getErrno())
+ .warn(Events.FALLOCATE_NOT_AVAILABLE);
+ }
+ }
+ }
+
+ this.bufferPool = bufferPool;
+ this.nativeBuffer = bufferPool.acquire();
+ }
+
+ @Override
+ public int logId() {
+ return id;
+ }
+
+ @Override
+ public void writeAt(long offset, ByteBuf buf) throws IOException {
+ checkArgument(Buffer.isAligned(offset),
+ "Offset to writeAt must be aligned to %d: %d is not", Buffer.ALIGNMENT, offset);
+ checkArgument(Buffer.isAligned(buf.readableBytes()),
+ "Buffer must write multiple of alignment bytes (%d), %d is not",
+ Buffer.ALIGNMENT, buf.readableBytes());
+ Buffer tmpBuffer = bufferPool.acquire();
+ int bytesToWrite = buf.readableBytes();
+ tmpBuffer.reset();
+ tmpBuffer.writeByteBuf(buf);
+ Future<?> f = writeExecutor.submit(() -> {
+ try {
+ int ret = nativeIO.pwrite(fd, tmpBuffer.pointer(), bytesToWrite, offset);
+ if (ret != bytesToWrite) {
+ throw new IOException(exMsg("Incomplete write")
+ .kv("filename", filename)
+ .kv("writeSize", bytesToWrite)
+ .kv("bytesWritten", ret)
+ .kv("offset", offset).toString());
+ }
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg("Write error")
+ .kv("filename", filename)
+ .kv("writeSize", bytesToWrite)
+ .kv("errno", ne.getErrno())
+ .kv("offset", offset).toString());
+ } finally {
+ bufferPool.release(tmpBuffer);
+ }
+ return null;
+ });
+ addOutstandingWrite(f);
+ }
+
+ @Override
+ public int writeDelimited(ByteBuf buf) throws IOException {
+ synchronized (bufferLock) {
+ if (!nativeBuffer.hasSpace(serializedSize(buf))) {
+ flushBuffer();
+ }
+
+ int readable = buf.readableBytes();
+ long bufferPosition = position() + Integer.BYTES;
+ if (bufferPosition > Integer.MAX_VALUE) {
+ throw new IOException(exMsg("Cannot write past max int")
+ .kv("filename", filename)
+ .kv("writeSize", readable)
+ .kv("position", bufferPosition)
+ .toString());
+ }
+ nativeBuffer.writeInt(readable);
+ nativeBuffer.writeByteBuf(buf);
+ return (int) bufferPosition;
+ }
+ }
+
+ @Override
+ public void position(long offset) throws IOException {
+ synchronized (bufferLock) {
+ if (nativeBuffer != null && nativeBuffer.position() > 0) {
+ flushBuffer();
+ }
+ if ((offset % Buffer.ALIGNMENT) != 0) {
+ throw new IOException(exMsg("offset must be multiple of alignment")
+ .kv("offset", offset)
+ .kv("alignment", Buffer.ALIGNMENT)
+ .toString());
+ }
+ this.offset = offset;
+ }
+ }
+
+ @Override
+ public long position() {
+ synchronized (bufferLock) {
+ return this.offset + (nativeBuffer != null ? nativeBuffer.position() : 0);
+ }
+ }
+
+ @Override
+ public void flush() throws IOException {
+ flushBuffer();
+
+ waitForOutstandingWrites();
+
+ try {
+ int ret = nativeIO.fsync(fd);
+ checkState(ret == 0, "Fsync should throw exception on non-zero return (%d)", ret);
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg(ne.getMessage())
+ .kv("file", filename)
+ .kv("errno", ne.getErrno()).toString());
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ synchronized (bufferLock) {
+ if (nativeBuffer != null && nativeBuffer.position() > 0) {
+ flushBuffer();
+ }
+ }
+
+ try {
+ int ret = nativeIO.close(fd);
+ checkState(ret == 0, "Close should throw exception on non-zero return (%d)", ret);
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg(ne.getMessage())
+ .kv("file", filename)
+ .kv("errno", ne.getErrno()).toString());
+ }
+ synchronized (bufferLock) {
+ bufferPool.release(nativeBuffer);
+ nativeBuffer = null;
+ }
+ }
+
+ private void addOutstandingWrite(Future<?> toAdd) throws IOException {
+ synchronized (outstandingWrites) {
+ outstandingWrites.add(toAdd);
+
+ Iterator<Future<?>> iter = outstandingWrites.iterator();
+ while (iter.hasNext()) { // clear out completed futures
+ Future<?> f = iter.next();
+ if (f.isDone()) {
+ waitForFuture(f);
+ iter.remove();
+ } else {
+ break;
+ }
+ }
+ }
+ }
+
+ private void waitForOutstandingWrites() throws IOException {
+ synchronized (outstandingWrites) {
+ Iterator<Future<?>> iter = outstandingWrites.iterator();
+ while (iter.hasNext()) { // clear out completed futures
+ Future<?> f = iter.next();
+ waitForFuture(f);
+ iter.remove();
+ }
+ }
+ }
+
+ private void waitForFuture(Future<?> f) throws IOException {
+ try {
+ f.get();
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ throw new IOException(ie);
+ } catch (Throwable t) {
+ if (t.getCause() instanceof IOException) {
+ throw (IOException) t.getCause();
+ } else {
+ throw new IOException(t);
+ }
+ }
+ }
+
+ private void flushBuffer() throws IOException {
+ synchronized (bufferLock) {
+ if (this.nativeBuffer != null) {
+ int bytesToWrite = this.nativeBuffer.padToAlignment();
+ if (bytesToWrite == 0) {
+ return;
+ }
+
+ Buffer bufferToFlush = this.nativeBuffer;
+ this.nativeBuffer = null;
+
+ long offsetToWrite = offset;
+ offset += bytesToWrite;
+
+ Future<?> f = writeExecutor.submit(() -> {
+ try {
+ if (bytesToWrite <= 0) {
+ return null;
+ }
+ int ret = nativeIO.pwrite(fd, bufferToFlush.pointer(), bytesToWrite, offsetToWrite);
+ if (ret != bytesToWrite) {
+ throw new IOException(exMsg("Incomplete write")
+ .kv("filename", filename)
+ .kv("pointer", bufferToFlush.pointer())
+ .kv("offset", offsetToWrite)
+ .kv("writeSize", bytesToWrite)
+ .kv("bytesWritten", ret).toString());
+ }
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg(ne.getMessage())
+ .kv("filename", filename)
+ .kv("offset", offsetToWrite)
+ .kv("writeSize", bytesToWrite)
+ .kv("pointer", bufferToFlush.pointer())
+ .kv("errno", ne.getErrno()).toString(), ne);
+ } finally {
+ bufferPool.release(bufferToFlush);
+ }
+ return null;
Review Comment:
I meant the code at https://github.com/apache/bookkeeper/blob/44f0f1ae9758fdfd8b9e3687b786a5f595c768d6/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/directentrylogger/DirectWriter.java#L118-L138 looks similar, can probably refactor, this will also guarantee "if (bytesToWrite <= 0)" check in both places
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [bookkeeper] hangc0276 commented on pull request #3263: BP-47 (task6): Direct I/O entrylogger support
Posted by GitBox <gi...@apache.org>.
hangc0276 commented on PR #3263:
URL: https://github.com/apache/bookkeeper/pull/3263#issuecomment-1159632349
> > The biggest difference is the padding. Direct I/O must write in aligned blocked.
> > Another difference in the format is that the header is now 4K rather than 1K. Again, this is to allow aligned rights. No changes are necessary to allow the existing entry logger to deal with the header change, as we create a dummy entry in the extra header space that the existing entry logger already knows to ignore.
>
> I think this calls to addition of backwards-compat tests, upgrade/downgrade scenario (data written with these changes can be read by previous version of BK and vice versa)
>
> https://github.com/apache/bookkeeper/tree/master/tests/backward-compat
@dlg99 I have added all the compat tests in `TestDirectEntryLoggerCompat` to ensure the DefaultEntryLogger and DirectEntryLogger switch between each other doesn't impact the entry read and write.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [bookkeeper] dlg99 commented on pull request #3263: BP-47 (task6): Direct I/O entrylogger support
Posted by GitBox <gi...@apache.org>.
dlg99 commented on PR #3263:
URL: https://github.com/apache/bookkeeper/pull/3263#issuecomment-1122756450
> The biggest difference is the padding. Direct I/O must write in aligned blocked.
> Another difference in the format is that the header is now 4K rather than 1K. Again, this is to allow aligned rights. No changes are necessary to allow the existing entry logger to deal with the header change, as we create a dummy entry in the extra header space that the existing entry logger already knows to ignore.
I think this calls to addition of backwards-compat tests, upgrade/downgrade scenario (data written with these changes can be read by previous version of BK and vice versa)
https://github.com/apache/bookkeeper/tree/master/tests/backward-compat
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [bookkeeper] dlg99 commented on a diff in pull request #3263: BP-47 (task6): Direct I/O entrylogger support
Posted by GitBox <gi...@apache.org>.
dlg99 commented on code in PR #3263:
URL: https://github.com/apache/bookkeeper/pull/3263#discussion_r902974846
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/directentrylogger/DirectEntryLoggerStats.java:
##########
@@ -0,0 +1,174 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.bookkeeper.bookie.storage.directentrylogger;
+import static org.apache.bookkeeper.bookie.BookKeeperServerStats.BOOKIE_ADD_ENTRY;
+import static org.apache.bookkeeper.bookie.BookKeeperServerStats.BOOKIE_READ_ENTRY;
+import static org.apache.bookkeeper.bookie.BookKeeperServerStats.BOOKIE_SCOPE;
+import static org.apache.bookkeeper.bookie.BookKeeperServerStats.CATEGORY_SERVER;
+
+import org.apache.bookkeeper.stats.Counter;
+import org.apache.bookkeeper.stats.OpStatsLogger;
+import org.apache.bookkeeper.stats.StatsLogger;
+import org.apache.bookkeeper.stats.annotations.StatsDoc;
+
+@StatsDoc(
+ name = BOOKIE_SCOPE,
+ category = CATEGORY_SERVER,
+ help = "Direct entry logger stats"
+)
+class DirectEntryLoggerStats {
+ private static final String ADD_ENTRY = "entrylog-add-entry";
+ private static final String READ_ENTRY = "entrylog-read-entry";
+ private static final String FLUSH = "entrylog-flush";
+ private static final String WRITER_FLUSH = "entrylog-writer-flush";
+ private static final String READ_BLOCK = "entrylog-read-block";
+ private static final String READER_OPEN = "entrylog-open-reader";
+ private static final String READER_CLOSE = "entrylog-close-reader";
+ private static final String CACHED_READER_SERVED_CLOSED = "entrylog-cached-reader-closed";
+
+ @StatsDoc(
+ name = ADD_ENTRY,
+ help = "Operation stats of adding entries to the entry log",
+ parent = BOOKIE_ADD_ENTRY
+ )
+ private final OpStatsLogger addEntryStats;
+
+ @StatsDoc(
+ name = READ_ENTRY,
+ help = "Operation stats of reading entries from the entry log",
+ parent = BOOKIE_READ_ENTRY
+ )
+ private final ThreadLocal<OpStatsLogger> readEntryStats;
Review Comment:
https://stackoverflow.com/a/2784059/2237794
You are getting per thread/per instance in this case instead of per thread
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [bookkeeper] hangc0276 commented on pull request #3263: BP-47 (task6): Direct I/O entrylogger support
Posted by GitBox <gi...@apache.org>.
hangc0276 commented on PR #3263:
URL: https://github.com/apache/bookkeeper/pull/3263#issuecomment-1162735157
rerun failure checks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [bookkeeper] hangc0276 commented on pull request #3263: BP-47 (task6): Direct I/O entrylogger support
Posted by GitBox <gi...@apache.org>.
hangc0276 commented on PR #3263:
URL: https://github.com/apache/bookkeeper/pull/3263#issuecomment-1163261788
rerun failure checks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [bookkeeper] hangc0276 commented on a diff in pull request #3263: BP-47 (task6): Direct I/O entrylogger support
Posted by GitBox <gi...@apache.org>.
hangc0276 commented on code in PR #3263:
URL: https://github.com/apache/bookkeeper/pull/3263#discussion_r903237152
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/directentrylogger/DirectWriter.java:
##########
@@ -0,0 +1,322 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.bookkeeper.bookie.storage.directentrylogger;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.bookkeeper.common.util.ExceptionMessageHelper.exMsg;
+
+import io.netty.buffer.ByteBuf;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+
+import org.apache.bookkeeper.common.util.nativeio.NativeIO;
+import org.apache.bookkeeper.common.util.nativeio.NativeIOException;
+import org.apache.bookkeeper.slogger.Slogger;
+import org.apache.commons.lang3.SystemUtils;
+
+class DirectWriter implements LogWriter {
+ final NativeIO nativeIO;
+ final int fd;
+ final int id;
+ final String filename;
+ final BufferPool bufferPool;
+ final ExecutorService writeExecutor;
+ final Object bufferLock = new Object();
+ final List<Future<?>> outstandingWrites = new ArrayList<Future<?>>();
+ Buffer nativeBuffer;
+ long offset;
+ private volatile boolean useFallocate = true;
+
+ DirectWriter(int id,
+ String filename,
+ long maxFileSize,
+ ExecutorService writeExecutor,
+ BufferPool bufferPool,
+ NativeIO nativeIO, Slogger slog) throws IOException {
+ checkArgument(maxFileSize > 0, "Max file size (%d) must be positive");
+ this.id = id;
+ this.filename = filename;
+ this.writeExecutor = writeExecutor;
+ this.nativeIO = nativeIO;
+
+ offset = 0;
+
+ try {
+ fd = nativeIO.open(filename,
+ NativeIO.O_CREAT | NativeIO.O_WRONLY | NativeIO.O_DIRECT,
+ 00755);
+ checkState(fd >= 0, "Open should have thrown exception, fd is invalid : %d", fd);
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg(ne.getMessage()).kv("file", filename)
+ .kv("errno", ne.getErrno()).toString(), ne);
+ }
+
+ if (useFallocate) {
+ if (!SystemUtils.IS_OS_LINUX) {
+ useFallocate = false;
+ slog.warn(Events.FALLOCATE_NOT_AVAILABLE);
+ } else {
+ try {
+ int ret = nativeIO.fallocate(fd, NativeIO.FALLOC_FL_ZERO_RANGE, 0, maxFileSize);
+ checkState(ret == 0, "Exception should have been thrown on non-zero ret: %d", ret);
+ } catch (NativeIOException ex) {
+ // fallocate(2) is not supported on all filesystems. Since this is an optimization, disable
+ // subsequent usage instead of failing the operation.
+ useFallocate = false;
+ slog.kv("message", ex.getMessage())
+ .kv("file", filename)
+ .kv("errno", ex.getErrno())
+ .warn(Events.FALLOCATE_NOT_AVAILABLE);
+ }
+ }
+ }
+
+ this.bufferPool = bufferPool;
+ this.nativeBuffer = bufferPool.acquire();
+ }
+
+ @Override
+ public int logId() {
+ return id;
+ }
+
+ @Override
+ public void writeAt(long offset, ByteBuf buf) throws IOException {
+ checkArgument(Buffer.isAligned(offset),
+ "Offset to writeAt must be aligned to %d: %d is not", Buffer.ALIGNMENT, offset);
+ checkArgument(Buffer.isAligned(buf.readableBytes()),
+ "Buffer must write multiple of alignment bytes (%d), %d is not",
+ Buffer.ALIGNMENT, buf.readableBytes());
+ Buffer tmpBuffer = bufferPool.acquire();
+ int bytesToWrite = buf.readableBytes();
+ tmpBuffer.reset();
+ tmpBuffer.writeByteBuf(buf);
+ Future<?> f = writeExecutor.submit(() -> {
+ try {
+ int ret = nativeIO.pwrite(fd, tmpBuffer.pointer(), bytesToWrite, offset);
+ if (ret != bytesToWrite) {
+ throw new IOException(exMsg("Incomplete write")
+ .kv("filename", filename)
+ .kv("writeSize", bytesToWrite)
+ .kv("bytesWritten", ret)
+ .kv("offset", offset).toString());
+ }
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg("Write error")
+ .kv("filename", filename)
+ .kv("writeSize", bytesToWrite)
+ .kv("errno", ne.getErrno())
+ .kv("offset", offset).toString());
+ } finally {
+ bufferPool.release(tmpBuffer);
+ }
+ return null;
+ });
+ addOutstandingWrite(f);
+ }
+
+ @Override
+ public int writeDelimited(ByteBuf buf) throws IOException {
+ synchronized (bufferLock) {
+ if (!nativeBuffer.hasSpace(serializedSize(buf))) {
+ flushBuffer();
+ }
+
+ int readable = buf.readableBytes();
+ long bufferPosition = position() + Integer.BYTES;
+ if (bufferPosition > Integer.MAX_VALUE) {
+ throw new IOException(exMsg("Cannot write past max int")
+ .kv("filename", filename)
+ .kv("writeSize", readable)
+ .kv("position", bufferPosition)
+ .toString());
+ }
+ nativeBuffer.writeInt(readable);
+ nativeBuffer.writeByteBuf(buf);
+ return (int) bufferPosition;
+ }
+ }
+
+ @Override
+ public void position(long offset) throws IOException {
+ synchronized (bufferLock) {
+ if (nativeBuffer != null && nativeBuffer.position() > 0) {
+ flushBuffer();
+ }
+ if ((offset % Buffer.ALIGNMENT) != 0) {
+ throw new IOException(exMsg("offset must be multiple of alignment")
+ .kv("offset", offset)
+ .kv("alignment", Buffer.ALIGNMENT)
+ .toString());
+ }
+ this.offset = offset;
+ }
+ }
+
+ @Override
+ public long position() {
+ synchronized (bufferLock) {
+ return this.offset + (nativeBuffer != null ? nativeBuffer.position() : 0);
+ }
+ }
+
+ @Override
+ public void flush() throws IOException {
+ flushBuffer();
+
+ waitForOutstandingWrites();
+
+ try {
+ int ret = nativeIO.fsync(fd);
+ checkState(ret == 0, "Fsync should throw exception on non-zero return (%d)", ret);
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg(ne.getMessage())
+ .kv("file", filename)
+ .kv("errno", ne.getErrno()).toString());
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ synchronized (bufferLock) {
+ if (nativeBuffer != null && nativeBuffer.position() > 0) {
+ flushBuffer();
+ }
+ }
+
+ try {
+ int ret = nativeIO.close(fd);
+ checkState(ret == 0, "Close should throw exception on non-zero return (%d)", ret);
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg(ne.getMessage())
+ .kv("file", filename)
+ .kv("errno", ne.getErrno()).toString());
+ }
+ synchronized (bufferLock) {
+ bufferPool.release(nativeBuffer);
+ nativeBuffer = null;
+ }
+ }
+
+ private void addOutstandingWrite(Future<?> toAdd) throws IOException {
+ synchronized (outstandingWrites) {
+ outstandingWrites.add(toAdd);
+
+ Iterator<Future<?>> iter = outstandingWrites.iterator();
+ while (iter.hasNext()) { // clear out completed futures
+ Future<?> f = iter.next();
+ if (f.isDone()) {
+ waitForFuture(f);
+ iter.remove();
+ } else {
+ break;
+ }
+ }
+ }
+ }
+
+ private void waitForOutstandingWrites() throws IOException {
+ synchronized (outstandingWrites) {
+ Iterator<Future<?>> iter = outstandingWrites.iterator();
+ while (iter.hasNext()) { // clear out completed futures
+ Future<?> f = iter.next();
+ waitForFuture(f);
+ iter.remove();
+ }
+ }
+ }
+
+ private void waitForFuture(Future<?> f) throws IOException {
+ try {
+ f.get();
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ throw new IOException(ie);
+ } catch (Throwable t) {
+ if (t.getCause() instanceof IOException) {
+ throw (IOException) t.getCause();
+ } else {
+ throw new IOException(t);
+ }
+ }
+ }
+
+ private void flushBuffer() throws IOException {
+ synchronized (bufferLock) {
+ if (this.nativeBuffer != null) {
+ int bytesToWrite = this.nativeBuffer.padToAlignment();
+ if (bytesToWrite == 0) {
+ return;
+ }
+
+ Buffer bufferToFlush = this.nativeBuffer;
+ this.nativeBuffer = null;
+
+ long offsetToWrite = offset;
+ offset += bytesToWrite;
+
+ Future<?> f = writeExecutor.submit(() -> {
+ try {
+ if (bytesToWrite <= 0) {
+ return null;
+ }
+ int ret = nativeIO.pwrite(fd, bufferToFlush.pointer(), bytesToWrite, offsetToWrite);
+ if (ret != bytesToWrite) {
+ throw new IOException(exMsg("Incomplete write")
+ .kv("filename", filename)
+ .kv("pointer", bufferToFlush.pointer())
+ .kv("offset", offsetToWrite)
+ .kv("writeSize", bytesToWrite)
+ .kv("bytesWritten", ret).toString());
+ }
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg(ne.getMessage())
+ .kv("filename", filename)
+ .kv("offset", offsetToWrite)
+ .kv("writeSize", bytesToWrite)
+ .kv("pointer", bufferToFlush.pointer())
+ .kv("errno", ne.getErrno()).toString(), ne);
+ } finally {
+ bufferPool.release(bufferToFlush);
+ }
+ return null;
Review Comment:
done
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/directentrylogger/DirectEntryLoggerStats.java:
##########
@@ -0,0 +1,174 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.bookkeeper.bookie.storage.directentrylogger;
+import static org.apache.bookkeeper.bookie.BookKeeperServerStats.BOOKIE_ADD_ENTRY;
+import static org.apache.bookkeeper.bookie.BookKeeperServerStats.BOOKIE_READ_ENTRY;
+import static org.apache.bookkeeper.bookie.BookKeeperServerStats.BOOKIE_SCOPE;
+import static org.apache.bookkeeper.bookie.BookKeeperServerStats.CATEGORY_SERVER;
+
+import org.apache.bookkeeper.stats.Counter;
+import org.apache.bookkeeper.stats.OpStatsLogger;
+import org.apache.bookkeeper.stats.StatsLogger;
+import org.apache.bookkeeper.stats.annotations.StatsDoc;
+
+@StatsDoc(
+ name = BOOKIE_SCOPE,
+ category = CATEGORY_SERVER,
+ help = "Direct entry logger stats"
+)
+class DirectEntryLoggerStats {
+ private static final String ADD_ENTRY = "entrylog-add-entry";
+ private static final String READ_ENTRY = "entrylog-read-entry";
+ private static final String FLUSH = "entrylog-flush";
+ private static final String WRITER_FLUSH = "entrylog-writer-flush";
+ private static final String READ_BLOCK = "entrylog-read-block";
+ private static final String READER_OPEN = "entrylog-open-reader";
+ private static final String READER_CLOSE = "entrylog-close-reader";
+ private static final String CACHED_READER_SERVED_CLOSED = "entrylog-cached-reader-closed";
+
+ @StatsDoc(
+ name = ADD_ENTRY,
+ help = "Operation stats of adding entries to the entry log",
+ parent = BOOKIE_ADD_ENTRY
+ )
+ private final OpStatsLogger addEntryStats;
+
+ @StatsDoc(
+ name = READ_ENTRY,
+ help = "Operation stats of reading entries from the entry log",
+ parent = BOOKIE_READ_ENTRY
+ )
+ private final ThreadLocal<OpStatsLogger> readEntryStats;
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [bookkeeper] hangc0276 commented on a diff in pull request #3263: BP-47 (task6): Direct I/O entrylogger support
Posted by GitBox <gi...@apache.org>.
hangc0276 commented on code in PR #3263:
URL: https://github.com/apache/bookkeeper/pull/3263#discussion_r895162580
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/directentrylogger/DirectWriter.java:
##########
@@ -0,0 +1,322 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.bookkeeper.bookie.storage.directentrylogger;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.bookkeeper.common.util.ExceptionMessageHelper.exMsg;
+
+import io.netty.buffer.ByteBuf;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+
+import org.apache.bookkeeper.common.util.nativeio.NativeIO;
+import org.apache.bookkeeper.common.util.nativeio.NativeIOException;
+import org.apache.bookkeeper.slogger.Slogger;
+import org.apache.commons.lang3.SystemUtils;
+
+class DirectWriter implements LogWriter {
+ final NativeIO nativeIO;
+ final int fd;
+ final int id;
+ final String filename;
+ final BufferPool bufferPool;
+ final ExecutorService writeExecutor;
+ final Object bufferLock = new Object();
+ final List<Future<?>> outstandingWrites = new ArrayList<Future<?>>();
+ Buffer nativeBuffer;
+ long offset;
+ private volatile boolean useFallocate = true;
+
+ DirectWriter(int id,
+ String filename,
+ long maxFileSize,
+ ExecutorService writeExecutor,
+ BufferPool bufferPool,
+ NativeIO nativeIO, Slogger slog) throws IOException {
+ checkArgument(maxFileSize > 0, "Max file size (%d) must be positive");
+ this.id = id;
+ this.filename = filename;
+ this.writeExecutor = writeExecutor;
+ this.nativeIO = nativeIO;
+
+ offset = 0;
+
+ try {
+ fd = nativeIO.open(filename,
+ NativeIO.O_CREAT | NativeIO.O_WRONLY | NativeIO.O_DIRECT,
+ 00755);
+ checkState(fd >= 0, "Open should have thrown exception, fd is invalid : %d", fd);
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg(ne.getMessage()).kv("file", filename)
+ .kv("errno", ne.getErrno()).toString(), ne);
+ }
+
+ if (useFallocate) {
+ if (!SystemUtils.IS_OS_LINUX) {
+ useFallocate = false;
+ slog.warn(Events.FALLOCATE_NOT_AVAILABLE);
+ } else {
+ try {
+ int ret = nativeIO.fallocate(fd, NativeIO.FALLOC_FL_ZERO_RANGE, 0, maxFileSize);
+ checkState(ret == 0, "Exception should have been thrown on non-zero ret: %d", ret);
+ } catch (NativeIOException ex) {
+ // fallocate(2) is not supported on all filesystems. Since this is an optimization, disable
+ // subsequent usage instead of failing the operation.
+ useFallocate = false;
+ slog.kv("message", ex.getMessage())
+ .kv("file", filename)
+ .kv("errno", ex.getErrno())
+ .warn(Events.FALLOCATE_NOT_AVAILABLE);
+ }
+ }
+ }
+
+ this.bufferPool = bufferPool;
+ this.nativeBuffer = bufferPool.acquire();
+ }
+
+ @Override
+ public int logId() {
+ return id;
+ }
+
+ @Override
+ public void writeAt(long offset, ByteBuf buf) throws IOException {
+ checkArgument(Buffer.isAligned(offset),
+ "Offset to writeAt must be aligned to %d: %d is not", Buffer.ALIGNMENT, offset);
+ checkArgument(Buffer.isAligned(buf.readableBytes()),
+ "Buffer must write multiple of alignment bytes (%d), %d is not",
+ Buffer.ALIGNMENT, buf.readableBytes());
+ Buffer tmpBuffer = bufferPool.acquire();
+ int bytesToWrite = buf.readableBytes();
+ tmpBuffer.reset();
+ tmpBuffer.writeByteBuf(buf);
+ Future<?> f = writeExecutor.submit(() -> {
+ try {
+ int ret = nativeIO.pwrite(fd, tmpBuffer.pointer(), bytesToWrite, offset);
+ if (ret != bytesToWrite) {
+ throw new IOException(exMsg("Incomplete write")
+ .kv("filename", filename)
+ .kv("writeSize", bytesToWrite)
+ .kv("bytesWritten", ret)
+ .kv("offset", offset).toString());
+ }
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg("Write error")
+ .kv("filename", filename)
+ .kv("writeSize", bytesToWrite)
+ .kv("errno", ne.getErrno())
+ .kv("offset", offset).toString());
+ } finally {
+ bufferPool.release(tmpBuffer);
+ }
+ return null;
+ });
+ addOutstandingWrite(f);
+ }
+
+ @Override
+ public int writeDelimited(ByteBuf buf) throws IOException {
+ synchronized (bufferLock) {
+ if (!nativeBuffer.hasSpace(serializedSize(buf))) {
+ flushBuffer();
+ }
+
+ int readable = buf.readableBytes();
+ long bufferPosition = position() + Integer.BYTES;
+ if (bufferPosition > Integer.MAX_VALUE) {
+ throw new IOException(exMsg("Cannot write past max int")
+ .kv("filename", filename)
+ .kv("writeSize", readable)
+ .kv("position", bufferPosition)
+ .toString());
+ }
+ nativeBuffer.writeInt(readable);
+ nativeBuffer.writeByteBuf(buf);
+ return (int) bufferPosition;
+ }
+ }
+
+ @Override
+ public void position(long offset) throws IOException {
+ synchronized (bufferLock) {
+ if (nativeBuffer != null && nativeBuffer.position() > 0) {
+ flushBuffer();
+ }
+ if ((offset % Buffer.ALIGNMENT) != 0) {
+ throw new IOException(exMsg("offset must be multiple of alignment")
+ .kv("offset", offset)
+ .kv("alignment", Buffer.ALIGNMENT)
+ .toString());
+ }
+ this.offset = offset;
+ }
+ }
+
+ @Override
+ public long position() {
+ synchronized (bufferLock) {
+ return this.offset + (nativeBuffer != null ? nativeBuffer.position() : 0);
+ }
+ }
+
+ @Override
+ public void flush() throws IOException {
+ flushBuffer();
+
+ waitForOutstandingWrites();
+
+ try {
+ int ret = nativeIO.fsync(fd);
+ checkState(ret == 0, "Fsync should throw exception on non-zero return (%d)", ret);
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg(ne.getMessage())
+ .kv("file", filename)
+ .kv("errno", ne.getErrno()).toString());
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ synchronized (bufferLock) {
+ if (nativeBuffer != null && nativeBuffer.position() > 0) {
+ flushBuffer();
+ }
+ }
+
+ try {
+ int ret = nativeIO.close(fd);
+ checkState(ret == 0, "Close should throw exception on non-zero return (%d)", ret);
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg(ne.getMessage())
+ .kv("file", filename)
+ .kv("errno", ne.getErrno()).toString());
+ }
+ synchronized (bufferLock) {
+ bufferPool.release(nativeBuffer);
+ nativeBuffer = null;
+ }
+ }
+
+ private void addOutstandingWrite(Future<?> toAdd) throws IOException {
+ synchronized (outstandingWrites) {
+ outstandingWrites.add(toAdd);
Review Comment:
Good idea. We can add a limiter for it in the following Pr. This Pr is huge enough, I have tried my best to keep it simple.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [bookkeeper] hangc0276 commented on a diff in pull request #3263: BP-47 (task6): Direct I/O entrylogger support
Posted by GitBox <gi...@apache.org>.
hangc0276 commented on code in PR #3263:
URL: https://github.com/apache/bookkeeper/pull/3263#discussion_r895162974
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/EntryLogIdsImpl.java:
##########
@@ -0,0 +1,157 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.bookkeeper.bookie.storage;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.apache.bookkeeper.bookie.LedgerDirsManager;
+import org.apache.bookkeeper.bookie.storage.directentrylogger.Events;
+import org.apache.bookkeeper.slogger.Slogger;
+
+/**
+ * EntryLogIdsImpl.
+ */
+public class EntryLogIdsImpl implements EntryLogIds {
+ public static final Pattern FILE_PATTERN = Pattern.compile("^([0-9a-fA-F]+)\\.log$");
+ public static final Pattern COMPACTED_FILE_PATTERN =
+ Pattern.compile("^([0-9a-fA-F]+)\\.log\\.([0-9a-fA-F]+)\\.compacted$");
+
+ private final LedgerDirsManager ledgerDirsManager;
+ private final Slogger slog;
+ private int nextId;
+ private int maxId;
+
+ public EntryLogIdsImpl(LedgerDirsManager ledgerDirsManager,
+ Slogger slog) throws IOException {
+ this.ledgerDirsManager = ledgerDirsManager;
+ this.slog = slog;
+ findLargestGap();
+ }
+
+ @Override
+ public int nextId() throws IOException {
+ while (true) {
+ synchronized (this) {
+ int current = nextId;
+ nextId++;
+ if (nextId == maxId) {
+ findLargestGap();
+ } else {
+ return current;
+ }
+ }
+ }
+ }
+
+ private void findLargestGap() throws IOException {
+ long start = System.nanoTime();
+ List<Integer> currentIds = new ArrayList<Integer>();
+
+ for (File ledgerDir : ledgerDirsManager.getAllLedgerDirs()) {
+ currentIds.addAll(logIdsInDirectory(ledgerDir));
+ currentIds.addAll(compactedLogIdsInDirectory(ledgerDir));
+ }
+
+ int[] gap = findLargestGap(currentIds);
+ nextId = gap[0];
+ maxId = gap[1];
+ slog.kv("dirs", ledgerDirsManager.getAllLedgerDirs())
+ .kv("nextId", nextId)
+ .kv("maxId", maxId)
+ .kv("durationMs", TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start))
+ .info(Events.ENTRYLOG_IDS_CANDIDATES_SELECTED);
+ }
+
+ /**
+ * O(nlogn) algorithm to find largest contiguous gap between
+ * integers in a passed list. n should be relatively small.
+ * Entry logs should be about 1GB in size, so even if the node
Review Comment:
Yes, it can only be guaranteed by `EntryLogManagerForSingleEntryLog`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [bookkeeper] hangc0276 commented on pull request #3263: BP-47 (task6): Direct I/O entrylogger support
Posted by GitBox <gi...@apache.org>.
hangc0276 commented on PR #3263:
URL: https://github.com/apache/bookkeeper/pull/3263#issuecomment-1159639125
All the comments have been addressed, please help take a look. This Pr is huge, we need more eyes.
@merlimat @eolivelli @dlg99 @ivankelly @rdhabalia @RaulGracia @reddycharan @Ghatage
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [bookkeeper] eolivelli commented on a diff in pull request #3263: BP-47 (task6): Direct I/O entrylogger support
Posted by GitBox <gi...@apache.org>.
eolivelli commented on code in PR #3263:
URL: https://github.com/apache/bookkeeper/pull/3263#discussion_r897607221
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/directentrylogger/DirectEntryLoggerStats.java:
##########
@@ -0,0 +1,174 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.bookkeeper.bookie.storage.directentrylogger;
+import static org.apache.bookkeeper.bookie.BookKeeperServerStats.BOOKIE_ADD_ENTRY;
+import static org.apache.bookkeeper.bookie.BookKeeperServerStats.BOOKIE_READ_ENTRY;
+import static org.apache.bookkeeper.bookie.BookKeeperServerStats.BOOKIE_SCOPE;
+import static org.apache.bookkeeper.bookie.BookKeeperServerStats.CATEGORY_SERVER;
+
+import org.apache.bookkeeper.stats.Counter;
+import org.apache.bookkeeper.stats.OpStatsLogger;
+import org.apache.bookkeeper.stats.StatsLogger;
+import org.apache.bookkeeper.stats.annotations.StatsDoc;
+
+@StatsDoc(
+ name = BOOKIE_SCOPE,
+ category = CATEGORY_SERVER,
+ help = "Direct entry logger stats"
+)
+class DirectEntryLoggerStats {
+ private static final String ADD_ENTRY = "entrylog-add-entry";
+ private static final String READ_ENTRY = "entrylog-read-entry";
+ private static final String FLUSH = "entrylog-flush";
+ private static final String WRITER_FLUSH = "entrylog-writer-flush";
+ private static final String READ_BLOCK = "entrylog-read-block";
+ private static final String READER_OPEN = "entrylog-open-reader";
+ private static final String READER_CLOSE = "entrylog-close-reader";
+ private static final String CACHED_READER_SERVED_CLOSED = "entrylog-cached-reader-closed";
+
+ @StatsDoc(
+ name = ADD_ENTRY,
+ help = "Operation stats of adding entries to the entry log",
+ parent = BOOKIE_ADD_ENTRY
+ )
+ private final OpStatsLogger addEntryStats;
+
+ @StatsDoc(
+ name = READ_ENTRY,
+ help = "Operation stats of reading entries from the entry log",
+ parent = BOOKIE_READ_ENTRY
+ )
+ private final ThreadLocal<OpStatsLogger> readEntryStats;
Review Comment:
ThreadLocal must be static, because it is actually not bound to a specific instance.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [bookkeeper] hangc0276 commented on a diff in pull request #3263: BP-47 (task6): Direct I/O entrylogger support
Posted by GitBox <gi...@apache.org>.
hangc0276 commented on code in PR #3263:
URL: https://github.com/apache/bookkeeper/pull/3263#discussion_r895160792
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/directentrylogger/DirectWriter.java:
##########
@@ -0,0 +1,322 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.bookkeeper.bookie.storage.directentrylogger;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.bookkeeper.common.util.ExceptionMessageHelper.exMsg;
+
+import io.netty.buffer.ByteBuf;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+
+import org.apache.bookkeeper.common.util.nativeio.NativeIO;
+import org.apache.bookkeeper.common.util.nativeio.NativeIOException;
+import org.apache.bookkeeper.slogger.Slogger;
+import org.apache.commons.lang3.SystemUtils;
+
+class DirectWriter implements LogWriter {
+ final NativeIO nativeIO;
+ final int fd;
+ final int id;
+ final String filename;
+ final BufferPool bufferPool;
+ final ExecutorService writeExecutor;
+ final Object bufferLock = new Object();
+ final List<Future<?>> outstandingWrites = new ArrayList<Future<?>>();
+ Buffer nativeBuffer;
+ long offset;
+ private volatile boolean useFallocate = true;
+
+ DirectWriter(int id,
+ String filename,
+ long maxFileSize,
+ ExecutorService writeExecutor,
+ BufferPool bufferPool,
+ NativeIO nativeIO, Slogger slog) throws IOException {
+ checkArgument(maxFileSize > 0, "Max file size (%d) must be positive");
+ this.id = id;
+ this.filename = filename;
+ this.writeExecutor = writeExecutor;
+ this.nativeIO = nativeIO;
+
+ offset = 0;
+
+ try {
+ fd = nativeIO.open(filename,
+ NativeIO.O_CREAT | NativeIO.O_WRONLY | NativeIO.O_DIRECT,
+ 00755);
Review Comment:
`644` is better. The current ledger log file's permission is `644`, we can keep the same as the previous one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [bookkeeper] dlg99 commented on a diff in pull request #3263: BP-47 (task6): Direct I/O entrylogger support
Posted by GitBox <gi...@apache.org>.
dlg99 commented on code in PR #3263:
URL: https://github.com/apache/bookkeeper/pull/3263#discussion_r869651292
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/EntryLogIdsImpl.java:
##########
@@ -0,0 +1,157 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.bookkeeper.bookie.storage;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.apache.bookkeeper.bookie.LedgerDirsManager;
+import org.apache.bookkeeper.bookie.storage.directentrylogger.Events;
+import org.apache.bookkeeper.slogger.Slogger;
+
+/**
+ * EntryLogIdsImpl.
+ */
+public class EntryLogIdsImpl implements EntryLogIds {
+ public static final Pattern FILE_PATTERN = Pattern.compile("^([0-9a-fA-F]+)\\.log$");
+ public static final Pattern COMPACTED_FILE_PATTERN =
+ Pattern.compile("^([0-9a-fA-F]+)\\.log\\.([0-9a-fA-F]+)\\.compacted$");
+
+ private final LedgerDirsManager ledgerDirsManager;
+ private final Slogger slog;
+ private int nextId;
+ private int maxId;
+
+ public EntryLogIdsImpl(LedgerDirsManager ledgerDirsManager,
+ Slogger slog) throws IOException {
+ this.ledgerDirsManager = ledgerDirsManager;
+ this.slog = slog;
+ findLargestGap();
+ }
+
+ @Override
+ public int nextId() throws IOException {
+ while (true) {
+ synchronized (this) {
+ int current = nextId;
+ nextId++;
+ if (nextId == maxId) {
+ findLargestGap();
+ } else {
+ return current;
+ }
+ }
+ }
+ }
+
+ private void findLargestGap() throws IOException {
+ long start = System.nanoTime();
+ List<Integer> currentIds = new ArrayList<Integer>();
+
+ for (File ledgerDir : ledgerDirsManager.getAllLedgerDirs()) {
+ currentIds.addAll(logIdsInDirectory(ledgerDir));
+ currentIds.addAll(compactedLogIdsInDirectory(ledgerDir));
+ }
+
+ int[] gap = findLargestGap(currentIds);
+ nextId = gap[0];
+ maxId = gap[1];
+ slog.kv("dirs", ledgerDirsManager.getAllLedgerDirs())
+ .kv("nextId", nextId)
+ .kv("maxId", maxId)
+ .kv("durationMs", TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start))
+ .info(Events.ENTRYLOG_IDS_CANDIDATES_SELECTED);
+ }
+
+ /**
+ * O(nlogn) algorithm to find largest contiguous gap between
+ * integers in a passed list. n should be relatively small.
+ * Entry logs should be about 1GB in size, so even if the node
Review Comment:
not guaranteed in case of entry log per ledger.
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/directentrylogger/DirectWriter.java:
##########
@@ -0,0 +1,322 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.bookkeeper.bookie.storage.directentrylogger;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.bookkeeper.common.util.ExceptionMessageHelper.exMsg;
+
+import io.netty.buffer.ByteBuf;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+
+import org.apache.bookkeeper.common.util.nativeio.NativeIO;
+import org.apache.bookkeeper.common.util.nativeio.NativeIOException;
+import org.apache.bookkeeper.slogger.Slogger;
+import org.apache.commons.lang3.SystemUtils;
+
+class DirectWriter implements LogWriter {
+ final NativeIO nativeIO;
+ final int fd;
+ final int id;
+ final String filename;
+ final BufferPool bufferPool;
+ final ExecutorService writeExecutor;
+ final Object bufferLock = new Object();
+ final List<Future<?>> outstandingWrites = new ArrayList<Future<?>>();
+ Buffer nativeBuffer;
+ long offset;
+ private volatile boolean useFallocate = true;
+
+ DirectWriter(int id,
+ String filename,
+ long maxFileSize,
+ ExecutorService writeExecutor,
+ BufferPool bufferPool,
+ NativeIO nativeIO, Slogger slog) throws IOException {
+ checkArgument(maxFileSize > 0, "Max file size (%d) must be positive");
+ this.id = id;
+ this.filename = filename;
+ this.writeExecutor = writeExecutor;
+ this.nativeIO = nativeIO;
+
+ offset = 0;
+
+ try {
+ fd = nativeIO.open(filename,
+ NativeIO.O_CREAT | NativeIO.O_WRONLY | NativeIO.O_DIRECT,
+ 00755);
Review Comment:
`00755` as in permissions? why 755 and not e.g. 600 or 640?
"755 means read and execute access for everyone and also write access for the owner of the file."
Why are we granting execute access to this file?
Why read access is needed to everyone?
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/directentrylogger/DirectWriter.java:
##########
@@ -0,0 +1,322 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.bookkeeper.bookie.storage.directentrylogger;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.bookkeeper.common.util.ExceptionMessageHelper.exMsg;
+
+import io.netty.buffer.ByteBuf;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+
+import org.apache.bookkeeper.common.util.nativeio.NativeIO;
+import org.apache.bookkeeper.common.util.nativeio.NativeIOException;
+import org.apache.bookkeeper.slogger.Slogger;
+import org.apache.commons.lang3.SystemUtils;
+
+class DirectWriter implements LogWriter {
+ final NativeIO nativeIO;
+ final int fd;
+ final int id;
+ final String filename;
+ final BufferPool bufferPool;
+ final ExecutorService writeExecutor;
+ final Object bufferLock = new Object();
+ final List<Future<?>> outstandingWrites = new ArrayList<Future<?>>();
+ Buffer nativeBuffer;
+ long offset;
+ private volatile boolean useFallocate = true;
+
+ DirectWriter(int id,
+ String filename,
+ long maxFileSize,
+ ExecutorService writeExecutor,
+ BufferPool bufferPool,
+ NativeIO nativeIO, Slogger slog) throws IOException {
+ checkArgument(maxFileSize > 0, "Max file size (%d) must be positive");
+ this.id = id;
+ this.filename = filename;
+ this.writeExecutor = writeExecutor;
+ this.nativeIO = nativeIO;
+
+ offset = 0;
+
+ try {
+ fd = nativeIO.open(filename,
+ NativeIO.O_CREAT | NativeIO.O_WRONLY | NativeIO.O_DIRECT,
+ 00755);
+ checkState(fd >= 0, "Open should have thrown exception, fd is invalid : %d", fd);
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg(ne.getMessage()).kv("file", filename)
+ .kv("errno", ne.getErrno()).toString(), ne);
+ }
+
+ if (useFallocate) {
+ if (!SystemUtils.IS_OS_LINUX) {
+ useFallocate = false;
+ slog.warn(Events.FALLOCATE_NOT_AVAILABLE);
+ } else {
+ try {
+ int ret = nativeIO.fallocate(fd, NativeIO.FALLOC_FL_ZERO_RANGE, 0, maxFileSize);
+ checkState(ret == 0, "Exception should have been thrown on non-zero ret: %d", ret);
+ } catch (NativeIOException ex) {
+ // fallocate(2) is not supported on all filesystems. Since this is an optimization, disable
+ // subsequent usage instead of failing the operation.
+ useFallocate = false;
+ slog.kv("message", ex.getMessage())
+ .kv("file", filename)
+ .kv("errno", ex.getErrno())
+ .warn(Events.FALLOCATE_NOT_AVAILABLE);
+ }
+ }
+ }
+
+ this.bufferPool = bufferPool;
+ this.nativeBuffer = bufferPool.acquire();
+ }
+
+ @Override
+ public int logId() {
+ return id;
+ }
+
+ @Override
+ public void writeAt(long offset, ByteBuf buf) throws IOException {
+ checkArgument(Buffer.isAligned(offset),
+ "Offset to writeAt must be aligned to %d: %d is not", Buffer.ALIGNMENT, offset);
+ checkArgument(Buffer.isAligned(buf.readableBytes()),
+ "Buffer must write multiple of alignment bytes (%d), %d is not",
+ Buffer.ALIGNMENT, buf.readableBytes());
+ Buffer tmpBuffer = bufferPool.acquire();
+ int bytesToWrite = buf.readableBytes();
+ tmpBuffer.reset();
+ tmpBuffer.writeByteBuf(buf);
+ Future<?> f = writeExecutor.submit(() -> {
+ try {
+ int ret = nativeIO.pwrite(fd, tmpBuffer.pointer(), bytesToWrite, offset);
+ if (ret != bytesToWrite) {
+ throw new IOException(exMsg("Incomplete write")
+ .kv("filename", filename)
+ .kv("writeSize", bytesToWrite)
+ .kv("bytesWritten", ret)
+ .kv("offset", offset).toString());
+ }
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg("Write error")
+ .kv("filename", filename)
+ .kv("writeSize", bytesToWrite)
+ .kv("errno", ne.getErrno())
+ .kv("offset", offset).toString());
+ } finally {
+ bufferPool.release(tmpBuffer);
+ }
+ return null;
+ });
+ addOutstandingWrite(f);
+ }
+
+ @Override
+ public int writeDelimited(ByteBuf buf) throws IOException {
+ synchronized (bufferLock) {
+ if (!nativeBuffer.hasSpace(serializedSize(buf))) {
+ flushBuffer();
+ }
+
+ int readable = buf.readableBytes();
+ long bufferPosition = position() + Integer.BYTES;
+ if (bufferPosition > Integer.MAX_VALUE) {
+ throw new IOException(exMsg("Cannot write past max int")
+ .kv("filename", filename)
+ .kv("writeSize", readable)
+ .kv("position", bufferPosition)
+ .toString());
+ }
+ nativeBuffer.writeInt(readable);
+ nativeBuffer.writeByteBuf(buf);
+ return (int) bufferPosition;
+ }
+ }
+
+ @Override
+ public void position(long offset) throws IOException {
+ synchronized (bufferLock) {
+ if (nativeBuffer != null && nativeBuffer.position() > 0) {
+ flushBuffer();
+ }
+ if ((offset % Buffer.ALIGNMENT) != 0) {
+ throw new IOException(exMsg("offset must be multiple of alignment")
+ .kv("offset", offset)
+ .kv("alignment", Buffer.ALIGNMENT)
+ .toString());
+ }
+ this.offset = offset;
+ }
+ }
+
+ @Override
+ public long position() {
+ synchronized (bufferLock) {
+ return this.offset + (nativeBuffer != null ? nativeBuffer.position() : 0);
+ }
+ }
+
+ @Override
+ public void flush() throws IOException {
+ flushBuffer();
+
+ waitForOutstandingWrites();
+
+ try {
+ int ret = nativeIO.fsync(fd);
+ checkState(ret == 0, "Fsync should throw exception on non-zero return (%d)", ret);
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg(ne.getMessage())
+ .kv("file", filename)
+ .kv("errno", ne.getErrno()).toString());
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ synchronized (bufferLock) {
+ if (nativeBuffer != null && nativeBuffer.position() > 0) {
+ flushBuffer();
+ }
+ }
+
Review Comment:
I think you should do `waitForOutstandingWrites();` here or call `flush()` instead of `flushBuffer()` to confirm that the last outstanding write (submitted e.g. by `flushBuffer()`) actually succeeded.
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/EntryLogIdsImpl.java:
##########
@@ -0,0 +1,157 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.bookkeeper.bookie.storage;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.apache.bookkeeper.bookie.LedgerDirsManager;
+import org.apache.bookkeeper.bookie.storage.directentrylogger.Events;
+import org.apache.bookkeeper.slogger.Slogger;
+
+/**
+ * EntryLogIdsImpl.
+ */
+public class EntryLogIdsImpl implements EntryLogIds {
+ public static final Pattern FILE_PATTERN = Pattern.compile("^([0-9a-fA-F]+)\\.log$");
+ public static final Pattern COMPACTED_FILE_PATTERN =
+ Pattern.compile("^([0-9a-fA-F]+)\\.log\\.([0-9a-fA-F]+)\\.compacted$");
+
+ private final LedgerDirsManager ledgerDirsManager;
+ private final Slogger slog;
+ private int nextId;
+ private int maxId;
+
+ public EntryLogIdsImpl(LedgerDirsManager ledgerDirsManager,
+ Slogger slog) throws IOException {
+ this.ledgerDirsManager = ledgerDirsManager;
+ this.slog = slog;
+ findLargestGap();
+ }
+
+ @Override
+ public int nextId() throws IOException {
+ while (true) {
+ synchronized (this) {
+ int current = nextId;
+ nextId++;
+ if (nextId == maxId) {
+ findLargestGap();
+ } else {
+ return current;
+ }
+ }
+ }
+ }
+
+ private void findLargestGap() throws IOException {
+ long start = System.nanoTime();
+ List<Integer> currentIds = new ArrayList<Integer>();
+
+ for (File ledgerDir : ledgerDirsManager.getAllLedgerDirs()) {
+ currentIds.addAll(logIdsInDirectory(ledgerDir));
+ currentIds.addAll(compactedLogIdsInDirectory(ledgerDir));
+ }
+
+ int[] gap = findLargestGap(currentIds);
+ nextId = gap[0];
+ maxId = gap[1];
+ slog.kv("dirs", ledgerDirsManager.getAllLedgerDirs())
+ .kv("nextId", nextId)
+ .kv("maxId", maxId)
+ .kv("durationMs", TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start))
+ .info(Events.ENTRYLOG_IDS_CANDIDATES_SELECTED);
+ }
+
+ /**
+ * O(nlogn) algorithm to find largest contiguous gap between
+ * integers in a passed list. n should be relatively small.
+ * Entry logs should be about 1GB in size, so even if the node
+ * stores a PB, there should be only 1000000 entry logs.
+ */
+ static int[] findLargestGap(List<Integer> currentIds) {
Review Comment:
use `Pair<>` instead of int[] because that's what it is
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/directentrylogger/DirectWriter.java:
##########
@@ -0,0 +1,322 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.bookkeeper.bookie.storage.directentrylogger;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.bookkeeper.common.util.ExceptionMessageHelper.exMsg;
+
+import io.netty.buffer.ByteBuf;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+
+import org.apache.bookkeeper.common.util.nativeio.NativeIO;
+import org.apache.bookkeeper.common.util.nativeio.NativeIOException;
+import org.apache.bookkeeper.slogger.Slogger;
+import org.apache.commons.lang3.SystemUtils;
+
+class DirectWriter implements LogWriter {
+ final NativeIO nativeIO;
+ final int fd;
+ final int id;
+ final String filename;
+ final BufferPool bufferPool;
+ final ExecutorService writeExecutor;
+ final Object bufferLock = new Object();
+ final List<Future<?>> outstandingWrites = new ArrayList<Future<?>>();
+ Buffer nativeBuffer;
+ long offset;
+ private volatile boolean useFallocate = true;
+
+ DirectWriter(int id,
+ String filename,
+ long maxFileSize,
+ ExecutorService writeExecutor,
+ BufferPool bufferPool,
+ NativeIO nativeIO, Slogger slog) throws IOException {
+ checkArgument(maxFileSize > 0, "Max file size (%d) must be positive");
+ this.id = id;
+ this.filename = filename;
+ this.writeExecutor = writeExecutor;
+ this.nativeIO = nativeIO;
+
+ offset = 0;
+
+ try {
+ fd = nativeIO.open(filename,
+ NativeIO.O_CREAT | NativeIO.O_WRONLY | NativeIO.O_DIRECT,
+ 00755);
+ checkState(fd >= 0, "Open should have thrown exception, fd is invalid : %d", fd);
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg(ne.getMessage()).kv("file", filename)
+ .kv("errno", ne.getErrno()).toString(), ne);
+ }
+
+ if (useFallocate) {
+ if (!SystemUtils.IS_OS_LINUX) {
+ useFallocate = false;
+ slog.warn(Events.FALLOCATE_NOT_AVAILABLE);
+ } else {
+ try {
+ int ret = nativeIO.fallocate(fd, NativeIO.FALLOC_FL_ZERO_RANGE, 0, maxFileSize);
+ checkState(ret == 0, "Exception should have been thrown on non-zero ret: %d", ret);
+ } catch (NativeIOException ex) {
+ // fallocate(2) is not supported on all filesystems. Since this is an optimization, disable
+ // subsequent usage instead of failing the operation.
+ useFallocate = false;
+ slog.kv("message", ex.getMessage())
+ .kv("file", filename)
+ .kv("errno", ex.getErrno())
+ .warn(Events.FALLOCATE_NOT_AVAILABLE);
+ }
+ }
+ }
+
+ this.bufferPool = bufferPool;
+ this.nativeBuffer = bufferPool.acquire();
+ }
+
+ @Override
+ public int logId() {
+ return id;
+ }
+
+ @Override
+ public void writeAt(long offset, ByteBuf buf) throws IOException {
+ checkArgument(Buffer.isAligned(offset),
+ "Offset to writeAt must be aligned to %d: %d is not", Buffer.ALIGNMENT, offset);
+ checkArgument(Buffer.isAligned(buf.readableBytes()),
+ "Buffer must write multiple of alignment bytes (%d), %d is not",
+ Buffer.ALIGNMENT, buf.readableBytes());
+ Buffer tmpBuffer = bufferPool.acquire();
+ int bytesToWrite = buf.readableBytes();
+ tmpBuffer.reset();
+ tmpBuffer.writeByteBuf(buf);
+ Future<?> f = writeExecutor.submit(() -> {
+ try {
+ int ret = nativeIO.pwrite(fd, tmpBuffer.pointer(), bytesToWrite, offset);
+ if (ret != bytesToWrite) {
+ throw new IOException(exMsg("Incomplete write")
+ .kv("filename", filename)
+ .kv("writeSize", bytesToWrite)
+ .kv("bytesWritten", ret)
+ .kv("offset", offset).toString());
+ }
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg("Write error")
+ .kv("filename", filename)
+ .kv("writeSize", bytesToWrite)
+ .kv("errno", ne.getErrno())
+ .kv("offset", offset).toString());
+ } finally {
+ bufferPool.release(tmpBuffer);
+ }
+ return null;
+ });
+ addOutstandingWrite(f);
+ }
+
+ @Override
+ public int writeDelimited(ByteBuf buf) throws IOException {
+ synchronized (bufferLock) {
+ if (!nativeBuffer.hasSpace(serializedSize(buf))) {
+ flushBuffer();
+ }
+
+ int readable = buf.readableBytes();
+ long bufferPosition = position() + Integer.BYTES;
+ if (bufferPosition > Integer.MAX_VALUE) {
+ throw new IOException(exMsg("Cannot write past max int")
+ .kv("filename", filename)
+ .kv("writeSize", readable)
+ .kv("position", bufferPosition)
+ .toString());
+ }
+ nativeBuffer.writeInt(readable);
+ nativeBuffer.writeByteBuf(buf);
+ return (int) bufferPosition;
+ }
+ }
+
+ @Override
+ public void position(long offset) throws IOException {
+ synchronized (bufferLock) {
+ if (nativeBuffer != null && nativeBuffer.position() > 0) {
+ flushBuffer();
+ }
+ if ((offset % Buffer.ALIGNMENT) != 0) {
+ throw new IOException(exMsg("offset must be multiple of alignment")
+ .kv("offset", offset)
+ .kv("alignment", Buffer.ALIGNMENT)
+ .toString());
+ }
+ this.offset = offset;
+ }
+ }
+
+ @Override
+ public long position() {
+ synchronized (bufferLock) {
+ return this.offset + (nativeBuffer != null ? nativeBuffer.position() : 0);
+ }
+ }
+
+ @Override
+ public void flush() throws IOException {
+ flushBuffer();
+
+ waitForOutstandingWrites();
+
+ try {
+ int ret = nativeIO.fsync(fd);
+ checkState(ret == 0, "Fsync should throw exception on non-zero return (%d)", ret);
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg(ne.getMessage())
+ .kv("file", filename)
+ .kv("errno", ne.getErrno()).toString());
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ synchronized (bufferLock) {
+ if (nativeBuffer != null && nativeBuffer.position() > 0) {
+ flushBuffer();
+ }
+ }
+
+ try {
+ int ret = nativeIO.close(fd);
+ checkState(ret == 0, "Close should throw exception on non-zero return (%d)", ret);
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg(ne.getMessage())
+ .kv("file", filename)
+ .kv("errno", ne.getErrno()).toString());
+ }
+ synchronized (bufferLock) {
+ bufferPool.release(nativeBuffer);
+ nativeBuffer = null;
+ }
+ }
+
+ private void addOutstandingWrite(Future<?> toAdd) throws IOException {
+ synchronized (outstandingWrites) {
+ outstandingWrites.add(toAdd);
Review Comment:
There is no limit on number of outstandingWrites.
Typically we start with that and end up adding some kind of limiter or throttler to prevent system overload.
Does it make sense to do it right now?
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/directentrylogger/DirectWriter.java:
##########
@@ -0,0 +1,322 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.bookkeeper.bookie.storage.directentrylogger;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.bookkeeper.common.util.ExceptionMessageHelper.exMsg;
+
+import io.netty.buffer.ByteBuf;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+
+import org.apache.bookkeeper.common.util.nativeio.NativeIO;
+import org.apache.bookkeeper.common.util.nativeio.NativeIOException;
+import org.apache.bookkeeper.slogger.Slogger;
+import org.apache.commons.lang3.SystemUtils;
+
+class DirectWriter implements LogWriter {
+ final NativeIO nativeIO;
+ final int fd;
+ final int id;
+ final String filename;
+ final BufferPool bufferPool;
+ final ExecutorService writeExecutor;
+ final Object bufferLock = new Object();
+ final List<Future<?>> outstandingWrites = new ArrayList<Future<?>>();
Review Comment:
can it be ConcurrentLinkedQueue to avoid synchronization later?
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/directentrylogger/DirectWriter.java:
##########
@@ -0,0 +1,322 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.bookkeeper.bookie.storage.directentrylogger;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.bookkeeper.common.util.ExceptionMessageHelper.exMsg;
+
+import io.netty.buffer.ByteBuf;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+
+import org.apache.bookkeeper.common.util.nativeio.NativeIO;
+import org.apache.bookkeeper.common.util.nativeio.NativeIOException;
+import org.apache.bookkeeper.slogger.Slogger;
+import org.apache.commons.lang3.SystemUtils;
+
+class DirectWriter implements LogWriter {
+ final NativeIO nativeIO;
+ final int fd;
+ final int id;
+ final String filename;
+ final BufferPool bufferPool;
+ final ExecutorService writeExecutor;
+ final Object bufferLock = new Object();
+ final List<Future<?>> outstandingWrites = new ArrayList<Future<?>>();
+ Buffer nativeBuffer;
+ long offset;
+ private volatile boolean useFallocate = true;
+
+ DirectWriter(int id,
+ String filename,
+ long maxFileSize,
+ ExecutorService writeExecutor,
+ BufferPool bufferPool,
+ NativeIO nativeIO, Slogger slog) throws IOException {
+ checkArgument(maxFileSize > 0, "Max file size (%d) must be positive");
+ this.id = id;
+ this.filename = filename;
+ this.writeExecutor = writeExecutor;
+ this.nativeIO = nativeIO;
+
+ offset = 0;
+
+ try {
+ fd = nativeIO.open(filename,
+ NativeIO.O_CREAT | NativeIO.O_WRONLY | NativeIO.O_DIRECT,
+ 00755);
+ checkState(fd >= 0, "Open should have thrown exception, fd is invalid : %d", fd);
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg(ne.getMessage()).kv("file", filename)
+ .kv("errno", ne.getErrno()).toString(), ne);
+ }
+
+ if (useFallocate) {
+ if (!SystemUtils.IS_OS_LINUX) {
+ useFallocate = false;
+ slog.warn(Events.FALLOCATE_NOT_AVAILABLE);
+ } else {
+ try {
+ int ret = nativeIO.fallocate(fd, NativeIO.FALLOC_FL_ZERO_RANGE, 0, maxFileSize);
+ checkState(ret == 0, "Exception should have been thrown on non-zero ret: %d", ret);
+ } catch (NativeIOException ex) {
+ // fallocate(2) is not supported on all filesystems. Since this is an optimization, disable
+ // subsequent usage instead of failing the operation.
+ useFallocate = false;
+ slog.kv("message", ex.getMessage())
+ .kv("file", filename)
+ .kv("errno", ex.getErrno())
+ .warn(Events.FALLOCATE_NOT_AVAILABLE);
+ }
+ }
+ }
+
+ this.bufferPool = bufferPool;
+ this.nativeBuffer = bufferPool.acquire();
+ }
+
+ @Override
+ public int logId() {
+ return id;
+ }
+
+ @Override
+ public void writeAt(long offset, ByteBuf buf) throws IOException {
+ checkArgument(Buffer.isAligned(offset),
+ "Offset to writeAt must be aligned to %d: %d is not", Buffer.ALIGNMENT, offset);
+ checkArgument(Buffer.isAligned(buf.readableBytes()),
+ "Buffer must write multiple of alignment bytes (%d), %d is not",
+ Buffer.ALIGNMENT, buf.readableBytes());
+ Buffer tmpBuffer = bufferPool.acquire();
+ int bytesToWrite = buf.readableBytes();
+ tmpBuffer.reset();
+ tmpBuffer.writeByteBuf(buf);
+ Future<?> f = writeExecutor.submit(() -> {
+ try {
+ int ret = nativeIO.pwrite(fd, tmpBuffer.pointer(), bytesToWrite, offset);
+ if (ret != bytesToWrite) {
+ throw new IOException(exMsg("Incomplete write")
+ .kv("filename", filename)
+ .kv("writeSize", bytesToWrite)
+ .kv("bytesWritten", ret)
+ .kv("offset", offset).toString());
+ }
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg("Write error")
+ .kv("filename", filename)
+ .kv("writeSize", bytesToWrite)
+ .kv("errno", ne.getErrno())
+ .kv("offset", offset).toString());
+ } finally {
+ bufferPool.release(tmpBuffer);
+ }
+ return null;
+ });
+ addOutstandingWrite(f);
+ }
+
+ @Override
+ public int writeDelimited(ByteBuf buf) throws IOException {
+ synchronized (bufferLock) {
+ if (!nativeBuffer.hasSpace(serializedSize(buf))) {
+ flushBuffer();
+ }
+
+ int readable = buf.readableBytes();
+ long bufferPosition = position() + Integer.BYTES;
+ if (bufferPosition > Integer.MAX_VALUE) {
+ throw new IOException(exMsg("Cannot write past max int")
+ .kv("filename", filename)
+ .kv("writeSize", readable)
+ .kv("position", bufferPosition)
+ .toString());
+ }
+ nativeBuffer.writeInt(readable);
+ nativeBuffer.writeByteBuf(buf);
+ return (int) bufferPosition;
+ }
+ }
+
+ @Override
+ public void position(long offset) throws IOException {
+ synchronized (bufferLock) {
+ if (nativeBuffer != null && nativeBuffer.position() > 0) {
+ flushBuffer();
+ }
+ if ((offset % Buffer.ALIGNMENT) != 0) {
+ throw new IOException(exMsg("offset must be multiple of alignment")
+ .kv("offset", offset)
+ .kv("alignment", Buffer.ALIGNMENT)
+ .toString());
+ }
+ this.offset = offset;
+ }
+ }
+
+ @Override
+ public long position() {
+ synchronized (bufferLock) {
+ return this.offset + (nativeBuffer != null ? nativeBuffer.position() : 0);
+ }
+ }
+
+ @Override
+ public void flush() throws IOException {
+ flushBuffer();
+
+ waitForOutstandingWrites();
+
+ try {
+ int ret = nativeIO.fsync(fd);
+ checkState(ret == 0, "Fsync should throw exception on non-zero return (%d)", ret);
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg(ne.getMessage())
+ .kv("file", filename)
+ .kv("errno", ne.getErrno()).toString());
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ synchronized (bufferLock) {
+ if (nativeBuffer != null && nativeBuffer.position() > 0) {
+ flushBuffer();
+ }
+ }
+
+ try {
+ int ret = nativeIO.close(fd);
+ checkState(ret == 0, "Close should throw exception on non-zero return (%d)", ret);
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg(ne.getMessage())
+ .kv("file", filename)
+ .kv("errno", ne.getErrno()).toString());
+ }
+ synchronized (bufferLock) {
+ bufferPool.release(nativeBuffer);
+ nativeBuffer = null;
+ }
+ }
+
+ private void addOutstandingWrite(Future<?> toAdd) throws IOException {
+ synchronized (outstandingWrites) {
+ outstandingWrites.add(toAdd);
+
+ Iterator<Future<?>> iter = outstandingWrites.iterator();
+ while (iter.hasNext()) { // clear out completed futures
+ Future<?> f = iter.next();
+ if (f.isDone()) {
+ waitForFuture(f);
+ iter.remove();
+ } else {
+ break;
+ }
+ }
+ }
+ }
+
+ private void waitForOutstandingWrites() throws IOException {
+ synchronized (outstandingWrites) {
+ Iterator<Future<?>> iter = outstandingWrites.iterator();
+ while (iter.hasNext()) { // clear out completed futures
+ Future<?> f = iter.next();
+ waitForFuture(f);
+ iter.remove();
+ }
+ }
+ }
+
+ private void waitForFuture(Future<?> f) throws IOException {
+ try {
+ f.get();
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ throw new IOException(ie);
+ } catch (Throwable t) {
+ if (t.getCause() instanceof IOException) {
+ throw (IOException) t.getCause();
+ } else {
+ throw new IOException(t);
+ }
+ }
+ }
+
+ private void flushBuffer() throws IOException {
+ synchronized (bufferLock) {
+ if (this.nativeBuffer != null) {
+ int bytesToWrite = this.nativeBuffer.padToAlignment();
+ if (bytesToWrite == 0) {
+ return;
+ }
+
+ Buffer bufferToFlush = this.nativeBuffer;
+ this.nativeBuffer = null;
+
+ long offsetToWrite = offset;
+ offset += bytesToWrite;
+
+ Future<?> f = writeExecutor.submit(() -> {
+ try {
+ if (bytesToWrite <= 0) {
+ return null;
+ }
+ int ret = nativeIO.pwrite(fd, bufferToFlush.pointer(), bytesToWrite, offsetToWrite);
+ if (ret != bytesToWrite) {
+ throw new IOException(exMsg("Incomplete write")
+ .kv("filename", filename)
+ .kv("pointer", bufferToFlush.pointer())
+ .kv("offset", offsetToWrite)
+ .kv("writeSize", bytesToWrite)
+ .kv("bytesWritten", ret).toString());
+ }
+ } catch (NativeIOException ne) {
+ throw new IOException(exMsg(ne.getMessage())
+ .kv("filename", filename)
+ .kv("offset", offsetToWrite)
+ .kv("writeSize", bytesToWrite)
+ .kv("pointer", bufferToFlush.pointer())
+ .kv("errno", ne.getErrno()).toString(), ne);
+ } finally {
+ bufferPool.release(bufferToFlush);
+ }
+ return null;
Review Comment:
same as lines 120+? consider refactoring
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [bookkeeper] hangc0276 commented on pull request #3263: BP-47 (task6): Direct I/O entrylogger support
Posted by GitBox <gi...@apache.org>.
hangc0276 commented on PR #3263:
URL: https://github.com/apache/bookkeeper/pull/3263#issuecomment-1163790241
rerun failure checks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [bookkeeper] hangc0276 commented on a diff in pull request #3263: BP-47 (task6): Direct I/O entrylogger support
Posted by GitBox <gi...@apache.org>.
hangc0276 commented on code in PR #3263:
URL: https://github.com/apache/bookkeeper/pull/3263#discussion_r901056980
##########
bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/directentrylogger/TestDirectEntryLoggerCompat.java:
##########
@@ -0,0 +1,651 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.bookkeeper.bookie.storage.directentrylogger;
+
+import static org.apache.bookkeeper.bookie.storage.EntryLogTestUtils.assertEntryEquals;
+import static org.apache.bookkeeper.bookie.storage.EntryLogTestUtils.logIdFromLocation;
+import static org.apache.bookkeeper.bookie.storage.EntryLogTestUtils.makeEntry;
+import static org.apache.bookkeeper.bookie.storage.EntryLogTestUtils.newLegacyEntryLogger;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.not;
+
+import com.google.common.util.concurrent.MoreExecutors;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.Unpooled;
+import java.io.File;
+import java.nio.charset.StandardCharsets;
+import org.apache.bookkeeper.bookie.DefaultEntryLogger;
+import org.apache.bookkeeper.bookie.EntryLogMetadata;
+import org.apache.bookkeeper.bookie.LedgerDirsManager;
+import org.apache.bookkeeper.bookie.storage.EntryLogger;
+import org.apache.bookkeeper.bookie.storage.MockEntryLogIds;
+import org.apache.bookkeeper.common.util.nativeio.NativeIOImpl;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.slogger.Slogger;
+import org.apache.bookkeeper.stats.NullStatsLogger;
+import org.apache.bookkeeper.test.TmpDirs;
+import org.apache.bookkeeper.util.DiskChecker;
+import org.junit.After;
+import org.junit.Test;
+
+/**
+ * TestDirectEntryLoggerCompat.
+ */
+public class TestDirectEntryLoggerCompat {
+ private final Slogger slog = Slogger.CONSOLE;
+
+ private static final long ledgerId1 = 1234;
+ private static final long ledgerId2 = 4567;
+ private static final long ledgerId3 = 7890;
+
+ private final TmpDirs tmpDirs = new TmpDirs();
+
+ @After
+ public void cleanup() throws Exception {
+ tmpDirs.cleanup();
+ }
+
+ @Test
+ public void testLegacyCanReadDirect() throws Exception {
+ File ledgerDir = tmpDirs.createNew("legacyCanRead", "ledgers");
+ File curDir = new File(ledgerDir, "current");
+ curDir.mkdirs();
+
+ ByteBuf e1 = makeEntry(ledgerId1, 1L, 1000);
+ ByteBuf e2 = makeEntry(ledgerId1, 2L, 1000);
+ ByteBuf e3 = makeEntry(ledgerId1, 3L, 1000);
+
+ long loc1, loc2, loc3;
+ try (EntryLogger elog = new DirectEntryLogger(
+ curDir, new MockEntryLogIds(),
+ new NativeIOImpl(),
+ ByteBufAllocator.DEFAULT,
+ MoreExecutors.newDirectExecutorService(),
+ MoreExecutors.newDirectExecutorService(),
+ 10 * 1024 * 1024, // 10MiB, max file size
+ 10 * 1024 * 1024, // max sane entry size
+ 1024 * 1024, // total write buffer size
+ 1024 * 1024, // total read buffer size
+ 64 * 1024, // read buffer size
+ 1, // numReadThreads
+ 300, // max fd cache time in seconds
+ slog, NullStatsLogger.INSTANCE)) {
+ loc1 = elog.addEntry(ledgerId1, e1.slice());
+ loc2 = elog.addEntry(ledgerId1, e2.slice());
+ loc3 = elog.addEntry(ledgerId1, e3.slice());
+ }
+
+ try (EntryLogger legacy = newLegacyEntryLogger(2000000, ledgerDir)) {
+ assertEntryEquals(legacy.readEntry(ledgerId1, 1L, loc1), e1);
+ assertEntryEquals(legacy.readEntry(ledgerId1, 2L, loc2), e2);
+ assertEntryEquals(legacy.readEntry(ledgerId1, 3L, loc3), e3);
+ }
+ }
+
+ @Test
+ public void testDirectCanReadLegacy() throws Exception {
+ File ledgerDir = tmpDirs.createNew("legacyCanRead", "ledgers");
+
+ ByteBuf e1 = makeEntry(ledgerId1, 1L, 1000);
+ ByteBuf e2 = makeEntry(ledgerId1, 2L, 1000);
+ ByteBuf e3 = makeEntry(ledgerId1, 3L, 1000);
+
+ long loc1, loc2, loc3;
+ try (EntryLogger legacy = newLegacyEntryLogger(2000000, ledgerDir)) {
+ loc1 = legacy.addEntry(ledgerId1, e1.slice());
+ loc2 = legacy.addEntry(ledgerId1, e2.slice());
+ loc3 = legacy.addEntry(ledgerId1, e3.slice());
+ legacy.flush();
+ }
+
+ try (EntryLogger elog = new DirectEntryLogger(
+ new File(ledgerDir, "current"), new MockEntryLogIds(),
+ new NativeIOImpl(),
+ ByteBufAllocator.DEFAULT,
+ MoreExecutors.newDirectExecutorService(),
+ MoreExecutors.newDirectExecutorService(),
+ 10 * 1024 * 1024, // 10MiB, max file size
+ 10 * 1024 * 1024, // max sane entry size
+ 1024 * 1024, // total write buffer size
+ 1024 * 1024, // total read buffer size
+ 64 * 1024, // read buffer size
+ 1, // numReadThreads
+ 300, // max fd cache time in seconds
+ slog, NullStatsLogger.INSTANCE)) {
+ assertEntryEquals(elog.readEntry(ledgerId1, 1L, loc1), e1);
+ assertEntryEquals(elog.readEntry(ledgerId1, 2L, loc2), e2);
+ assertEntryEquals(elog.readEntry(ledgerId1, 3L, loc3), e3);
+ }
+ }
+
+ @Test
+ public void testLegacyCanReadDirectAfterMultipleRolls() throws Exception {
+ File ledgerDir = tmpDirs.createNew("legacyCanRead", "ledgers");
+ File curDir = new File(ledgerDir, "current");
+ curDir.mkdirs();
+
+ ByteBuf e1 = makeEntry(ledgerId1, 1L, 4000);
+ ByteBuf e2 = makeEntry(ledgerId1, 2L, 4000);
+ ByteBuf e3 = makeEntry(ledgerId1, 3L, 4000);
+
+ long loc1, loc2, loc3;
+ try (EntryLogger elog = new DirectEntryLogger(
+ curDir, new MockEntryLogIds(),
+ new NativeIOImpl(),
+ ByteBufAllocator.DEFAULT,
+ MoreExecutors.newDirectExecutorService(),
+ MoreExecutors.newDirectExecutorService(),
+ 6000, // max file size
+ 10 * 1024 * 1024, // max sane entry size
+ 1024 * 1024, // total write buffer size
+ 1024 * 1024, // total read buffer size
+ 64 * 1024, // read buffer size
+ 1, // numReadThreads
+ 300, // max fd cache time in seconds
+ slog, NullStatsLogger.INSTANCE)) {
+ loc1 = elog.addEntry(ledgerId1, e1.slice());
+ loc2 = elog.addEntry(ledgerId1, e2.slice());
+ loc3 = elog.addEntry(ledgerId1, e3.slice());
+ }
+
+ try (EntryLogger legacy = newLegacyEntryLogger(2000000, ledgerDir)) {
+ assertEntryEquals(legacy.readEntry(ledgerId1, 1L, loc1), e1);
+ assertEntryEquals(legacy.readEntry(ledgerId1, 2L, loc2), e2);
+ assertEntryEquals(legacy.readEntry(ledgerId1, 3L, loc3), e3);
+ }
+ }
+
+ @Test
+ public void testLegacyCanReadMetadataOfDirectWithIndexWritten() throws Exception {
+ File ledgerDir = tmpDirs.createNew("legacyCanReadMeta", "ledgers");
+ File curDir = new File(ledgerDir, "current");
+ curDir.mkdirs();
+
+ ByteBuf e1 = makeEntry(ledgerId1, 1L, 1000);
+ ByteBuf e2 = makeEntry(ledgerId2, 2L, 2000);
+ ByteBuf e3 = makeEntry(ledgerId1, 3L, 3000);
+ ByteBuf e4 = makeEntry(ledgerId1, 4L, 4000);
+
+ int maxFileSize = 1000 + 2000 + 3000 + (Integer.BYTES * 3) + 4096;
+ long loc1, loc2, loc3, loc4;
+ try (EntryLogger elog = new DirectEntryLogger(
+ curDir, new MockEntryLogIds(),
+ new NativeIOImpl(),
+ ByteBufAllocator.DEFAULT,
+ MoreExecutors.newDirectExecutorService(),
+ MoreExecutors.newDirectExecutorService(),
+ maxFileSize, // max file size
+ 10 * 1024 * 1024, // max sane entry size
+ 1024 * 1024, // total write buffer size
+ 1024 * 1024, // total read buffer size
+ 64 * 1024, // read buffer size
+ 1, // numReadThreads
+ 300, // max fd cache time in seconds
+ slog, NullStatsLogger.INSTANCE)) {
+ loc1 = elog.addEntry(ledgerId1, e1);
+ loc2 = elog.addEntry(ledgerId2, e2);
+ loc3 = elog.addEntry(ledgerId1, e3);
+ loc4 = elog.addEntry(ledgerId1, e4);
+ }
+
+ try (EntryLogger legacy = newLegacyEntryLogger(
+ maxFileSize, // size of first 3 entries + header
+ ledgerDir)) {
+ int logId = logIdFromLocation(loc1);
+ assertThat(logId, equalTo(logIdFromLocation(loc2)));
+ assertThat(logId, equalTo(logIdFromLocation(loc3)));
+ assertThat(logId, not(equalTo(logIdFromLocation(loc4))));
+
+ EntryLogMetadata meta = legacy.getEntryLogMetadata(logId);
+
+ assertThat(meta.getEntryLogId(), equalTo((long) logId));
+ assertThat(meta.getTotalSize(), equalTo(1000L + 2000 + 3000 + (Integer.BYTES * 3)));
+ assertThat(meta.getRemainingSize(), equalTo(meta.getTotalSize()));
+ assertThat(meta.getLedgersMap().get(ledgerId1), equalTo(1000L + 3000L + (Integer.BYTES * 2)));
+ assertThat(meta.getLedgersMap().get(ledgerId2), equalTo(2000L + Integer.BYTES));
+ }
+ }
+
+ @Test
+ public void testLegacyCanReadMetadataOfDirectWithNoIndexWritten() throws Exception {
+ File ledgerDir = tmpDirs.createNew("legacyCanReadMeta", "ledgers");
+ File curDir = new File(ledgerDir, "current");
+ curDir.mkdirs();
+
+ ByteBuf e1 = makeEntry(ledgerId1, 1L, 1000);
+ ByteBuf e2 = makeEntry(ledgerId2, 2L, 2000);
+ ByteBuf e3 = makeEntry(ledgerId1, 3L, 3000);
+ ByteBuf e4 = makeEntry(ledgerId1, 4L, 4000);
+
+ int maxFileSize = 1000 + 2000 + 3000 + (Integer.BYTES * 3) + 4096;
+ long loc1, loc2, loc3;
+ try (EntryLogger elog = new DirectEntryLogger(
+ curDir, new MockEntryLogIds(),
+ new NativeIOImpl(),
+ ByteBufAllocator.DEFAULT,
+ MoreExecutors.newDirectExecutorService(),
+ MoreExecutors.newDirectExecutorService(),
+ maxFileSize * 10, // max file size
+ 10 * 1024 * 1024, // max sane entry size
+ 1024 * 1024, // total write buffer size
+ 1024 * 1024, // total read buffer size
+ 64 * 1024, // read buffer size
+ 1, // numReadThreads
+ 300, // max fd cache time in seconds
+ slog, NullStatsLogger.INSTANCE)) {
+ loc1 = elog.addEntry(ledgerId1, e1);
+ loc2 = elog.addEntry(ledgerId2, e2);
+ loc3 = elog.addEntry(ledgerId1, e3);
+ }
+
+ try (EntryLogger legacy = newLegacyEntryLogger(
+ maxFileSize, // size of first 3 entries + header
+ ledgerDir)) {
+ int logId = logIdFromLocation(loc1);
+ assertThat(logId, equalTo(logIdFromLocation(loc2)));
+ assertThat(logId, equalTo(logIdFromLocation(loc3)));
+ EntryLogMetadata meta = legacy.getEntryLogMetadata(logId);
+
+ assertThat(meta.getEntryLogId(), equalTo((long) logId));
+ assertThat(meta.getTotalSize(), equalTo(1000L + 2000 + 3000 + (Integer.BYTES * 3)));
+ assertThat(meta.getRemainingSize(), equalTo(meta.getTotalSize()));
+ assertThat(meta.getLedgersMap().get(ledgerId1), equalTo(1000L + 3000L + (Integer.BYTES * 2)));
+ assertThat(meta.getLedgersMap().get(ledgerId2), equalTo(2000L + Integer.BYTES));
+ }
+ }
+
+ @Test
+ public void testDirectCanReadMetadataAndScanFromLegacy() throws Exception {
+ File ledgerDir = tmpDirs.createNew("directCanReadLegacyMeta", "ledgers");
+ File curDir = new File(ledgerDir, "current");
+ curDir.mkdirs();
+
+ ByteBuf e1 = makeEntry(ledgerId1, 1L, 1000);
+ ByteBuf e2 = makeEntry(ledgerId2, 2L, 2000);
+ ByteBuf e3 = makeEntry(ledgerId1, 3L, 3000);
+ ByteBuf e4 = makeEntry(ledgerId1, 4L, 4000);
+
+ int maxFileSize = 1000 + 2000 + 3000 + (Integer.BYTES * 3) + 4096;
+ long loc1, loc2, loc3, loc4;
+ try (EntryLogger legacy = newLegacyEntryLogger(
+ maxFileSize, // size of first 3 entries + header
+ ledgerDir)) {
+ loc1 = legacy.addEntry(ledgerId1, e1);
+ loc2 = legacy.addEntry(ledgerId2, e2);
+ loc3 = legacy.addEntry(ledgerId1, e3);
+ loc4 = legacy.addEntry(ledgerId1, e4); // should force a roll
+ }
+
+ try (DirectEntryLogger elog = new DirectEntryLogger(
+ curDir, new MockEntryLogIds(),
+ new NativeIOImpl(),
+ ByteBufAllocator.DEFAULT,
+ MoreExecutors.newDirectExecutorService(),
+ MoreExecutors.newDirectExecutorService(),
+ maxFileSize * 10, // max file size
+ 10 * 1024 * 1024, // max sane entry size
+ 1024 * 1024, // total write buffer size
+ 1024 * 1024, // total read buffer size
+ 64 * 1024, // read buffer size
+ 1, // numReadThreads
+ 300, // max fd cache time in seconds
+ slog, NullStatsLogger.INSTANCE)) {
+ int logId = logIdFromLocation(loc1);
+ assertThat(logId, equalTo(logIdFromLocation(loc2)));
+ assertThat(logId, equalTo(logIdFromLocation(loc3)));
+ assertThat(logId, not(equalTo(logIdFromLocation(loc4))));
+
+ EntryLogMetadata metaRead = elog.readEntryLogIndex(logId);
+ assertThat(metaRead.getEntryLogId(), equalTo((long) logId));
+ assertThat(metaRead.getTotalSize(), equalTo(1000L + 2000 + 3000 + (Integer.BYTES * 3)));
+ assertThat(metaRead.getRemainingSize(), equalTo(metaRead.getTotalSize()));
+ assertThat(metaRead.getLedgersMap().get(ledgerId1), equalTo(1000L + 3000L + (Integer.BYTES * 2)));
+ assertThat(metaRead.getLedgersMap().get(ledgerId2), equalTo(2000L + Integer.BYTES));
+
+ EntryLogMetadata metaScan = elog.scanEntryLogMetadata(logId, null);
+ assertThat(metaScan.getEntryLogId(), equalTo((long) logId));
+ assertThat(metaScan.getTotalSize(), equalTo(1000L + 2000 + 3000 + (Integer.BYTES * 3)));
+ assertThat(metaScan.getRemainingSize(), equalTo(metaScan.getTotalSize()));
+ assertThat(metaScan.getLedgersMap().get(ledgerId1), equalTo(1000L + 3000L + (Integer.BYTES * 2)));
+ assertThat(metaScan.getLedgersMap().get(ledgerId2), equalTo(2000L + Integer.BYTES));
+ }
+ }
+
+ // step1: default is DirectEntryLogger, write entries, read entries
+ // step2: change DirectEntryLogger to DefaultEntryLogger, write entries, and read all entries both written
+ // by DirectEntryLogger and DefaultEntryLogger
+ // step3: change DefaultEntryLogger to DirectEntryLogger, write entries, and read all entries written by
+ // DirectEntryLogger, DefaultEntryLogger and DirectEntryLogger.
+ // DirectEntryLogger -> DefaultEntryLogge -> DirectEntryLogger.
+ @Test
+ public void testCompatFromDirectToDefaultToDirectLogger() throws Exception {
Review Comment:
@dlg99 I have been added the compat test here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [bookkeeper] hangc0276 commented on a diff in pull request #3263: BP-47 (task6): Direct I/O entrylogger support
Posted by GitBox <gi...@apache.org>.
hangc0276 commented on code in PR #3263:
URL: https://github.com/apache/bookkeeper/pull/3263#discussion_r901057111
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/directentrylogger/DirectEntryLoggerStats.java:
##########
@@ -0,0 +1,174 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.bookkeeper.bookie.storage.directentrylogger;
+import static org.apache.bookkeeper.bookie.BookKeeperServerStats.BOOKIE_ADD_ENTRY;
+import static org.apache.bookkeeper.bookie.BookKeeperServerStats.BOOKIE_READ_ENTRY;
+import static org.apache.bookkeeper.bookie.BookKeeperServerStats.BOOKIE_SCOPE;
+import static org.apache.bookkeeper.bookie.BookKeeperServerStats.CATEGORY_SERVER;
+
+import org.apache.bookkeeper.stats.Counter;
+import org.apache.bookkeeper.stats.OpStatsLogger;
+import org.apache.bookkeeper.stats.StatsLogger;
+import org.apache.bookkeeper.stats.annotations.StatsDoc;
+
+@StatsDoc(
+ name = BOOKIE_SCOPE,
+ category = CATEGORY_SERVER,
+ help = "Direct entry logger stats"
+)
+class DirectEntryLoggerStats {
+ private static final String ADD_ENTRY = "entrylog-add-entry";
+ private static final String READ_ENTRY = "entrylog-read-entry";
+ private static final String FLUSH = "entrylog-flush";
+ private static final String WRITER_FLUSH = "entrylog-writer-flush";
+ private static final String READ_BLOCK = "entrylog-read-block";
+ private static final String READER_OPEN = "entrylog-open-reader";
+ private static final String READER_CLOSE = "entrylog-close-reader";
+ private static final String CACHED_READER_SERVED_CLOSED = "entrylog-cached-reader-closed";
+
+ @StatsDoc(
+ name = ADD_ENTRY,
+ help = "Operation stats of adding entries to the entry log",
+ parent = BOOKIE_ADD_ENTRY
+ )
+ private final OpStatsLogger addEntryStats;
+
+ @StatsDoc(
+ name = READ_ENTRY,
+ help = "Operation stats of reading entries from the entry log",
+ parent = BOOKIE_READ_ENTRY
+ )
+ private final ThreadLocal<OpStatsLogger> readEntryStats;
Review Comment:
Those stats which use ThreadLocal need the passed parameter `stats`, It is hard to make them `static`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [bookkeeper] hangc0276 commented on a diff in pull request #3263: BP-47 (task6): Direct I/O entrylogger support
Posted by GitBox <gi...@apache.org>.
hangc0276 commented on code in PR #3263:
URL: https://github.com/apache/bookkeeper/pull/3263#discussion_r895160289
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/directentrylogger/DirectWriter.java:
##########
@@ -0,0 +1,322 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.bookkeeper.bookie.storage.directentrylogger;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.bookkeeper.common.util.ExceptionMessageHelper.exMsg;
+
+import io.netty.buffer.ByteBuf;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+
+import org.apache.bookkeeper.common.util.nativeio.NativeIO;
+import org.apache.bookkeeper.common.util.nativeio.NativeIOException;
+import org.apache.bookkeeper.slogger.Slogger;
+import org.apache.commons.lang3.SystemUtils;
+
+class DirectWriter implements LogWriter {
+ final NativeIO nativeIO;
+ final int fd;
+ final int id;
+ final String filename;
+ final BufferPool bufferPool;
+ final ExecutorService writeExecutor;
+ final Object bufferLock = new Object();
+ final List<Future<?>> outstandingWrites = new ArrayList<Future<?>>();
Review Comment:
Maybe not. There are two methods `addOutstandingWrite` and `waitForOutstandingWrites` using the `outstandingWrites`. It does not just add and remove operation but uses an iterator to traverse the list content. We need synchronization to keep it in order.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [bookkeeper] eolivelli commented on a diff in pull request #3263: BP-47 (task6): Direct I/O entrylogger support
Posted by GitBox <gi...@apache.org>.
eolivelli commented on code in PR #3263:
URL: https://github.com/apache/bookkeeper/pull/3263#discussion_r867762916
##########
pom.xml:
##########
@@ -905,8 +906,22 @@
<trimStackTrace>false</trimStackTrace>
<forkedProcessTimeoutInSeconds>1800</forkedProcessTimeoutInSeconds>
<rerunFailingTestsCount>${testRetryCount}</rerunFailingTestsCount>
+ <useSystemClassLoader>false</useSystemClassLoader>
+ <argLine>-Djava.library.path=target/natives</argLine>
Review Comment:
this should be useless, isn't it ?
##########
pom.xml:
##########
@@ -905,8 +906,22 @@
<trimStackTrace>false</trimStackTrace>
<forkedProcessTimeoutInSeconds>1800</forkedProcessTimeoutInSeconds>
<rerunFailingTestsCount>${testRetryCount}</rerunFailingTestsCount>
+ <useSystemClassLoader>false</useSystemClassLoader>
+ <argLine>-Djava.library.path=target/natives</argLine>
</configuration>
</plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-failsafe-plugin</artifactId>
+ <version>${maven-failsafe-plugin.version}</version>
+ <executions>
Review Comment:
you don't need to add this "executions" flag, otherwise we are going to execute failsafe every time you run Maven.
it is automatically pre-configured, it makes sense to declare it only in order to set a version
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/DefaultEntryLogger.java:
##########
@@ -1086,7 +1087,9 @@ EntryLogMetadata extractEntryLogMetadataFromIndex(long entryLogId) throws IOExce
bc.read(sizeBuffer.get(), offset);
int ledgersMapSize = sizeBuffer.get().readInt();
-
+ if (ledgersMapSize <= 0) {
Review Comment:
why are we adding this check ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [bookkeeper] eolivelli commented on a diff in pull request #3263: BP-47 (task6): Direct I/O entrylogger support
Posted by GitBox <gi...@apache.org>.
eolivelli commented on code in PR #3263:
URL: https://github.com/apache/bookkeeper/pull/3263#discussion_r868863306
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/DefaultEntryLogger.java:
##########
@@ -1086,7 +1087,9 @@ EntryLogMetadata extractEntryLogMetadataFromIndex(long entryLogId) throws IOExce
bc.read(sizeBuffer.get(), offset);
int ledgersMapSize = sizeBuffer.get().readInt();
-
+ if (ledgersMapSize <= 0) {
Review Comment:
so this case was not possible before this change ?
btw, it does no harm.
thanks for clarification
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org