| 135 | butil::atomic<int> g_thread_offset(0); |
| 136 | |
| 137 | static void* replay_thread(void* arg) { |
| 138 | ChannelGroup* chan_group = static_cast<ChannelGroup*>(arg); |
| 139 | const int thread_offset = g_thread_offset.fetch_add(1, butil::memory_order_relaxed); |
| 140 | double req_rate = FLAGS_qps / (double)FLAGS_thread_num; |
| 141 | brpc::SerializedRequest req; |
| 142 | brpc::NsheadMessage nshead_req; |
| 143 | int64_t last_expected_time = butil::monotonic_time_ns(); |
| 144 | const int64_t interval = (int64_t) (1000000000L / req_rate); |
| 145 | // the max tolerant delay between end_time and expected_time. 10ms or 10 intervals |
| 146 | int64_t max_tolerant_delay = std::max((int64_t) 10000000L, 10 * interval); |
| 147 | for (int i = 0; !brpc::IsAskedToQuit() && i < FLAGS_times; ++i) { |
| 148 | brpc::SampleIterator it(FLAGS_dir); |
| 149 | int j = 0; |
| 150 | for (brpc::SampledRequest* sample = it.Next(); |
| 151 | !brpc::IsAskedToQuit() && sample != NULL; sample = it.Next(), ++j) { |
| 152 | std::unique_ptr<brpc::SampledRequest> sample_guard(sample); |
| 153 | if ((j % FLAGS_thread_num) != thread_offset) { |
| 154 | continue; |
| 155 | } |
| 156 | brpc::Channel* chan = |
| 157 | chan_group->channel(sample->meta.protocol_type()); |
| 158 | if (chan == NULL) { |
| 159 | LOG(ERROR) << "No channel on protocol=" |
| 160 | << sample->meta.protocol_type(); |
| 161 | continue; |
| 162 | } |
| 163 | |
| 164 | brpc::Controller* cntl = new brpc::Controller; |
| 165 | req.Clear(); |
| 166 | |
| 167 | google::protobuf::Message* req_ptr = &req; |
| 168 | cntl->reset_sampled_request(sample_guard.release()); |
| 169 | if (sample->meta.protocol_type() == brpc::PROTOCOL_HTTP) { |
| 170 | brpc::HttpMessage http_message; |
| 171 | http_message.ParseFromIOBuf(sample->request); |
| 172 | cntl->http_request().Swap(http_message.header()); |
| 173 | if (!FLAGS_http_host.empty()) { |
| 174 | // reset Host in header |
| 175 | cntl->http_request().SetHeader("Host", FLAGS_http_host); |
| 176 | } |
| 177 | cntl->request_attachment() = http_message.body().movable(); |
| 178 | req_ptr = NULL; |
| 179 | } else if (sample->meta.protocol_type() == brpc::PROTOCOL_NSHEAD) { |
| 180 | nshead_req.Clear(); |
| 181 | memcpy(&nshead_req.head, sample->meta.nshead().c_str(), sample->meta.nshead().length()); |
| 182 | nshead_req.body = sample->request; |
| 183 | req_ptr = &nshead_req; |
| 184 | } else if (sample->meta.attachment_size() > 0) { |
| 185 | sample->request.cutn( |
| 186 | &req.serialized_data(), |
| 187 | sample->request.size() - sample->meta.attachment_size()); |
| 188 | cntl->request_attachment() = sample->request.movable(); |
| 189 | } else { |
| 190 | req.serialized_data() = sample->request.movable(); |
| 191 | } |
| 192 | g_sent_count << 1; |
| 193 | const int64_t start_time = butil::cpuwide_time_us(); |
| 194 | if (FLAGS_qps <= 0) { |
nothing calls this directly
no test coverage detected