Skip to content

Commit 6773c12

Browse files
LiuRuoyu01LiuRuoyu01
committed
[feat][sdk]Support merge 2PC process
1 parent de385f7 commit 6773c12

7 files changed

Lines changed: 22 additions & 45 deletions

File tree

‎include/dingosdk/client.h‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -254,13 +254,11 @@ class Transaction {
254254
// maybe multiple invoke, when out_kvs.size < limit is over.
255255
Status Scan(const std::string& start_key, const std::string& end_key, uint64_t limit, std::vector<KVPair>& kvs);
256256

257-
// If return status is ok, then call Commit
258-
// else try to precommit or rollback depends on status code
257+
// NOTE: PreCommit deprecated, use Commit() includes PreWriteAndCommit internally
259258
Status PreCommit();
260259

261-
// NOTE: Caller should first call PreCommit, when PreCommit success then call Commit
262-
// If return status is ok or rolledback, txn is end
263-
// other status, caller should retry
260+
// NOTE: If return status is ok , txn is end
261+
// other status, caller should txn->rollback
264262
Status Commit();
265263

266264
Status Rollback();

‎src/benchmark/operation.cc‎

Lines changed: 0 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -382,12 +382,6 @@ Operation::Result BaseOperation::KvTxnPut(const std::vector<sdk::KVPair>& kvs) {
382382
}
383383
}
384384

385-
result.status = txn->PreCommit();
386-
if (!result.status.IsOK()) {
387-
LOG(ERROR) << fmt::format("pre commit transaction failed, error: {}", result.status.ToString());
388-
goto end;
389-
}
390-
391385
result.status = txn->Commit();
392386
if (!result.status.IsOK()) {
393387
LOG(ERROR) << fmt::format("commit transaction failed, error: {}", result.status.ToString());
@@ -458,12 +452,6 @@ Operation::Result BaseOperation::KvTxnBatchPut(const std::vector<sdk::KVPair>& k
458452
goto end;
459453
}
460454

461-
result.status = txn->PreCommit();
462-
if (!result.status.IsOK()) {
463-
LOG(ERROR) << fmt::format("pre commit transaction failed, error: {}", result.status.ToString());
464-
goto end;
465-
}
466-
467455
result.status = txn->Commit();
468456
if (!result.status.IsOK()) {
469457
LOG(ERROR) << fmt::format("commit transaction failed, error: {}", result.status.ToString());
@@ -501,12 +489,6 @@ Operation::Result BaseOperation::KvTxnGet(const std::vector<std::string>& keys)
501489
}
502490
}
503491

504-
result.status = txn->PreCommit();
505-
if (!result.status.IsOK()) {
506-
LOG(ERROR) << fmt::format("pre commit transaction failed, error: {}", result.status.ToString());
507-
goto end;
508-
}
509-
510492
result.status = txn->Commit();
511493
if (!result.status.IsOK()) {
512494
LOG(ERROR) << fmt::format("commit transaction failed, error: {}", result.status.ToString());
@@ -544,12 +526,6 @@ Operation::Result BaseOperation::KvTxnBatchGet(const std::vector<std::vector<std
544526
}
545527
}
546528

