chore: 19096: Improve DataFileWriter: cache FileChannel and deprecate itemsCount

Fixes: #19096
Reviewed-by: Anthony Petrov <[email protected]>, Artem Ananev <[email protected]>, Roger Barker <[email protected]>
Signed-off-by: Artur Kugal <[email protected]>

Author: akugal

platform-sdk/swirlds-base/src/testFixtures/java/com/swirlds/base/test/fixtures/util/DataUtils.java

@@ -0,0 +1,43 @@
+// SPDX-License-Identifier: Apache-2.0
+package com.swirlds.base.test.fixtures.util;
+
+import java.util.Random;
+import java.util.UUID;
+
+/**
+ * Utility class to generate some primitive data for testing purposes.
+ */
+public final class DataUtils {
+
+    private DataUtils() {
+        // Prevent instantiation
+    }
+
+    public static int[] shuffle(Random random, final int[] array) {
+        if (random == null) {
+            random = new Random();
+        }
+        final int count = array.length;
+        for (int i = count; i > 1; i--) {
+            swap(array, i - 1, random.nextInt(i));
+        }
+        return array;
+    }
+
+    public static void swap(final int[] array, final int i, final int j) {
+        final int temp = array[i];
+        array[i] = array[j];
+        array[j] = temp;
+    }
+
+    public static byte[] randomUtf8Bytes(final int n) {
+        final byte[] data = new byte[n];
+        int i = 0;
+        while (i < n) {
+            final byte[] rnd = UUID.randomUUID().toString().getBytes();
+            System.arraycopy(rnd, 0, data, i, Math.min(rnd.length, n - 1 - i));
+            i += rnd.length;
+        }
+        return data;
+    }
+}

platform-sdk/swirlds-merkledb/build.gradle.kts

