[C#] Add Status.Record (#659)

* Add Status.Record; update OperationStatus with all Advanced Status codes * Update Remote for status.Pending => .IsPending
microsoft · Feb 22, 2022 · 34a7576 · 34a7576
1 parent 69f8eda
commit 34a7576
Show file tree

Hide file tree

Showing 69 changed files with 306 additions and 286 deletions.
diff --git a/cs/playground/AsyncStress/FasterWrapper.cs b/cs/playground/AsyncStress/FasterWrapper.cs
@@ -56,15 +56,15 @@ public FasterWrapper(bool isRefType, bool useLargeLog, bool useOsReadBuffering =
         async ValueTask<int> CompleteAsync(FasterKV<Key, Value>.UpsertAsyncResult<Value, Value, Empty> result)
         {
             var numPending = 0;
-            for (; result.Status.Pending; ++numPending)
+            for (; result.Status.IsPending; ++numPending)
                 result = await result.CompleteAsync().ConfigureAwait(false);
             return numPending;
         }
 
         async ValueTask<int> CompleteAsync(FasterKV<Key, Value>.RmwAsyncResult<Value, Value, Empty> result)
         {
             var numPending = 0;
-            for (; result.Status.Pending; ++numPending)
+            for (; result.Status.IsPending; ++numPending)
                 result = await result.CompleteAsync().ConfigureAwait(false);
             return numPending;
         }
@@ -82,7 +82,7 @@ public void Upsert(Key key, Value value)
             if (!_sessionPool.TryGet(out var session))
                 session = _sessionPool.GetAsync().GetAwaiter().GetResult();
             var status = session.Upsert(key, value);
-            Assert.False(status.Pending);
+            Assert.False(status.IsPending);
             _sessionPool.Return(session);
         }
 
@@ -109,7 +109,7 @@ public void RMW(Key key, Value value)
             if (!_sessionPool.TryGet(out var session))
                 session = _sessionPool.GetAsync().GetAwaiter().GetResult();
             var status = session.RMW(key, value);
-            Assert.False(status.Pending);
+            Assert.False(status.IsPending);
             _sessionPool.Return(session);
         }
 
@@ -137,7 +137,7 @@ public async ValueTask RMWChunkAsync((Key, Value)[] chunk, int offset, int count
             if (!_sessionPool.TryGet(out var session))
                 session = _sessionPool.GetAsync().GetAwaiter().GetResult();
             var result = session.Read(key);
-            if (result.status.Pending)
+            if (result.status.IsPending)
             {
                 session.CompletePendingWithOutputs(out var completedOutputs, wait: true);
                 int count = 0;

diff --git a/cs/playground/AsyncStress/SerializedFasterWrapper.cs b/cs/playground/AsyncStress/SerializedFasterWrapper.cs
@@ -121,7 +121,7 @@ public void Update<TUpdater, TAsyncResult>(TUpdater updater, Key key, Value valu
                 }
             }
 
-            Assert.False(status.Pending);
+            Assert.False(status.IsPending);
             _sessionPool.Return(session);
         }
 
@@ -198,7 +198,7 @@ public async ValueTask UpdateChunkAsync<TUpdater, TAsyncResult>(TUpdater updater
                 }
             }
 
