Homework4

.codeclimate.yml

@@ -49,14 +49,14 @@ plugins:
   sonar-java:
     enabled: true
     config:
-      sonar.java.source: "11"
+      sonar.java.source: "21"
     checks:
       squid:S00112:
         enabled: false
 checks:
   argument-count:
     config:
-      threshold: 5 # No more than 5 arguments
+      threshold: 6 # No more than 6 arguments
   method-lines:
     config:
       threshold: 80 # No more than 2 screens

README.md

@@ -57,3 +57,15 @@ $ ./gradlew clean test
 * `upsert` с `value == null` подразумевает удаление: такие записи не должны возвращаться в `get`.
 ### Report
 Когда всё будет готово, присылайте pull request в ветку `main` со своей реализацией на review. Не забывайте **отвечать на комментарии в PR** и **исправлять замечания**!
+
+## Этап 4. Compact (deadline 2023-11-01 23:59:59 MSK)
+Требуется реализовать метод `compact()` в DAO:
+* Все SSTable'ы должны быть слиты в один, включая как таблицы на диске, так и таблицу в памяти
+* Старые должны быть удалены
+* Должна сохраниться только самая свежая версия, старые записи должны быть удалены
+* Удалённые записи (`upsert` с `null`-значением) не должны сохраняться при `сompact()`
+* В рамках данного задания гарантируется, что сразу после `compact()` будет вызван метод `close()`, а значит работа с этим DAO больше производиться не будет. Но в последствии может быть открыто новое `DAO` c тем же конфигом. И на нём тоже может быть вызван `compact`.
+* `close()` обязан быть идемпотентным -- повторный `close()` не должен приводить к отрицательным последствиям, включая потерю производительности. На текущем этапе можно считать, что конкуретных вызовов `close()` нет.
+* Тесты будут появляться и дополняться
+### Report
+Когда всё будет готово, присылайте pull request в ветку `main` со своей реализацией на review. Не забывайте **отвечать на комментарии в PR** и **исправлять замечания**!

src/main/java/ru/vk/itmo/Dao.java

