Skip to content

Commit

Permalink
Adding FLUSHALL command (#553)
Browse files Browse the repository at this point in the history 
* Adding FLUSHALL command

* Adding comment

* Fix
  • Loading branch information
@TalZaccai
TalZaccai committed Aug 2, 2024
1 parent b629644 commit 98772c0
Show file tree
Hide file tree
Showing 10 changed files with 180 additions and 77 deletions.
149 changes: 89 additions & 60 deletions libs/server/Resp/BasicCommands.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -888,74 +888,31 @@ private bool NetworkQUIT()
/// <summary>
/// FLUSHDB [ASYNC|SYNC] [UNSAFETRUNCATELOG]
/// </summary>
private bool NetworkFLUSHDB(int count)
private bool NetworkFLUSHDB()
{
if (count > 2)
if (parseState.count > 2)
{
return AbortWithWrongNumberOfArguments(nameof(RespCommand.FLUSHDB), count);
return AbortWithWrongNumberOfArguments(nameof(RespCommand.FLUSHDB), parseState.count);
}

var unsafeTruncateLog = false;
var async = false;
var sync = false;
var syntaxError = false;

for (var i = 0; i < count; i++)
{
var nextToken = parseState.GetArgSliceByRef(i).ReadOnlySpan;

if (nextToken.EqualsUpperCaseSpanIgnoringCase(CmdStrings.UNSAFETRUNCATELOG))
{
if (unsafeTruncateLog)
{
syntaxError = true;
break;
}

unsafeTruncateLog = true;
}
else if (nextToken.EqualsUpperCaseSpanIgnoringCase(CmdStrings.ASYNC))
{
if (sync || async)
{
syntaxError = true;
break;
}

async = true;
}
else if (nextToken.EqualsUpperCaseSpanIgnoringCase(CmdStrings.SYNC))
{
if (sync || async)
{
syntaxError = true;
break;
}
FlushDb(RespCommand.FLUSHDB);

sync = true;
}
else
{
syntaxError = true;
break;
}
}
return true;
}

if (syntaxError)
/// <summary>
/// FLUSHALL [ASYNC|SYNC] [UNSAFETRUNCATELOG]
/// </summary>
private bool NetworkFLUSHALL()
{
if (parseState.count > 2)
{
while (!RespWriteUtils.WriteError(CmdStrings.RESP_SYNTAX_ERROR, ref dcurr, dend))
SendAndReset();
return true;
return AbortWithWrongNumberOfArguments(nameof(RespCommand.FLUSHALL), parseState.count);
}

if (async)
Task.Run(() => FlushDB(unsafeTruncateLog)).ConfigureAwait(false);
else
FlushDB(unsafeTruncateLog);

logger?.LogInformation("Running flushDB " + (async ? "async" : "sync") + (unsafeTruncateLog ? " with unsafetruncatelog." : ""));
while (!RespWriteUtils.WriteDirect(CmdStrings.RESP_OK, ref dcurr, dend))
SendAndReset();
// Since Garnet currently only supports a single database,
// FLUSHALL and FLUSHDB share the same logic
FlushDb(RespCommand.FLUSHALL);

return true;
}
Expand Down Expand Up @@ -1485,7 +1442,79 @@ void ProcessHelloCommand(byte? respProtocolVersion, ReadOnlySpan<byte> username,
SendAndReset();
}

void FlushDB(bool unsafeTruncateLog)
/// <summary>
/// Common logic for FLUSHDB and FLUSHALL
/// </summary>
/// <param name="cmd">RESP command (FLUSHDB / FLUSHALL)</param>
void FlushDb(RespCommand cmd)
{
Debug.Assert(cmd is RespCommand.FLUSHDB or RespCommand.FLUSHALL);
var unsafeTruncateLog = false;
var async = false;
var sync = false;
var syntaxError = false;

var count = parseState.count;
for (var i = 0; i < count; i++)
{
var nextToken = parseState.GetArgSliceByRef(i).ReadOnlySpan;

if (nextToken.EqualsUpperCaseSpanIgnoringCase(CmdStrings.UNSAFETRUNCATELOG))
{
if (unsafeTruncateLog)
{
syntaxError = true;
break;
}

unsafeTruncateLog = true;
}
else if (nextToken.EqualsUpperCaseSpanIgnoringCase(CmdStrings.ASYNC))
{
if (sync || async)
{
syntaxError = true;
break;
}

async = true;
}
else if (nextToken.EqualsUpperCaseSpanIgnoringCase(CmdStrings.SYNC))
{
if (sync || async)
{
syntaxError = true;
break;
}

sync = true;
}
else
{
syntaxError = true;
break;
}
}

if (syntaxError)
{
while (!RespWriteUtils.WriteError(CmdStrings.RESP_SYNTAX_ERROR, ref dcurr, dend))
SendAndReset();
return;
}

if (async)
Task.Run(() => ExecuteFlushDb(unsafeTruncateLog)).ConfigureAwait(false);
else
ExecuteFlushDb(unsafeTruncateLog);

logger?.LogInformation(
$"Running {nameof(cmd)} {(async ? "async" : "sync")} {(unsafeTruncateLog ? " with unsafetruncatelog." : string.Empty)}");
while (!RespWriteUtils.WriteDirect(CmdStrings.RESP_OK, ref dcurr, dend))
SendAndReset();
}

void ExecuteFlushDb(bool unsafeTruncateLog)
{
storeWrapper.store.Log.ShiftBeginAddress(storeWrapper.store.Log.TailAddress, truncateLog: unsafeTruncateLog);
storeWrapper.objectStore?.Log.ShiftBeginAddress(storeWrapper.objectStore.Log.TailAddress, truncateLog: unsafeTruncateLog);
Expand Down
1 change: 1 addition & 0 deletions libs/server/Resp/CmdStrings.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,7 @@ static partial class CmdStrings
public static ReadOnlySpan<byte> MONITOR => "MONITOR"u8;
public static ReadOnlySpan<byte> monitor => "monitor"u8;
public static ReadOnlySpan<byte> COMMITAOF => "COMMITAOF"u8;
public static ReadOnlySpan<byte> FLUSHALL => "FLUSHALL"u8;
public static ReadOnlySpan<byte> FLUSHDB => "FLUSHDB"u8;
public static ReadOnlySpan<byte> FORCEGC => "FORCEGC"u8;
public static ReadOnlySpan<byte> MATCH => "MATCH"u8;
Expand Down
5 changes: 5 additions & 0 deletions libs/server/Resp/Parser/RespCommand.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,7 @@ public enum RespCommand : byte
DECRBY,
DEL,
EXPIRE,
FLUSHALL,
FLUSHDB,
GEOADD,
GETDEL,
Expand Down Expand Up @@ -1608,6 +1609,10 @@ private RespCommand SlowParseCommand(ref int count, ref ReadOnlySpan<byte> speci
{
return RespCommand.COMMITAOF;
}
else if (command.SequenceEqual(CmdStrings.FLUSHALL))
{
return RespCommand.FLUSHALL;
}
else if (command.SequenceEqual(CmdStrings.FLUSHDB))
{
return RespCommand.FLUSHDB;
Expand Down
35 changes: 26 additions & 9 deletions libs/server/Resp/RespCommandsInfo.json
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1210,8 +1210,8 @@
"SubCommands": null
},
{
"Command": "CustomRawStringCmd",
"Name": "CustomRawStringCmd",
"Command": "CustomObjCmd",
"Name": "CustomObjCmd",
"IsInternal": true,
"Arity": 1,
"Flags": "None",
Expand All @@ -1224,8 +1224,8 @@
"SubCommands": null
},
{
"Command": "CustomObjCmd",
"Name": "CustomObjCmd",
"Command": "CustomProcedure",
"Name": "CustomProcedure",
"IsInternal": true,
"Arity": 1,
"Flags": "None",
Expand All @@ -1238,8 +1238,8 @@
"SubCommands": null
},
{
"Command": "CustomTxn",
"Name": "CustomTxn",
"Command": "CustomRawStringCmd",
"Name": "CustomRawStringCmd",
"IsInternal": true,
"Arity": 1,
"Flags": "None",
Expand All @@ -1252,15 +1252,15 @@
"SubCommands": null
},
{
"Command": "CustomProcedure",
"Name": "CustomProcedure",
"Command": "CustomTxn",
"Name": "CustomTxn",
"IsInternal": true,
"Arity": 1,
"Flags": "None",
"FirstKey": 0,
"LastKey": 0,
"Step": 0,
"AclCategories": "Garnet, Custom, Dangerous, Slow",
"AclCategories": "Dangerous, Slow, Garnet, Custom",
"Tips": null,
"KeySpecifications": null,
"SubCommands": null
Expand Down Expand Up @@ -1489,6 +1489,23 @@
"KeySpecifications": null,
"SubCommands": null
},
{
"Command": "FLUSHALL",
"Name": "FLUSHALL",
"IsInternal": false,
"Arity": -1,
"Flags": "Write",
"FirstKey": 0,
"LastKey": 0,
"Step": 0,
"AclCategories": "Dangerous, KeySpace, Slow, Write",
"Tips": [
"request_policy:all_shards",
"response_policy:all_succeeded"
],
"KeySpecifications": null,
"SubCommands": null
},
{
"Command": "FLUSHDB",
"Name": "FLUSHDB",
Expand Down
3 changes: 2 additions & 1 deletion libs/server/Resp/RespServerSession.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -477,7 +477,8 @@ private bool ProcessBasicCommands<TGarnetApi>(RespCommand cmd, ref TGarnetApi st
RespCommand.INFO => NetworkINFO(parseState.count),
RespCommand.HELLO => NetworkHELLO(parseState.count),
RespCommand.TIME => NetworkTIME(parseState.count),
RespCommand.FLUSHDB => NetworkFLUSHDB(parseState.count),
RespCommand.FLUSHALL => NetworkFLUSHALL(),
RespCommand.FLUSHDB => NetworkFLUSHDB(),
RespCommand.AUTH => NetworkAUTH(parseState.count),
RespCommand.MEMORY_USAGE => NetworkMemoryUsage(parseState.count, ref storageApi),
RespCommand.ACL_CAT => NetworkAclCat(parseState.count),
Expand Down
1 change: 1 addition & 0 deletions playground/CommandInfoUpdater/SupportedCommand.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,7 @@ public class SupportedCommand
new("EXISTS", RespCommand.EXISTS),
new("EXPIRE", RespCommand.EXPIRE),
new("FAILOVER", RespCommand.FAILOVER),
new("FLUSHALL", RespCommand.FLUSHALL),
new("FLUSHDB", RespCommand.FLUSHDB),
new("FORCEGC", RespCommand.FORCEGC),
new("GEOADD", RespCommand.GEOADD),
Expand Down
21 changes: 21 additions & 0 deletions test/Garnet.test/Resp/ACL/RespCommandTests.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2672,6 +2672,27 @@ static async Task DoFlushDBAsyncAsync(GarnetClient client)
}
}

[Test]
public async Task FlushAllACLsAsync()
{
await CheckCommandsAsync(
"FLUSHALL",
[DoFlushAllAsync, DoFlushAllAsyncAsync]
);

static async Task DoFlushAllAsync(GarnetClient client)
{
string val = await client.ExecuteForStringResultAsync("FLUSHALL");
Assert.AreEqual("OK", val);
}

static async Task DoFlushAllAsyncAsync(GarnetClient client)
{
string val = await client.ExecuteForStringResultAsync("FLUSHALL", ["ASYNC"]);
Assert.AreEqual("OK", val);
}
}

[Test]
public async Task ForceGCACLsAsync()
{
Expand Down
Loading

0 comments on commit 98772c0

Please sign in to comment.