Merge pull request #74 from koculu/73-enhancement-create-evicttodisk-method-in-maintainer

koculu · web-flow · commit b2357dae8061 · 2024-08-17T01:47:59.000+02:00
Add EvictToDisk and refactor IMaintainer.
diff --git a/README.md b/README.md
@@ -143,8 +143,8 @@ Note: For small data you don't need a maintainer.
   // 2. Read/Write data
   zoneTree.Upsert(39, "Hello ZoneTree!");
 
-  // 3. Complete maintainer running tasks.
-  maintainer.CompleteRunningTasks();
+  // 3. Wait for background threads.
+  maintainer.WaitForBackgroundThreads();
 ```
 
 ## How to delete keys?
diff --git a/src/Playground/Benchmark/OldTests.cs b/src/Playground/Benchmark/OldTests.cs
@@ -59,7 +59,7 @@ public static void Insert(WriteAheadLogMode mode, int count)
             zoneTree.Maintenance.MoveMutableSegmentForward();
             zoneTree.Maintenance.StartMergeOperation()?.Join();
         }
-        basicMaintainer.CompleteRunningTasks();
+        basicMaintainer.WaitForBackgroundThreads();
         new StatsCollector().LogWithColor(
             "Merged in:",
             stopWatch.ElapsedMilliseconds,
@@ -183,7 +183,7 @@ public static void InsertSingleAndMerge(WriteAheadLogMode mode, int count, int k
         zoneTree.Maintenance.MoveMutableSegmentForward();
         zoneTree.Maintenance.StartMergeOperation()?.Join();
 
-        basicMaintainer.CompleteRunningTasks();
+        basicMaintainer.WaitForBackgroundThreads();
         new StatsCollector().LogWithColor(
             "Merged in:",
             stopWatch.ElapsedMilliseconds,
@@ -222,7 +222,7 @@ public static void Iterate(WriteAheadLogMode mode, int count)
             "Completed in:",
             stopWatch.ElapsedMilliseconds,
             ConsoleColor.Green);
-        basicMaintainer.CompleteRunningTasks();
+        basicMaintainer.WaitForBackgroundThreads();
     }
 
     public static void MultipleIterate(WriteAheadLogMode mode, int count, int iteratorCount)
@@ -262,7 +262,7 @@ public static void MultipleIterate(WriteAheadLogMode mode, int count, int iterat
             "Completed in:",
             stopWatch.ElapsedMilliseconds,
             ConsoleColor.Green);
-        basicMaintainer.CompleteRunningTasks();
+        basicMaintainer.WaitForBackgroundThreads();
     }
 
     private static IZoneTree<int, int> OpenOrCreateZoneTree(WriteAheadLogMode mode, string dataPath)
diff --git a/src/Playground/Benchmark/StevesChallenge.cs b/src/Playground/Benchmark/StevesChallenge.cs
@@ -71,7 +71,7 @@ public void Insert(IStatsCollector stats)
             zoneTree.Maintenance.MoveMutableSegmentForward();
             zoneTree.Maintenance.StartMergeOperation()?.Join();
         }
-        maintainer.CompleteRunningTasks();
+        maintainer.WaitForBackgroundThreads();
 
         stats.AddStage("Merged In", ConsoleColor.DarkCyan);
     }
@@ -100,7 +100,7 @@ public void Iterate(IStatsCollector stats)
         stats.AddStage(
             "Iterated in",
             ConsoleColor.Green);
-        maintainer.CompleteRunningTasks();
+        maintainer.WaitForBackgroundThreads();
     }
 
 }
diff --git a/src/Playground/Benchmark/ZoneTreeTest1.cs b/src/Playground/Benchmark/ZoneTreeTest1.cs
@@ -51,7 +51,7 @@ public void Insert(IStatsCollector stats)
             zoneTree.Maintenance.MoveMutableSegmentForward();
             zoneTree.Maintenance.StartMergeOperation()?.Join();
         }
-        maintainer.CompleteRunningTasks();
+        maintainer.WaitForBackgroundThreads();
 
         stats.AddStage("Merged In", ConsoleColor.DarkCyan);
     }
@@ -82,6 +82,6 @@ public void Iterate(IStatsCollector stats)
         stats.AddStage(
             "Iterated in",
             ConsoleColor.Green);
-        maintainer.CompleteRunningTasks();
+        maintainer.WaitForBackgroundThreads();
     }
 }
diff --git a/src/Playground/Benchmark/ZoneTreeTest2.cs b/src/Playground/Benchmark/ZoneTreeTest2.cs
@@ -54,7 +54,7 @@ public void Insert(IStatsCollector stats)
             zoneTree.Maintenance.MoveMutableSegmentForward();
             zoneTree.Maintenance.StartMergeOperation()?.Join();
         }
-        maintainer.CompleteRunningTasks();
+        maintainer.WaitForBackgroundThreads();
 
         stats.AddStage("Merged In", ConsoleColor.DarkCyan);
     }
@@ -85,6 +85,6 @@ public void Iterate(IStatsCollector stats)
         stats.AddStage(
             "Iterated in",
             ConsoleColor.Green);
-        maintainer.CompleteRunningTasks();
+        maintainer.WaitForBackgroundThreads();
     }
 }
diff --git a/src/Playground/Benchmark/ZoneTreeTest3.cs b/src/Playground/Benchmark/ZoneTreeTest3.cs
@@ -51,7 +51,7 @@ public void Insert(IStatsCollector stats)
             zoneTree.Maintenance.ZoneTree.Maintenance.MoveMutableSegmentForward();
             zoneTree.Maintenance.ZoneTree.Maintenance.StartMergeOperation()?.Join();
         }
-        maintainer.CompleteRunningTasks();
+        maintainer.WaitForBackgroundThreads();
 
         stats.AddStage("Merged In", ConsoleColor.DarkCyan);
     }
@@ -82,6 +82,6 @@ public void Iterate(IStatsCollector stats)
         stats.AddStage(
             "Iterated in",
             ConsoleColor.Green);
-        maintainer.CompleteRunningTasks();
+        maintainer.WaitForBackgroundThreads();
     }
 }
diff --git a/src/Playground/Test1.cs b/src/Playground/Test1.cs
@@ -63,7 +63,7 @@ public static void SeveralParallelTransactions()
             });
         }
         Console.WriteLine("Elapsed: " + stopWatch.ElapsedMilliseconds);
-        basicMaintainer.CompleteRunningTasks();
+        basicMaintainer.WaitForBackgroundThreads();
         zoneTree.Maintenance.SaveMetaData();
     }
 
@@ -110,7 +110,7 @@ public static void TestReverseIterator(
                         return true;
                     });
             }
-            maintainer.CompleteRunningTasks();
+            maintainer.WaitForBackgroundThreads();
         }
 
         upload.Stop();
@@ -247,7 +247,7 @@ public static void TestIteratorBehavior(
         });
         t1.Wait();
         zoneTree1.Maintenance.TryCancelMergeOperation();
-        maintainer.CompleteRunningTasks();
+        maintainer.WaitForBackgroundThreads();
     }
 
 
diff --git a/src/ZoneTree/Core/ZoneTreeMaintainer.cs b/src/ZoneTree/Core/ZoneTreeMaintainer.cs
@@ -1,4 +1,5 @@
 ﻿using System.Collections.Concurrent;
+using System.Threading;
 using Tenray.ZoneTree.Logger;
 using Tenray.ZoneTree.Segments;
 using Tenray.ZoneTree.Segments.Disk;
@@ -36,12 +37,6 @@ public sealed class ZoneTreeMaintainer<TKey, TValue> : IMaintainer, IDisposable
     /// </summary>
     public IZoneTreeMaintenance<TKey, TValue> Maintenance { get; }
 
-    /// <inheritdoc/>
-    public int MinimumSparseArrayLength { get; set; }
-
-    /// <inheritdoc/>
-    public int SparseArrayStepLength { get; set; } = 1_000;
-
     /// <inheritdoc/>
     public int ThresholdForMergeOperationStart { get; set; } = 0;
 
@@ -105,17 +100,27 @@ void AttachEvents()
         Maintenance.OnDiskSegmentCreated += OnDiskSegmentCreated;
         Maintenance.OnMergeOperationEnded += OnMergeOperationEnded;
         Maintenance.OnZoneTreeIsDisposing += OnZoneTreeIsDisposing;
+        Maintenance.OnBottomSegmentsMergeOperationEnded += OnBottomSegmentsMergeOperationEnded;
     }
 
+
     void OnZoneTreeIsDisposing(IZoneTreeMaintenance<TKey, TValue> zoneTree)
     {
         Trace("ZoneTree is disposing. ZoneTreeMaintainer disposal started.");
         PeriodicTimerCancellationTokenSource.Cancel();
-        CompleteRunningTasks();
+        WaitForBackgroundThreads();
         Dispose();
         Trace("ZoneTreeMaintainer is disposed.");
     }
 
+    void OnBottomSegmentsMergeOperationEnded(
+        IZoneTreeMaintenance<TKey, TValue> zoneTree,
+        MergeResult mergeResult)
+    {
+        Trace(mergeResult.ToString());
+        MergerThreads.Remove(Environment.CurrentManagedThreadId, out _);
+    }
+
     void OnMergeOperationEnded(
         IZoneTreeMaintenance<TKey, TValue> zoneTree,
         MergeResult mergeResult)
@@ -168,7 +173,8 @@ void OnMutableSegmentMovedForward(IZoneTreeMaintenance<TKey, TValue> zoneTree)
             StartMerge();
     }
 
-    void StartMerge()
+    /// <inheritdoc/>
+    public void StartMerge()
     {
         lock (this)
         {
@@ -182,14 +188,39 @@ void StartMerge()
         }
     }
 
+
+    /// <inheritdoc/>
+    public void StartBottomSegmentsMerge(
+        int fromIndex = 0, int toIndex = int.MaxValue)
+    {
+        lock (this)
+        {
+            var mergerThread = Maintenance
+                .StartBottomSegmentsMergeOperation(fromIndex, toIndex);
+            if (mergerThread == null)
+                return;
+            MergerThreads.AddOrUpdate(
+                mergerThread.ManagedThreadId,
+                mergerThread,
+                (key, value) => mergerThread);
+        }
+    }
+
     /// <inheritdoc/>
-    public void TryCancelRunningTasks()
+    public void TryCancelBackgroundThreads()
     {
         Maintenance.TryCancelMergeOperation();
+        Maintenance.TryCancelBottomSegmentsMergeOperation();
     }
 
     /// <inheritdoc/>
-    public void CompleteRunningTasks()
+    public void WaitForBackgroundThreads()
+    {
+        WaitForBackgroundThreadsAsync().Wait();
+    }
+
+    /// <inheritdoc/>
+    public async Task WaitForBackgroundThreadsAsync()
     {
         while (true)
         {
@@ -203,7 +234,7 @@ public void CompleteRunningTasks()
                 if (t.ThreadState == ThreadState.Stopped)
                     MergerThreads.TryRemove(a.Key, out var _);
                 else
-                    t.Join();
+                    await Task.Run(() => t.Join());
             }
             Trace("Wait ended");
         }
@@ -245,6 +276,13 @@ void Trace(string msg)
         Logger.LogTrace(msg);
     }
 
+    /// <inheritdoc/>
+    public void EvictToDisk()
+    {
+        Maintenance.MoveMutableSegmentForward();
+        StartMerge();
+    }
+
     /// <summary>
     /// Disposes this maintainer.
     /// </summary>
@@ -256,5 +294,6 @@ public void Dispose()
         Maintenance.OnDiskSegmentCreated -= OnDiskSegmentCreated;
         Maintenance.OnMergeOperationEnded -= OnMergeOperationEnded;
         Maintenance.OnZoneTreeIsDisposing -= OnZoneTreeIsDisposing;
+        Maintenance.OnBottomSegmentsMergeOperationEnded -= OnBottomSegmentsMergeOperationEnded;
     }
 }
diff --git a/src/ZoneTree/IMaintainer.cs b/src/ZoneTree/IMaintainer.cs
@@ -5,27 +5,11 @@
 /// merge operations and memory compaction.
 /// </summary>
 /// <remarks>
-/// You must complete or cancel all pending tasks of this maintainer
+/// You must complete or cancel all pending threads of this maintainer
 /// before disposing.
 /// </remarks>
 public interface IMaintainer : IDisposable
 {
-    /// <summary>
-    /// Minimum sparse array length when a new disk segment is created.
-    /// Default value is 0.
-    /// </summary>
-    int MinimumSparseArrayLength { get; set; }
-
-    /// <summary>
-    /// Configures sparse array step length when the disk segment length is bigger than
-    /// MinimumSparseArrayLength * SparseArrayStepLength.
-    /// The default value is 1000.
-    /// <remarks>The sparse array length reduce binary lookup range on disk segment
-    /// to reduce IO.
-    /// </remarks>
-    /// </summary>
-    int SparseArrayStepLength { get; set; }
-
     /// <summary>
     /// Starts merge operation when records count
     /// in read-only segments exceeds this value.
@@ -61,12 +45,43 @@ public interface IMaintainer : IDisposable
     TimeSpan InactiveBlockCacheCleanupInterval { get; set; }
 
     /// <summary>
-    /// Tries cancel running tasks.
+    /// Tries cancel background threads.
+    /// </summary>
+    void TryCancelBackgroundThreads();
+
+    /// <summary>
+    /// Blocks the calling thread until all background threads have completed their execution.
+    /// </summary>
+    void WaitForBackgroundThreads();
+
+    /// <summary>
+    /// Asynchronously waits for all background threads to complete.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous wait operation.</returns>
+    Task WaitForBackgroundThreadsAsync();
+
+    /// <summary>
+    /// Evicts all in-memory data to disk by moving the mutable segment forward and initiating a merge process.
+    /// </summary>
+    /// <remarks>
+    /// This method is responsible for freeing up memory in the LSM tree by moving data from the mutable in-memory segment to disk storage. 
+    /// It first advances the current mutable segment to a new state, ensuring that any data currently in memory is prepared for disk storage. 
+    /// Afterward, it starts the merging process, which combines the in-memory data with existing on-disk data to maintain the integrity 
+    /// and efficiency of the LSM tree structure.
+    /// </remarks>
+    void EvictToDisk();
+
+    /// <summary>
+    /// Initiates the merge process in a new thread.
     /// </summary>
-    void TryCancelRunningTasks();
+    void StartMerge();
 
     /// <summary>
-    /// Waits until all running tasks are completed.
+    /// Initiates a merge of selected bottom segments into a single bottom disk segment.
     /// </summary>
-    void CompleteRunningTasks();
+    /// <param name="fromIndex">The lower bound</param>
+    /// <param name="toIndex">The upper bound</param>
+    /// <returns></returns>
+    void StartBottomSegmentsMerge(
+        int fromIndex = 0, int toIndex = int.MaxValue);
 }
diff --git a/src/ZoneTree/docs/ZoneTree/README-NUGET.md b/src/ZoneTree/docs/ZoneTree/README-NUGET.md
@@ -141,8 +141,8 @@ Note: For small data you don't need a maintainer.
   // 2. Read/Write data
   zoneTree.Upsert(39, "Hello ZoneTree!");
 
-  // 3. Complete maintainer running tasks.
-  maintainer.CompleteRunningTasks();
+  // 3. Wait for background threads.
+  maintainer.WaitForBackgroundThreads();
 ```
 
 ## How to delete keys?
