Platform specific prefetcher

src/Paprika.Tests/Store/BasePageTests.cs

@@ -27,8 +27,9 @@ internal class TestBatchContext(uint batchId, Stack<DbAddress>? reusable = null)
 
         public override Page GetAt(DbAddress address) => _address2Page[address];
 
-        public override void Prefetch(DbAddress address)
-        { }
+        public override void Prefetch(ReadOnlySpan<DbAddress> addresses)
+        {
+        }
 
         public override DbAddress GetAddress(Page page) => _page2Address[page.Raw];
 
@@ -106,4 +107,4 @@ public TestBatchContext Next()
     }
 
     internal static TestBatchContext NewBatch(uint batchId) => new(batchId);
-}
+}

src/Paprika/Platform.cs

@@ -0,0 +1,96 @@
+using System.ComponentModel;
+using System.Diagnostics;
+using System.Runtime.CompilerServices;
+using System.Runtime.InteropServices;
+
+namespace Paprika;
+
+public static class Platform
+{
+    public static void Prefetch(ReadOnlySpan<UIntPtr> addresses, UIntPtr size) => Manager.SchedulePrefetch(addresses, size);
+
+    private static readonly IMemoryManager Manager =
+        IsPosix() ? new PosixMemoryManager() : new WindowsMemoryManager();
+
+    private static bool IsPosix() => RuntimeInformation.IsOSPlatform(OSPlatform.Linux) ||
+                                     RuntimeInformation.IsOSPlatform(OSPlatform.OSX);
+
+    private sealed class PosixMemoryManager : IMemoryManager
+    {
+        [Flags]
+        private enum Advice : int
+        {
+            // ReSharper disable InconsistentNaming
+            /// <summary>
+            ///  Expect access in the near future. (Hence, it might be a good idea to read some pages ahead.)
+            /// </summary>
+            MADV_WILLNEED = 0x3,
+
+            /// <summary>
+            ///  Do not expect access in the near future.
+            /// (For the time being, the application is finished with the given range,
+            /// so the kernel can free resources associated with it.)
+            /// </summary>
+            MADV_DONTNEED = 0x4,
+            // ReSharper restore InconsistentNaming
+        }
+
+        [DllImport("LIBC_6", SetLastError = true)]
+        static extern int madvise(IntPtr addr, UIntPtr length, Advice advice);
+
+        public void SchedulePrefetch(ReadOnlySpan<UIntPtr> addresses, UIntPtr length)
+        {
+            // TODO:
+        }
+    }
+
+    private sealed class WindowsMemoryManager : IMemoryManager
+    {
+        [DllImport("kernel32.dll", SetLastError = true)]
+        private static extern unsafe bool PrefetchVirtualMemory(IntPtr hProcess, ulong numberOfEntries,
+            Win32MemoryRangeEntry* entries, int flags);
+
+        [StructLayout(LayoutKind.Sequential)]
+        private struct Win32MemoryRangeEntry
+        {
+            /// <summary>
+            /// Starting address of the memory range
+            /// </summary>
+            public IntPtr VirtualAddress;
+
+            /// <summary>
+            /// Size of the memory range in bytes
+            /// </summary>
+            public UIntPtr NumberOfBytes;
+        }
+
+        [SkipLocalsInit]
+        public unsafe void SchedulePrefetch(ReadOnlySpan<UIntPtr> addresses, UIntPtr length)
+        {
+            var count = addresses.Length;
+            var ptr = stackalloc Win32MemoryRangeEntry[count];
+            var span = new Span<Win32MemoryRangeEntry>(ptr, count);
+
+            for (var i = 0; i < span.Length; i++)
+            {
+                span[i].VirtualAddress = (IntPtr)addresses[i];
+                span[i].NumberOfBytes = length;
+            }
+
+            const int reserved = 0;
+
+            if (PrefetchVirtualMemory(Process.GetCurrentProcess().Handle, (ulong)count, ptr, reserved) == false)
+            {
+                throw new Win32Exception(Marshal.GetLastWin32Error());
+            }
+        }
+    }
+
+    private interface IMemoryManager
+    {
+        /// <summary>
+        /// Schedules an OS dependent prefetch.
+        /// </summary>
+        void SchedulePrefetch(ReadOnlySpan<UIntPtr> addresses, UIntPtr length);
+    }
+}

