Cleanup and updates.

microsoft · Oct 7, 2019 · 089d545 · 089d545
1 parent 915f01e
commit 089d545
Show file tree

Hide file tree

Showing 4 changed files with 47 additions and 49 deletions.
diff --git a/cs/benchmark/FasterYcsbBenchmark.cs b/cs/benchmark/FasterYcsbBenchmark.cs
@@ -24,8 +24,13 @@ public enum Op : ulong
             ReadModifyWrite = 2
         }
 
-        const bool kUseSyntheticData = false;
+#if DEBUG
+        const bool kUseSmallData = true;
+        const bool kUseSyntheticData = true;
+#else
         const bool kUseSmallData = false;
+        const bool kUseSyntheticData = false;
+#endif
         const long kInitCount = kUseSmallData ? 2500480 : 250000000;
         const long kTxnCount = kUseSmallData ? 10000000 : 1000000000;
         const int kMaxKey = kUseSmallData ? 1 << 22 : 1 << 28;
@@ -432,7 +437,7 @@ void DoContinuousMeasurements()
         }
 #endif
 
-        #region Load Data
+#region Load Data
 
         private unsafe void LoadDataFromFile(string filePath)
         {
@@ -574,7 +579,7 @@ private void LoadSyntheticData()
             Console.WriteLine("loaded " + kTxnCount + " txns.");
 
         }
-        #endregion
+#endregion
 
 
     }

diff --git a/cs/src/core/ClientSession/ClientSession.cs b/cs/src/core/ClientSession/ClientSession.cs
@@ -4,6 +4,7 @@
 #pragma warning disable 0162
 
 using System;
+using System.Runtime.CompilerServices;
 using System.Threading.Tasks;
 
 namespace FASTER.core
@@ -34,14 +35,10 @@ internal ClientSession(
             this.fht = fht;
             this.ctx = ctx;
             this.supportAsync = supportAsync;
-            if (supportAsync)
-            {
-                fht.UseRelaxedCPR();
-            }
-            else
-            {
+
+            // Session runs on a single thread
+            if (!supportAsync)
                 UnsafeResumeThread();
-            }
         }
 
         /// <summary>