diff --git a/src/ZoneTree/docs/ZoneTree/guide/quick-start.md b/src/ZoneTree/docs/ZoneTree/guide/quick-start.md
@@ -61,7 +61,7 @@ for (var i = 0; i < 10_000_000; ++i){
 }
  
 // Ensure maintainer merge operations are completed before tree disposal.
-maintainer.CompleteRunningTasks();
+maintainer.WaitForBackgroundThreads();
 ```
 
 ## How to merge data to the disk segment manually?

Original file line number	Diff line number	Diff line change
`@@ -71,7 +71,7 @@ public void Insert(IStatsCollector stats)`
`71`	`71`	`zoneTree.Maintenance.MoveMutableSegmentForward();`
`72`	`72`	`zoneTree.Maintenance.StartMergeOperation()?.Join();`
`73`	`73`	`}`
`74`		`- maintainer.CompleteRunningTasks();`
	`74`	`+ maintainer.WaitForBackgroundThreads();`
`75`	`75`
`76`	`76`	`stats.AddStage("Merged In", ConsoleColor.DarkCyan);`
`77`	`77`	`}`
`@@ -100,7 +100,7 @@ public void Iterate(IStatsCollector stats)`
`100`	`100`	`stats.AddStage(`
`101`	`101`	`"Iterated in",`
`102`	`102`	`ConsoleColor.Green);`
`103`		`- maintainer.CompleteRunningTasks();`
	`103`	`+ maintainer.WaitForBackgroundThreads();`
`104`	`104`	`}`
`105`	`105`
`106`	`106`	`}`
Original file line number	Diff line number	Diff line change
`@@ -51,7 +51,7 @@ public void Insert(IStatsCollector stats)`
`51`	`51`	`zoneTree.Maintenance.MoveMutableSegmentForward();`
`52`	`52`	`zoneTree.Maintenance.StartMergeOperation()?.Join();`
`53`	`53`	`}`
`54`		`- maintainer.CompleteRunningTasks();`
	`54`	`+ maintainer.WaitForBackgroundThreads();`
`55`	`55`
`56`	`56`	`stats.AddStage("Merged In", ConsoleColor.DarkCyan);`
`57`	`57`	`}`
`@@ -82,6 +82,6 @@ public void Iterate(IStatsCollector stats)`
`82`	`82`	`stats.AddStage(`
`83`	`83`	`"Iterated in",`
`84`	`84`	`ConsoleColor.Green);`
`85`		`- maintainer.CompleteRunningTasks();`
	`85`	`+ maintainer.WaitForBackgroundThreads();`
`86`	`86`	`}`
`87`	`87`	`}`
Original file line number	Diff line number	Diff line change
`@@ -54,7 +54,7 @@ public void Insert(IStatsCollector stats)`
`54`	`54`	`zoneTree.Maintenance.MoveMutableSegmentForward();`
`55`	`55`	`zoneTree.Maintenance.StartMergeOperation()?.Join();`
`56`	`56`	`}`
`57`		`- maintainer.CompleteRunningTasks();`
	`57`	`+ maintainer.WaitForBackgroundThreads();`
`58`	`58`
`59`	`59`	`stats.AddStage("Merged In", ConsoleColor.DarkCyan);`
`60`	`60`	`}`
`@@ -85,6 +85,6 @@ public void Iterate(IStatsCollector stats)`
`85`	`85`	`stats.AddStage(`
`86`	`86`	`"Iterated in",`
`87`	`87`	`ConsoleColor.Green);`
`88`		`- maintainer.CompleteRunningTasks();`
	`88`	`+ maintainer.WaitForBackgroundThreads();`
`89`	`89`	`}`
`90`	`90`	`}`
Original file line number	Diff line number	Diff line change
`@@ -63,7 +63,7 @@ public static void SeveralParallelTransactions()`
`63`	`63`	`});`
`64`	`64`	`}`
`65`	`65`	`Console.WriteLine("Elapsed: " + stopWatch.ElapsedMilliseconds);`
`66`		`- basicMaintainer.CompleteRunningTasks();`
	`66`	`+ basicMaintainer.WaitForBackgroundThreads();`
`67`	`67`	`zoneTree.Maintenance.SaveMetaData();`
`68`	`68`	`}`
`69`	`69`
`@@ -110,7 +110,7 @@ public static void TestReverseIterator(`
`110`	`110`	`return true;`
`111`	`111`	`});`
`112`	`112`	`}`
`113`		`- maintainer.CompleteRunningTasks();`
	`113`	`+ maintainer.WaitForBackgroundThreads();`
`114`	`114`	`}`
`115`	`115`
`116`	`116`	`upload.Stop();`
`@@ -247,7 +247,7 @@ public static void TestIteratorBehavior(`
`247`	`247`	`});`
`248`	`248`	`t1.Wait();`
`249`	`249`	`zoneTree1.Maintenance.TryCancelMergeOperation();`
`250`		`- maintainer.CompleteRunningTasks();`
	`250`	`+ maintainer.WaitForBackgroundThreads();`
`251`	`251`	`}`
`252`	`252`
`253`	`253`