+ * block 0 (aka rootBlock):
+ * [‹next block number; 8 bytes›,
+ * ‹last block number; 8 bytes›,
+ * ‹byte encoded values›]
+ * block 1:
+ * [‹next block number; 8 bytes›,
+ * ‹not used ; 8 bytes›,
+ * ‹byte encoded values›]
+ * ...
+ * block n (the last block):
+ * [‹next block number; 8 bytes; value is {@link #NO_LAST_BLOCK}›,
+ * ‹not used ; 8 bytes›,
+ * ‹byte encoded values›]
+ *
+ */
+public class BSFile implements AutoCloseable {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(BSFile.class);
+
+ /*
+ * The last disk block of this sequence. This is the block new values will be
+ * appended to.
+ */
+ private DiskBlock buffer;
+
+ private int offsetInBuffer = 0;
+
+ private boolean dirty = false;
+
+ private static final ThreadLocal+ * If the encoded values do not fit into the buffer, then 0 is returned. The + * caller will have to provide a new buffer with more space. + * + * @param value the value of the measurement, non-negative + * @param buffer + * @param offsetInBuffer + * @return number of bytes appended to the provided buffer + */ + public int encodeInto(final long value, final byte[] buffer, final int offsetInBuffer) { + + assert value >= 0 : "value must be non-negative"; + + final int bytesNeeded = computeNumberOfEncodedBytes(value); + + // check if the encoded bytes fit into the provided buffer and copy them into + // the buffer if they fit + if (bytesNeeded <= buffer.length - offsetInBuffer) { + + // encode time and value into temporary buffers + final byte[] tmpBuffer = TMP_BUFFER.get(); + final int valueIndex = encode(value, tmpBuffer); + System.arraycopy(tmpBuffer, valueIndex, buffer, offsetInBuffer, bytesNeeded); + + return bytesNeeded; + } + + // return 0 if the encoded bytes do not fit + // the caller will have to provide a new buffer + return 0; + } + + static int computeNumberOfEncodedBytes(final long value) { + + // the first byte stores 6 bit, the continuation bytes store 7 bits: + // 2^6-1 = 63 -> 1 byte + // 2^13-1 = 8191 -> 2 byte + // 2^20-1 = 1048575 -> 3 byte + // 2^27-1 = 134217727 -> 4 byte + // 2^34-1 = 17179869183 -> 5 byte + // 2^41-1 = 2199023255551 -> 6 byte + // 2^48-1 = 281474976710655-> 7 byte + // 2^55-1 = 36028797018963967-> 8 byte + // 2^62-1 = 4611686018427387903-> 9 byte + // 2^69-1 = 590295810358705651711 -> 10 byte + + final int highestOneBit = 64 - Long.numberOfLeadingZeros(value); + + // 1 2 3 4 5 6 -> 1 + // 7 8 9 10 11 12 13 -> 2 + // 14 15 16 17 18 19 20 -> 3 + // 21 22 23 24 25 26 27 -> 4 + return highestOneBit / 7 + 1; + } + + /** + * Encodes the value into the buffer. + *
+ * The buffer is filled from the end, so that the encoded bytes will be in
+ * {@code Arrays.copyOfRange(buffer, index, buffer.length)}
+ *
+ * @param value the value to encode
+ * @param buffer the value will be encoded into this buffer. The length must be
+ * at least 10 bytes.
+ * @return index of the value start
+ */
+ private int encode(final long value, final byte[] buffer) {
+ int index = buffer.length - 1;
+
+ final long maxFirstByteValue = 63;
+ long val = value;
+ while (val > maxFirstByteValue) {
+ // handles continuation bytes
+ buffer[index] = (byte) ((val & CONTINUATION_DATA_BITS) | CONTINUATION_PREFIX);
+ index--;
+ val = val >> 7; // shift by number of value bits
+ }
+
+ buffer[index] = (byte) (val | VALUE_PREFIX);
+
+ return index;
+ }
+
+ public LongList decode(final byte[] buffer) {
+
+ final LongList result = new LongList();
+ decodeInto(buffer, result);
+ return result;
+ }
+
+ private boolean isContinuationByte(final byte b) {
+ return (b & CONTINUATION_PREFIX_BITS) == CONTINUATION_PREFIX;
+ }
+
+ public void decodeInto(final byte[] buffer, final LongList bufferedLongs) {
+ for (int i = 0; i < buffer.length; i++) {
+
+ if ((buffer[i] & VALUE_PREFIX) == VALUE_PREFIX) {
+ long val = buffer[i] & VALUE_DATA_BITS;
+
+ while (i + 1 < buffer.length) {
+
+ // if ((buffer[i + 1] & CONTINUATION_PREFIX_BITS) == CONTINUATION_PREFIX) {
+ if (isContinuationByte(buffer[i + 1])) {
+ val = val << CONTINUATION_NUM_DATA_BITS;
+ val = val | (buffer[i + 1] & CONTINUATION_DATA_BITS);
+ i++;
+ } else {
+ break;
+ }
+ }
+
+ bufferedLongs.add(val);
+ } else if (buffer[i] != 0) {
+ assert false;
+ } else {
+ assert buffer[i] == 0;
+ // No value, not event the value 0, can be encoded as the null byte.
+ // Therefore the sequences are null-terminated
+ break;
+ }
+ }
+ }
+
+// public void encode(final LongList values, final byte[] buffer) {
+//
+// int offsetInBuffer = 0;
+//
+// for (int i = 0; i < values.size(); i++) {
+//
+// final long value = values.get(i);
+// final int encodedBytes = encodeInto(value, buffer, offsetInBuffer);
+//
+// if (encodedBytes == 0 || offsetInBuffer >= 509) {
+// System.out.println("first header block is full");
+// }
+//
+// offsetInBuffer += encodedBytes;
+// }
+// }
+
+}
diff --git a/block-storage/src/main/java/org/lucares/pdb/diskstorage/DiskBlock.java b/block-storage/src/main/java/org/lucares/pdb/diskstorage/DiskBlock.java
new file mode 100644
index 0000000..35b4653
--- /dev/null
+++ b/block-storage/src/main/java/org/lucares/pdb/diskstorage/DiskBlock.java
@@ -0,0 +1,118 @@
+package org.lucares.pdb.diskstorage;
+
+import java.nio.MappedByteBuffer;
+
+import org.lucares.collections.LongList;
+import org.lucares.pdb.blockstorage.intsequence.LongSequenceEncoderDecoder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DiskBlock {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(DiskBlock.class);
+
+ protected static final int NEXT_POINTER_OFFSET = 0;
+ public static final long NO_NEXT_POINTER = 0;
+ private static final int LAST_BLOCK_POINTER_POSITION = 8;
+ public static final long NO_LAST_BLOCK = 0;
+ private static final int INT_SEQUENCE_OFFSET = 8 // next block pointer
+ + 8; // last block pointer;
+
+ private byte[] buffer = null;
+ private final long blockNumber;
+ private long nextBlockNumber = 0;
+ private long lastBlockNumber = 0;
+
+ private final MappedByteBuffer byteBuffer;
+
+ public DiskBlock(final long blockNumber, final MappedByteBuffer byteBuffer) {
+ this.blockNumber = blockNumber;
+ this.byteBuffer = byteBuffer;
+ }
+
+ public byte[] getBuffer() {
+
+ if (buffer == null) {
+ this.buffer = new byte[byteBuffer.capacity() - INT_SEQUENCE_OFFSET];
+ byteBuffer.position(INT_SEQUENCE_OFFSET);
+ byteBuffer.get(buffer);
+ }
+
+ return buffer;
+ }
+
+ public long getBlockNumber() {
+ return blockNumber;
+ }
+
+ public void setNextBlockNumber(final long nextBlockNumber) {
+ this.nextBlockNumber = nextBlockNumber;
+ }
+
+// public void replace(final byte[] serialized) {
+// if (buffer.length != serialized.length) {
+// throw new IllegalArgumentException(
+// "existing buffer has length " + buffer.length + ", but new buffer has length " + serialized.length);
+// }
+// buffer = serialized;
+// }
+
+ private void writeBufferToByteBuffer() {
+ byteBuffer.position(INT_SEQUENCE_OFFSET);
+ byteBuffer.put(buffer);
+ }
+
+ private void writeBlockHeader() {
+ byteBuffer.putLong(NEXT_POINTER_OFFSET, nextBlockNumber);
+ byteBuffer.putLong(LAST_BLOCK_POINTER_POSITION, lastBlockNumber);
+ }
+
+ public void writeAsync() {
+ final long start = System.nanoTime();
+ writeBlockHeader();
+ writeBufferToByteBuffer();
+ final long duration = System.nanoTime() - start;
+ LOGGER.trace("write() of block={}: {}ms", blockNumber, duration / 1_000_000.0);
+ }
+
+ public void force() {
+ final long start = System.nanoTime();
+ byteBuffer.force();
+ LOGGER.trace("force of block={}: {}ms", blockNumber, (System.nanoTime() - start) / 1_000_000.0);
+ }
+
+ public long getLastBlockPointer() {
+
+ if (lastBlockNumber <= 0) {
+ lastBlockNumber = byteBuffer.getLong(LAST_BLOCK_POINTER_POSITION);
+ }
+
+ return lastBlockNumber;
+ }
+
+ public long getNextBlockNumber() {
+ if (nextBlockNumber <= 0) {
+ nextBlockNumber = byteBuffer.getLong(NEXT_POINTER_OFFSET);
+ }
+ return nextBlockNumber;
+ }
+
+ public void setLastBlockNumber(final long lastBlockNumber) {
+ this.lastBlockNumber = lastBlockNumber;
+ }
+
+ public void writeLastBlockNumber(final long lastBlockNumber) {
+ this.lastBlockNumber = lastBlockNumber;
+ byteBuffer.putLong(LAST_BLOCK_POINTER_POSITION, lastBlockNumber);
+ }
+
+ @Override
+ public String toString() {
+ final LongList bufferDecoded = new LongSequenceEncoderDecoder().decode(buffer);
+ return "DiskBlock[" + blockNumber + ", bufferDecoded=" + bufferDecoded + "]";
+ }
+//
+// public void putNoNextBlockNumber() {
+// nextBlockNumber = NO_NEXT_POINTER;
+// }
+}
diff --git a/block-storage/src/main/java/org/lucares/pdb/diskstorage/DiskStorage.java b/block-storage/src/main/java/org/lucares/pdb/diskstorage/DiskStorage.java
new file mode 100644
index 0000000..f249b50
--- /dev/null
+++ b/block-storage/src/main/java/org/lucares/pdb/diskstorage/DiskStorage.java
@@ -0,0 +1,90 @@
+package org.lucares.pdb.diskstorage;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.MappedByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.channels.FileChannel.MapMode;
+import java.nio.channels.FileLock;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DiskStorage implements AutoCloseable {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(DiskStorage.class);
+
+ public static final int BLOCK_SIZE = 512;
+
+ private final FileChannel fileChannel;
+
+ public DiskStorage(final Path databaseFile) throws IOException {
+ fileChannel = FileChannel.open(databaseFile, StandardOpenOption.READ, StandardOpenOption.WRITE,
+ StandardOpenOption.CREATE);
+ }
+
+ public DiskBlock getDiskBlock(final long blockNumber) throws IOException {
+
+ // block numbers start with 1, so that the uninitialized value
+ // (0) means 'no block'. That way we do not have to write data to a newly
+ // created block, which reduces IO.
+ final long position = (blockNumber - 1) * BLOCK_SIZE;
+
+ final long start = System.nanoTime();
+
+ try (final FileLock lock = fileChannel.lock(position, BLOCK_SIZE, true)) {
+
+ final MappedByteBuffer byteBuffer = fileChannel.map(MapMode.READ_WRITE, position, BLOCK_SIZE);
+
+ return new DiskBlock(blockNumber, byteBuffer);
+
+ } finally {
+ LOGGER.trace("read block={}: {}ms", blockNumber, (System.nanoTime() - start) / 1_000_000.0);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ fileChannel.force(true);
+ fileChannel.close();
+ }
+
+ public long getNumBlocks() throws IOException {
+ return fileChannel.size() / BLOCK_SIZE;
+ }
+
+ public long[] appendNewBlocks(final int numNewBlocks) throws IOException {
+
+ final long[] result = new long[numNewBlocks];
+ synchronized (fileChannel) {
+ for (int i = 0; i < numNewBlocks; i++) {
+ final long blockNumber = appendNewBlock();
+ result[i] = blockNumber;
+ }
+ }
+ return result;
+ }
+
+ public long appendNewBlock() throws IOException {
+
+ final byte[] buffer = new byte[BLOCK_SIZE];
+ final ByteBuffer src = ByteBuffer.wrap(buffer);
+
+ synchronized (fileChannel) {
+ // block numbers start with 1, so that the uninitialized value
+ // (0) means 'no block'. That way we do not have to write
+ // data to a newly created block, which reduces IO.
+ final long blockNumber = getNumBlocks() + 1;
+ fileChannel.write(src, fileChannel.size());
+ return blockNumber;
+ }
+ }
+
+ public DiskBlock getNewBlock() throws IOException {
+ final long blockNumber = appendNewBlock();
+ return getDiskBlock(blockNumber);
+ }
+
+}
diff --git a/block-storage/src/test/java/org/lucares/pdb/blockstorage/BSFileTest.java b/block-storage/src/test/java/org/lucares/pdb/blockstorage/BSFileTest.java
new file mode 100644
index 0000000..7baf8ab
--- /dev/null
+++ b/block-storage/src/test/java/org/lucares/pdb/blockstorage/BSFileTest.java
@@ -0,0 +1,135 @@
+package org.lucares.pdb.blockstorage;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.LongStream;
+
+import org.lucares.collections.LongList;
+import org.lucares.pdb.diskstorage.DiskStorage;
+import org.lucares.utils.file.FileUtils;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+@Test
+public class BSFileTest {
+
+ private Path dataDirectory;
+
+ @BeforeMethod
+ public void beforeMethod() throws IOException {
+ dataDirectory = Files.createTempDirectory("pdb");
+ }
+
+ @AfterMethod
+ public void afterMethod() throws IOException {
+ FileUtils.delete(dataDirectory);
+ }
+
+ public void testBlockStorage() throws Exception {
+ final Path file = dataDirectory.resolve("data.int.db");
+ final int numLongs = 1000;
+ long blockNumber = -1;
+
+ long start = System.nanoTime();
+ //
+ try (final DiskStorage ds = new DiskStorage(file)) {
+
+ try (final BSFile bsFile = BSFile.newFile(ds)) {
+
+ blockNumber = bsFile.getRootBlockNumber();
+
+ for (long i = 0; i < numLongs / 2; i++) {
+ bsFile.append(i);
+ }
+ }
+ try (final BSFile bsFile = BSFile.existingFile(blockNumber, ds)) {
+
+ for (long i = numLongs / 2; i < numLongs; i++) {
+ bsFile.append(i);
+ }
+ }
+ }
+ System.out.println("duration write: " + (System.nanoTime() - start) / 1_000_000.0 + "ms");
+
+ start = System.nanoTime();
+ try (final DiskStorage ds = new DiskStorage(file)) {
+ final BSFile bsFile = BSFile.existingFile(blockNumber, ds);
+ final long[] actualLongs = bsFile.stream().toArray();
+ final long[] expectedLongs = LongStream.rangeClosed(0, numLongs - 1).toArray();
+ Assert.assertEquals(actualLongs, expectedLongs);
+ }
+ System.out.println("duration read: " + (System.nanoTime() - start) / 1_000_000.0 + "ms");
+ }
+
+ public void testBlockStorageMultithreading() throws Exception {
+ final ExecutorService pool = Executors.newCachedThreadPool();
+
+ final Path file = dataDirectory.resolve("data.int.db");
+
+ final int threads = 50;
+ final int values = 10000;
+ final Map