src/Paprika/Store/AbandonedList.cs

@@ -139,6 +139,13 @@ public bool TryGet(out DbAddress reused, uint minBatchId, IBatchContext batch)
 
         if (current.TryPop(out reused))
         {
+            // Schedule prefetching next if possible
+            if (current.TryPeek(out var next, out _))
+            {
+                Debug.Assert(next.IsNull == false, "Next should not be NULL here");
+                batch.Prefetch(next);
+            }
+
             return true;
         }
 

src/Paprika/Store/BatchContextBase.cs

@@ -10,7 +10,7 @@ abstract class BatchContextBase(uint batchId) : IBatchContext
     public uint BatchId { get; } = batchId;
 
     public abstract Page GetAt(DbAddress address);
-    public abstract void Prefetch(DbAddress address);
+    public abstract void Prefetch(ReadOnlySpan<DbAddress> addresses);
 
     public abstract DbAddress GetAddress(Page page);
 

src/Paprika/Store/IBatchContext.cs

@@ -1,5 +1,7 @@
 using System.Diagnostics;
 using System.Diagnostics.CodeAnalysis;
+using System.Runtime.CompilerServices;
+using System.Runtime.InteropServices;
 using Paprika.Crypto;
 
 namespace Paprika.Store;
@@ -126,29 +128,22 @@ public interface IPageResolver
     /// </summary>
     Page GetAt(DbAddress address);
 
-    void Prefetch(DbAddress address);
+    void Prefetch(DbAddress address) => Prefetch(MemoryMarshal.CreateReadOnlySpan(ref address, 1));
 
-    void Prefetch(ReadOnlySpan<DbAddress> addresses)
-    {
-        foreach (var bucket in addresses)
-        {
-            if (!bucket.IsNull)
-            {
-                Prefetch(bucket);
-            }
-        }
-    }
+    void Prefetch(ReadOnlySpan<DbAddress> addresses);
 
+    [SkipLocalsInit]
     void Prefetch<TAddressList>(in TAddressList addresses)
         where TAddressList : struct, DbAddressList.IDbAddressList
     {
+        Span<DbAddress> span = stackalloc DbAddress[TAddressList.Length];
+
+        // Copy all
         for (var i = 0; i < TAddressList.Length; i++)
         {
-            var addr = addresses[i];
-            if (!addr.IsNull)
-            {
-                Prefetch(addr);
-            }
+            span[i] = addresses[i];
         }
+
+        Prefetch(span);
     }
 }

src/Paprika/Store/PageManagers/MemoryMappedPageManager.cs

@@ -2,7 +2,10 @@
 using System.Diagnostics;
 using System.Diagnostics.Metrics;
 using System.IO.MemoryMappedFiles;
+using System.Runtime.CompilerServices;
 using System.Runtime.InteropServices;
+using System.Threading.Channels;
+using System.Xml.Serialization;
 using Paprika.Utils;
 
 namespace Paprika.Store.PageManagers;
@@ -24,6 +27,18 @@ public sealed class MemoryMappedPageManager : PointerPageManager
     private readonly MemoryMappedViewAccessor _whole;
     private readonly unsafe byte* _ptr;
 
+    // Prefetcher
+    private const int PrefetcherCapacity = 1000;
+    private readonly Channel<DbAddress> _prefetches = Channel.CreateBounded<DbAddress>(new BoundedChannelOptions(PrefetcherCapacity)
+    {
+        FullMode = BoundedChannelFullMode.DropOldest,
+        SingleReader = true,
+        SingleWriter = false,
+        Capacity = PrefetcherCapacity,
+        AllowSynchronousContinuations = false
+    });
+    private readonly Task _prefetcher;
+
     // Flusher section
     private readonly Stack<PageMemoryOwner> _owners = new();
     private readonly List<PageMemoryOwner> _ownersUsed = new();
@@ -72,6 +87,8 @@ public unsafe MemoryMappedPageManager(long size, byte historyDepth, string dir,
         _meter = new Meter("Paprika.Store.PageManager");
         _fileWrites = _meter.CreateHistogram<int>("File writes", "Syscall", "Actual numbers of file writes issued");
         _writeTime = _meter.CreateHistogram<int>("Write time", "ms", "Time spent in writing");
+
+        _prefetcher = Task.Factory.StartNew(RunPrefetcher);
     }
 
     public static string GetPaprikaFilePath(string dir) => System.IO.Path.Combine(dir, PaprikaFileName);
