[C#] Fix missing Sunlock on nested CopyToTail lock (#857)

* Fix missing Sunlock on nested CopyToTail lock

* Improve handling of Sealed records in Ephemeral locking

---------

Co-authored-by: Badrish Chandramouli <badrishc@microsoft.com>

Author: TedHartMS

cs/src/core/Index/Common/RecordInfo.cs

@@ -141,7 +141,8 @@ public bool TryLockExclusive()
             for (; ; Thread.Yield())
             {
                 long expected_word = word;
-                Debug.Assert(!IsClosedWord(expected_word), "Should not be X locking closed records, pt 1");
+                if (IsClosedWord(expected_word))
+                    return false;
                 if ((expected_word & kExclusiveLockBitMask) == 0)
                 {
                     if (expected_word == Interlocked.CompareExchange(ref word, expected_word | kExclusiveLockBitMask, expected_word))
@@ -198,7 +199,8 @@ public bool TryLockShared()
             for (; ; Thread.Yield())
             {
                 long expected_word = word;
-                Debug.Assert(!IsClosedWord(expected_word), "Should not be S locking closed records");
+                if (IsClosedWord(expected_word))
+                    return false;
                 if (((expected_word & kExclusiveLockBitMask) == 0) // not exclusively locked
                     && (expected_word & kSharedLockMaskInWord) != kSharedLockMaskInWord) // shared lock is not full
                 {

cs/src/core/Index/FASTER/Implementation/ConditionalCopyToTail.cs

@@ -12,7 +12,7 @@ public unsafe partial class FasterKV<Key, Value> : FasterBase, IFasterKV<Key, Va
         private OperationStatus ConditionalCopyToTail<Input, Output, Context, FasterSession>(FasterSession fasterSession,
                 ref PendingContext<Input, Output, Context> pendingContext,
                 ref Key key, ref Input input, ref Value value, ref Output output, ref Context userContext, long lsn,
-                ref OperationStackContext<Key, Value> stackCtx, WriteReason writeReason)
+                ref OperationStackContext<Key, Value> stackCtx, WriteReason writeReason, bool callerHasLock = false)
             where FasterSession : IFasterSession<Key, Value, Input, Output, Context>
         {
             // We are called by one of ReadFromImmutable, CompactionConditionalCopyToTail, or ContinueConditionalCopyToTail, and stackCtx is set up for the first try.
@@ -22,7 +22,7 @@ private OperationStatus ConditionalCopyToTail<Input, Output, Context, FasterSess
             {
                 // ConditionalCopyToTail is different in regard to locking from the usual procedures, in that if we find a source record we don't lock--we exit with success.
                 // So we only do LockTable-based locking and only when we are about to insert at the tail.
-                if (TryTransientSLock<Input, Output, Context, FasterSession>(fasterSession, ref key, ref stackCtx, out OperationStatus status))
+                if (callerHasLock || TryTransientSLock<Input, Output, Context, FasterSession>(fasterSession, ref key, ref stackCtx, out OperationStatus status))
                 {
                     try
                     {
@@ -32,7 +32,8 @@ private OperationStatus ConditionalCopyToTail<Input, Output, Context, FasterSess
                     finally
                     {
                         stackCtx.HandleNewRecordOnException(this);
-                        TransientSUnlock<Input, Output, Context, FasterSession>(fasterSession, ref key, ref stackCtx);
+                        if (!callerHasLock)
+                            TransientSUnlock<Input, Output, Context, FasterSession>(fasterSession, ref key, ref stackCtx);
                     }
                 }
                 if (!HandleImmediateRetryStatus(status, fasterSession, ref pendingContext))

cs/src/core/Index/FASTER/Implementation/ContinuePending.cs

@@ -114,7 +114,7 @@ internal OperationStatus ContinuePendingRead<Input, Output, Context, FasterSessi
                         {
                             if (pendingContext.readCopyOptions.CopyTo == ReadCopyTo.MainLog)
                                 status = ConditionalCopyToTail(fasterSession, ref pendingContext, ref key, ref pendingContext.input.Get(), ref value, ref pendingContext.output,
-                                                               ref pendingContext.userContext, pendingContext.serialNum, ref stackCtx, WriteReason.CopyToTail);
+                                                               ref pendingContext.userContext, pendingContext.serialNum, ref stackCtx, WriteReason.CopyToTail, callerHasLock: true);
                             else if (pendingContext.readCopyOptions.CopyTo == ReadCopyTo.ReadCache && !stackCtx.recSrc.HasReadCacheSrc
                                     && TryCopyToReadCache(fasterSession, ref pendingContext, ref key, ref pendingContext.input.Get(), ref value, ref stackCtx))
                                 status |= OperationStatus.COPIED_RECORD_TO_READ_CACHE;

cs/src/core/Index/FASTER/Implementation/InternalRead.cs

@@ -286,7 +286,8 @@ private OperationStatus ReadFromImmutableRegion<Input, Output, Context, FasterSe
                     if (pendingContext.readCopyOptions.CopyFrom != ReadCopyFrom.AllImmutable)
                         return OperationStatus.SUCCESS;
                     if (pendingContext.readCopyOptions.CopyTo == ReadCopyTo.MainLog)
-                        return ConditionalCopyToTail(fasterSession, ref pendingContext, ref key, ref input, ref recordValue, ref output, ref userContext, lsn, ref stackCtx, WriteReason.CopyToTail);
+                        return ConditionalCopyToTail(fasterSession, ref pendingContext, ref key, ref input, ref recordValue, ref output, ref userContext, lsn, ref stackCtx,
+                                                     WriteReason.CopyToTail, callerHasLock: true);
                     if (pendingContext.readCopyOptions.CopyTo == ReadCopyTo.ReadCache
                             && TryCopyToReadCache(fasterSession, ref pendingContext, ref key, ref input, ref recordValue, ref stackCtx))
                         return OperationStatus.SUCCESS | OperationStatus.COPIED_RECORD_TO_READ_CACHE;