-            if (result.Item1.Pending)
+            if (result.Item1.IsPending)
             {
                 session.CompletePendingWithOutputs(out var completedOutputs, wait: true);
                 int count = 0;

diff --git a/cs/playground/AsyncStress/SerializedUpdaters.cs b/cs/playground/AsyncStress/SerializedUpdaters.cs
@@ -26,7 +26,7 @@ public ValueTask<FasterKV<SpanByte, SpanByte>.UpsertAsyncResult<SpanByte, SpanBy
             public async ValueTask<int> CompleteAsync(FasterKV<SpanByte, SpanByte>.UpsertAsyncResult<SpanByte, SpanByteAndMemory, Empty> result)
             {
                 var numPending = 0;
-                for (; result.Status.Pending; ++numPending)
+                for (; result.Status.IsPending; ++numPending)
                     result = await result.CompleteAsync().ConfigureAwait(false);
                 return numPending;
             }
@@ -43,7 +43,7 @@ public ValueTask<FasterKV<SpanByte, SpanByte>.RmwAsyncResult<SpanByte, SpanByteA
             public async ValueTask<int> CompleteAsync(FasterKV<SpanByte, SpanByte>.RmwAsyncResult<SpanByte, SpanByteAndMemory, Empty> result)
             {
                 var numPending = 0;
-                for (; result.Status.Pending; ++numPending)
+                for (; result.Status.IsPending; ++numPending)
                     result = await result.CompleteAsync().ConfigureAwait(false);
                 return numPending;
             }

diff --git a/cs/playground/CacheStoreConcurrent/Program.cs b/cs/playground/CacheStoreConcurrent/Program.cs
@@ -175,7 +175,7 @@ private static void RandomReadWorkload(int threadid)
                 var key = new CacheKey(k);
                 var status = hts.Read(ref key, ref output);
 
-                if (status.Pending)
+                if (status.IsPending)
                 {
                     statusPending++;
                 }

diff --git a/cs/playground/ClassRecoveryDurability/Program.cs b/cs/playground/ClassRecoveryDurability/Program.cs
@@ -90,7 +90,7 @@ static void Filldb(int stopAfterIteration = 5)
 
                         // Console.WriteLine("add=" + i);
 
-                        if (!addStatus.InPlaceUpdatedRecord)
+                        if (!addStatus.Record.InPlaceUpdated)
                             throw new Exception();
                     }
 
@@ -109,7 +109,7 @@ static void Filldb(int stopAfterIteration = 5)
                                 var deleteStatus = session.Delete(ref deteletKey, context2, 1);
                                 // Console.WriteLine("delete=" + i);
 
-                                if (!deleteStatus.InPlaceUpdatedRecord)
+                                if (!deleteStatus.Record.InPlaceUpdated)
                                     throw new Exception();
                             }
                         }
@@ -195,7 +195,7 @@ static void TestData(Storedb store, int  count)
                             var deleteStatus = session.Read(ref readKey, ref input1, ref output1, context, 1);
                             //Console.WriteLine("test delete=" + i);
 