@@ -64,11 +64,18 @@ default Iterator<E> all() {
      */
     void upsert(E entry);
 
-    /*
+    /**
      * Persists data (no-op by default).
      */
     default void flush() throws IOException {
-        //by default do nothing
+        // Do nothing
+    }
+
+    /**
+     * Compacts data (no-op by default).
+     */
+    default void compact() throws IOException {
+        // Do nothing
     }
 
     /*

src/main/java/ru/vk/itmo/dyagayalexandra/Storage.java

@@ -0,0 +1,123 @@
+package ru.vk.itmo.dyagayalexandra;
+
+import ru.vk.itmo.BaseEntry;
+import ru.vk.itmo.Config;
+import ru.vk.itmo.Dao;
+import ru.vk.itmo.Entry;
+
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.lang.foreign.MemorySegment;
+import java.lang.foreign.ValueLayout;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.NavigableMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+
+public class Storage implements Dao<MemorySegment, Entry<MemorySegment>> {
+
+    protected final NavigableMap<MemorySegment, Entry<MemorySegment>> dataStorage;
+    private final Path path;
+    private FileInputStream fis;
+    private BufferedInputStream reader;
+
+    public Storage() {
+        dataStorage = new ConcurrentSkipListMap<>(new MemorySegmentComparator());
+        path = null;
+    }
+
+    public Storage(Config config) {
+        dataStorage = new ConcurrentSkipListMap<>(new MemorySegmentComparator());
+        path = config.basePath().resolve("storage.txt");
+        try {
+            fis = new FileInputStream(String.valueOf(path));
+            reader = new BufferedInputStream(fis);
+        } catch (IOException e) {
+            throw new IllegalArgumentException("Failed to open the file.", e);
+        }
+    }
+
+    @Override
+    public Iterator<Entry<MemorySegment>> get(MemorySegment from, MemorySegment to) {
+        Collection<Entry<MemorySegment>> values;
+        if (from == null && to == null) {
+            values = dataStorage.values();
+        } else if (from == null) {
+            values = dataStorage.headMap(to).values();
+        } else if (to == null) {
+            values = dataStorage.tailMap(from).values();
+        } else {
+            values = dataStorage.subMap(from, to).values();
+        }
+
+        return values.iterator();
+    }
+
+    @Override
+    public void upsert(Entry<MemorySegment> entry) {
+        dataStorage.put(entry.key(), entry);
+    }
+
+    @Override
+    public Entry<MemorySegment> get(MemorySegment key) {
+        try {
+            return getEntry(key);
+        } catch (IOException ex) {
+            return null;
+        }
+    }
+
+    @Override
+    public void flush() throws IOException {
+        if (!Files.exists(path)) {
+            Files.createFile(path);
+        }
+
+        try (FileOutputStream fos = new FileOutputStream(String.valueOf(path));
+             BufferedOutputStream writer = new BufferedOutputStream(fos)) {
+            for (var entry : dataStorage.values()) {
+                writer.write((byte) entry.key().byteSize());
+                writer.write(entry.key().toArray(ValueLayout.JAVA_BYTE));
+                writer.write((byte) entry.value().byteSize());
+                writer.write(entry.value().toArray(ValueLayout.JAVA_BYTE));
+            }
+        }
+
+        fis.close();
+        reader.close();
+    }
+
+    private Entry<MemorySegment> getEntry(MemorySegment key) throws IOException {
+        Entry<MemorySegment> entry = dataStorage.get(key);
+        if (entry != null) {
+            return entry;
+        }
+
+        Entry<MemorySegment> result = null;
+        while (reader.available() != 0 && result == null) {
+            int keyLength = reader.read();
+            if (keyLength != key.byteSize()) {
+                reader.skipNBytes(keyLength);
+                reader.skipNBytes(reader.read());
+                continue;
+            }
+
+            byte[] keyBytes = reader.readNBytes(keyLength);
+            if (!key.equals(MemorySegment.ofArray(keyBytes))) {
+                reader.skipNBytes(reader.read());
+                continue;
+            }
+
+            int valueLength = reader.read();
+            byte[] valueBytes = reader.readNBytes(valueLength);
+            result = new BaseEntry<>(key, MemorySegment.ofArray(valueBytes));
+        }
+
+        return null;
+    }
+}

src/main/java/ru/vk/itmo/grunskiialexey/MemorySegmentDao.java

@@ -0,0 +1,170 @@
+package ru.vk.itmo.grunskiialexey;
+
+import ru.vk.itmo.BaseEntry;
+import ru.vk.itmo.Config;
+import ru.vk.itmo.Dao;
+import ru.vk.itmo.Entry;
+
+import java.io.IOException;
+import java.lang.foreign.Arena;
+import java.lang.foreign.MemorySegment;
+import java.lang.foreign.ValueLayout;
+import java.nio.channels.FileChannel;
+import java.nio.file.Files;
+import java.nio.file.NoSuchFileException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.NavigableMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+
+public class MemorySegmentDao implements Dao<MemorySegment, Entry<MemorySegment>> {
+    private final Comparator<MemorySegment> comparator = (o1, o2) -> {
+        long firstMismatch = o1.mismatch(o2);
+        if (firstMismatch == -1) {
+            return 0;
+        }
+        if (firstMismatch == o1.byteSize()) {
+            return -1;
+        }
+        if (firstMismatch == o2.byteSize()) {
+            return 1;
+        }
+
+        byte byte1 = o1.get(ValueLayout.JAVA_BYTE, firstMismatch);
+        byte byte2 = o2.get(ValueLayout.JAVA_BYTE, firstMismatch);
+        return Byte.compare(byte1, byte2);
+    };
+    private final NavigableMap<MemorySegment, Entry<MemorySegment>> data = new ConcurrentSkipListMap<>(comparator);
+    private final Path filePath;
+    private final Arena arena;
+    private final MemorySegment page;
+
+    public MemorySegmentDao(Config config) throws IOException {
+        this.filePath = Paths.get(config.basePath().toString(), "file.db");
+        arena = Arena.ofShared();
+
+        long size;
+        try {
+            size = Files.size(filePath);
+        } catch (NoSuchFileException e) {
+            page = MemorySegment.NULL;
+            return;
+        }
+
+        MemorySegment currentPage = null;
+        try (FileChannel channel = FileChannel.open(filePath, StandardOpenOption.READ)) {
+            currentPage = channel.map(FileChannel.MapMode.READ_ONLY, 0, size, arena);
+        } catch (IOException e) {
+            arena.close();
+        } finally {
+            page = currentPage;
+        }
+    }
+
+    @Override
+    public Iterator<Entry<MemorySegment>> get(MemorySegment from, MemorySegment to) {
+        if (from == null && to == null) {
+            return data.values().iterator();
+        } else if (from == null) {
+            return data.headMap(to).values().iterator();
+        } else if (to == null) {
+            return data.tailMap(from).values().iterator();
+        } else {
+            return data.subMap(from, to).values().iterator();
+        }
+    }
+
+    @Override
+    public Entry<MemorySegment> get(MemorySegment key) {
+        if (data.containsKey(key)) {
+            return data.get(key);
+        }
+        if (page.equals(MemorySegment.NULL)) {
+            return null;
+        }
+
+        long offset = 0;
+        while (offset < page.byteSize()) {
+            int keyLength = page.get(ValueLayout.JAVA_INT, offset);
+            offset += 4;
+            MemorySegment resultKey = MemorySegment.ofArray(new byte[keyLength]);
+            MemorySegment.copy(page, ValueLayout.JAVA_BYTE, offset, resultKey, ValueLayout.JAVA_BYTE, 0, keyLength);
+            offset += correctAlignedSize(keyLength);
+
+            int valueLength = page.get(ValueLayout.JAVA_INT, offset);
+            offset += 4;
+            MemorySegment resultValue = MemorySegment.ofArray(new byte[valueLength]);
+            MemorySegment.copy(page, ValueLayout.JAVA_BYTE, offset, resultValue, ValueLayout.JAVA_BYTE, 0, valueLength);
+            offset += correctAlignedSize(valueLength);
+
+            if (resultKey.mismatch(key) == -1) {
+                return new BaseEntry<>(
+                        resultKey,
+                        resultValue
+                );
+            }
+        }
+        return null;
+    }
+
+    @Override
+    public void upsert(Entry<MemorySegment> entry) {
+        if (entry.value() == null) {
+            data.remove(entry.key());
+        } else {
+            data.put(entry.key(), entry);
+        }
+    }
+
+    @Override
+    public void close() throws IOException {
+        if (!arena.scope().isAlive()) {
+            return;
+        }
+
+        arena.close();
+
+        try (
+                FileChannel channel = FileChannel.open(
+                        filePath,
+                        StandardOpenOption.CREATE,
+                        StandardOpenOption.READ,
+                        StandardOpenOption.WRITE
+                );
+                Arena writeArena = Arena.ofConfined()
+        ) {
+            long allSize = data.values().stream().mapToLong(entry ->
+                    correctAlignedSize(entry.key().byteSize()) + correctAlignedSize(entry.value().byteSize()) + 2 * 4
+            ).sum();
+            MemorySegment writePage = channel.map(FileChannel.MapMode.READ_WRITE, 0, allSize, writeArena);
+
+            long offset = 0;
+            for (Entry<MemorySegment> entry : data.values()) {
+                long keyLength = entry.key().byteSize();
+                writePage.set(ValueLayout.JAVA_INT, offset, (int) keyLength);
+                offset += 4;
+                MemorySegment.copy(
+                        entry.key(), ValueLayout.JAVA_BYTE, 0,
+                        writePage, ValueLayout.JAVA_BYTE, offset, keyLength
+                );
+                offset += correctAlignedSize(keyLength);
+
+                long valueLength = entry.value().byteSize();
+                writePage.set(ValueLayout.JAVA_INT, offset, (int) valueLength);
+                offset += 4;
+                MemorySegment.copy(
+                        entry.value(), ValueLayout.JAVA_BYTE, 0,
+                        writePage, ValueLayout.JAVA_BYTE, offset, valueLength
+                );
+                offset += correctAlignedSize(valueLength);
+            }
+        }
+    }
+
+    private long correctAlignedSize(long offset) {
+        return offset % 4 == 0 ? offset : offset + 4 - (offset % 4);
+    }
+}