ДЗ№4. Тяпуев Дмитрий. Магистратура. Политех

src/main/java/ru/vk/itmo/test/tyapuevdmitrij/DaoFactoryImplementation.java

@@ -10,7 +10,7 @@
 import java.lang.foreign.ValueLayout;
 import java.nio.charset.StandardCharsets;
 
-@DaoFactory(stage = 2)
+@DaoFactory(stage = 4)
 public class DaoFactoryImplementation implements DaoFactory.Factory<MemorySegment, Entry<MemorySegment>> {
 
     @Override

src/main/java/ru/vk/itmo/tyapuevdmitrij/InMemoryDao.java

@@ -1,6 +1,5 @@
 package ru.vk.itmo.tyapuevdmitrij;
 
-import ru.vk.itmo.BaseEntry;
 import ru.vk.itmo.Config;
 import ru.vk.itmo.Dao;
 import ru.vk.itmo.Entry;
@@ -9,19 +8,15 @@
 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.Path;
-import java.nio.file.StandardOpenOption;
 import java.util.Comparator;
-import java.util.EnumSet;
 import java.util.Iterator;
-import java.util.Map;
 import java.util.concurrent.ConcurrentNavigableMap;
 import java.util.concurrent.ConcurrentSkipListMap;
 
 public class InMemoryDao implements Dao<MemorySegment, Entry<MemorySegment>> {
-    private final Comparator<MemorySegment> memorySegmentComparator = (segment1, segment2) -> {
+
+    private static final Comparator<MemorySegment> memorySegmentComparator = (segment1, segment2) -> {
         long offset = segment1.mismatch(segment2);
         if (offset == -1) {
             return 0;
@@ -34,29 +29,40 @@ public class InMemoryDao implements Dao<MemorySegment, Entry<MemorySegment>> {
         }
         return segment1.get(ValueLayout.JAVA_BYTE, offset) - segment2.get(ValueLayout.JAVA_BYTE, offset);
     };
-    private final ConcurrentNavigableMap<MemorySegment, Entry<MemorySegment>> memTable
-            = new ConcurrentSkipListMap<>(memorySegmentComparator);
+    private final ConcurrentNavigableMap<MemorySegment, Entry<MemorySegment>> memTable =
+            new ConcurrentSkipListMap<>(memorySegmentComparator);
+
+    private final Arena readArena;
     private final Path ssTablePath;
-    private final MemorySegment ssTable;
-    private static final String SS_TABLE_FILE_NAME = "ssTable";
+    private long ssTablesEntryQuantity;
+    private boolean compacted;
+    private final Storage storage;
 
     public InMemoryDao() {
         ssTablePath = null;
-        ssTable = null;
+        readArena = null;
+        storage = null;
     }
 
     public InMemoryDao(Config config) {
-        ssTablePath = config.basePath().resolve(SS_TABLE_FILE_NAME);
-        ssTable = getReadBufferFromSsTable();
+        ssTablePath = config.basePath();
+        readArena = Arena.ofShared();
+        storage = new Storage(ssTablePath, readArena);
     }
 
     @Override
     public Iterator<Entry<MemorySegment>> get(MemorySegment from, MemorySegment to) {
+        return storage.range(getMemTableIterator(from, to), from, to, memorySegmentComparator);
+    }
+
+    private Iterator<Entry<MemorySegment>> getMemTableIterator(MemorySegment from, MemorySegment to) {
         if (from == null && to == null) {
             return memTable.values().iterator();
-        } else if (from == null) {
+        }
+        if (from == null) {
             return memTable.headMap(to).values().iterator();
-        } else if (to == null) {
+        }
+        if (to == null) {
             return memTable.tailMap(from).values().iterator();
         }
         return memTable.subMap(from, to).values().iterator();
@@ -65,10 +71,13 @@ public Iterator<Entry<MemorySegment>> get(MemorySegment from, MemorySegment to)
     @Override
     public Entry<MemorySegment> get(MemorySegment key) {
         Entry<MemorySegment> value = memTable.get(key);
-        if (value != null || ssTable == null) {
+        if (memTable.containsKey(key) && value.value() == null) {
+            return null;
+        }
+        if (value != null || storage.ssTables == null) {
             return value;
         }
-        return getSsTableDataByKey(key);
+        return storage.getSsTableDataByKey(key, memorySegmentComparator);
     }
 
     @Override
@@ -77,78 +86,54 @@ public void upsert(Entry<MemorySegment> entry) {
     }
 
     @Override
-    public void flush() throws IOException {
-        throw new UnsupportedOperationException("");
+    public void compact() throws IOException {
+        if (storage.ssTablesQuantity == 0 && memTable.isEmpty()) {
+            return;
+        }
+        Iterator<Entry<MemorySegment>> dataIterator = get(null, null);
+        MemorySegment buffer = NmapBuffer.getWriteBufferToSsTable(getCompactionTableByteSize(),
+                ssTablePath,
+                storage.ssTablesQuantity);
+        long bufferByteSize = buffer.byteSize();
+        buffer.set(ValueLayout.JAVA_LONG_UNALIGNED, bufferByteSize - Long.BYTES, ssTablesEntryQuantity);
+        long[] offsets = new long[2];
+        offsets[1] = bufferByteSize - Long.BYTES - ssTablesEntryQuantity * 2L * Long.BYTES;
+        while (dataIterator.hasNext()) {
+            storage.writeEntryAndIndexesToCompactionTable(buffer, dataIterator.next(), offsets);
+        }
+        StorageHelper.deleteOldSsTables(ssTablePath, storage.ssTablesQuantity);
+        StorageHelper.renameCompactedSsTable(ssTablePath);
+        compacted = true;
     }
 
     @Override
     public void close() throws IOException {
-        MemorySegment buffer = getWriteBufferToSsTable();
-        writeMemTableDataToFile(buffer);
-    }
-
-    public long getSsTableDataByteSize() {
-        long ssTableDataByteSize = 0;
-        for (Map.Entry<MemorySegment, Entry<MemorySegment>> entry : memTable.entrySet()) {
-            ssTableDataByteSize += entry.getKey().byteSize();
-            ssTableDataByteSize += entry.getValue().value().byteSize();
+        if (compacted) {
+            return;
         }
-        return ssTableDataByteSize + memTable.size() * Long.BYTES * 2L;
-    }
-
-    public MemorySegment getWriteBufferToSsTable() throws IOException {
-        MemorySegment buffer;
-        try (FileChannel channel = FileChannel.open(ssTablePath, EnumSet.of(
-                StandardOpenOption.READ,
-                StandardOpenOption.WRITE,
-                StandardOpenOption.CREATE,
-                StandardOpenOption.TRUNCATE_EXISTING))) {
-            buffer = channel.map(FileChannel.MapMode.READ_WRITE, 0, getSsTableDataByteSize(), Arena.ofAuto());
+        if (memTable.isEmpty()) {
+            return;
         }
-        return buffer;
-    }
-
-    public void writeMemTableDataToFile(MemorySegment buffer) {
-        long offset = 0;
-        for (Entry<MemorySegment> entry : memTable.values()) {
-            buffer.set(ValueLayout.JAVA_LONG_UNALIGNED, offset, entry.key().byteSize());
-            offset += Long.BYTES;
-            MemorySegment.copy(entry.key(), 0, buffer, offset, entry.key().byteSize());
-            offset += entry.key().byteSize();
-            buffer.set(ValueLayout.JAVA_LONG_UNALIGNED, offset, entry.value().byteSize());
-            offset += Long.BYTES;
-            MemorySegment.copy(entry.value(), 0, buffer, offset, entry.value().byteSize());
-            offset += entry.value().byteSize();
+        if (!readArena.scope().isAlive()) {
+            return;
         }
-    }
+        readArena.close();
+        storage.save(memTable.values(), ssTablePath);
 
-    public final MemorySegment getReadBufferFromSsTable() {
-        MemorySegment buffer;
-        try (FileChannel channel = FileChannel.open(ssTablePath, StandardOpenOption.READ)) {
-            buffer = channel.map(FileChannel.MapMode.READ_ONLY, 0, Files.size(ssTablePath), Arena.ofAuto());
-        } catch (IOException e) {
-            buffer = null;
-        }
-        return buffer;
     }
 
-    public Entry<MemorySegment> getSsTableDataByKey(MemorySegment key) {
-        long offset = 0;
-        MemorySegment valueFromSsTable;
-        while (offset < ssTable.byteSize()) {
-            long keyByteSize = ssTable.get(ValueLayout.JAVA_LONG_UNALIGNED, offset);
-            offset += Long.BYTES;
-            long keysMismatch;
-            keysMismatch = MemorySegment.mismatch(ssTable, offset, offset + keyByteSize, key, 0, key.byteSize());
-            offset += keyByteSize;
-            long valueByteSize = ssTable.get(ValueLayout.JAVA_LONG_UNALIGNED, offset);
-            offset += Long.BYTES;
-            valueFromSsTable = ssTable.asSlice(offset, valueByteSize);
-            if (keysMismatch == -1) {
-                return new BaseEntry<>(key, valueFromSsTable);
-            }
-            offset += valueByteSize;
+    private long getCompactionTableByteSize() {
+        Iterator<Entry<MemorySegment>> dataIterator = get(null, null);
+        long compactionTableByteSize = 0;
+        long countEntry = 0;
+        while (dataIterator.hasNext()) {
+            Entry<MemorySegment> entry = dataIterator.next();
+            compactionTableByteSize += entry.key().byteSize();
+            compactionTableByteSize += entry.value().byteSize();
+            countEntry++;
         }
-        return null; 
+        ssTablesEntryQuantity = countEntry;
+        return compactionTableByteSize + countEntry * 4L * Long.BYTES + Long.BYTES;
     }
+
 }

src/main/java/ru/vk/itmo/tyapuevdmitrij/MergeIterator.java

@@ -0,0 +1,145 @@
+package ru.vk.itmo.tyapuevdmitrij;
+
+import ru.vk.itmo.Entry;
+
+import java.lang.foreign.MemorySegment;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+import java.util.PriorityQueue;
+
+public class MergeIterator implements Iterator<Entry<MemorySegment>> {
+
+    private final PriorityQueue<PeekIterator> priorityQueue;
+    private final Comparator<Entry<MemorySegment>> comparator;
+    PeekIterator tableIterator;
+
+    private static class PeekIterator implements Iterator<Entry<MemorySegment>> {
+
+        public final int id;
+        private final Iterator<Entry<MemorySegment>> delegate;
+        private Entry<MemorySegment> memorySegmentsEntry;
+
+        private PeekIterator(int id, Iterator<Entry<MemorySegment>> delegate) {
+            this.id = id;
+            this.delegate = delegate;
+        }
+
+        @Override
+        public boolean hasNext() {
+            if (memorySegmentsEntry == null) {
+                return delegate.hasNext();
+            }
+            return true;
+        }
+
+        @Override
+        public Entry<MemorySegment> next() {
+            if (!hasNext()) {
+                throw new NoSuchElementException();
+            }
+            Entry<MemorySegment> peek = peek();
+            this.memorySegmentsEntry = null;
+            return peek;
+        }
+
+        private Entry<MemorySegment> peek() {
+            if (memorySegmentsEntry == null) {
+                if (!delegate.hasNext()) {
+                    return null;
+                }
+                memorySegmentsEntry = delegate.next();
+            }
+            return memorySegmentsEntry;
+        }
+    }
+
+    public MergeIterator(Collection<Iterator<Entry<MemorySegment>>> iterators,
+                         Comparator<Entry<MemorySegment>> comparator) {
+        this.comparator = comparator;
+        Comparator<PeekIterator> peekComp = (o1, o2) -> comparator.compare(o1.peek(), o2.peek());
+        priorityQueue = new PriorityQueue<>(
+                iterators.size(),
+                peekComp.thenComparing(o -> -o.id)
+        );
+
+        int id = 0;
+        for (Iterator<Entry<MemorySegment>> iterator : iterators) {
+            if (iterator.hasNext()) {
+                priorityQueue.add(new PeekIterator(id++, iterator));
+            }
+        }
+    }
+
+    private PeekIterator peek() {
+        while (tableIterator == null) {
+            tableIterator = priorityQueue.poll();
+            if (tableIterator == null) {
+                return null;
+            }
+            peekFromPriorityQueue();
+            if (tableIterator.peek() == null) {
+                tableIterator = null;
+                continue;
+            }
+
+            if (skip(tableIterator.peek())) {
+                tableIterator.next();
+                if (tableIterator.hasNext()) {
+                    priorityQueue.add(tableIterator);
+                }
+                tableIterator = null;
+            }
+        }
+
+        return tableIterator;
+    }
+
+    private void peekFromPriorityQueue() {
+        while (true) {
+            PeekIterator next = priorityQueue.peek();
+            if (next == null) {
+                break;
+            }
+            int compare = comparator.compare(tableIterator.peek(), next.peek());
+            if (compare == 0) {
+                PeekIterator poll = priorityQueue.poll();
+                if (poll != null) {
+                    poll.next();
+                    if (poll.hasNext()) {
+                        priorityQueue.add(poll);
+                    }
+                }
+            } else {
+                break;
+            }
+        }
+    }
+
+    protected boolean skip(Entry<MemorySegment> entry) {
+        if (entry != null) {
+            return entry.value() == null;
+        }
+        return false;
+    }
+
+    @Override
+    public boolean hasNext() {
+        return peek() != null;
+    }
+
+    @Override
+    public Entry<MemorySegment> next() {
+        PeekIterator entryIterator = peek();
+        if (entryIterator == null) {
+            throw new NoSuchElementException();
+        }
+        Entry<MemorySegment> next = entryIterator.next();
+        this.tableIterator = null;
+        if (entryIterator.hasNext()) {
+            priorityQueue.add(entryIterator);
+        }
+        return next;
+    }
+}