@@ -16,7 +16,10 @@ tasks.withType<JavaCompile>().configureEach {
 
 mainModuleInfo { annotationProcessor("com.swirlds.config.processor") }
 
-jmhModuleInfo { requires("jmh.core") }
+jmhModuleInfo {
+    requires("jmh.core")
+    requires("com.swirlds.base.test.fixtures")
+}
 
 testModuleInfo {
     requires("com.swirlds.common.test.fixtures")

platform-sdk/swirlds-merkledb/src/hammer/java/com/swirlds/merkledb/files/DataFileReaderHammerTest.java

@@ -54,7 +54,7 @@ void interruptedReadsHammerTest() throws Exception {
         final ExecutorService exec = Executors.newFixedThreadPool(readerThreads);
         final Random rand = new Random();
         final MerkleDbConfig dbConfig = CONFIGURATION.getConfigData(MerkleDbConfig.class);
-        final DataFileMetadata metadata = new DataFileMetadata(itemCount, 0, Instant.now(), INITIAL_COMPACTION_LEVEL);
+        final DataFileMetadata metadata = new DataFileMetadata(0, Instant.now(), INITIAL_COMPACTION_LEVEL);
         final DataFileReader dataReader = new DataFileReader(dbConfig, tempFile, metadata);
         final AtomicInteger activeReaders = new AtomicInteger(readerThreads);
         final AtomicReferenceArray<Thread> threads = new AtomicReferenceArray<>(readerThreads);

platform-sdk/swirlds-merkledb/src/jmh/java/com/swirlds/benchmark/DataFileWriterBenchmark.java

@@ -0,0 +1,138 @@
+// SPDX-License-Identifier: Apache-2.0
+package com.swirlds.benchmark;
+
+import static com.swirlds.base.units.UnitConstants.MEBIBYTES_TO_BYTES;
+
+import com.hedera.pbj.runtime.io.buffer.BufferedData;
+import com.swirlds.base.test.fixtures.util.DataUtils;
+import com.swirlds.common.io.utility.FileUtils;
+import com.swirlds.merkledb.files.DataFileWriter;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Instant;
+import java.util.Arrays;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Warmup;
+
+/**
+ * This benchmarks can be used to measure performance of {@link DataFileWriter} class.
+ * It uses pregenerated list of data items to measure only writing speed.
+ */
+@State(Scope.Benchmark)
+public class DataFileWriterBenchmark {
+
+    /**
+     * Size of the buffer in MB - <b>should not</b> impact on performance with the current implementation with mapped buffer.
+     */
+    @Param({"16", "64", "128", "256"})
+    public int bufferSizeMb;
+
+    /**
+     * Maximum size of the file to write in MB.
+     */
+    @Param({"50", "200", "500"})
+    public int maxFileSizeMb;
+
+    /**
+     * Number of sample data items to pre-generate before each iteration.
+     */
+    @Param({"10"})
+    public int sampleSize;
+
+    /**
+     * Range of the sample data items in bytes, chosen randomly from the sample.
+     */
+    // first is test for small data items like hashes
+    @Param({"56-56", "300-1000", "1024-8192"})
+    public String sampleRangeBytes;
+
+    // Runtime variables
+    private Random random;
+    private BufferedData[] sampleData;
+    private long maxFileSize;
+
+    private Path benchmarkDir;
+    private DataFileWriter dataFileWriter;
+
+    @Setup(Level.Trial)
+    public void setupGlobal() throws IOException {
+        random = new Random(1234);
+        benchmarkDir = Files.createTempDirectory("dataFileWriterBenchmark");
+
+        maxFileSize = maxFileSizeMb * MEBIBYTES_TO_BYTES;
+
+        // Generate sample data
+        String[] range = sampleRangeBytes.split("-");
+        int sampleMinLength = Integer.parseInt(range[0]);
+        int sampleMaxLength = Integer.parseInt(range[1]);
+
+        sampleData = new BufferedData[sampleSize];
+        for (int i = 0; i < sampleSize; i++) {
+            sampleData[i] =
+                    BufferedData.wrap(DataUtils.randomUtf8Bytes(random.nextInt(sampleMinLength, sampleMaxLength + 1)));
+        }
+
+        System.out.println("Sample data sizes in bytes: "
+                + Arrays.toString(Arrays.stream(sampleData)
+                        .mapToLong(BufferedData::length)
+                        .toArray()));
+    }
+
+    @TearDown(Level.Trial)
+    public void tearDownGlobal() throws IOException {
+        if (benchmarkDir != null) {
+            FileUtils.deleteDirectory(benchmarkDir);
+        }
+    }
+
+    @Setup(Level.Invocation)
+    public void setup() throws IOException {
+        dataFileWriter =
+                new DataFileWriter("test", benchmarkDir, 1, Instant.now(), 1, bufferSizeMb * MEBIBYTES_TO_BYTES);
+    }
+
+    @TearDown(Level.Invocation)
+    public void tearDown() throws IOException {
+        dataFileWriter.close();
+    }
+
+    @Benchmark
+    @BenchmarkMode(Mode.AverageTime)
+    @Measurement(iterations = 1, time = 3, timeUnit = TimeUnit.SECONDS)
+    @OutputTimeUnit(TimeUnit.MILLISECONDS)
+    @Fork(value = 1, warmups = 0)
+    @Warmup(iterations = 0)
+    public void writeInFile() throws IOException {
+        long fileSize = 0;
+        BufferedData data;
+
+        while (true) {
+            data = getRandomData();
+            fileSize += data.length();
+            if (fileSize > maxFileSize) {
+                break;
+            }
+
+            dataFileWriter.storeDataItem(data);
+            data.flip();
+        }
+    }
+
+    private BufferedData getRandomData() {
+        return sampleData[random.nextInt(sampleSize)];
+    }
+}

platform-sdk/swirlds-merkledb/src/main/java/com/swirlds/merkledb/files/DataFileCollection.java

@@ -319,7 +319,7 @@ public void close() throws IOException {
         // finish writing if we still are
         final DataFileWriter currentDataFileForWriting = currentDataFileWriter.getAndSet(null);
         if (currentDataFileForWriting != null) {
-            currentDataFileForWriting.finishWriting();
+            currentDataFileForWriting.close();
         }
         // calling startSnapshot causes the metadata file to be written
         saveMetadata(storeDir);
@@ -401,7 +401,7 @@ public DataFileReader endWriting(final long minimumValidKey, final long maximumV
             throw new IOException("Tried to end writing when we never started writing.");
         }
         // finish writing the file and write its footer
-        dataWriter.finishWriting();
+        dataWriter.close();
         final DataFileReader dataReader = currentDataFileReader.getAndSet(null);
         if (logger.isTraceEnabled()) {
             final DataFileMetadata metadata = dataReader.getMetadata();

platform-sdk/swirlds-merkledb/src/main/java/com/swirlds/merkledb/files/DataFileCompactor.java

@@ -318,7 +318,7 @@ private void startNewCompactionFile(int compactionLevel) throws IOException {
      * @throws IOException If an I/O error occurs
      */
     private void finishCurrentCompactionFile() throws IOException {
-        currentWriter.get().finishWriting();
+        currentWriter.get().close();
         currentWriter.set(null);
         // Now include the file in future compactions
         currentReader.get().setFileCompleted();

platform-sdk/swirlds-merkledb/src/main/java/com/swirlds/merkledb/files/DataFileIterator.java

@@ -16,8 +16,6 @@
 import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
 import java.util.Objects;
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
 
 /**
  * Iterator class for iterating over data items in a data file created by {@link  DataFileWriter}.
@@ -30,8 +28,6 @@
  */
 public final class DataFileIterator implements AutoCloseable {
 
-    private static final Logger logger = LogManager.getLogger(DataFileIterator.class);
-
     /** Input stream this iterator is reading from */
     private final BufferedInputStream inputStream;
     /** Readable data on top of the input stream */
@@ -113,20 +109,11 @@ public boolean next() throws IOException {
             throw new IllegalStateException("Cannot read from a closed iterator");
         }
 
-        // Have we reached the end?
-        if (currentDataItem >= metadata.getDataItemCount() - 1) {
-            if (in.hasRemaining()) {
-                logger.warn(
-                        "Data file has more data than expected items={}. {}", metadata.getDataItemCount(), toString());
-            }
-            dataItemBuffer = null;
-            return false;
-        }
-
         while (in.hasRemaining()) {
             currentDataItemFilePosition = in.position();
             final int tag = in.readVarInt(false);
             final int fieldNum = tag >> TAG_FIELD_OFFSET;
+
             if (fieldNum == FIELD_DATAFILE_ITEMS.number()) {
                 final int currentDataItemSize = in.readVarInt(false);
                 dataItemBuffer = fillBuffer(currentDataItemSize);
@@ -140,7 +127,7 @@ public boolean next() throws IOException {
             }
         }
 
-        throw new IllegalStateException("Reached the end of data file while expecting more data items");
+        return false;
     }
 
     /**