547-
result.status = txn->PreCommit();
548-
if (!result.status.IsOK()) {
549-
LOG(ERROR) << fmt::format("pre commit transaction failed, error: {}", result.status.ToString());
550-
goto end;
551-
}
552-
553529
result.status = txn->Commit();
554530
if (!result.status.IsOK()) {
555531
LOG(ERROR) << fmt::format("commit transaction failed, error: {}", result.status.ToString());

‎src/sdk/client.cc‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -174,8 +174,7 @@ Status Client::NewRawKV(RawKV** raw_kv) {
174174
}
175175

176176
Status Client::NewTransaction(const TransactionOptions& options, Transaction** txn) {
177-
auto txn_impl =
178-
std::make_shared<TxnImpl>(*data_->stub, options,data_->stub->GetTxnManager());
177+
auto txn_impl = std::make_shared<TxnImpl>(*data_->stub, options, data_->stub->GetTxnManager());
179178
Transaction* tmp_txn = new Transaction(new Transaction::Data(*data_->stub, txn_impl));
180179
Status s = tmp_txn->Begin();
181180
if (!s.ok()) {
@@ -187,7 +186,7 @@ Status Client::NewTransaction(const TransactionOptions& options, Transaction** t
187186
delete tmp_txn;
188187
return s;
189188
}
190-
189+
191190
*txn = tmp_txn;
192191
return s;
193192
}
@@ -596,7 +595,7 @@ Status Transaction::Scan(const std::string& start_key, const std::string& end_ke
596595

597596
Status Transaction::PreCommit() { return data_->impl->PreCommit(); }
598597

599-
Status Transaction::Commit() { return data_->impl->Commit(); }
598+
Status Transaction::Commit() { return data_->impl->PreWriteAndCommit(); }
600599

601600
Status Transaction::Rollback() { return data_->impl->Rollback(); }
602601

‎src/sdk/transaction/txn_impl.cc‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -425,6 +425,17 @@ Status TxnImpl::DoScan(const std::string& start_key, const std::string& end_key,
425425
return Status::OK();
426426
}
427427

428+
Status TxnImpl::PreWriteAndCommit() {
429+
Status s = DoPreCommit();
430+
if (!s.ok()) {
431+
return s;
432+
} else if (s.ok() && is_one_pc_.load()) {
433+
return Status::OK();
434+
}
435+
s = DoCommit();
436+
return s;
437+
}
438+
428439
void TxnImpl::ScheduleHeartBeat() {
429440
stub_.GetActuator()->Schedule(
430441
[shared_this = shared_from_this(), start_ts = start_ts_.load(), primary_key = buffer_->GetPrimaryKey()] {
@@ -527,6 +538,7 @@ Status TxnImpl::PreCommit1PC() {
527538
}
528539
return status;
529540
}
541+
530542
Status TxnImpl::PreCommit2PC() {
531543
DINGO_LOG(DEBUG) << fmt::format("[sdk.txn.{}] precommit primary key, pk({}).", ID(),
532544
StringToHex(buffer_->GetPrimaryKey()));

‎src/sdk/transaction/txn_impl.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -121,8 +121,8 @@ class TxnImpl : public std::enable_shared_from_this<TxnImpl> {
121121
// maybe multiple invoke, when out_kvs.size < limit is over.
122122
Status Scan(const std::string& start_key, const std::string& end_key, uint64_t limit, std::vector<KVPair>& out_kvs);
123123

124+
Status PreWriteAndCommit();
124125
Status PreCommit();
125-
126126
Status Commit();
127127

128128
Status Rollback();

‎test/integration_test/test_txn.cc‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -70,7 +70,7 @@ TYPED_TEST(TxnTest, TxnMultiThread1PC) {
7070
std::string key = Helper::EncodeTxnKey(FLAGS_txn_key_prefix);
7171
std::string value = "0";
7272
txn->Put(key, value);
73-
status = txn->PreCommit();
73+
status = txn->Commit();
7474
EXPECT_EQ(true, status.IsOK()) << status.ToString();
7575
delete txn;
7676
}
@@ -103,10 +103,11 @@ TYPED_TEST(TxnTest, TxnMultiThread1PC) {
103103
LOG(INFO) << fmt::format("before_commit txn_id {}, value: {}, success count {}", txn->ID(), value,
104104
success_count.load());
105105
txn->Put(key, value);
106-
status = txn->PreCommit();
106+
status = txn->Commit();
107107
if (!status.IsOK()) {
108108
LOG(WARNING) << fmt::format("Failed to pre-commit transaction id {} in thread {}: {}", txn->ID(), i,
109109
status.ToString());
110+
txn->Rollback();
110111
delete txn;
111112
continue;
112113
}

‎test/unit_test/sdk/transaction/test_txn_manager.cc‎

Lines changed: 0 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -78,13 +78,10 @@ TEST_F(SDKTxnManagerTest, TransactionManagerEmpty) {
7878
EXPECT_GT(txn2->ID(), 0);
7979
EXPECT_GT(txn3->ID(), 0);
8080

81-
txn1->PreCommit();
8281
txn1->Commit();
8382

84-
txn2->PreCommit();
8583
txn2->Commit();
8684

87-
txn3->PreCommit();
8885
txn3->Commit();
8986
}
9087

@@ -97,21 +94,18 @@ TEST_F(SDKTxnManagerTest, TransactionManagerWithData1pc) {
9794
{
9895
auto txn1 = NewTransaction(options);
9996
txn1->Put("a", "a");
100-
txn1->PreCommit();
10197
txn1->Commit();
10298
}
10399

104100
{
105101
auto txn2 = NewTransaction(options);
106102
txn2->Put("c", "c");
107-
txn2->PreCommit();
108103
txn2->Commit();
109104
}
110105

111106
{
112107
auto txn3 = NewTransaction(options);
113108
txn3->PutIfAbsent("d", "d");
114-
txn3->PreCommit();
115109
txn3->Commit();
116110
}
117111
}
@@ -126,23 +120,20 @@ TEST_F(SDKTxnManagerTest, TransactionManagerWithData2pc) {
126120
auto txn1 = NewTransaction(options);
127121
txn1->Put("a", "a");
128122
txn1->Put("d", "d");
129-
txn1->PreCommit();
130123
txn1->Commit();
131124
}
132125

133126
{
134127
auto txn2 = NewTransaction(options);
135128
txn2->Put("b", "b");
136129
txn2->Put("e", "e");
137-
txn2->PreCommit();
138130
txn2->Commit();
139131
}
140132

141133
{
142134
auto txn3 = NewTransaction(options);
143135
txn3->PutIfAbsent("a", "newa");
144136
txn3->PutIfAbsent("d", "newd");
145-
txn3->PreCommit();
146137
txn3->Commit();
147138
}
148139
}

0 commit comments

Comments
 (0)