@@ -80,6 +97,51 @@ public unsafe MemoryMappedPageManager(long size, byte historyDepth, string dir,
 
     protected override unsafe void* Ptr => _ptr;
 
+    public override void Prefetch(ReadOnlySpan<DbAddress> addresses)
+    {
+        var writer = _prefetches.Writer;
+
+        foreach (var address in addresses)
+        {
+            if (address.IsNull == false)
+            {
+                writer.TryWrite(address);
+            }
+        }
+    }
+
+    private async Task RunPrefetcher()
+    {
+        var reader = _prefetches.Reader;
+
+        while (await reader.WaitToReadAsync())
+        {
+            PrefetchImpl(reader, this);
+        }
+
+        [SkipLocalsInit]
+        static void PrefetchImpl(ChannelReader<DbAddress> reader, MemoryMappedPageManager manager)
+        {
+            const int maxPrefetch = 128;
+
+            Span<UIntPtr> span = stackalloc UIntPtr[maxPrefetch];
+            var i = 0;
+
+            for (; i < maxPrefetch; i++)
+            {
+                if (reader.TryRead(out var address) == false)
+                    break;
+
+                span[i] = manager.GetAt(address).Raw;
+            }
+
+            if (i > 0)
+            {
+                Platform.Prefetch(span[..i], Page.PageSize);
+            }
+        }
+    }
+
     public override async ValueTask FlushPages(ICollection<DbAddress> dbAddresses, CommitOptions options)
     {
         if (_options == PersistenceOptions.MMapOnly)
@@ -182,6 +244,9 @@ public override void ForceFlush()
 
     public override void Dispose()
     {
+        _prefetches.Writer.Complete();
+        _prefetcher.Wait();
+
         _meter.Dispose();
 
         _whole.SafeMemoryMappedViewHandle.ReleasePointer();

src/Paprika/Store/PageManagers/PointerPageManager.cs

@@ -11,18 +11,21 @@ public abstract unsafe class PointerPageManager(long size) : IPageManager
 
     protected abstract void* Ptr { get; }
 
-    public void Prefetch(DbAddress address)
+    public virtual void Prefetch(ReadOnlySpan<DbAddress> addresses)
     {
         if (Sse.IsSupported)
         {
-            if (address.IsNull || address.Raw > (uint)MaxPage)
+            foreach (var address in addresses)
             {
-                return;
-            }
+                if (address.IsNull || address.Raw > (uint)MaxPage)
+                {
+                    return;
+                }
 
-            // Fetch to L2 cache as we don't know if will need it
-            // So don't pollute L1 cache
-            Sse.Prefetch1((byte*)Ptr + address.FileOffset);
+                // Fetch to L2 cache as we don't know if will need it
+                // So don't pollute L1 cache
+                Sse.Prefetch1((byte*)Ptr + address.FileOffset);
+            }
         }
     }
 

src/Paprika/Store/PagedDb.cs

@@ -126,7 +126,7 @@ public static PagedDb MemoryMappedDb(long size, byte historyDepth, string direct
             new MemoryMappedPageManager(size, historyDepth, directory,
                 flushToDisk ? PersistenceOptions.FlushFile : PersistenceOptions.MMapOnly), historyDepth);
 
-    public void Prefetch(DbAddress address) => _manager.Prefetch(address);
+    public void Prefetch(ReadOnlySpan<DbAddress> addresses) => _manager.Prefetch(addresses);
 
     private void ReportReads(long number) => _reads.Add(number);
 
@@ -506,7 +506,7 @@ public IDictionary<Keccak, uint> IdCache
             }
         }
 
-        public void Prefetch(DbAddress address) => db.Prefetch(address);
+        public void Prefetch(ReadOnlySpan<DbAddress> addresses) => db.Prefetch(addresses);
 
         public Page GetAt(DbAddress address) => db._manager.GetAt(address);
 
@@ -678,7 +678,7 @@ public override Page GetAt(DbAddress address)
             return page;
         }
 
-        public override void Prefetch(DbAddress address) => _db.Prefetch(address);
+        public override void Prefetch(ReadOnlySpan<DbAddress> addresses) => _db.Prefetch(addresses);
 
         public override DbAddress GetAddress(Page page) => _db.GetAddress(page);