oceanbase

1
/**
2
 * Copyright (c) 2023 OceanBase
3
 * OceanBase CE is licensed under Mulan PubL v2.
4
 * You can use this software according to the terms and conditions of the Mulan PubL v2.
5
 * You may obtain a copy of Mulan PubL v2 at:
6
 *          http://license.coscl.org.cn/MulanPubL-2.0
7
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
8
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
9
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
10
 * See the Mulan PubL v2 for more details.
11
 */
12
#include <gtest/gtest.h>
13
#include <gmock/gmock.h>
14
#define private public
15
#include "observer/table/ob_table_connection_mgr.h"
16
#include "lib/time/ob_time_utility.h"
17
#include "rpc/ob_request.h"
18
#include "rpc/frame/ob_net_easy.cpp"
19
#include "share/ob_thread_pool.h"
20
#include "lib/container/ob_array.h"
21
using namespace oceanbase::table;
22
using namespace oceanbase::common;
23
using namespace oceanbase::rpc;
24
using namespace oceanbase::share;
25

26
namespace oceanbase {
27
namespace observer {
28
class TestTableConnection : public::testing::Test
29
{
30
public:
31
  TestTableConnection() {}
32
  virtual ~TestTableConnection() {}
33
  virtual void SetUp() {}
34
  virtual void TearDown() {}
35
};
36

37
#define CONS_FAKE_REQUEST(ob_req, ob_addr)\
38
  ObRequest ob_req(ObRequest::OB_RPC);\
39
  easy_request_t ez_req;\
40
  easy_message_session_t ms;\
41
  easy_connection_t c;\
42
  easy_addr_t ez_addr = __to_ez_addr(ob_addr);\
43
  c.addr = ez_addr;\
44
  ms.c = &c;\
45
  ez_req.ms = &ms;\
46
  ob_req.ez_req_ = &ez_req;
47

48
#define CONS_FAKE_EZ_CONN(ez_conn, ob_addr)\
49
  easy_connection_t ez_conn;\
50
  easy_addr_t ez_conn_addr = __to_ez_addr(ob_addr);\
51
  ez_conn.addr = ez_conn_addr;\
52
53
// mock a connection which used by multiply thread
54
class MockObTableRequestSender  : public ObThreadPool
55
{
56
public:
57
  MockObTableRequestSender() {}
58
  MockObTableRequestSender(const ObAddr &client_addr, int64_t thread_cnt, int64_t tenant_id,
59
                                    int64_t database_id, int64_t user_id, int64_t request_times)
60
  : client_addr_(client_addr),
61
    thread_cnt_(thread_cnt),
62
    tenant_id_(tenant_id),
63
    database_id_(database_id),
64
    user_id_(user_id),
65
    request_times_(request_times)
66
  {
67
    set_thread_count(thread_cnt_);
68
  }
69

70
  virtual ~MockObTableRequestSender() {};
71
  virtual void run1();
72
  TO_STRING_KV(K_(client_addr), K_(thread_cnt), K_(tenant_id), K_(database_id), K_(user_id), K_(request_times));
73
private:
74
  ObAddr client_addr_;
75
  int64_t thread_cnt_;
76
  int64_t tenant_id_;
77
  int64_t database_id_;
78
  int64_t user_id_;
79
  int64_t request_times_;
80
};
81

82
void MockObTableRequestSender::run1()
83
{
84
  CONS_FAKE_REQUEST(req, client_addr_);
85
  for(int64_t i = 0; i < request_times_; i++) {
86
    ASSERT_EQ(OB_SUCCESS, ObTableConnectionMgr::get_instance().update_table_connection(&req, tenant_id_, database_id_, user_id_));
87
  }
88
}
89

90
// mock the ObSrvRpcHandler::on_close
91
class MockObTableConnCloser  : public ObThreadPool
92
{
93
public:
94
  explicit MockObTableConnCloser(const ObAddr &client_addr)
95
  : client_addr_(client_addr)
96
  {
97
    set_thread_count(1);
98
  }
99
  virtual ~MockObTableConnCloser() {};
100
  virtual void run1();
101
  TO_STRING_KV(K_(client_addr));
102
private:
103
  const ObAddr client_addr_;
104
};
105

106
void MockObTableConnCloser::run1()
107
{
108
  CONS_FAKE_EZ_CONN(ez_conn, client_addr_)
109
  ObTableConnectionMgr::get_instance().on_conn_close(&ez_conn);
110
}
111

112
// test correctness of table connection init
113
TEST_F(TestTableConnection, test_table_connection)
114
{
115
  // fake connection
116
  ObTableConnection conn;
117
  ObAddr addr(2130706433, 6001);
118
  int64_t tenant_id = 1001;
119
  int64_t database_id = 99999;
120
  int64_t user_id = 99999;
121
  ASSERT_EQ(OB_SUCCESS, conn.init(addr, tenant_id, database_id, user_id));
122
  ASSERT_EQ(addr, conn.get_addr());
123
  ASSERT_EQ(tenant_id, conn.get_tenant_id());
124
  ASSERT_EQ(database_id, conn.get_database_id());
125
  ASSERT_EQ(user_id, conn.get_user_id());
126

127
  int64_t first_active_time = conn.get_first_active_time();
128
  ASSERT_EQ(first_active_time, conn.get_last_active_time());
129
  int64_t new_active_time = ObTimeUtility::current_time();
130
  conn.update_last_active_time(new_active_time);
131
  ASSERT_EQ(first_active_time, conn.get_first_active_time());
132
  ASSERT_EQ(new_active_time, conn.get_last_active_time());
133

134
  ObAddr invalid_addr;
135
  ASSERT_EQ(OB_INVALID_ARGUMENT, conn.init(invalid_addr, tenant_id, database_id, user_id));
136
}
137

138
// test basic interface correctness in connection mgr
139
TEST_F(TestTableConnection, test_connection_mgr)
140
{
141
  ObAddr addr(2130706433, 6001);
142
  int64_t tenant_id = 1001;
143
  int64_t database_id = 99999;
144
  int64_t user_id = 99999;
145
  CONS_FAKE_REQUEST(req, addr);
146
  ASSERT_EQ(OB_SUCCESS, ObTableConnectionMgr::get_instance().update_table_connection(&req, tenant_id, database_id, user_id));
147
  ASSERT_EQ(1, ObTableConnectionMgr::get_instance().connection_map_.size());
148
  CONS_FAKE_EZ_CONN(ez_conn, addr);
149
  ObTableConnectionMgr::get_instance().on_conn_close(&ez_conn);
150
  ASSERT_EQ(0, ObTableConnectionMgr::get_instance().connection_map_.size());
151
}
152

153
// send mutliply request using single connection concurrently
154
// the size of connection map should be equal to the size of req_sender
155
// after closer execute, the size of connection map should be equal to 0
156
TEST_F(TestTableConnection, test_multi_request)
157
{
158
  const int64_t start = ObTimeUtility::current_time();
159
  const int64_t thread_cnt = 100;
160
  int64_t tenant_id = 1001;
161
  int64_t database_id = 99999;
162
  int64_t user_id = 99999;
163
  int64_t request_times  = 100;
164
  const ObAddr addr(2130706433, 1);
165
  MockObTableRequestSender req_sender(addr, thread_cnt, tenant_id, database_id, user_id, request_times);
166
  ASSERT_EQ(OB_SUCCESS, req_sender.start());
167
  req_sender.wait();
168
  ASSERT_EQ(1, ObTableConnectionMgr::get_instance().connection_map_.size());
169
  MockObTableConnCloser closer(addr);
170
  ASSERT_EQ(OB_SUCCESS, closer.start());
171
  closer.wait();
172
  ASSERT_EQ(0, ObTableConnectionMgr::get_instance().connection_map_.size());
173
  const int64_t duration = ObTimeUtility::current_time() - start;
174
  printf("time elapsed: %ldms\n", duration/1000);
175
}
176

177
// send mutliply request using different connection concurrently
178
// the size of connection map should be equal to the size of req_sender
179
// after closer execute, the size of connection map should be equal to 0
180
TEST_F(TestTableConnection, test_multi_connection)
181
{
182
  const int64_t start = ObTimeUtility::current_time();
183
  ObArenaAllocator alloc;
184
  const int64_t thread_cnt = 8;
185
  int64_t tenant_id = 1001;
186
  int64_t database_id = 99999;
187
  int64_t user_id = 99999;
188
  int64_t request_times  = 100;
189
  int64_t conn_cnt = 200;
190
  const int32_t ip = 2130706433;
191
  ObArray<MockObTableRequestSender *> senders;
192
  ObArray<MockObTableConnCloser *> closers;
193
  for (int64_t i = 1; i <= conn_cnt; i++) {
194
    MockObTableRequestSender *sender = OB_NEWx(MockObTableRequestSender, (&alloc), ObAddr(ip, i), thread_cnt,
195
                                               tenant_id, database_id, user_id, request_times);
196
    ASSERT_NE(nullptr, sender);
197
    ASSERT_EQ(OB_SUCCESS, senders.push_back(sender));
198

199
    MockObTableConnCloser *closer = OB_NEWx(MockObTableConnCloser, (&alloc), ObAddr(ip, i));
200
    ASSERT_NE(nullptr, closer);
201
    ASSERT_EQ(OB_SUCCESS, closers.push_back(closer));
202
  }
203
  int64_t curr_mem = ObMallocAllocator::get_instance()->get_tenant_hold(OB_SERVER_TENANT_ID);
204
  printf("current server tenant mem before sending request: %ld\n", curr_mem);
205

206
  const int64_t req_start = ObTimeUtility::current_time();
207
  for (int i = 0; i < conn_cnt; i++) {
208
    ASSERT_EQ(OB_SUCCESS, senders.at(i)->start());
209
  }
210
  for (int i = 0; i < conn_cnt; i++) {
211
    senders.at(i)->wait();
212
  }
213
  ASSERT_EQ(conn_cnt, ObTableConnectionMgr::get_instance().connection_map_.size());
214

215
  curr_mem = ObMallocAllocator::get_instance()->get_tenant_hold(OB_SERVER_TENANT_ID);
216
  printf("current server tenant mem after sending request: %ld\n", curr_mem);
217

218
  for (int i = 0; i < conn_cnt; i++) {
219
    ASSERT_EQ(OB_SUCCESS, closers.at(i)->start());
220
  }
221
  for (int i = 0; i < conn_cnt; i++) {
222
    closers.at(i)->wait();
223
  }
224
  ASSERT_EQ(0, ObTableConnectionMgr::get_instance().connection_map_.size());
225

226
  const int64_t end = ObTimeUtility::current_time();
227
  printf("request time elapsed: %ldms\n", (end - req_start)/1000);
228
  printf("all time elapsed: %ldms\n", (end - start)/1000);
229
}
230

231
} // namespace observer
232
} // namespace oceanbase
233

234
int main(int argc, char** argv)
235
{
236
  oceanbase::common::ObLogger::get_logger().set_log_level("INFO");
237
  OB_LOGGER.set_log_level("INFO");
238
  OB_LOGGER.set_file_name("test_table_connection.log", true);
239
  ::testing::InitGoogleTest(&argc, argv);
240
  return RUN_ALL_TESTS();
241
}
242