forked from etcd-cpp-apiv3/etcd-cpp-apiv3
-
Notifications
You must be signed in to change notification settings - Fork 0
/
TransactionTest.cpp
130 lines (107 loc) · 3.26 KB
/
TransactionTest.cpp
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
#define CATCH_CONFIG_MAIN
#include <catch.hpp>
#include <chrono>
#include <iostream>
#include <thread>
#include "etcd/Client.hpp"
#include "etcd/v3/Transaction.hpp"
static const std::string etcd_url =
etcdv3::detail::resolve_etcd_endpoints("http://127.0.0.1:2379");
TEST_CASE("setup") {
etcd::Client etcd(etcd_url);
etcd.rmdir("/test", true).wait();
}
TEST_CASE("add a new key") {
etcd::Client etcd(etcd_url);
etcd.rmdir("/test", true).wait();
// do some put using txn
{
etcdv3::Transaction txn;
txn.setup_put("/test/x1", "1");
txn.setup_put("/test/x2", "2");
txn.setup_put("/test/x3", "3");
etcd::Response resp = etcd.txn(txn).get();
REQUIRE(0 == resp.error_code());
}
// check the value
{
etcd::Response resp = etcd.get("/test/x1").get();
REQUIRE(0 == resp.error_code());
CHECK(resp.value().as_string() == "1");
resp = etcd.get("/test/x2").get();
REQUIRE(0 == resp.error_code());
CHECK(resp.value().as_string() == "2");
resp = etcd.get("/test/x3").get();
REQUIRE(0 == resp.error_code());
CHECK(resp.value().as_string() == "3");
}
// do some put and delete using txn
{
etcdv3::Transaction txn;
// setup the conditions
txn.add_compare_value("/test/x1", "1");
txn.add_compare_value("/test/x2", "2");
txn.setup_put("/test/x1", "111");
txn.setup_delete("/test/x2");
txn.setup_delete("/test/x3");
txn.setup_put("/test/x4", "4");
etcd::Response resp = etcd.txn(txn).get();
REQUIRE(0 == resp.error_code());
}
// check the value
{
etcd::Response resp = etcd.get("/test/x1").get();
REQUIRE(0 == resp.error_code());
CHECK(resp.value().as_string() == "111");
resp = etcd.get("/test/x2").get();
REQUIRE(etcdv3::ERROR_KEY_NOT_FOUND == resp.error_code());
resp = etcd.get("/test/x3").get();
REQUIRE(etcdv3::ERROR_KEY_NOT_FOUND == resp.error_code());
resp = etcd.get("/test/x4").get();
REQUIRE(0 == resp.error_code());
CHECK(resp.value().as_string() == "4");
}
}
TEST_CASE("fetch & add") {
etcd::Client etcd(etcd_url);
etcd.rmdir("/test", true).wait();
etcd.set("/test/key", "0").wait();
auto fetch_and_add = [](etcd::Client& client,
std::string const& key) -> void {
auto value = stoi(client.get(key).get().value().as_string());
while (true) {
auto txn = etcdv3::Transaction();
txn.setup_compare_and_swap(key, std::to_string(value),
std::to_string(value + 1));
etcd::Response resp = client.txn(txn).get();
if (resp.is_ok()) {
break;
}
value = stoi(resp.value().as_string());
}
};
// run 1000 times
const size_t rounds = 100;
std::atomic_size_t counter(0);
std::vector<std::thread> threads;
for (size_t i = 0; i < 10; ++i) {
threads.emplace_back([&]() {
while (counter.fetch_add(1) < rounds) {
fetch_and_add(etcd, "/test/key");
}
});
}
for (auto& thr : threads) {
thr.join();
}
// check the value
{
etcd::Response resp = etcd.get("/test/key").get();
REQUIRE(0 == resp.error_code());
CHECK(resp.value().as_string() == std::to_string(rounds));
}
}
TEST_CASE("cleanup") {
etcd::Client etcd(etcd_url);
REQUIRE(0 == etcd.rmdir("/test", true).get().error_code());
}