Loading...
Searching...
No Matches
transaction_executor_test.cpp
Go to the documentation of this file.
1/*
2 * Licensed to the Apache Software Foundation (ASF) under one
3 * or more contributor license agreements. See the NOTICE file
4 * distributed with this work for additional information
5 * regarding copyright ownership. The ASF licenses this file
6 * to you under the Apache License, Version 2.0 (the
7 * "License"); you may not use this file except in compliance
8 * with the License. You may obtain a copy of the License at
9 *
10 * http://www.apache.org/licenses/LICENSE-2.0
11 *
12 * Unless required by applicable law or agreed to in writing,
13 * software distributed under the License is distributed on an
14 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15 * KIND, either express or implied. See the License for the
16 * specific language governing permissions and limitations
17 * under the License.
18 */
19
20#include "platform/consensus/execution/transaction_executor.h"
21
22#include <gmock/gmock.h>
23#include <gtest/gtest.h>
24
25#include <future>
26
27#include "common/test/test_macros.h"
28#include "executor/common/mock_transaction_manager.h"
29
30namespace resdb {
31namespace {
32
33using ::resdb::testing::EqualsProto;
34using ::testing::Invoke;
35
36ResDBConfig GetResDBConfig() {
37 return ResDBConfig({ReplicaInfo()}, ReplicaInfo());
38}
39
40TEST(TransactionExecutorTest, ExecuteOne) {
41 std::promise<bool> done;
42 std::future<bool> done_future = done.get_future();
43
44 ResDBConfig config = GetResDBConfig();
45 Request request;
46 request.set_seq(1);
47
48 BatchUserRequest batch_request;
49 batch_request.add_user_requests()->mutable_request()->set_data("execute_1");
50 batch_request.SerializeToString(request.mutable_data());
51
52 SystemInfo system_info(config);
53 auto mock_executor = std::make_unique<MockTransactionExecutorDataImpl>();
54
55 EXPECT_CALL(*mock_executor, ExecuteData)
56 .WillOnce(Invoke([&](const std::string& input) {
57 EXPECT_EQ(input, "execute_1");
58 done.set_value(true);
59 return nullptr;
60 }));
61 TransactionExecutor executor(
62 config,
63 [&](std::unique_ptr<Request>, std::unique_ptr<BatchUserResponse> resp) {},
64 &system_info, std::move(mock_executor));
65
66 EXPECT_EQ(executor.Commit(std::make_unique<Request>(request)), 0);
67
68 done_future.get();
69}
70
71TEST(TransactionExecutorTest, MaxPendingExecuteSeq) {
72 Stats::GetGlobalStats()->Stop();
73 ResDBConfig config = GetResDBConfig();
74
75 Request request;
76 request.set_seq(1);
77
78 BatchUserRequest batch_request;
79 batch_request.add_user_requests()->mutable_request()->set_data("execute_1");
80 batch_request.SerializeToString(request.mutable_data());
81
82 std::mutex mutex;
83 std::condition_variable cv;
84 std::atomic<int> call_num = 0;
85 SystemInfo system_info(config);
86 TransactionExecutor executor(
87 config,
88 [&](std::unique_ptr<Request>, std::unique_ptr<BatchUserResponse> resp) {},
89 &system_info, std::make_unique<MockTransactionManager>());
90
91 executor.SetSeqUpdateNotifyFunc([&](uint64_t seq) {
92 std::unique_lock<std::mutex> lk(mutex);
93 cv.notify_one();
94 call_num++;
95 });
96
97 EXPECT_EQ(executor.Commit(std::make_unique<Request>(request)), 0);
98 {
99 std::unique_lock<std::mutex> lk(mutex);
100 cv.wait(lk, [&]() { return call_num >= 1; });
101 }
102 EXPECT_LE(executor.GetMaxPendingExecutedSeq(), 1);
103
104 request.set_seq(3);
105 EXPECT_EQ(executor.Commit(std::make_unique<Request>(request)), 0);
106 EXPECT_EQ(executor.GetMaxPendingExecutedSeq(), 1);
107
108 request.set_seq(2);
109 EXPECT_EQ(executor.Commit(std::make_unique<Request>(request)), 0);
110 {
111 std::unique_lock<std::mutex> lk(mutex);
112 cv.wait(lk, [&]() { return call_num == 3; });
113 }
114 EXPECT_EQ(executor.GetMaxPendingExecutedSeq(), 3);
115}
116
117TEST(TransactionExecutorTest, MaxPendingExecuteSeqOnOutOfOrder) {
118 Stats::GetGlobalStats()->Stop();
119 ResDBConfig config = GetResDBConfig();
120
121 Request request;
122 request.set_seq(1);
123
124 BatchUserRequest batch_request;
125 batch_request.add_user_requests()->mutable_request()->set_data("execute_1");
126 batch_request.SerializeToString(request.mutable_data());
127
128 std::mutex mutex;
129 std::condition_variable cv;
130 std::atomic<int> call_num = 0;
131 SystemInfo system_info(config);
132 TransactionExecutor executor(
133 config,
134 [&](std::unique_ptr<Request>, std::unique_ptr<BatchUserResponse> resp) {},
135 &system_info, std::make_unique<MockTransactionManager>(true));
136
137 executor.SetSeqUpdateNotifyFunc([&](uint64_t seq) {
138 std::unique_lock<std::mutex> lk(mutex);
139 cv.notify_one();
140 call_num++;
141 });
142
143 EXPECT_EQ(executor.Commit(std::make_unique<Request>(request)), 0);
144 {
145 std::unique_lock<std::mutex> lk(mutex);
146 cv.wait(lk, [&]() { return call_num >= 1; });
147 }
148 EXPECT_LE(executor.GetMaxPendingExecutedSeq(), 1);
149
150 request.set_seq(3);
151 EXPECT_EQ(executor.Commit(std::make_unique<Request>(request)), 0);
152 EXPECT_EQ(executor.GetMaxPendingExecutedSeq(), 1);
153
154 request.set_seq(2);
155 EXPECT_EQ(executor.Commit(std::make_unique<Request>(request)), 0);
156 {
157 std::unique_lock<std::mutex> lk(mutex);
158 cv.wait(lk, [&]() { return call_num == 3; });
159 }
160 EXPECT_EQ(executor.GetMaxPendingExecutedSeq(), 3);
161}
162
163TEST(TransactionExecutorTest, ExecuteWaitInOrder) {
164 std::promise<bool> done;
165 std::future<bool> done_future = done.get_future();
166
167 ResDBConfig config = GetResDBConfig();
168
169 SystemInfo system_info(config);
170 auto mock_executor = std::make_unique<MockTransactionExecutorDataImpl>();
171
172 EXPECT_CALL(*mock_executor, ExecuteData)
173 .WillOnce(Invoke([&](const std::string& input) {
174 EXPECT_EQ(input, "execute_1");
175 return nullptr;
176 }))
177 .WillOnce(Invoke([&](const std::string& input) {
178 EXPECT_EQ(input, "execute_2");
179 done.set_value(true);
180 return nullptr;
181 }));
182
183 TransactionExecutor executor(
184 config,
185 [&](std::unique_ptr<Request>, std::unique_ptr<BatchUserResponse> resp) {},
186 &system_info, std::move(mock_executor));
187
188 Request request;
189 request.set_seq(2);
190 BatchUserRequest batch_request;
191 batch_request.add_user_requests()->mutable_request()->set_data("execute_2");
192 batch_request.SerializeToString(request.mutable_data());
193 EXPECT_EQ(executor.Commit(std::make_unique<Request>(request)), 0);
194
195 request.set_seq(1);
196 batch_request.clear_user_requests();
197 batch_request.add_user_requests()->mutable_request()->set_data("execute_1");
198 batch_request.SerializeToString(request.mutable_data());
199 EXPECT_EQ(executor.Commit(std::make_unique<Request>(request)), 0);
200
201 done_future.get();
202}
203
204TEST(TransactionExecutorTest, ExecuteWaitOutOfOrder) {
205 std::promise<bool> done;
206 std::future<bool> done_future = done.get_future();
207
208 ResDBConfig config = GetResDBConfig();
209
210 SystemInfo system_info(config);
211 auto mock_executor = std::make_unique<MockTransactionManager>(true);
212
213 EXPECT_CALL(*mock_executor, ExecuteBatch)
214 .WillOnce(Invoke([&](const BatchUserRequest& request) {
215 EXPECT_EQ(request.seq(), 4);
216 return nullptr;
217 }))
218 .WillOnce(Invoke([&](const BatchUserRequest& request) {
219 EXPECT_EQ(request.seq(), 1);
220 done.set_value(true);
221 return nullptr;
222 }));
223
224 TransactionExecutor executor(
225 config,
226 [&](std::unique_ptr<Request>, std::unique_ptr<BatchUserResponse> resp) {},
227 &system_info, std::move(mock_executor));
228
229 Request request;
230 request.set_seq(4);
231 BatchUserRequest batch_request;
232 batch_request.add_user_requests()->mutable_request()->set_data("execute_2");
233 batch_request.SerializeToString(request.mutable_data());
234 EXPECT_EQ(executor.Commit(std::make_unique<Request>(request)), 0);
235
236 request.set_seq(1);
237 batch_request.clear_user_requests();
238 batch_request.add_user_requests()->mutable_request()->set_data("execute_1");
239 batch_request.SerializeToString(request.mutable_data());
240 EXPECT_EQ(executor.Commit(std::make_unique<Request>(request)), 0);
241
242 done_future.get();
243}
244
245TEST(TransactionExecutorTest, CallBack) {
246 std::promise<bool> done;
247 std::future<bool> done_future = done.get_future();
248
249 ResDBConfig config = GetResDBConfig();
250
251 Request request;
252 request.set_seq(1);
253 BatchUserRequest batch_request;
254 batch_request.add_user_requests()->mutable_request()->set_data("execute_1");
255 batch_request.SerializeToString(request.mutable_data());
256
257 SystemInfo system_info(config);
258 auto mock_executor = std::make_unique<MockTransactionExecutorDataImpl>();
259
260 EXPECT_CALL(*mock_executor, ExecuteData)
261 .WillOnce(Invoke([&](const std::string& input) {
262 EXPECT_EQ(input, "execute_1");
263 return nullptr;
264 }));
265
266 TransactionExecutor executor(
267 config,
268 [&](std::unique_ptr<Request> call_request,
269 std::unique_ptr<BatchUserResponse> resp) {
270 EXPECT_EQ(call_request->seq(), 1);
271 done.set_value(true);
272 },
273 &system_info, std::move(mock_executor));
274
275 EXPECT_EQ(executor.Commit(std::make_unique<Request>(request)), 0);
276
277 done_future.get();
278}
279
280} // namespace
281
282} // namespace resdb
resdb::Stats::GetGlobalStats
static Stats * GetGlobalStats(int sleep_seconds=5)
Definition stats.cpp:29
resdb::Stats::Stop
void Stop()
Definition stats.cpp:67
mock_transaction_manager.h
resdb::TEST
TEST(NetworkUtilsTest, GetDNSName)
Definition network_utils_test.cpp:30
test_macros.h
transaction_executor.h