@@ -55,24 +52,10 @@ internal ClientSession(
         public void Dispose()
         {
             CompletePending(true);
-            fht.DisposeClientSession(ID);
-        }
-
-        /// <summary>
-        /// Resume session on current thread
-        /// Call SuspendThread before any async op
-        /// </summary>
-        internal void UnsafeResumeThread()
-        {
-            fht.ResumeSession(ctx);
-        }
 
-        /// <summary>
-        /// Suspend session on current thread
-        /// </summary>
-        internal void UnsafeSuspendThread()
-        {
-            fht.SuspendSession();
+            // Session runs on a single thread
+            if (!supportAsync)
+                UnsafeSuspendThread();
         }
 
         /// <summary>
@@ -84,6 +67,7 @@ internal void UnsafeSuspendThread()
         /// <param name="userContext"></param>
         /// <param name="serialNo"></param>
         /// <returns></returns>
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
         public Status Read(ref Key key, ref Input input, ref Output output, Context userContext, long serialNo)
         {
             if (supportAsync) UnsafeResumeThread();
@@ -100,6 +84,7 @@ public Status Read(ref Key key, ref Input input, ref Output output, Context user
         /// <param name="userContext"></param>
         /// <param name="serialNo"></param>
         /// <returns></returns>
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
         public Status Upsert(ref Key key, ref Value desiredValue, Context userContext, long serialNo)
         {
             if (supportAsync) UnsafeResumeThread();
@@ -116,6 +101,7 @@ public Status Upsert(ref Key key, ref Value desiredValue, Context userContext, l
         /// <param name="userContext"></param>
         /// <param name="serialNo"></param>
         /// <returns></returns>
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
         public Status RMW(ref Key key, ref Input input, Context userContext, long serialNo)
         {
             if (supportAsync) UnsafeResumeThread();
@@ -131,6 +117,7 @@ public Status RMW(ref Key key, ref Input input, Context userContext, long serial
         /// <param name="userContext"></param>
         /// <param name="serialNo"></param>
         /// <returns></returns>
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
         public Status Delete(ref Key key, Context userContext, long serialNo)
         {
             if (supportAsync) UnsafeResumeThread();
@@ -189,10 +176,31 @@ public bool CompleteCheckpoint(bool spinWait = false)
         /// Complete the ongoing checkpoint (if any)
         /// </summary>
         /// <returns></returns>
-        internal async ValueTask CompleteCheckpointAsync()
+        public async ValueTask CompleteCheckpointAsync()
         {
             if (!supportAsync) throw new NotSupportedException();
             await fht.CompleteCheckpointAsync(ctx, this);
         }
+
+        /// <summary>
+        /// Resume session on current thread
+        /// Call SuspendThread before any async op
+        /// </summary>
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        internal void UnsafeResumeThread()
+        {
+            fht.epoch.Acquire();
+            fht.InternalRefresh(ctx);
+        }
+
+        /// <summary>
+        /// Suspend session on current thread
+        /// </summary>
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        internal void UnsafeSuspendThread()
+        {
+            fht.epoch.Release();
+        }
+
     }
 }
diff --git a/cs/src/core/ClientSession/FASTERAsync.cs b/cs/src/core/ClientSession/FASTERAsync.cs
@@ -38,9 +38,7 @@ internal async ValueTask CompletePendingAsync(ClientSession<Key, Value, Input, O
 
                     if (clientSession.ctx.prevCtx.retryRequests.Count > 0)
                     {
-                        clientSession.UnsafeResumeThread();
                         CompleteRetryRequests(clientSession.ctx.prevCtx, clientSession.ctx, clientSession);
-                        clientSession.UnsafeSuspendThread();
                     }
 
                     done &= (clientSession.ctx.prevCtx.ioPendingRequests.Count == 0);

diff --git a/cs/src/core/ClientSession/FASTERClientSession.cs b/cs/src/core/ClientSession/FASTERClientSession.cs
@@ -25,6 +25,9 @@ public unsafe partial class FasterKV<Key, Value, Input, Output, Context, Functio
         /// <returns></returns>
         public ClientSession<Key, Value, Input, Output, Context, Functions> StartClientSession(bool supportAsync = true)
         {
+            if (supportAsync)
+                UseRelaxedCPR();
+
             Guid guid = Guid.NewGuid();
             var ctx = new FasterExecutionContext();
             InitContext(ctx, guid);
@@ -52,6 +55,9 @@ public ClientSession<Key, Value, Input, Output, Context, Functions> StartClientS
         /// <returns></returns>
         public ClientSession<Key, Value, Input, Output, Context, Functions> ContinueClientSession(Guid guid, out CommitPoint cp, bool supportAsync = true)
         {
+            if (supportAsync)
+                UseRelaxedCPR();
+
             cp = InternalContinue(guid, out FasterExecutionContext ctx);
             if (cp.UntilSerialNo == -1)
                 throw new Exception($"Unable to find session {guid} to recover");
@@ -74,25 +80,6 @@ internal void DisposeClientSession(Guid guid)
         {
             lock (_activeSessions)
                 _activeSessions.Remove(guid);
-            epoch.Release();
-        }
-
-        /// <summary>
-        /// Resume session with FASTER
-        /// </summary>
-        /// <param name="ctx">Context</param>
-        internal void ResumeSession(FasterExecutionContext ctx)
-        {
-            epoch.Resume();
-            InternalRefresh(ctx);
-        }
-
-        /// <summary>
-        /// Suspend session with FASTER
-        /// </summary>
-        internal void SuspendSession()
-        {
-            epoch.Suspend();
         }
     }
 }