Exposing FASTER log as a first-class abstraction

cs/FASTER.sln

@@ -44,6 +44,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "devices", "devices", "{A6B1
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "FASTER.devices.AzureStorageDevice", "src\devices\AzureStorageDevice\FASTER.devices.AzureStorageDevice.csproj", "{E571E686-01A0-44D5-BFF5-B7678284258B}"
 EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "FasterLogSample", "playground\FasterLogSample\FasterLogSample.csproj", "{25C5C6B6-4A8A-46DD-88C1-EB247033FE58}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -156,6 +158,12 @@ Global
 		{E571E686-01A0-44D5-BFF5-B7678284258B}.Release|Any CPU.Build.0 = Release|Any CPU
 		{E571E686-01A0-44D5-BFF5-B7678284258B}.Release|x64.ActiveCfg = Release|x64
 		{E571E686-01A0-44D5-BFF5-B7678284258B}.Release|x64.Build.0 = Release|x64
+		{25C5C6B6-4A8A-46DD-88C1-EB247033FE58}.Debug|Any CPU.ActiveCfg = Debug|x64
+		{25C5C6B6-4A8A-46DD-88C1-EB247033FE58}.Debug|x64.ActiveCfg = Debug|x64
+		{25C5C6B6-4A8A-46DD-88C1-EB247033FE58}.Debug|x64.Build.0 = Debug|x64
+		{25C5C6B6-4A8A-46DD-88C1-EB247033FE58}.Release|Any CPU.ActiveCfg = Release|x64
+		{25C5C6B6-4A8A-46DD-88C1-EB247033FE58}.Release|x64.ActiveCfg = Release|x64
+		{25C5C6B6-4A8A-46DD-88C1-EB247033FE58}.Release|x64.Build.0 = Release|x64
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -176,6 +184,7 @@ Global
 		{7EBB5ADF-D9EA-4B8B-AAE7-C48A98EBF780} = {E6026D6A-01C5-4582-B2C1-64751490DABE}
 		{A6B14415-D316-4955-BE5F-725BB2DEBEBE} = {28800357-C8CE-4CD0-A2AD-D4A910ABB496}
 		{E571E686-01A0-44D5-BFF5-B7678284258B} = {A6B14415-D316-4955-BE5F-725BB2DEBEBE}
+		{25C5C6B6-4A8A-46DD-88C1-EB247033FE58} = {E6026D6A-01C5-4582-B2C1-64751490DABE}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {A0750637-2CCB-4139-B25E-F2CE740DCFAC}

cs/playground/FasterLogSample/App.config

@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="utf-8"?>
+<configuration>
+    <startup> 
+        <supportedRuntime version="v4.0" sku=".NETFramework,Version=v4.6" />
+    </startup>
+</configuration>

cs/playground/FasterLogSample/FasterLogSample.csproj

@@ -0,0 +1,38 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>netcoreapp2.2</TargetFramework>
+    <Platforms>x64</Platforms>
+    <RuntimeIdentifier>win7-x64</RuntimeIdentifier>
+  </PropertyGroup>
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
+    <RootNamespace>StructSample</RootNamespace>
+    <ErrorReport>prompt</ErrorReport>
+    <RestoreProjectStyle>PackageReference</RestoreProjectStyle>
+    <Prefer32Bit>true</Prefer32Bit>
+  </PropertyGroup>
+
+  <PropertyGroup Condition="'$(Configuration)' == 'Debug'">
+    <DefineConstants>TRACE;DEBUG</DefineConstants>
+    <DebugType>full</DebugType>
+    <DebugSymbols>true</DebugSymbols>
+    <OutputPath>bin\x64\Debug\</OutputPath>
+  </PropertyGroup>
+  <PropertyGroup Condition="'$(Configuration)' == 'Release'">
+    <DefineConstants>TRACE</DefineConstants>
+    <DebugType>pdbonly</DebugType>
+    <Optimize>true</Optimize>
+    <OutputPath>bin\x64\Release\</OutputPath>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <None Include="App.config" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\src\core\FASTER.core.csproj" />
+  </ItemGroup>
+</Project>

