oceanbase

test_ob_concurrent_seq_queue.cpp
198 строк · 5.7 Кб
Перенос по словам
1
/**
2
 * Copyright (c) 2021 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

13
#include <gtest/gtest.h>
14
#include "ob_concurrent_seq_queue.h"
15

16
#include "share/ob_define.h"
17
#include "lib/time/ob_time_utility.h"
18
#include "lib/atomic/ob_atomic.h"
19

20
namespace oceanbase
21
{
22
namespace common
23
{
24
class TestConSeqQueue : public ::testing::Test
25
{
26
public:
27
  static const int64_t RUN_TIME = 1L * 60L * 60L * 1000L * 1000L;
28
  static const int64_t THREAD_NUM = 20;
29
  static const int64_t STAT_INTERVAL = 5 * 1000 * 1000;
30
public:
31
  TestConSeqQueue() {}
32
  ~TestConSeqQueue() {}
33

34
  virtual void SetUp()
35
  {
36
    ASSERT_EQ(0, queue_.init(1024, ObMemAttr(OB_SERVER_TENANT_ID, ObNewModIds::TEST)));
37
    produce_seq_ = 0;
38
    consume_seq_ = 0;
39
    consume_thread_counter_ = 0;
40
    consume_task_count_ = 0;
41
    last_stat_time_ = 0;
42
    last_consume_task_count_ = 0;
43
    stop_flag_ = false;
44
  }
45
  virtual void TearDown()
46
  {
47
    queue_.destroy();
48
  }
49
  static void *produce_thread_func(void *args);
50
  static void *consume_thread_func(void *args);
51
  void run_produce();
52
  void run_consume();
53

54
public:
55
  pthread_t produce_threads_[THREAD_NUM];
56
  pthread_t consume_threads_[THREAD_NUM];
57
  int64_t consume_thread_counter_;
58
  ObConcurrentSeqQueue queue_;
59
  int64_t produce_seq_ CACHE_ALIGNED;
60
  int64_t consume_seq_ CACHE_ALIGNED;
61
  int64_t consume_task_count_ CACHE_ALIGNED;
62
  int64_t last_consume_task_count_ CACHE_ALIGNED;
63
  int64_t last_stat_time_ CACHE_ALIGNED;
64

65
  volatile bool stop_flag_ CACHE_ALIGNED;
66
};
67

68
TEST_F(TestConSeqQueue, basic)
69
{
70
  ObConcurrentSeqQueue queue;
71
  void *data = 0;
72

73
  EXPECT_EQ(0, queue.init(1024, ObMemAttr(OB_SERVER_TENANT_ID, ObNewModIds::TEST)));
74

75
  EXPECT_EQ(0, queue.push((void*)0, 0, 0));
76
  EXPECT_EQ(0, queue.push((void*)1, 1, 0));
77
  EXPECT_EQ(0, queue.push((void*)2, 2, 0));
78

79
  EXPECT_EQ(0, queue.pop(data, 0, 0));
80
  EXPECT_EQ(0, (int64_t)data);
81
  EXPECT_EQ(0, queue.pop(data, 1, 0));
82
  EXPECT_EQ(1, (int64_t)data);
83
  EXPECT_EQ(0, queue.pop(data, 2, 0));
84
  EXPECT_EQ(2, (int64_t)data);
85

86
  // Failed to push and pop elements with the same serial number again
87
  EXPECT_NE(0, queue.push((void*)0, 0, 0));
88
  EXPECT_NE(0, queue.push((void*)1, 1, 0));
89
  EXPECT_NE(0, queue.push((void*)2, 2, 0));
90
  EXPECT_NE(0, queue.pop(data, 0, 0));
91
  EXPECT_NE(0, queue.pop(data, 1, 0));
92
  EXPECT_NE(0, queue.pop(data, 2, 0));
93
}
94

95
void *TestConSeqQueue::produce_thread_func(void *args)
96
{
97
  if (NULL != args) {
98
    ((TestConSeqQueue *)args)->run_produce();
99
  }
100

101
  return NULL;
102
}
103

104
void TestConSeqQueue::run_produce()
105
{
106
  int ret = OB_SUCCESS;
107
  int64_t batch_count = 1000;
108

109
  while (OB_SUCCESS == ret && ! stop_flag_) {
110
    for (int64_t index = 0; OB_SUCCESS == ret && index < batch_count; index++) {
111
      int64_t seq = ATOMIC_FAA(&produce_seq_, 1);
112
      while (! stop_flag_ && OB_TIMEOUT == (ret = queue_.push((void*)seq, seq, 1 * 1000 * 1000)));
113
      if (! stop_flag_) {
114
        EXPECT_EQ(OB_SUCCESS, ret);
115
      }
116
    }
117
  }
118
}
119

120
void *TestConSeqQueue::consume_thread_func(void *args)
121
{
122
  if (NULL != args) {
123
    ((TestConSeqQueue *)args)->run_consume();
124
  }
125

126
  return NULL;
127
}
128

129
void TestConSeqQueue::run_consume()
130
{
131
  int ret = OB_SUCCESS;
132
  int64_t batch_count = 1000;
133
  int64_t end_time = ObTimeUtility::current_time();
134

135
  int64_t thread_index = ATOMIC_FAA(&consume_thread_counter_, 0);
136

137
  while (OB_SUCCESS == ret && !stop_flag_) {
138
    for (int64_t index = 0; OB_SUCCESS == ret && index < batch_count; index++) {
139
      int64_t seq = ATOMIC_FAA(&consume_seq_, 1);
140
      void *data = NULL;
141
      while (! stop_flag_ && OB_TIMEOUT == (ret = queue_.pop(data, seq, 1 * 1000 * 1000)));
142
      if (! stop_flag_) {
143
        EXPECT_EQ(OB_SUCCESS, ret);
144
        EXPECT_EQ(seq, (int64_t)data);
145
        ATOMIC_INC(&consume_task_count_);
146
      }
147
    }
148

149
    int64_t cur_time = ObTimeUtility::current_time();
150
    if (OB_UNLIKELY(0 == thread_index) && cur_time - last_stat_time_ > STAT_INTERVAL) {
151
      int64_t task_count = ATOMIC_LOAD(&consume_task_count_);
152
      int64_t consume_seq = ATOMIC_LOAD(&consume_seq_);
153
      int64_t produce_seq = ATOMIC_LOAD(&produce_seq_);
154
      if (0 != last_stat_time_) {
155
        int64_t delta_task_count = task_count - last_consume_task_count_;
156
        int64_t delta_time_sec = (cur_time - last_stat_time_)/1000000;
157
        LIB_LOG(INFO, "STAT", "POP_TPS", delta_task_count/delta_time_sec, K(delta_task_count),
158
            K(consume_seq), K(produce_seq), K(INT32_MAX));
159
      }
160

161
      last_stat_time_ = cur_time;
162
      last_consume_task_count_ = task_count;
163
    }
164

165
    if (end_time - cur_time <= 0) {
166
      stop_flag_ = true;
167
    }
168
  }
169
}
170

171
TEST_F(TestConSeqQueue, thread)
172
{
173
  for (int64_t index = 0; index < THREAD_NUM; index++) {
174
    ASSERT_EQ(0, pthread_create(produce_threads_ + index, NULL, produce_thread_func, this));
175
  }
176
  for (int64_t index = 0; index < THREAD_NUM; index++) {
177
    ASSERT_EQ(0, pthread_create(consume_threads_ + index, NULL, consume_thread_func, this));
178
  }
179
  for (int64_t index = 0; index < THREAD_NUM; index++) {
180
    pthread_join(produce_threads_[index], NULL);
181
    produce_threads_[index] = 0;
182
  }
183
  for (int64_t index = 0; index < THREAD_NUM; index++) {
184
    pthread_join(consume_threads_[index], NULL);
185
    consume_threads_[index] = 0;
186
  }
187
}
188

189
}
190
}
191

192
int main(int argc, char **argv)
193
{
194
  oceanbase::common::ObLogger::get_logger().set_log_level("INFO");
195
  OB_LOGGER.set_log_level("INFO");
196
  testing::InitGoogleTest(&argc, argv);
197
  return RUN_ALL_TESTS();
198
}
199
oceanbase

Использование cookies
