feat: support redis transaction #10
+1,322
−246
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Binary file not shown.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -126,6 +126,7 @@ class PClient : public std::enable_shared_from_this<PClient> { | |
| } | ||
| } | ||
|
|
||
| bool CheckTransation(std::vector<std::string>& param); | ||
| bool Watch(int dbno, const std::string& key); | ||
| bool NotifyDirty(int dbno, const std::string& key); | ||
| bool Exec(); | ||
|
|
@@ -239,8 +240,8 @@ class PClient : public std::enable_shared_from_this<PClient> { | |
| std::unordered_set<std::string> pattern_channels_; | ||
|
|
||
| uint32_t flag_ = 0; | ||
| std::unordered_map<int32_t, std::unordered_set<std::string>> watch_keys_; | ||
| std::vector<std::vector<std::string>> queue_cmds_; | ||
|
|
||
| // blocked list | ||
| std::unordered_set<std::string> waiting_keys_; | ||
|
|
@@ -262,7 +263,6 @@ class PClient : public std::enable_shared_from_this<PClient> { | |
| time_t last_auth_ = 0; | ||
|
|
||
| ClientState state_; | ||
| uint64_t net_id_ = 0; | ||
| int8_t net_thread_index_ = 0; | ||
| net::SocketAddr addr_; | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -20,6 +20,7 @@ | |
| #include "cmd_set.h" | ||
| #include "cmd_zset.h" | ||
| #include "std_string.h" | ||
| #include "transaction.h" | ||
|
|
||
| namespace kiwi { | ||
|
|
||
|
|
@@ -51,6 +52,10 @@ CmdTableManager::CmdTableManager() { | |
| void CmdTableManager::InitCmdTable() { | ||
| std::unique_lock wl(mutex_); | ||
|
|
||
| if (cmds_->size() != 0) { | ||
| return; | ||
| } | ||
|
|
||
| // admin | ||
| ADD_COMMAND_GROUP(Config, -2); | ||
| ADD_SUBCOMMAND(Config, Get, -3); | ||
|
|
@@ -198,6 +203,13 @@ void CmdTableManager::InitCmdTable() { | |
| ADD_COMMAND(ZRevrank, 3); | ||
| ADD_COMMAND(ZRem, -3); | ||
| ADD_COMMAND(ZIncrby, 4); | ||
|
|
||
| // multi | ||
| ADD_COMMAND(Multi, 1); | ||
| ADD_COMMAND(Watch, -2); | ||
| ADD_COMMAND(UnWatch, 1); | ||
| ADD_COMMAND(Exec, 1); | ||
| ADD_COMMAND(Discard, 1); | ||
| } | ||
|
|
||
| std::pair<BaseCmd*, CmdRes> CmdTableManager::GetCommand(const std::string& cmdName, PClient* client) { | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -35,14 +35,17 @@ void CmdWorkThreadPoolWorker::Work() { | |
| if (!cmdPtr) { | ||
| if (ret == CmdRes::kUnknownCmd) { | ||
| task->Client()->SetRes(CmdRes::kUnknownCmd, fmt::format("unknown command '{}'", param[0])); | ||
| task->Client()->FlagExecWrong(); | ||
| WARN("client IP:{},port:{} unknown command '{}'", task->Client()->PeerIP(), task->Client()->PeerPort(), | ||
| param[0]); | ||
| } else if (ret == CmdRes::kUnknownSubCmd) { | ||
| task->Client()->SetRes(CmdRes::kUnknownSubCmd, task->Client()->argv_[1]); | ||
| task->Client()->FlagExecWrong(); | ||
| WARN("client IP:{},port:{} unknown sub command '{}'", task->Client()->PeerIP(), task->Client()->PeerPort(), | ||
| task->Client()->argv_[1]); | ||
| } else { | ||
| task->Client()->SetRes(CmdRes::kWrongNum, param[0]); | ||
| task->Client()->FlagExecWrong(); | ||
| WARN("client IP:{},port:{} unknown command '{}'", task->Client()->PeerIP(), task->Client()->PeerPort(), | ||
| param[0]); | ||
| } | ||
|
|
@@ -52,10 +55,16 @@ void CmdWorkThreadPoolWorker::Work() { | |
|
|
||
| if (!cmdPtr->CheckArg(task->Client()->ParamsSize())) { | ||
| task->Client()->SetRes(CmdRes::kWrongNum, param[0]); | ||
| task->Client()->FlagExecWrong(); | ||
| g_kiwi->PushWriteTask(task->Client()); | ||
| continue; | ||
| } | ||
|
|
||
| // check transaction | ||
| if (task->Client()->CheckTransation(param)) { | ||
| continue; | ||
| } | ||
|
|
||
|
Comment on lines
+63
to
+67
There was a problem hiding this comment. 💡 Codebase verification Transaction check implementation is correct but contains a typo The transaction check implementation is properly integrated with appropriate error handling and state management. However, there is a consistent typo in the method name:
🔗 Analysis chainVerify transaction check implementation. The new transaction check is added but its behavior needs verification, especially around error handling and state management. Run this script to verify the transaction check implementation: 🏁 Scripts executedThe following scripts were executed for the analysis: Script: #!/bin/bash
# Description: Verify transaction check implementation and error handling
# Test: Search for transaction check usage and related error handling
ast-grep --pattern 'CheckTransation($$$)'
Length of output: 43 Script: #!/bin/bash
# Search for both spellings and Client class implementation
echo "=== Searching for Transaction methods ==="
rg -p "CheckTransaction|CheckTransation" -t cpp -t hpp
echo -e "\n=== Looking for Client class definition ==="
ast-grep --pattern 'class Client {
$$$
}'
echo -e "\n=== Searching for transaction-related code ==="
rg -p "transaction|Transaction" -t cpp -t hpp -C 2
Length of output: 516 Script: #!/bin/bash
# Search with corrected file types and broader patterns
echo "=== Searching for Client class and methods ==="
rg -p "class.*Client" -g "*.{h,cc}"
echo -e "\n=== Searching for transaction check methods ==="
rg -p "Check.*Trans.*" -g "*.{h,cc}" -C 2
echo -e "\n=== Searching for transaction-related code ==="
rg -p "transaction|Transaction" -g "*.{h,cc}" -C 2
echo -e "\n=== Looking for the specific file content ==="
cat src/cmd_thread_pool_worker.cc
Length of output: 12283 |
||
| auto cmdstat_map = task->Client()->GetCommandStatMap(); | ||
| CommandStatistics statistics; | ||
| if (cmdstat_map->find(param[0]) == cmdstat_map->end()) { | ||
|
|
@@ -71,6 +80,8 @@ void CmdWorkThreadPoolWorker::Work() { | |
| (*cmdstat_map)[param[0]].cmd_count_.fetch_add(1); | ||
| (*cmdstat_map)[param[0]].cmd_time_consuming_.fetch_add(task->Client()->GetTimeStat()->GetTotalTime()); | ||
|
|
||
| task->Client()->FeedMonitors(param); | ||
|
|
||
| g_kiwi->PushWriteTask(task->Client()); | ||
| } | ||
| } | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -34,6 +34,12 @@ void Resp2Encode::SetRes(CmdRes ret, const std::string& content) { | |
| case CmdRes::kInvalidBitOffsetInt: | ||
| SetLineString("-ERR bit offset is not an integer or out of range"); | ||
| break; | ||
| case CmdRes::kInvalidBitPosArgument: | ||
| SetLineString("-ERR The bit argument must be 1 or 0."); | ||
| break; | ||
| case CmdRes::kWrongBitOpNotNum: | ||
| SetLineString("-ERR BITOP NOT must be called with a single source key."); | ||
| break; | ||
| case CmdRes::kInvalidFloat: | ||
| SetLineString("-ERR value is not a valid float"); | ||
| break; | ||
|
|
@@ -88,6 +94,18 @@ void Resp2Encode::SetRes(CmdRes ret, const std::string& content) { | |
| AppendStringRaw( | ||
| fmt::format("-WRONGTYPE Operation against a key holding the wrong kind of value {}\r\n", content)); | ||
| break; | ||
| case CmdRes::kDirtyExec: | ||
| AppendStringRaw("-ERR EXECABORT Transaction discarded because of previous errors."); | ||
| AppendStringRaw(CRLF); | ||
| break; | ||
| case CmdRes::kPErrorWatch: | ||
| AppendStringRaw("-ERR WATCH inside MULTI is not allowed"); | ||
| AppendStringRaw(CRLF); | ||
| break; | ||
| case CmdRes::kQueued: | ||
| AppendStringRaw("+QUEUED"); | ||
| AppendStringRaw(CRLF); | ||
| break; | ||
| case CmdRes::kNoAuth: | ||
| SetLineString("-NOAUTH Authentication required"); | ||
| break; | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -18,6 +18,8 @@ enum class CmdRes : std::int8_t { | |
| kInvalidInt, | ||
| kInvalidBitInt, | ||
| kInvalidBitOffsetInt, | ||
| kInvalidBitPosArgument, | ||
| kWrongBitOpNotNum, | ||
| kInvalidFloat, | ||
| kOverFlow, | ||
| kNotFound, | ||
|
|
@@ -38,6 +40,9 @@ enum class CmdRes : std::int8_t { | |
| kInvalidCursor, | ||
| kWrongLeader, | ||
| kMultiKey, | ||
| kDirtyExec, | ||
| kPErrorWatch, | ||
| kQueued, | ||
| kNoAuth, | ||
| }; | ||
|
|
||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -17,6 +17,7 @@ | |
| #include "db.h" | ||
| #include "std/log.h" | ||
| #include "std/std_string.h" | ||
| #include "transaction.h" | ||
|
|
||
| namespace kiwi { | ||
|
|
||
|
|
@@ -76,4 +77,34 @@ void Store::HandleTaskSpecificDB(const TasksVector& tasks) { | |
| } | ||
| }); | ||
| } | ||
|
|
||
| void Propagate(const std::vector<PString>& params, int dbno) { | ||
| assert(!params.empty()); | ||
| // | ||
| // if (!g_dirtyKeys.empty()) { | ||
| // for (const auto& k : g_dirtyKeys) { | ||
| // PTransaction::Instance().NotifyDirty(PSTORE.GetDBNumber(), k); | ||
| // | ||
| // } | ||
| // g_dirtyKeys.clear(); | ||
| // } else if (params.size() > 1) { | ||
| // PTransaction::Instance().NotifyDirty(PSTORE.GetDBNumber(), params[1]); | ||
| // } | ||
| if (params.size() > 1) { | ||
| PTransaction::Instance().NotifyDirty(dbno, params[1]); | ||
| } | ||
| } | ||
|
Comment on lines
+81
to
+96
There was a problem hiding this comment. Validate In Modify the code to include a size check: void Propagate(const std::vector<PString>& params, int dbno) {
if (params.size() > 1) {
PTransaction::Instance().NotifyDirty(dbno, params[1]);
} else {
// Handle insufficient parameters
}
} |
||
|
|
||
| void Propagate(int dbno, const std::vector<PString>& params) { | ||
| PTransaction::Instance().NotifyDirtyAll(dbno); | ||
| Propagate(params, dbno); | ||
| } | ||
|
|
||
| void signalModifiedKey(const std::vector<PString>& keys, int dbno) { | ||
| if (keys.size() > 1) { | ||
| for (const auto& key : keys) { | ||
| PTransaction::Instance().NotifyDirty(dbno, key); | ||
| } | ||
| } | ||
|
Comment on lines
+103
to
+108
There was a problem hiding this comment. Ensure all modified keys are signaled regardless of count In Update the condition to include all non-empty key vectors: void signalModifiedKey(const std::vector<PString>& keys, int dbno) {
if (!keys.empty()) {
for (const auto& key : keys) {
PTransaction::Instance().NotifyDirty(dbno, key);
}
}
} |
||
| } | ||
| } // namespace kiwi | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Ensure thread-safe access to the global command table manager.
The global
cmd_table_managervariable could lead to race conditions in a multi-threaded environment. Consider: