Skip to content

Commit

Permalink
nit
Browse files Browse the repository at this point in the history
  • Loading branch information
@badrishc
badrishc committed Nov 11, 2024
1 parent 108ca79 commit 2db3c27
Show file tree
Hide file tree
Showing 6 changed files with 16 additions and 16 deletions.
4 changes: 2 additions & 2 deletions libs/server/AOF/AofHeader.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -35,10 +35,10 @@ struct AofHeader
[FieldOffset(3)]
public byte procedureId;
/// <summary>
/// Session version
/// Store version
/// </summary>
[FieldOffset(4)]
public long sessionVersion;
public long storeVersion;
/// <summary>
/// Session ID
/// </summary>
Expand Down
8 changes: 4 additions & 4 deletions libs/server/AOF/AofProcessor.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -206,14 +206,14 @@ public unsafe void ProcessAofRecordInternal(byte* ptr, int length, bool asReplic
case AofEntryType.MainStoreCheckpointCommit:
if (asReplica)
{
if (header.sessionVersion > storeWrapper.store.CurrentVersion)
if (header.storeVersion > storeWrapper.store.CurrentVersion)
storeWrapper.TakeCheckpoint(false, StoreType.Main, logger);
}
break;
case AofEntryType.ObjectStoreCheckpointCommit:
if (asReplica)
{
if (header.sessionVersion > storeWrapper.objectStore.CurrentVersion)
if (header.storeVersion > storeWrapper.objectStore.CurrentVersion)
storeWrapper.TakeCheckpoint(false, StoreType.Object, logger);
}
break;
Expand Down Expand Up @@ -388,8 +388,8 @@ bool SkipRecord(AofHeader header)

return storeType switch
{
AofStoreType.MainStoreType => header.sessionVersion <= storeWrapper.store.CurrentVersion - 1,
AofStoreType.ObjectStoreType => header.sessionVersion <= storeWrapper.objectStore.CurrentVersion - 1,
AofStoreType.MainStoreType => header.storeVersion <= storeWrapper.store.CurrentVersion - 1,
AofStoreType.ObjectStoreType => header.storeVersion <= storeWrapper.objectStore.CurrentVersion - 1,
AofStoreType.TxnType => false,
AofStoreType.ReplicationType => false,
AofStoreType.CheckpointType => false,
Expand Down
6 changes: 3 additions & 3 deletions libs/server/Storage/Functions/MainStore/PrivateMethods.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -678,7 +678,7 @@ void WriteLogUpsert(ref SpanByte key, ref RawStringInput input, ref SpanByte val
input.header.flags |= RespInputFlags.Deterministic;

functionsState.appendOnlyFile.Enqueue(
new AofHeader { opType = AofEntryType.StoreUpsert, sessionVersion = version, sessionID = sessionId },
new AofHeader { opType = AofEntryType.StoreUpsert, storeVersion = version, sessionID = sessionId },
ref key, ref value, ref input, out _);
}

Expand All @@ -694,7 +694,7 @@ void WriteLogRMW(ref SpanByte key, ref RawStringInput input, long version, int s
input.header.flags |= RespInputFlags.Deterministic;

functionsState.appendOnlyFile.Enqueue(
new AofHeader { opType = AofEntryType.StoreRMW, sessionVersion = version, sessionID = sessionId },
new AofHeader { opType = AofEntryType.StoreRMW, storeVersion = version, sessionID = sessionId },
ref key, ref input, out _);
}

Expand All @@ -707,7 +707,7 @@ void WriteLogDelete(ref SpanByte key, long version, int sessionID)
{
if (functionsState.StoredProcMode) return;
SpanByte def = default;
functionsState.appendOnlyFile.Enqueue(new AofHeader { opType = AofEntryType.StoreDelete, sessionVersion = version, sessionID = sessionID }, ref key, ref def, out _);
functionsState.appendOnlyFile.Enqueue(new AofHeader { opType = AofEntryType.StoreDelete, storeVersion = version, sessionID = sessionID }, ref key, ref def, out _);
}

BitFieldCmdArgs GetBitFieldArguments(ref RawStringInput input)
Expand Down
6 changes: 3 additions & 3 deletions libs/server/Storage/Functions/ObjectStore/PrivateMethods.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ void WriteLogUpsert(ref byte[] key, ref ObjectInput input, ref IGarnetObject val
var valSB = SpanByte.FromPinnedPointer(valPtr, valueBytes.Length);

functionsState.appendOnlyFile.Enqueue(
new AofHeader { opType = AofEntryType.ObjectStoreUpsert, sessionVersion = version, sessionID = sessionID },
new AofHeader { opType = AofEntryType.ObjectStoreUpsert, storeVersion = version, sessionID = sessionID },
ref keySB, ref valSB, out _);
}
}
Expand All @@ -57,7 +57,7 @@ void WriteLogRMW(ref byte[] key, ref ObjectInput input, long version, int sessio
var sbKey = SpanByte.FromPinnedPointer(keyPtr, key.Length);

functionsState.appendOnlyFile.Enqueue(
new AofHeader { opType = AofEntryType.ObjectStoreRMW, sessionVersion = version, sessionID = sessionID },
new AofHeader { opType = AofEntryType.ObjectStoreRMW, storeVersion = version, sessionID = sessionID },
ref sbKey, ref input, out _);
}
}
Expand All @@ -75,7 +75,7 @@ void WriteLogDelete(ref byte[] key, long version, int sessionID)
var keySB = SpanByte.FromPinnedPointer(ptr, key.Length);
SpanByte valSB = default;