-                            if (deleteStatus.Pending)
+                            if (deleteStatus.IsPending)
                             {
                                 session.CompletePending(true);
                                 context.FinalizeRead(ref deleteStatus, ref output1);
@@ -215,7 +215,7 @@ static void TestData(Storedb store, int  count)
                             var addStatus = session.Read(ref readKey, ref input1, ref output1, context, 1);
                             //Console.WriteLine("test add=" + i);
 
-                            if (addStatus.Pending)
+                            if (addStatus.IsPending)
                             {
                                 session.CompletePending(true);
                                 context.FinalizeRead(ref addStatus, ref output1);

diff --git a/cs/playground/SumStore/ConcurrencyTest.cs b/cs/playground/SumStore/ConcurrencyTest.cs
@@ -156,7 +156,7 @@ public void Test()
                 Input input = default;
                 Output output = default;
                 var status = session.Read(ref inputArray[i].adId, ref input, ref output, Empty.Default, i);
-                if (status.Pending)
+                if (status.IsPending)
                     throw new NotImplementedException();
 
                 inputArray[i].numClicks = output.value;

diff --git a/cs/remote/src/FASTER.server/BinaryServerSession.cs b/cs/remote/src/FASTER.server/BinaryServerSession.cs
@@ -167,7 +167,7 @@ private unsafe void ProcessBatch(byte* buf, int offset)
                         hrw.Write(message, ref dcurr, (int)(dend - dcurr));
                         Write(ref status, ref dcurr, (int)(dend - dcurr));
 
-                        if (status.Pending)
+                        if (status.IsPending)
                             Write(pendingSeqNo++, ref dcurr, (int)(dend - dcurr));
                         else if (status.Found)
                             serializer.SkipOutput(ref dcurr);
@@ -186,7 +186,7 @@ private unsafe void ProcessBatch(byte* buf, int offset)
 
                         hrw.Write(message, ref dcurr, (int)(dend - dcurr));
                         Write(ref status, ref dcurr, (int)(dend - dcurr));
-                        if (status.Pending)
+                        if (status.IsPending)
                             Write(pendingSeqNo++, ref dcurr, (int)(dend - dcurr));
                         else if (status.IsCompletedSuccessfully)
                             serializer.SkipOutput(ref dcurr);
@@ -273,7 +273,7 @@ private unsafe void Publish(ref byte* keyPtr, int keyLength, ref byte* valPtr, r
             if (valPtr == null)
                 status = session.Read(ref key, ref serializer.ReadInputByRef(ref inputPtr), ref serializer.AsRefOutput(outputDcurr, (int)(dend - dcurr)), ctx, 0);
 
-            if (!status.Pending)
+            if (!status.IsPending)
             {
                 // Write six bytes (message | status | sid)
                 hrw.Write(message, ref dcurr, (int)(dend - dcurr));

diff --git a/cs/remote/src/FASTER.server/WebsocketServerSession.cs b/cs/remote/src/FASTER.server/WebsocketServerSession.cs
@@ -384,7 +384,7 @@ private unsafe bool ProcessBatch(byte* buf, int length, int offset)
                             hrw.Write(message, ref dcurr, (int)(dend - dcurr));
                             Write(ref status, ref dcurr, (int)(dend - dcurr));
 
-                            if (status.Pending)
+                            if (status.IsPending)
                                 Write(pendingSeqNo++, ref dcurr, (int)(dend - dcurr));
                             else if (status.Found)
                                 serializer.SkipOutput(ref dcurr);
@@ -403,7 +403,7 @@ private unsafe bool ProcessBatch(byte* buf, int length, int offset)
 
                             hrw.Write(message, ref dcurr, (int)(dend - dcurr));
                             Write(ref status, ref dcurr, (int)(dend - dcurr));
-                            if (status.Pending)
+                            if (status.IsPending)
                                 Write(pendingSeqNo++, ref dcurr, (int)(dend - dcurr));
 
                             if (subscribeKVBroker != null)
@@ -589,7 +589,7 @@ private unsafe void Publish(ref byte* keyPtr, int keyLength, ref byte* valPtr, r
             if (valPtr == null)
                 status = session.Read(ref key, ref serializer.ReadInputByRef(ref inputPtr), ref serializer.AsRefOutput(outputDcurr, (int)(dend - dcurr)), ctx, 0);
 
-            if (!status.Pending)
+            if (!status.IsPending)
             {
                 // Write six bytes (message | status | sid)
                 hrw.Write(message, ref dcurr, (int)(dend - dcurr));

diff --git a/cs/samples/AzureBackedStore/Program.cs b/cs/samples/AzureBackedStore/Program.cs
@@ -52,7 +52,7 @@ static void Main()
 
                 var status = s.Read(ref key, ref output, context);
 
-                if (!status.Pending)
+                if (!status.IsPending)
                 {
                     if (status.Found && output == context)
                         Console.WriteLine("Success!");
@@ -94,7 +94,7 @@ static void Main()
 
                 var status = s.Read(ref key, ref output, context);
 
-                if (!status.Pending)
+                if (!status.IsPending)
                 {
                     if (status.Found && output == context)
                         Console.WriteLine("Success!");

diff --git a/cs/samples/CacheStore/Program.cs b/cs/samples/CacheStore/Program.cs
@@ -144,7 +144,7 @@ private static void RandomReadWorkload(FasterKV<CacheKey, CacheValue> store, int
                 var key = new CacheKey(k);
                 var status = s.Read(ref key, ref output);
 
-                if (status.Pending)
+                if (status.IsPending)
                 {
                     statusPending++;
                     if (statusPending % 100 == 0)
@@ -186,7 +186,7 @@ private static void InteractiveReadWorkload(FasterKV<CacheKey, CacheValue> store
 
                 context.ticks = Stopwatch.GetTimestamp();
                 var status = s.Read(ref key, ref output, context);
-                if (status.Pending)
+                if (status.IsPending)
                 {
                     s.CompletePending(true);
                 }

diff --git a/cs/samples/HelloWorld/Program.cs b/cs/samples/HelloWorld/Program.cs
@@ -147,7 +147,7 @@ static void DiskSample()
             // Reads from disk will return PENDING status, result available via either asynchronous IFunctions callback
             // or on this thread via CompletePendingWithOutputs, shown below
             status = session.Read(ref key, ref output);
-            if (status.Pending)
+            if (status.IsPending)
             {
                 session.CompletePendingWithOutputs(out var iter, true);
                 while (iter.Next())

diff --git a/cs/samples/ReadAddress/VersionedReadApp.cs b/cs/samples/ReadAddress/VersionedReadApp.cs
@@ -157,7 +157,7 @@ private static void ScanStore(FasterKV<Key, Value> store, int keyValue)
                 var status = session.Read(ref key, ref input, ref output, ref recordMetadata, ReadFlags.SkipCopyReads, serialNo: maxLap + 1);
 
                 // This will wait for each retrieved record; not recommended for performance-critical code or when retrieving multiple records unless necessary.
-                if (status.Pending)
+                if (status.IsPending)
                 {
                     session.CompletePendingWithOutputs(out var completedOutputs, wait: true);
                     using (completedOutputs)

diff --git a/cs/samples/StoreAsyncApi/Program.cs b/cs/samples/StoreAsyncApi/Program.cs
@@ -87,7 +87,7 @@ static async Task AsyncOperator(int id)
                         if (asyncUpsert)
                         {
                             var r = await session.UpsertAsync(ref key, ref value, context, seqNo++);
-                            while (r.Status.Pending)
+                            while (r.Status.IsPending)
                                 r = await r.CompleteAsync();
                         }
                         else
@@ -130,7 +130,7 @@ static async Task AsyncOperator(int id)
                             for (int i = 0; i < batchSize; i++)
                             {
                                 var r = await taskBatch[i];
-                                while (r.Status.Pending)
+                                while (r.Status.IsPending)
                                     r = await r.CompleteAsync();
                             }
                         }

diff --git a/cs/samples/StoreCustomTypes/Program.cs b/cs/samples/StoreCustomTypes/Program.cs
@@ -78,7 +78,7 @@ static void Main()
             status = s.Read(ref key, ref input, ref g2, context, 0);
 
             // We will receive the result via ReadCompletionCallback in Functions 
-            if (!status.Pending)
+            if (!status.IsPending)
                 Console.WriteLine("Error!");
 
             // End session when done

diff --git a/cs/samples/StoreDiskReadBenchmark/Program.cs b/cs/samples/StoreDiskReadBenchmark/Program.cs
@@ -155,7 +155,7 @@ static async Task AsyncReadOperator(int id)
                         var result = session.Read(ref key, ref input, ref output, Empty.Default, 0);
                         if (readBatching)
                         {
-                            if (!result.Pending)
+                            if (!result.IsPending)
                             {
                                 if (output.value.vfield1 != key.key)
                                 {
@@ -166,7 +166,7 @@ static async Task AsyncReadOperator(int id)
                         }
                         else
                         {
-                            if (result.Pending)
+                            if (result.IsPending)
                             {
                                 session.CompletePending(true);
                             }

diff --git a/cs/samples/StoreVarLenTypes/AsciiSumSample.cs b/cs/samples/StoreVarLenTypes/AsciiSumSample.cs
@@ -59,7 +59,7 @@ public static void Run()
                 store.Log.FlushAndEvict(true); // Flush and evict all records to disk
                 var _status = s.RMW(_key, _input); // CopyUpdater to 270 (due to immutable source value on disk)
 
-                if (_status.Pending)
+                if (_status.IsPending)
                 {
                     Console.WriteLine("Error!");
                     return;

diff --git a/cs/samples/StoreVarLenTypes/MemoryByteSample.cs b/cs/samples/StoreVarLenTypes/MemoryByteSample.cs
@@ -61,7 +61,7 @@ public static void Run()
                 var expectedValue = valueMem.Slice(0, valLen);
                 expectedValue.Span.Fill((byte)valLen);
 
-                if (status.Pending)
+                if (status.IsPending)
                     s.CompletePending(true);
                 else
                 {

diff --git a/cs/samples/StoreVarLenTypes/MemoryIntSample.cs b/cs/samples/StoreVarLenTypes/MemoryIntSample.cs
@@ -61,7 +61,7 @@ public static void Run()
                 var expectedValue = valueMem.Slice(0, valLen);
                 expectedValue.Span.Fill(valLen);
 
-                if (status.Pending)
+                if (status.IsPending)
                     s.CompletePending(true);
                 else
                 {

diff --git a/cs/samples/StoreVarLenTypes/SpanByteSample.cs b/cs/samples/StoreVarLenTypes/SpanByteSample.cs
@@ -91,7 +91,7 @@ public static void Run()
                 var expectedValue = valueMem.Slice(0, valLen);
                 expectedValue.Fill((byte)valLen);
 
-                if (status.Pending)
+                if (status.IsPending)
                 {
                     s.CompletePending(true);
                 }

diff --git a/cs/src/core/Async/DeleteAsync.cs b/cs/src/core/Async/DeleteAsync.cs
@@ -69,13 +69,13 @@ internal DeleteAsyncResult(FasterKV<Key, Value> fasterKV, IFasterSession<Key, Va
             /// <summary>Complete the Delete operation, issuing additional allocation asynchronously if needed. It is usually preferable to use Complete() instead of this.</summary>
             /// <returns>ValueTask for Delete result. User needs to await again if result status is Status.PENDING.</returns>
             public ValueTask<DeleteAsyncResult<Input, Output, Context>> CompleteAsync(CancellationToken token = default)
-                => this.Status.Pending
+                => this.Status.IsPending
                     ? updateAsyncInternal.CompleteAsync(token)
                     : new ValueTask<DeleteAsyncResult<Input, Output, Context>>(new DeleteAsyncResult<Input, Output, Context>(this.Status));
 
             /// <summary>Complete the Delete operation, issuing additional I/O synchronously if needed.</summary>
             /// <returns>Status of Delete operation</returns>
-            public Status Complete() => this.Status.Pending ? updateAsyncInternal.Complete().Status : this.Status;
+            public Status Complete() => this.Status.IsPending ? updateAsyncInternal.Complete().Status : this.Status;
         }
 
         [MethodImpl(MethodImplOptions.AggressiveInlining)]

diff --git a/cs/src/core/Async/RMWAsync.cs b/cs/src/core/Async/RMWAsync.cs
@@ -32,7 +32,7 @@ public Status DoFastOperation(FasterKV<Key, Value> fasterKV, ref PendingContext<
                                                      pendingContext.serialNum, asyncOp, out flushEvent, out newDiskRequest);
                 output = pendingContext.output;
 
-                if (status.Pending && !newDiskRequest.IsDefault())
+                if (status.IsPending && !newDiskRequest.IsDefault())
                 {
                     flushEvent = default;
                     this.diskRequest = newDiskRequest;
@@ -56,7 +56,7 @@ public bool CompletePendingIO(IFasterSession<Key, Value, Input, Output, Context>
                 var status = completedOutputs.Next() ? completedOutputs.Current.Status : new(StatusCode.Error);
                 completedOutputs.Dispose();
                 this.diskRequest = default;
-                return !status.Pending;
+                return !status.IsPending;
             }
 
             /// <inheritdoc/>
@@ -109,7 +109,7 @@ internal RmwAsyncResult(FasterKV<Key, Value> fasterKV, IFasterSession<Key, Value
             /// <summary>Complete the RMW operation, issuing additional (rare) I/O asynchronously if needed. It is usually preferable to use Complete() instead of this.</summary>
             /// <returns>ValueTask for RMW result. User needs to await again if result status is pending.</returns>
             public ValueTask<RmwAsyncResult<Input, TOutput, Context>> CompleteAsync(CancellationToken token = default) 
-                => this.Status.Pending
+                => this.Status.IsPending
                     ? updateAsyncInternal.CompleteAsync(token)
                     : new ValueTask<RmwAsyncResult<Input, TOutput, Context>>(new RmwAsyncResult<Input, TOutput, Context>(this.Status, this.Output, this.RecordMetadata));
 
@@ -122,7 +122,7 @@ public ValueTask<RmwAsyncResult<Input, TOutput, Context>> CompleteAsync(Cancella
             /// <returns>Status of RMW operation</returns>
             public (Status status, TOutput output) Complete(out RecordMetadata recordMetadata)
             {
-                if (!this.Status.Pending)
+                if (!this.Status.IsPending)
                 {
                     recordMetadata = this.RecordMetadata;
                     return (this.Status, this.Output);
@@ -154,7 +154,7 @@ internal ValueTask<RmwAsyncResult<Input, Output, Context>> RmwAsync<Input, Outpu
             {
                 Output output = default;
                 var status = CallInternalRMW(fasterSession, currentCtx, ref pcontext, ref key, ref input, ref output, context, serialNo, asyncOp: true, out flushEvent, out diskRequest);
-                if (!status.Pending)
+                if (!status.IsPending)
                     return new ValueTask<RmwAsyncResult<Input, Output, Context>>(new RmwAsyncResult<Input, Output, Context>(status, output, new RecordMetadata(pcontext.recordInfo, pcontext.logicalAddress)));
             }
             finally