Тверитин Александр, Магистратура ИТМО "Распределенные веб-сервисы", lamport TimeStamp

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

@@ -5,4 +5,5 @@ public record BaseEntry<Data>(Data key, Data value) implements Entry<Data> {
     public String toString() {
         return "{" + key + ":" + value + "}";
     }
+
 }

src/main/java/ru/vk/itmo/reference/LamportTimeStamp.java

@@ -0,0 +1,19 @@
+package ru.vk.itmo.reference;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+public class LamportTimeStamp {
+    private final AtomicLong currentTimeStamp = new AtomicLong(0);
+
+    public long getCurrentTimeStamp() {
+        return currentTimeStamp.get();
+    }
+
+    public void setCurrentTimeStamp(long initialTimeStamp) {
+        currentTimeStamp.set(initialTimeStamp);
+    }
+
+    public long incrementAndGet() {
+        return currentTimeStamp.incrementAndGet();
+    }
+}

src/main/java/ru/vk/itmo/reference/LiveFilteringIterator.java

@@ -11,18 +11,18 @@
  *
  * @author incubos
  */
-final class LiveFilteringIterator implements Iterator<Entry<MemorySegment>> {
-    private final Iterator<Entry<MemorySegment>> delegate;
-    private Entry<MemorySegment> next;
+final class LiveFilteringIterator implements Iterator<TimeStampEntry> {
+    private final Iterator<TimeStampEntry> delegate;
+    private TimeStampEntry next;
 
-    LiveFilteringIterator(final Iterator<Entry<MemorySegment>> delegate) {
+    LiveFilteringIterator(final Iterator<TimeStampEntry> delegate) {
         this.delegate = delegate;
         skipTombstones();
     }
 
     private void skipTombstones() {
         while (delegate.hasNext()) {
-            final Entry<MemorySegment> entry = delegate.next();
+            final TimeStampEntry entry = delegate.next();
             if (entry.value() != null) {
                 this.next = entry;
                 break;
@@ -36,13 +36,13 @@ public boolean hasNext() {
     }
 
     @Override
-    public Entry<MemorySegment> next() {
+    public TimeStampEntry next() {
         if (!hasNext()) {
             throw new NoSuchElementException();
         }
 
         // Consume
-        final Entry<MemorySegment> result = next;
+        final TimeStampEntry result = next;
         next = null;
 
         skipTombstones();

src/main/java/ru/vk/itmo/reference/LiveFilteringIteratorAdapter.java

@@ -0,0 +1,24 @@
+package ru.vk.itmo.reference;
+
+import ru.vk.itmo.Entry;
+
+import java.lang.foreign.MemorySegment;
+import java.util.Iterator;
+
+public class LiveFilteringIteratorAdapter implements Iterator<Entry<MemorySegment>> {
+    private final LiveFilteringIterator liveFilteringIterator;
+
+    public LiveFilteringIteratorAdapter(LiveFilteringIterator liveFilteringIterator) {
+        this.liveFilteringIterator = liveFilteringIterator;
+    }
+
+    @Override
+    public boolean hasNext() {
+        return liveFilteringIterator.hasNext();
+    }
+
+    @Override
+    public Entry<MemorySegment> next() {
+        return liveFilteringIterator.next().getClearEntry();
+    }
+}

src/main/java/ru/vk/itmo/reference/MemTable.java

@@ -13,15 +13,15 @@
  * @author incubos
  */
 final class MemTable {
-    private final NavigableMap<MemorySegment, Entry<MemorySegment>> map =
+    private final NavigableMap<MemorySegment, TimeStampEntry> map =
             new ConcurrentSkipListMap<>(
                     MemorySegmentComparator.INSTANCE);
 
     boolean isEmpty() {
         return map.isEmpty();
     }
 
-    Iterator<Entry<MemorySegment>> get(
+    Iterator<TimeStampEntry> get(
             final MemorySegment from,
             final MemorySegment to) {
         if (from == null && to == null) {
@@ -39,11 +39,18 @@ Iterator<Entry<MemorySegment>> get(
         }
     }
 
+    /// Тут мы пока просто игнорим таймсетмп
     Entry<MemorySegment> get(final MemorySegment key) {
-        return map.get(key);
+        TimeStampEntry timeStampEntry = map.get(key);
+
+        if (timeStampEntry != null) {
+            return timeStampEntry.getClearEntry();
+        }
+
+        return null;
     }
 
     Entry<MemorySegment> upsert(final Entry<MemorySegment> entry) {
-        return map.put(entry.key(), entry);
+        return map.put(entry.key(), new TimeStampEntry(entry));
     }
 }

src/main/java/ru/vk/itmo/reference/MergingEntryIterator.java

@@ -3,24 +3,20 @@
 import ru.vk.itmo.Entry;
 
 import java.lang.foreign.MemorySegment;
-import java.util.Iterator;
-import java.util.List;
-import java.util.NoSuchElementException;
-import java.util.PriorityQueue;
-import java.util.Queue;
+import java.util.*;
 
 /**
  * Merges entry {@link Iterator}s.
  *
  * @author incubos
  */
-final class MergingEntryIterator implements Iterator<Entry<MemorySegment>> {
-    private final Queue<WeightedPeekingEntryIterator> iterators;
+final class MergingEntryIterator implements Iterator<TimeStampEntry> {
+    private final List<WeightedPeekingEntryIterator> iterators;
 
     MergingEntryIterator(final List<WeightedPeekingEntryIterator> iterators) {
         assert iterators.stream().allMatch(WeightedPeekingEntryIterator::hasNext);
 
-        this.iterators = new PriorityQueue<>(iterators);
+        this.iterators = new ArrayList<>(iterators);
     }
 
     @Override
@@ -29,44 +25,40 @@ public boolean hasNext() {
     }
 
     @Override
-    public Entry<MemorySegment> next() {
-        if (!hasNext()) {
-            throw new NoSuchElementException();
-        }
-
-        final WeightedPeekingEntryIterator top = iterators.remove();
-        final Entry<MemorySegment> result = top.next();
+    public TimeStampEntry next() {
+        TimeStampEntry nextElement = null;
 
-        if (top.hasNext()) {
-            // Not exhausted
-            iterators.add(top);
-        }
+        iterators.removeIf(currentIterator -> !currentIterator.hasNext());
 
-        // Remove older versions of the key
-        while (true) {
-            final WeightedPeekingEntryIterator iterator = iterators.peek();
-            if (iterator == null) {
-                // Nothing left
-                break;
-            }
+        int numberOfIterator = 0;
+        int counter = 0;
 
-            // Skip entries with the same key
-            final Entry<MemorySegment> entry = iterator.peek();
-            if (MemorySegmentComparator.INSTANCE.compare(result.key(), entry.key()) != 0) {
-                // Reached another key
-                break;
+        for(WeightedPeekingEntryIterator iterator: iterators) {
+            if (nextElement == null) {
+                nextElement = iterator.peek();
+            } else {
+                TimeStampEntry nextElementCandidate = iterator.peek();
+                if (MemorySegmentComparator.INSTANCE.compare(nextElement.key(), nextElementCandidate.key()) > 0) {
+                    nextElement = nextElementCandidate;
+                    numberOfIterator = counter;
+                } else if (MemorySegmentComparator.INSTANCE.compare(nextElement.key(), nextElementCandidate.key()) == 0) {
+                    if(nextElementCandidate.timeStamp() > nextElement.timeStamp()) {
+                        nextElement = nextElementCandidate;
+                        iterators.get(numberOfIterator).next();
+                        numberOfIterator = counter;
+                    } else {
+                        iterators.get(counter).next();
+                    }
+                }
             }
+            counter += 1;
+        }
 
-            // Drop
-            iterators.remove();
-            // Skip
-            iterator.next();
-            if (iterator.hasNext()) {
-                // Not exhausted
-                iterators.add(iterator);
-            }
+        nextElement = iterators.get(numberOfIterator).next();
+        if (!iterators.get(numberOfIterator).hasNext()) {
+            iterators.remove(numberOfIterator);
         }
 
-        return result;
+        return nextElement;
     }
 }

src/main/java/ru/vk/itmo/reference/ReferenceDao.java

@@ -66,10 +66,11 @@ public ReferenceDao(final Config config) throws IOException {
     public Iterator<Entry<MemorySegment>> get(
             final MemorySegment from,
             final MemorySegment to) {
-        return new LiveFilteringIterator(
+        LiveFilteringIterator liveFilteringIterator =  new LiveFilteringIterator(
                 tableSet.get(
                         from,
                         to));
+        return new LiveFilteringIteratorAdapter(liveFilteringIterator);
     }
 
     @Override

src/main/java/ru/vk/itmo/reference/SSTable.java

@@ -88,7 +88,13 @@ private long getLength(final long offset) {
                 offset);
     }
 
-    Iterator<Entry<MemorySegment>> get(
+    private long getTimeStamp(final long offset) {
+        return data.get(
+                ValueLayout.OfLong.JAVA_LONG_UNALIGNED,
+                offset);
+    }
+
+    Iterator<TimeStampEntry> get(
             final MemorySegment from,
             final MemorySegment to) {
         assert from == null || to == null || MemorySegmentComparator.INSTANCE.compare(from, to) <= 0;
@@ -134,29 +140,32 @@ Iterator<Entry<MemorySegment>> get(
         return new SliceIterator(fromOffset, toOffset);
     }
 
-    Entry<MemorySegment> get(final MemorySegment key) {
-        final long entry = entryBinarySearch(key);
-        if (entry < 0) {
+    TimeStampEntry get(final MemorySegment key) {
+        final long entryIndex = entryBinarySearch(key);
+        if (entryIndex < 0) {
             return null;
         }
 
         // Skip key (will reuse the argument)
-        long offset = entryOffset(entry);
+        long offset = entryOffset(entryIndex);
         offset += Long.BYTES + key.byteSize();
+        long timeStamp = getTimeStamp(offset);
+        offset += Long.BYTES;
+
         // Extract value length
         final long valueLength = getLength(offset);
         if (valueLength == SSTables.TOMBSTONE_VALUE_LENGTH) {
             // Tombstone encountered
-            return new BaseEntry<>(key, null);
+            return new TimeStampEntry(new BaseEntry<>(key, null), timeStamp);
         } else {
             // Get value
             offset += Long.BYTES;
             final MemorySegment value = data.asSlice(offset, valueLength);
-            return new BaseEntry<>(key, value);
+            return new TimeStampEntry(new BaseEntry<>(key, value), timeStamp);
         }
     }
 
-    private final class SliceIterator implements Iterator<Entry<MemorySegment>> {
+    private final class SliceIterator implements Iterator<TimeStampEntry> {
         private long offset;
         private final long toOffset;
 
@@ -173,7 +182,7 @@ public boolean hasNext() {
         }
 
         @Override
-        public Entry<MemorySegment> next() {
+        public TimeStampEntry next() {
             if (!hasNext()) {
                 throw new NoSuchElementException();
             }
@@ -186,18 +195,22 @@ public Entry<MemorySegment> next() {
             final MemorySegment key = data.asSlice(offset, keyLength);
             offset += keyLength;
 
+            // Read timeStamp
+            final long timeStamp = getTimeStamp(offset);
+            offset += Long.BYTES;
+
             // Read value length
             final long valueLength = getLength(offset);
             offset += Long.BYTES;
 
             // Read value
             if (valueLength == SSTables.TOMBSTONE_VALUE_LENGTH) {
                 // Tombstone encountered
-                return new BaseEntry<>(key, null);
+                return new TimeStampEntry(new BaseEntry<>(key, null), timeStamp);
             } else {
                 final MemorySegment value = data.asSlice(offset, valueLength);
                 offset += valueLength;
-                return new BaseEntry<>(key, value);
+                return new TimeStampEntry(new BaseEntry<>(key, value), timeStamp);
             }
         }
     }

src/main/java/ru/vk/itmo/reference/SSTableWriter.java

@@ -40,7 +40,7 @@ final class SSTableWriter {
     void write(
             final Path baseDir,
             final int sequence,
-            final Iterator<Entry<MemorySegment>> entries) throws IOException {
+            final Iterator<TimeStampEntry> entries) throws IOException {
         // Write to temporary files
         final Path tempIndexName = SSTables.tempIndexName(baseDir, sequence);
         final Path tempDataName = SSTables.tempDataName(baseDir, sequence);
@@ -71,11 +71,14 @@ void write(
                 writeLong(entryOffset, index);
 
                 // Then write the entry
-                final Entry<MemorySegment> entry = entries.next();
+                final TimeStampEntry entry = entries.next();
                 entryOffset += writeEntry(entry, data);
             }
         }
 
+        //index: entryOffset_1|entryOffset_2|entryOffset_3
+        //data: entryOffset_1|entryOffset_2|entryOffset_3
+
         // Publish files atomically
         // FIRST index, LAST data
         final Path indexName =
@@ -132,12 +135,14 @@ private void writeSegment(
      * @return written bytes
      */
     private long writeEntry(
-            final Entry<MemorySegment> entry,
+            final TimeStampEntry entry,
             final OutputStream os) throws IOException {
         final MemorySegment key = entry.key();
         final MemorySegment value = entry.value();
         long result = 0L;
 
+        //Long_KeySize_1|Key_data_1|Long_timestamp|Long_ValueSize_1|Value_data_1
+
         // Key size
         writeLong(key.byteSize(), os);
         result += Long.BYTES;
@@ -146,6 +151,10 @@ private long writeEntry(
         writeSegment(key, os);
         result += key.byteSize();
 
+        // TimeStamp
+        writeLong(entry.timeStamp(), os);
+        result += Long.BYTES;
+
         // Value size and possibly value
         if (value == null) {
             // Tombstone