Merge pull request #100 from koculu/99-bug-the-order-of-opindexes-are…

…-not-guranteed-to-be-incremental-during-mutable-segment-switch MoveMutableSegmentForward should wait until all writes to the previous mutable segment finalized.
koculu · Oct 16, 2024 · 82774e0 · 82774e0
2 parents 7c97c3d + 99199b1
commit 82774e0
Show file tree

Hide file tree

Showing 13 changed files with 263 additions and 38 deletions.
diff --git a/src/Playground/Benchmark/OldTests.cs b/src/Playground/Benchmark/OldTests.cs
@@ -125,7 +125,7 @@ public static void ShowBottomSegments(WriteAheadLogMode mode, int count)
         var iterate = false;
         if (iterate)
         {
-            var random = new Random();
+            var random = Random.Shared;
             Parallel.For(0, 750000, (i) =>
             {
                 var key = random.Next(0, 999_999_999);

diff --git a/src/Playground/DeadlockFinder.cs b/src/Playground/DeadlockFinder.cs
@@ -20,7 +20,7 @@ void TreeLoop(int taskNo)
             const int upsertLogFrequency = 1_000_000;
             const int iterationLogFrequency = 10_000_000;
             const int iterationYieldFrequency = 1000;
-            var rand = new Random();
+            var rand = Random.Shared;
 
             void resetTree()
             {

diff --git a/src/ZoneTree.UnitTests/AtomicUpdateTests.cs b/src/ZoneTree.UnitTests/AtomicUpdateTests.cs
@@ -29,7 +29,7 @@ public void IntIntAtomicIncrement(WriteAheadLogMode walMode)
         }
         data.Maintenance.MoveMutableSegmentForward();
         data.Maintenance.StartMergeOperation().Join();
-        var random = new Random();
+        var random = Random.Shared;
         var off = -1;
         Parallel.For(0, 1001, (x) =>
         {
@@ -100,7 +100,7 @@ public void IntIntAtomicIncrementForBTree(WriteAheadLogMode walMode)
         {
             data.Upsert(i, i + i);
         }
-        var random = new Random();
+        var random = Random.Shared;
         var off = -1;
         Parallel.For(0, 1001, (x) =>
         {
@@ -166,7 +166,7 @@ public void IntIntMutableSegmentOnlyAtomicIncrement(WriteAheadLogMode walMode)
         {
             data.Upsert(i, i + i);
         }
-        var random = new Random();
+        var random = Random.Shared;
         var off = -1;
         Parallel.For(0, 1001, (x) =>
         {
@@ -228,7 +228,7 @@ public void IntIntMutableSegmentSeveralUpserts(WriteAheadLogMode walMode)
             .ConfigureWriteAheadLogOptions(x => x.WriteAheadLogMode = walMode)
             .OpenOrCreate();
         var n = 1000;
-        var random = new Random();
+        var random = Random.Shared;
         Parallel.For(0, 1000, (x) =>
         {
             try

diff --git a/src/ZoneTree.UnitTests/ExceptionlessTransactionTests.cs b/src/ZoneTree.UnitTests/ExceptionlessTransactionTests.cs
@@ -82,7 +82,7 @@ public async Task TransactionWithFluentAPI(int compactionThreshold)
 
         zoneTree.Maintenance.TransactionLog.CompactionThreshold = compactionThreshold;
 
-        var random = new Random();
+        var random = Random.Shared;
         await Parallel.ForEachAsync(Enumerable.Range(0, 1000), async (x, cancel) =>
         {
             using var transaction =

diff --git a/src/ZoneTree.UnitTests/IteratorTests.cs b/src/ZoneTree.UnitTests/IteratorTests.cs
@@ -240,7 +240,7 @@ public void IntIntIteratorParallelInserts()
         if (Directory.Exists(dataPath))
             Directory.Delete(dataPath, true);
 
-        var random = new Random();
+        var random = Random.Shared;
         var insertCount = 100000;
         var iteratorCount = 1000;
 
@@ -289,7 +289,7 @@ public void IntIntReverseIteratorParallelInserts(bool reverse)
         if (Directory.Exists(dataPath))
             Directory.Delete(dataPath, true);
 
-        var random = new Random();
+        var random = Random.Shared;
         var insertCount = 1000000;
         var iteratorCount = 1000;
 
@@ -352,7 +352,7 @@ public void IntIntSnapshotIteratorParallelInserts()
         if (Directory.Exists(dataPath))
             Directory.Delete(dataPath, true);
 
-        var random = new Random();
+        var random = Random.Shared;
         var insertCount = 100000;
         var iteratorCount = 1000;
 
@@ -512,7 +512,7 @@ public void SeekIteratorsAfterMerge(
         if (merge)
             zoneTree.Maintenance.StartMergeOperation()?.Join();
 
-        var random = new Random();
+        var random = Random.Shared;
         DoPrefixSearch(zoneTree, list, random);
         zoneTree.Maintenance.DiskSegment.InitSparseArray(100);
         DoPrefixSearch(zoneTree, list, random);
@@ -604,7 +604,7 @@ public void SeekIteratorsAfterMergeReload(
                     x.MaximumRecordCount = maximumRecordCount;
             })
             .Open();
-        var random = new Random();
+        var random = Random.Shared;
         DoPrefixSearch(zoneTree, list, random);
         zoneTree.Maintenance.DiskSegment.InitSparseArray(100);
         DoPrefixSearch(zoneTree, list, random);

diff --git a/src/ZoneTree.UnitTests/ReplicatorTests.cs b/src/ZoneTree.UnitTests/ReplicatorTests.cs
@@ -4,15 +4,6 @@ namespace Tenray.ZoneTree.UnitTests;
 
 public sealed class ReplicatorTests
 {
-    [Test]
-    public void TestReplicator2()
-    {
-        for (int i = 0; i < 100; i++)
-        {
-            TestReplicator();
-        }
-    }
-
     [Test]
     public void TestReplicator()
     {
@@ -21,14 +12,17 @@ public void TestReplicator()
             Directory.Delete(dataPath, true);
         var recordCount = 50_000;
         var keyCount = 15_000;
+        var maxMemory = 2_000;
         void CreateData()
         {
             using var zoneTree = new ZoneTreeFactory<int, int>()
                 .SetDataDirectory(dataPath + "/source")
+                .SetMutableSegmentMaxItemCount(maxMemory)
                 .OpenOrCreate();
 
             using var replica = new ZoneTreeFactory<int, int>()
                 .SetDataDirectory(dataPath + "/replica")
+                .SetMutableSegmentMaxItemCount(maxMemory)
                 .OpenOrCreate();
 
             using var replicator = new Replicator<int, int>(replica, dataPath + "/replica-op-index");
@@ -84,4 +78,83 @@ void TestEqual()
         CreateData();
         TestEqual();
     }
-}
+
+    [Test]
+    public void TestReplicator2()
+    {
+        for (int i = 0; i < 5; i++)
+        {
+            var dataPath = "data/TestReplicator";
+            if (Directory.Exists(dataPath))
+                Directory.Delete(dataPath, true);
+
+            var recordCount = 50_000;
+            var keyCount = 15_000;
+            var maxMemory = 2_000;
+
+            void CreateData()
+            {
+                using var zoneTree = new ZoneTreeFactory<int, int>()
+                    .SetDataDirectory(dataPath + "/source")
+                    .SetMutableSegmentMaxItemCount(maxMemory)
+                    .OpenOrCreate();
+
+                using var replica = new ZoneTreeFactory<int, int>()
+                    .SetDataDirectory(dataPath + "/replica")
+                    .SetMutableSegmentMaxItemCount(maxMemory)
+                    .OpenOrCreate();
+
+                using var replicator = new Replicator<int, int>(replica, dataPath + "/replica-op-index");
+                using var maintainer1 = zoneTree.CreateMaintainer();
+                using var maintainer2 = replica.CreateMaintainer();
+                int replicated = 0;
+                var k = 0;
+                Parallel.For(0, recordCount, (i) =>
+                {
+                    var key = i % keyCount;
+                    var value = Interlocked.Increment(ref k);
+                    var opIndex = zoneTree.Upsert(key, value);
+                    Task.Run(() =>
+                    {
+                        replicator.OnUpsert(key, value, opIndex);
+                        Interlocked.Increment(ref replicated);
+                    });
+                });
+                while (replicated < recordCount) Task.Delay(500).Wait();
+                maintainer1.EvictToDisk();
+                maintainer2.EvictToDisk();
+                maintainer1.WaitForBackgroundThreads();
+                maintainer2.WaitForBackgroundThreads();
+            }
+
+            void TestEqual()
+            {
+                using var zoneTree = new ZoneTreeFactory<int, int>()
+                    .SetDataDirectory(dataPath + "/source")
+                    .Open();
+
+                using var replica = new ZoneTreeFactory<int, int>()
+                    .SetDataDirectory(dataPath + "/replica")
+                    .Open();
+
+                using var iterator1 = zoneTree.CreateIterator();
+                using var iterator2 = replica.CreateIterator();
+                var i = 0;
+                while (true)
+                {
+                    var n1 = iterator1.Next();
+                    var n2 = iterator2.Next();
+                    Assert.That(n2, Is.EqualTo(n1));
+                    if (!n1) break;
+                    Assert.That(iterator2.Current, Is.EqualTo(iterator1.Current));
+                    ++i;
+                }
+                Assert.That(i, Is.EqualTo(keyCount));
+                zoneTree.Maintenance.Drop();
+                replica.Maintenance.Drop();
+            }
+            CreateData();
+            TestEqual();
+        }
+    }
+}
diff --git a/src/ZoneTree.UnitTests/SafeBplusTreeTests.cs b/src/ZoneTree.UnitTests/SafeBplusTreeTests.cs
@@ -22,13 +22,13 @@ public void BTreeIteration(BTreeLockMode lockMode)
             tree.TryInsert(i, i + i, out _);
 
         var iterator = new BTreeSeekableIterator<int, int>(tree);
-        var j = 0; 
+        var j = 0;
         while (iterator.Next())
         {
             Assert.That(iterator.CurrentKey, Is.EqualTo(j));
             Assert.That(iterator.CurrentValue, Is.EqualTo(j + j));
             ++j;
-        } 
+        }
 
         iterator.SeekEnd();
         j = n - 1;
@@ -92,7 +92,7 @@ public void BTreeLowerOrEqualBound(BTreeLockMode lockMode)
             Assert.That(GetLastNodeSmallerOrEqual(iterator, 4), Is.EqualTo(3));
             Assert.That(GetLastNodeSmallerOrEqual(iterator, 3), Is.EqualTo(3));
             Assert.Throws<IndexOutOfRangeException>(
-                () => GetLastNodeSmallerOrEqual(iterator ,- 1));
+                () => GetLastNodeSmallerOrEqual(iterator, -1));
             Assert.That(GetLastNodeSmallerOrEqual(iterator, 10), Is.EqualTo(9));
             Assert.That(GetLastNodeSmallerOrEqual(iterator, 9), Is.EqualTo(9));
             Assert.That(GetLastNodeSmallerOrEqual(iterator, 1), Is.EqualTo(1));
@@ -135,7 +135,7 @@ int GetFirstNodeGreaterOrEqual(BTreeSeekableIterator<int, int> iterator, int key
     [TestCase(BTreeLockMode.NodeLevelMonitor)]
     public void BTreeIteratorParallelInserts(BTreeLockMode lockMode)
     {
-        var random = new Random();
+        var random = Random.Shared;
         var insertCount = 100000;
         var iteratorCount = 1000;
 
@@ -190,7 +190,7 @@ public void BTreeIteratorParallelInserts(BTreeLockMode lockMode)
     [TestCase(BTreeLockMode.NodeLevelMonitor)]
     public void BTreeReverseIteratorParallelInserts(BTreeLockMode lockMode)
     {
-        var random = new Random();
+        var random = Random.Shared;
         var insertCount = 100000;
         var iteratorCount = 1550;
 
@@ -203,7 +203,7 @@ public void BTreeReverseIteratorParallelInserts(BTreeLockMode lockMode)
             {
                 var key = random.Next();
                 tree.AddOrUpdate(key,
-                    AddOrUpdateResult  (ref int x) =>
+                    AddOrUpdateResult (ref int x) =>
                     {
                         x = key + key;
                         return AddOrUpdateResult.ADDED;
@@ -253,7 +253,7 @@ public void BTreeReverseIteratorParallelInserts(BTreeLockMode lockMode)
     [TestCase(BTreeLockMode.NodeLevelMonitor)]
     public void IntIntDuplicateRecords(BTreeLockMode lockMode)
     {
-        var random = new Random();
+        var random = Random.Shared;
         var insertCount = 1000000;
         var iteratorCount = 1000;
 
@@ -297,7 +297,7 @@ public void IntIntDuplicateRecords(BTreeLockMode lockMode)
     [TestCase(BTreeLockMode.NodeLevelMonitor)]
     public void IntIntDuplicateReverseRecords(BTreeLockMode lockMode)
     {
-        var random = new Random();
+        var random = Random.Shared;
         var insertCount = 1000000;
         var iteratorCount = 1000;