Skip to content

Commit f1965de

Browse files
chejingeForestLHcheniujhAlexStocks
authored
Feature/txn (#1585) (#2124)
* Feature/txn (#1585) * fix: fix select cmd return inconsistent with redis Signed-off-by: Hao Lee <1838249551@qq.com> * refactor:modified lock style while involve db level Signed-off-by: Hao Lee <1838249551@qq.com> * feature:txn basic Signed-off-by: Hao Lee <1838249551@qq.com> * fix:merge upstream Signed-off-by: Hao Lee <1838249551@qq.com> * feature:txn udpate Signed-off-by: Hao Lee <1838249551@qq.com> * feature:add txn for pika(#1446) todo:test txn. Just to verify the feasibility of the program. Signed-off-by: Hao Lee <1838249551@qq.com> * update unwatch cmd Add comments Signed-off-by: Hao Lee <1838249551@qq.com> * clear watched key when connection closed Signed-off-by: Hao Lee <1838249551@qq.com> * merge upstream code Signed-off-by: Hao Lee <1838249551@qq.com> * update Signed-off-by: Hao Lee <1838249551@qq.com> * feature: add txn for pika completely Signed-off-by: Hao Lee <1838249551@qq.com> * add set txn failed for modified watch key Signed-off-by: Hao Lee <1838249551@qq.com> * update:reduce the particle size of the lock in txn Signed-off-by: Hao Lee <1838249551@qq.com> * chore:remove redundant comment Signed-off-by: Hao Lee <1838249551@qq.com> * test:add go ci test for txn Signed-off-by: Hao Lee <1838249551@qq.com> * fix compile error for linux Signed-off-by: Hao Lee <1838249551@qq.com> * update txn go ci test Signed-off-by: Hao Lee <1838249551@qq.com> * update txn for block list pop command Signed-off-by: Hao Lee <1838249551@qq.com> * Improve blpop-related in Redis transactions Signed-off-by: Hao Lee <1838249551@qq.com> * blpop_txn_fix * add some test for go test txn Signed-off-by: Hao Lee <1838249551@qq.com> * update txn integration test Signed-off-by: Hao Lee <1838249551@qq.com> * txn change class to struct Signed-off-by: Hao Lee <1838249551@qq.com> * txn:use weak ptr instead of shared ptr in Cmd Signed-off-by: Hao Lee <1838249551@qq.com> --------- Signed-off-by: Hao Lee <1838249551@qq.com> Co-authored-by: cheniujh <1271435567@qq.com> * FNT * fix:txn compile error in ubuntu (#2128) Signed-off-by: LeeHao <1838249551@qq.com> * using func instead of class private member (#2130) * using func instead of class private member --------- Signed-off-by: Hao Lee <1838249551@qq.com> Signed-off-by: LeeHao <1838249551@qq.com> Co-authored-by: LeeHao <39085999+ForestLH@users.noreply.github.com> Co-authored-by: cheniujh <1271435567@qq.com> Co-authored-by: Xin.Zh <dragoncharlie@foxmail.com>
1 parent 4c140cb commit f1965de

7 files changed

Lines changed: 23 additions & 17 deletions

File tree

‎include/pika_command.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -530,7 +530,7 @@ class Cmd : public std::enable_shared_from_this<Cmd> {
530530
protected:
531531
CmdRes res_;
532532
PikaCmdArgsType argv_;
533-
std::string db_name_{};
533+
std::string db_name_;
534534

535535
std::weak_ptr<net::NetConn> conn_;
536536
std::weak_ptr<std::string> resp_;

‎include/pika_transaction.h‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -18,9 +18,9 @@ class MultiCmd : public Cmd {
1818
Cmd* Clone() override { return new MultiCmd(*this); }
1919
void Split(std::shared_ptr<Slot> slot, const HintKeys& hint_keys) override {}
2020
void Merge() override {}
21+
2122
private:
2223
void DoInitial() override;
23-
2424
};
2525

2626
class ExecCmd : public Cmd {
@@ -56,6 +56,7 @@ class ExecCmd : public Cmd {
5656
std::vector<CmdInfo> list_cmd_;
5757
std::vector<std::string> keys_;
5858
};
59+
5960
class DiscardCmd : public Cmd {
6061
public:
6162
DiscardCmd(const std::string& name, int arity, uint16_t flag) : Cmd(name, arity, flag) {}
@@ -67,8 +68,6 @@ class DiscardCmd : public Cmd {
6768
void DoInitial() override;
6869
};
6970

70-
71-
7271
class WatchCmd : public Cmd {
7372
public:
7473
WatchCmd(const std::string& name, int arity, uint16_t flag) : Cmd(name, arity, flag) {}

‎src/pika_admin.cc‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -519,6 +519,7 @@ void FlushallCmd::FlushAllWithoutLock() {
519519
res_.SetRes(CmdRes::kOk);
520520
}
521521
}
522+
522523
void FlushallCmd::DoWithoutLock(std::shared_ptr<Slot> slot) {
523524
if (!slot) {
524525
LOG(INFO) << "Flushall, but Slot not found";

‎src/pika_client_conn.cc‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -274,6 +274,7 @@ void PikaClientConn::TryWriteResp() {
274274
NotifyEpoll(true);
275275
}
276276
}
277+
277278
void PikaClientConn::PushCmdToQue(std::shared_ptr<Cmd> cmd) {
278279
txn_cmd_que_.push(cmd);
279280
}
@@ -287,6 +288,7 @@ bool PikaClientConn::IsTxnFailed() {
287288
std::lock_guard<std::mutex> lg(txn_state_mu_);
288289
return txn_state_[TxnStateBitMask::WatchFailed] | txn_state_[TxnStateBitMask::InitCmdFailed];
289290
}
291+
290292
bool PikaClientConn::IsTxnInitFailed() {
291293
std::lock_guard<std::mutex> lg(txn_state_mu_);
292294
return txn_state_[TxnStateBitMask::InitCmdFailed];
@@ -306,6 +308,7 @@ void PikaClientConn::SetTxnWatchFailState(bool is_failed) {
306308
std::lock_guard<std::mutex> lg(txn_state_mu_);
307309
txn_state_[TxnStateBitMask::WatchFailed] = is_failed;
308310
}
311+
309312
void PikaClientConn::SetTxnInitFailState(bool is_failed) {
310313
std::lock_guard<std::mutex> lg(txn_state_mu_);
311314
txn_state_[TxnStateBitMask::InitCmdFailed] = is_failed;
@@ -364,6 +367,7 @@ void PikaClientConn::SetAllTxnFailed() {
364367
}
365368
}
366369
}
370+
367371
void PikaClientConn::SetTxnFailedFromDBs(std::string db_name) {
368372
auto dispatcher = dynamic_cast<net::DispatchThread *>(server_thread());
369373
if (dispatcher != nullptr) {
@@ -375,6 +379,7 @@ void PikaClientConn::SetTxnFailedFromDBs(std::string db_name) {
375379
}
376380
}
377381
}
382+
378383
void PikaClientConn::ExitTxn() {
379384
if (IsInTxn()) {
380385
RemoveWatchedKeys();
@@ -384,7 +389,6 @@ void PikaClientConn::ExitTxn() {
384389
}
385390
}
386391

387-
388392
void PikaClientConn::ExecRedisCmd(const PikaCmdArgsType& argv, std::shared_ptr<std::string>& resp_ptr) {
389393
// get opt
390394
std::string opt = argv[0];

‎src/pika_slot.cc‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -484,6 +484,7 @@ bool Slot::FlushSubDB(const std::string& db_name) {
484484
std::lock_guard rwl(db_rwlock_);
485485
return FlushSubDBWithoutLock(db_name);
486486
}
487+
487488
bool Slot::FlushSubDBWithoutLock(const std::string& db_name) {
488489
std::lock_guard l(bgsave_protector_);
489490
if (bgsave_info_.bgsaving) {
@@ -510,6 +511,7 @@ bool Slot::FlushSubDBWithoutLock(const std::string& db_name) {
510511
g_pika_server->PurgeDir(del_dbpath);
511512
return true;
512513
}
514+
513515
void Slot::InitKeyScan() {
514516
key_scan_info_.start_time = time(nullptr);
515517
char s_time[32];
@@ -542,4 +544,3 @@ Status Slot::GetKeyNum(std::vector<storage::KeyInfo>* key_info) {
542544
key_scan_info_.duration = static_cast<int32_t>(time(nullptr) - key_scan_info_.start_time);
543545
return Status::OK();
544546
}
545-

‎src/pika_transaction.cc‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,14 +3,16 @@
33
// LICENSE file in the root directory of this source tree. An additional grant
44
// of patent rights can be found in the PATENTS file in the same directory.
55

6-
#include "include/pika_transaction.h"
76
#include <memory>
7+
8+
#include "include/pika_transaction.h"
89
#include "include/pika_admin.h"
910
#include "include/pika_client_conn.h"
1011
#include "include/pika_define.h"
1112
#include "include/pika_list.h"
1213
#include "include/pika_rm.h"
1314
#include "include/pika_server.h"
15+
#include "include/pika_transaction.h"
1416

1517
extern std::unique_ptr<PikaServer> g_pika_server;
1618
extern std::unique_ptr<PikaReplicaManager> g_pika_rm;
@@ -37,7 +39,6 @@ void MultiCmd::DoInitial() {
3739
}
3840
}
3941

40-
4142
void ExecCmd::Do(std::shared_ptr<Slot> slot) {
4243
auto conn = GetConn();
4344
auto client_conn = std::dynamic_pointer_cast<PikaClientConn>(conn);

‎tests/integration/txn_test.go‎

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@ var _ = Describe("Text Txn", func() {
3333
})
3434
Describe("test watch", func() {
3535
It("basic watch", func() {
36-
txnClient.Watch(ctx, func(tx *redis.Tx) error { // 这个func相当于就是被一对watch和unwatch所包含了
36+
txnClient.Watch(ctx, func(tx *redis.Tx) error { // including in a pair of watch and unwatch
3737
pipe := tx.TxPipeline()
3838
cmdClient.Set(ctx, "key", "1", 0)
3939
pipe.Set(ctx, "key", "2", 0)
@@ -61,7 +61,7 @@ var _ = Describe("Text Txn", func() {
6161
return nil
6262
}, "key")
6363
})
64-
// 在事务中有另一个事务来使用flushdb清除db1的数据，不会影响到watch的这个db的key的事务执行
64+
// Having another transaction in the transaction to clear the data in db1 using flushdb will not affect the transaction execution of the key in this db of watch
6565
It("test watch1", func() {
6666
watchKey := "key"
6767
watchkeyValue := "value"
@@ -81,7 +81,7 @@ var _ = Describe("Text Txn", func() {
8181
return nil
8282
}, watchKey)
8383
})
84-
// 测试watch的key有多个类型
84+
// multiple types of keys for testing watch
8585
It("test watch multi type key", func() {
8686
watchKey := "key"
8787
watchkeyValue := "value"
@@ -94,7 +94,7 @@ var _ = Describe("Text Txn", func() {
9494
}, watchKey)
9595
Expect(err).To(HaveOccurred())
9696
})
97-
//// 测试flushall命令会使watch的key失败
97+
// Testing the flushall command will cause watch's key to fail
9898
It("txn failed cause of flushall", func() {
9999
watchKey := "key"
100100
watchkeyValue := "value"
@@ -114,7 +114,7 @@ var _ = Describe("Text Txn", func() {
114114
return nil
115115
}, watchKey)
116116
})
117-
// 测试select命令
117+
// test 'select' command
118118
It("select in txn", func() {
119119
watchKey := "key"
120120
noExist := "noExist"
@@ -126,20 +126,20 @@ var _ = Describe("Text Txn", func() {
126126
Expect(intCmd.Err()).NotTo(HaveOccurred())
127127

128128
err := txnClient.Watch(ctx, func(tx *redis.Tx) error {
129-
tx.Select(ctx, 1) // 这个是和txnClient.Watch使用的一个端口
129+
tx.Select(ctx, 1) // this command used the same port with txnClient.Watch
130130
tx.Watch(ctx, watchKey)
131131
cmdClient.Set(ctx, watchKey, watchkeyValue, 0)
132132
pipeline := tx.TxPipeline()
133133
pipeline.Set(ctx, watchKey, modifiedValue, 0)
134134
pipeline.Get(ctx, watchKey)
135-
cmders, _ := pipeline.Exec(ctx) // 这个也是和txnClient.Watch使用的一个端口
135+
cmders, _ := pipeline.Exec(ctx) // using the same port with txnClient.Watch
136136
AssertEqualRedisString(modifiedValue, cmders[1])
137137
return nil
138138
}, noExist)
139139
Expect(err).NotTo(HaveOccurred())
140140
})
141141

142-
// 测试执行在事务中执行命令时不阻塞其他普通命令的执行
142+
// The test execution does not block the execution of other ordinary commands when executing commands in transactions
143143
It("test txn no block other cmd", func() {
144144
pipe := txnClient.TxPipeline()
145145
pipe.Get(ctx, "key")
@@ -183,7 +183,7 @@ var _ = Describe("Text Txn", func() {
183183
})
184184
})
185185
Describe("Test Unwatch", func() {
186-
// 测试unwatch的基本功能，unwatch之后事务应该不受影响
186+
// test unwatch
187187
It("unwatch1", func() {
188188
watchKey := "key"
189189
watchkeyValue := "value"

0 commit comments

Comments
 (0)