cs/playground/FasterLogSample/Program.cs

@@ -0,0 +1,236 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT license.
+
+using System;
+using System.Diagnostics;
+using System.Threading;
+using System.Threading.Tasks;
+using FASTER.core;
+
+namespace FasterLogSample
+{
+    public class Program
+    {
+        // Entry length can be between 1 and ((1 << FasterLogSettings.PageSizeBits) - 4)
+        const int entryLength = 1 << 10;
+        static readonly byte[] staticEntry = new byte[entryLength];
+        static readonly ReadOnlySpanBatch spanBatch = new ReadOnlySpanBatch(10);
+        static FasterLog log;
+
+        static void ReportThread()
+        {
+            long lastTime = 0;
+            long lastValue = log.TailAddress;
+            Stopwatch sw = new Stopwatch();
+            sw.Start();
+
+            while (true)
+            {
+                Thread.Sleep(5000);
+                var nowTime = sw.ElapsedMilliseconds;
+                var nowValue = log.TailAddress;
+
+                Console.WriteLine("Throughput: {0} MB/sec, Tail: {1}",
+                    (nowValue - lastValue) / (1000 * (nowTime - lastTime)), nowValue);
+                lastTime = nowTime;
+                lastValue = nowValue;
+            }
+        }
+
+        static void CommitThread()
+        {
+            while (true)
+            {
+                Thread.Sleep(5);
+                log.FlushAndCommit(true);
+
+                // Async version
+                // await Task.Delay(5);
+                // await log.FlushAndCommitAsync();
+            }
+        }
+
+        static void AppendThread()
+        {
+            while (true)
+            {
+                // TryAppend - can be used with throttling/back-off
+                // Accepts byte[] and ReadOnlySpan<byte>
+                while (!log.TryAppend(staticEntry, out _)) ;
+
+                // Synchronous blocking append
+                // Accepts byte[] and ReadOnlySpan<byte>
+                // log.Append(entry);
+
+                // Batched append - batch must fit on one page
+                // while (!log.TryAppend(spanBatch, out _)) ;
+            }
+        }
+
+        static void ScanThread()
+        {
+            Random r = new Random();
+
+            Thread.Sleep(5000);
+
+            byte[] entry = new byte[entryLength];
+            for (int i = 0; i < entryLength; i++)
+            {
+                entry[i] = (byte)i;
+            }
+
+            var entrySpan = new Span<byte>(entry);
+
+            long lastAddress = 0;
+            Span<byte> result;
+            using (var iter = log.Scan(0, long.MaxValue))
+            {
+                while (true)
+                {
+                    while (!iter.GetNext(out result, out int length))
+                    {
+                        iter.WaitAsync().GetAwaiter().GetResult();
+                    }
+
+                    if (!result.SequenceEqual(entrySpan))
+                    {
+                        if (result.Length != entrySpan.Length)
+                            throw new Exception("Invalid entry found, expected length " + entrySpan.Length + ", actual length " + result.Length);
+                        else
+                            throw new Exception("Invalid entry found at offset " + FindDiff(result, entrySpan));
+                    }
+
+                    // Re-insert entry with small probability
+                    if (r.Next(100) < 10)
+                    {
+                        log.Append(result);
+                    }
+
+                    if (iter.CurrentAddress - lastAddress > 500_000_000)
+                    {
+                        log.TruncateUntil(iter.CurrentAddress);
+                        lastAddress = iter.CurrentAddress;
+                    }
+                }
+            }
+        }
+
+        private static int FindDiff(Span<byte> b1, Span<byte> b2)
+        {
+            for (int i = 0; i < b1.Length; i++)
+            {
+                if (b1[i] != b2[i])
+                {
+                    return i;
+                }
+            }
+            return 0;
+        }
+
+        /// <summary>
+        /// Main program entry point
+        /// </summary>
+        /// <param name="args"></param>
+        static void Main(string[] args)
+        {
+            bool sync = true;
+            var device = Devices.CreateLogDevice("D:\\logs\\hlog.log");
+            log = new FasterLog(new FasterLogSettings { LogDevice = device });
+
+            // Populate entry being inserted
+            for (int i = 0; i < entryLength; i++)
+            {
+                staticEntry[i] = (byte)i;
+            }
+
+            if (sync)
+            {
+                // Append thread: create as many as needed
+                new Thread(new ThreadStart(AppendThread)).Start();
+
+                // Threads for scan, reporting, commit
+                var t1 = new Thread(new ThreadStart(ScanThread));
+                var t2 = new Thread(new ThreadStart(ReportThread));
+                var t3 = new Thread(new ThreadStart(CommitThread));
+                t1.Start(); t2.Start(); t3.Start();
+                t1.Join(); t2.Join(); t3.Join();
+            }
+            else
+            {
+                // Async version of demo: expect lower performance
+                // particularly for small payload sizes
+
+                const int NumParallelTasks = 10_000;
+                ThreadPool.SetMinThreads(2 * Environment.ProcessorCount, 2 * Environment.ProcessorCount);
+                TaskScheduler.UnobservedTaskException += (object sender, UnobservedTaskExceptionEventArgs e) =>
+                {
+                    Console.WriteLine($"Unobserved task exception: {e.Exception}");
+                    e.SetObserved();
+                };
+
+                Task[] tasks = new Task[NumParallelTasks];
+                for (int i = 0; i < NumParallelTasks; i++)
+                {
+                    int local = i;
+                    tasks[i] = Task.Run(() => AppendAsync(local));
+                }
+
+                // Threads for scan, reporting, commit
+                var t1 = new Thread(new ThreadStart(ScanThread));
+                var t2 = new Thread(new ThreadStart(ReportThread));
+                var t3 = new Thread(new ThreadStart(CommitThread));
+                t1.Start(); t2.Start(); t3.Start();
+                t1.Join(); t2.Join(); t3.Join();
+
+                Task.WaitAll(tasks);
+            }
+        }
+
+        static async Task AppendAsync(int id)
+        {
+            bool batched = false;
+
+            await Task.Yield();
+
+            if (!batched)
+            {
+                // Unbatched version - append each item with commit
+                // Needs high parallelism (NumParallelTasks) for perf
+                while (true)
+                {
+                    try
+                    {
+                        await log.AppendAsync(staticEntry);
+                    }
+                    catch (Exception ex)
+                    {
+                        Console.WriteLine($"{nameof(AppendAsync)}({id}): {ex}");
+                    }
+                }
+            }
+            else
+            {
+                // Batched version - we append many entries to memory,
+                // then wait for commit periodically
+                int count = 0;
+                while (true)
+                {
+                    await log.AppendToMemoryAsync(staticEntry);
+                    if (count++ % 100 == 0)
+                    {
+                        await log.WaitForCommitAsync();
+                    }
+                }
+            }
+        }
+
+        private struct ReadOnlySpanBatch : IReadOnlySpanBatch
+        {
+            private readonly int batchSize;
+            public ReadOnlySpanBatch(int batchSize) => this.batchSize = batchSize;
+            public ReadOnlySpan<byte> Get(int index) => staticEntry;
+            public int TotalEntries() => batchSize;
+        }
+
+    }
+}

cs/playground/FasterLogSample/Properties/AssemblyInfo.cs

@@ -0,0 +1,22 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT license.
+
+using System.Reflection;
+using System.Runtime.CompilerServices;
+using System.Runtime.InteropServices;
+
+// General Information about an assembly is controlled through the following
+// set of attributes. Change these attribute values to modify the information
+// associated with an assembly.
+[assembly: AssemblyDescription("")]
+[assembly: AssemblyCopyright("Copyright ©  2017")]
+[assembly: AssemblyTrademark("")]
+[assembly: AssemblyCulture("")]
+
+// Setting ComVisible to false makes the types in this assembly not visible
+// to COM components.  If you need to access a type in this assembly from
+// COM, set the ComVisible attribute to true on that type.
+[assembly: ComVisible(false)]
+
+// The following GUID is for the ID of the typelib if this project is exposed to COM
+[assembly: Guid("17bdd0a5-98e5-464a-8a00-050d9ff4c562")]