-
Notifications
You must be signed in to change notification settings - Fork 570
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[C#] Extension to allow Read of all versions of key in FasterKV log (#…
…347) * Remove obsolete/duplicate ClientSession.ReadAsyncResult * Improve ReadAddress App, add ReadAddressTests * Move RecordInfo to PendingContext only; fix TakeFullCheckpoint calls in test/sample * Fix to CPR handling for Read(.., startAddress, ..) * Mark old FunctionsBase.ReadCompletionCallback(...) form obsolete and have the new form call the old one so existing implementations continue to work * Rename ReadAddress sample to VersionedRead * Merge IAdvancedFunctions (changes to ClientSession.cs and FASTERThread.cs were already there) * Merge startAddress and RecordInfo args on Read() overload * Implement ReadAtAddress (pass address instead of key) * Full IAdvancedFunctions implementation (with ReadCompletionCallback(...., RecordInfo) * Narrow NewSession API - Make FasterKV methods taking Functions type parameter internal; FasterKV.For(f) should be used instead, to get IFunctions vs. IAdvancedFunctions overloading - Remove parameterless FasterKV.For<>() * Make IClientSession and non-Advanced ClientSession Read(..., ref RecordInfo) operations DEBUG-only * ReadAtAddress-related fixes * Add logicalAddresses to IAdvancedFunctions * Add some Checkpoint methods to IFasterKV; add RecordAccessor member to FasterKV; add comments * Add more ReadAtAddress/NoKey UTs; tighten up startAddress handling in InternalRead * Add ReadFlags to the new Read-by-address overloads; move readcache check to internal * Remove an obsolete line * Fix ReadAddressTests to use / rather than \\ for Linux compat Co-authored-by: Badrish Chandramouli <badrishc@microsoft.com>
- Loading branch information
Showing
27 changed files
with
3,111 additions
and
395 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,60 @@ | ||
| // Copyright (c) Microsoft Corporation. All rights reserved. | ||
| // Licensed under the MIT license. | ||
|
|
||
| using FASTER.core; | ||
| using System.Threading; | ||
|
|
||
| namespace ReadAddress | ||
| { | ||
| public struct Key | ||
| { | ||
| public long key; | ||
|
|
||
| public Key(long first) => key = first; | ||
|
|
||
| public override string ToString() => key.ToString(); | ||
|
|
||
| internal class Comparer : IFasterEqualityComparer<Key> | ||
| { | ||
| public long GetHashCode64(ref Key key) => Utility.GetHashCode(key.key); | ||
|
|
||
| public bool Equals(ref Key k1, ref Key k2) => k1.key == k2.key; | ||
| } | ||
| } | ||
|
|
||
| public struct Value | ||
| { | ||
| public long value; | ||
|
|
||
| public Value(long first) => value = first; | ||
|
|
||
| public override string ToString() => value.ToString(); | ||
| } | ||
|
|
||
| public class Context | ||
| { | ||
| public RecordInfo recordInfo; | ||
| public Status status; | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Callback for FASTER operations | ||
| /// </summary> | ||
| public class Functions : AdvancedSimpleFunctions<Key, Value, Context> | ||
| { | ||
| // Return false to force a chain of values. | ||
| public override bool ConcurrentWriter(ref Key key, ref Value src, ref Value dst, long address) => false; | ||
|
|
||
| public override bool InPlaceUpdater(ref Key key, ref Value input, ref Value value, long address) => false; | ||
|
|
||
| // Track the recordInfo for its PreviousAddress. | ||
| public override void ReadCompletionCallback(ref Key key, ref Value input, ref Value output, Context ctx, Status status, RecordInfo recordInfo) | ||
| { | ||
| if (!(ctx is null)) | ||
| { | ||
| ctx.recordInfo = recordInfo; | ||
| ctx.status = status; | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,12 @@ | ||
| <Project Sdk="Microsoft.NET.Sdk"> | ||
|
|
||
| <PropertyGroup> | ||
| <OutputType>Exe</OutputType> | ||
| <TargetFramework>netcoreapp3.1</TargetFramework> | ||
| </PropertyGroup> | ||
|
|
||
| <ItemGroup> | ||
| <ProjectReference Include="..\..\src\core\FASTER.core.csproj" /> | ||
| </ItemGroup> | ||
|
|
||
| </Project> |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,206 @@ | ||
| // Copyright (c) Microsoft Corporation. All rights reserved. | ||
| // Licensed under the MIT license. | ||
|
|
||
| using FASTER.core; | ||
| using System; | ||
| using System.Diagnostics; | ||
| using System.IO; | ||
| using System.Threading; | ||
| using System.Threading.Tasks; | ||
|
|
||
| namespace ReadAddress | ||
| { | ||
| class VersionedReadApp | ||
| { | ||
| // Number of keys in store | ||
| const int numKeys = 1000; | ||
| const int keyMod = 100; | ||
| const int maxLap = numKeys / keyMod; | ||
| const int deleteLap = maxLap / 2; | ||
|
|
||
| const string ReadCacheArg = "--readcache"; | ||
| static bool useReadCache = false; | ||
| const string CheckpointsArg = "--checkpoints"; | ||
| static bool useCheckpoints = true; | ||
| const string RMWArg = "--rmw"; | ||
| static bool useRMW = false; | ||
|
|
||
| private static void Usage() | ||
| { | ||
| Console.WriteLine("Reads 'linked lists' of records for each key by backing up the previous-address chain, including showing record versions"); | ||
| Console.WriteLine("Usage:"); | ||
| Console.WriteLine($" {ReadCacheArg}: use Read Cache; default = {useReadCache}"); | ||
| Console.WriteLine($" {CheckpointsArg}: issue periodic checkpoints during load; default = {useCheckpoints}"); | ||
| Console.WriteLine($" {RMWArg}: issue periodic checkpoints during load; default = {useRMW}"); | ||
| } | ||
|
|
||
| static async Task<int> Main(string[] args) | ||
| { | ||
| for (var ii = 0; ii < args.Length; ++ii) | ||
| { | ||
| var arg = args[ii]; | ||
| if (arg.ToLower() == ReadCacheArg) | ||
| { | ||
| useReadCache = true; | ||
| continue; | ||
| } | ||
| if (arg.ToLower() == CheckpointsArg) | ||
| { | ||
| useCheckpoints = true; | ||
| continue; | ||
| } | ||
| if (arg.ToLower() == RMWArg) | ||
| { | ||
| useRMW = true; | ||
| continue; | ||
| } | ||
| Console.WriteLine($"Unknown option: {arg}"); | ||
| Usage(); | ||
| return -1; | ||
| } | ||
|
|
||
| var (store, log, path) = CreateStore(); | ||
| await PopulateStore(store); | ||
|
|
||
| const int keyToScan = 42; | ||
| ScanStore(store, keyToScan); | ||
| var cts = new CancellationTokenSource(); | ||
| await ScanStoreAsync(store, keyToScan, cts.Token); | ||
|
|
||
| // Clean up | ||
| store.Dispose(); | ||
| log.Dispose(); | ||
|
|
||
| // Delete the created files | ||
| try { new DirectoryInfo(path).Delete(true); } catch { } | ||
|
|
||
| Console.WriteLine("Press <ENTER> to end"); | ||
| Console.ReadLine(); | ||
| return 0; | ||
| } | ||
|
|
||
| private static (FasterKV<Key, Value>, IDevice, string) CreateStore() | ||
| { | ||
| var path = Path.GetTempPath() + "FasterReadAddressSample\\"; | ||
| var log = Devices.CreateLogDevice(path + "hlog.log"); | ||
|
|
||
| var logSettings = new LogSettings | ||
| { | ||
| LogDevice = log, | ||
| ObjectLogDevice = new NullDevice(), | ||
| ReadCacheSettings = useReadCache ? new ReadCacheSettings() : null, | ||
| // Use small-footprint values | ||
| PageSizeBits = 12, // (4K pages) | ||
| MemorySizeBits = 20 // (1M memory for main log) | ||
| }; | ||
|
|
||
| var store = new FasterKV<Key, Value>( | ||
| size: 1L << 20, | ||
| logSettings: logSettings, | ||
| checkpointSettings: new CheckpointSettings { CheckpointDir = path }, | ||
| serializerSettings: null, | ||
| comparer: new Key.Comparer() | ||
| ); | ||
| return (store, log, path); | ||
| } | ||
|
|
||
| private async static Task PopulateStore(FasterKV<Key, Value> store) | ||
| { | ||
| // Start session with FASTER | ||
| using var s = store.For(new Functions()).NewSession<Functions>(); | ||
| Console.WriteLine($"Writing {numKeys} keys to FASTER", numKeys); | ||
|
|
||
| Stopwatch sw = new Stopwatch(); | ||
| sw.Start(); | ||
| var prevLap = 0; | ||
| for (int ii = 0; ii < numKeys; ii++) | ||
| { | ||
| // lap is used to illustrate the changing values | ||
| var lap = ii / keyMod; | ||
|
|
||
| if (useCheckpoints && lap != prevLap) | ||
| { | ||
| await store.TakeFullCheckpointAsync(CheckpointType.FoldOver); | ||
| prevLap = lap; | ||
| } | ||
|
|
||
| var key = new Key(ii % keyMod); | ||
|
|
||
| var value = new Value(key.key + (lap * numKeys * 100)); | ||
| if (useRMW) | ||
| s.RMW(ref key, ref value, serialNo: lap); | ||
| else | ||
| s.Upsert(ref key, ref value, serialNo: lap); | ||
|
|
||
| // Illustrate that deleted records can be shown as well (unless overwritten by in-place operations, which are not done here) | ||
| if (lap == deleteLap) | ||
| s.Delete(ref key, serialNo: lap); | ||
| } | ||
| sw.Stop(); | ||
| double numSec = sw.ElapsedMilliseconds / 1000.0; | ||
| Console.WriteLine("Total time to upsert {0} elements: {1:0.000} secs ({2:0.00} inserts/sec)", numKeys, numSec, numKeys / numSec); | ||
| } | ||
|
|
||
| private static void ScanStore(FasterKV<Key, Value> store, int keyValue) | ||
| { | ||
| // Start session with FASTER | ||
| using var session = store.For(new Functions()).NewSession<Functions>(); | ||
|
|
||
| Console.WriteLine($"Sync scanning records for key {keyValue}"); | ||
|
|
||
| var output = default(Value); | ||
| var input = default(Value); | ||
| var key = new Key(keyValue); | ||
| RecordInfo recordInfo = default; | ||
| var context = new Context(); | ||
| int version = int.MaxValue; | ||
| for (int lap = 9; /* tested in loop */; --lap) | ||
| { | ||
| var status = session.Read(ref key, ref input, ref output, ref recordInfo, userContext: context, serialNo: maxLap + 1); | ||
| if (status == Status.PENDING) | ||
| { | ||
| // This will spin CPU for each retrieved record; not recommended for performance-critical code or when retrieving chains for multiple records. | ||
| session.CompletePending(spinWait: true); | ||
| recordInfo = context.recordInfo; | ||
| status = context.status; | ||
| } | ||
| if (!ProcessRecord(store, status, recordInfo, lap, ref output, ref version)) | ||
| break; | ||
| } | ||
| } | ||
|
|
||
| private static async Task ScanStoreAsync(FasterKV<Key, Value> store, int keyValue, CancellationToken cancellationToken) | ||
| { | ||
| // Start session with FASTER | ||
| using var session = store.For(new Functions()).NewSession<Functions>(); | ||
|
|
||
| Console.WriteLine($"Async scanning records for key {keyValue}"); | ||
|
|
||
| var input = default(Value); | ||
| var key = new Key(keyValue); | ||
| RecordInfo recordInfo = default; | ||
| int version = int.MaxValue; | ||
| for (int lap = 9; /* tested in loop */; --lap) | ||
| { | ||
| var readAsyncResult = await session.ReadAsync(ref key, ref input, recordInfo.PreviousAddress, default, serialNo: maxLap + 1, cancellationToken: cancellationToken); | ||
| cancellationToken.ThrowIfCancellationRequested(); | ||
| var (status, output) = readAsyncResult.Complete(out recordInfo); | ||
| if (!ProcessRecord(store, status, recordInfo, lap, ref output, ref version)) | ||
| break; | ||
| } | ||
| } | ||
|
|
||
| private static bool ProcessRecord(FasterKV<Key, Value> store, Status status, RecordInfo recordInfo, int lap, ref Value output, ref int previousVersion) | ||
| { | ||
| Debug.Assert((status == Status.NOTFOUND) == recordInfo.Tombstone); | ||
| Debug.Assert((lap == deleteLap) == recordInfo.Tombstone); | ||
| var value = recordInfo.Tombstone ? "<deleted>" : output.value.ToString(); | ||
| Debug.Assert(previousVersion >= recordInfo.Version); | ||
| Console.WriteLine($" {value}; Version = {recordInfo.Version}; PrevAddress: {recordInfo.PreviousAddress}"); | ||
|
|
||
| // Check for end of loop | ||
| previousVersion = recordInfo.Version; | ||
| return recordInfo.PreviousAddress >= store.Log.BeginAddress; | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.