functionsState.appendOnlyFile.Enqueue(new AofHeader { opType = AofEntryType.ObjectStoreDelete, sessionVersion = version, sessionID = sessionID }, ref keySB, ref valSB, out _);
functionsState.appendOnlyFile.Enqueue(new AofHeader { opType = AofEntryType.ObjectStoreDelete, storeVersion = version, sessionID = sessionID }, ref keySB, ref valSB, out _);
}
}

Expand Down
2 changes: 1 addition & 1 deletion libs/server/StoreWrapper.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -400,7 +400,7 @@ public void EnqueueCommit(bool isMainStore, long version)
AofHeader header = new()
{
opType = isMainStore ? AofEntryType.MainStoreCheckpointCommit : AofEntryType.ObjectStoreCheckpointCommit,
sessionVersion = version,
storeVersion = version,
sessionID = -1
};
appendOnlyFile?.Enqueue(header, out _);
Expand Down
6 changes: 3 additions & 3 deletions libs/server/Transaction/TransactionManager.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -237,14 +237,14 @@ internal void Log(byte id, ref CustomProcedureInput procInput)
{
Debug.Assert(functionsState.StoredProcMode);

appendOnlyFile?.Enqueue(new AofHeader { opType = AofEntryType.StoredProcedure, procedureId = id, sessionVersion = basicContext.Session.Version, sessionID = basicContext.Session.ID }, ref procInput, out _);
appendOnlyFile?.Enqueue(new AofHeader { opType = AofEntryType.StoredProcedure, procedureId = id, storeVersion = basicContext.Session.Version, sessionID = basicContext.Session.ID }, ref procInput, out _);
}

internal void Commit(bool internal_txn = false)
{
if (appendOnlyFile != null && !functionsState.StoredProcMode)
{
appendOnlyFile.Enqueue(new AofHeader { opType = AofEntryType.TxnCommit, sessionVersion = basicContext.Session.Version, sessionID = basicContext.Session.ID }, out _);
appendOnlyFile.Enqueue(new AofHeader { opType = AofEntryType.TxnCommit, storeVersion = basicContext.Session.Version, sessionID = basicContext.Session.ID }, out _);
}
if (!internal_txn)
watchContainer.Reset();
Expand Down Expand Up @@ -335,7 +335,7 @@ internal bool Run(bool internal_txn = false, bool fail_fast_on_lock = false, Tim

if (appendOnlyFile != null && !functionsState.StoredProcMode)
{
appendOnlyFile.Enqueue(new AofHeader { opType = AofEntryType.TxnStart, sessionVersion = basicContext.Session.Version, sessionID = basicContext.Session.ID }, out _);
appendOnlyFile.Enqueue(new AofHeader { opType = AofEntryType.TxnStart, storeVersion = basicContext.Session.Version, sessionID = basicContext.Session.ID }, out _);
}

state = TxnState.Running;
Expand Down

0 comments on commit 2db3c27

Please sign in to comment.