| 284 | } |
| 285 | |
| 286 | void KeepSyncThread::SyncCursorID() { |
| 287 | QLVerb("KeepSyncThread::SyncCursorID"); |
| 288 | |
| 289 | comm::RetCode ret; |
| 290 | |
| 291 | SyncCtrl *sync{impl_->store->GetSyncCtrl()}; |
| 292 | phxpaxos::Node *node{impl_->store->GetNode()}; |
| 293 | |
| 294 | const int topic_id{impl_->store->GetTopicID()}; |
| 295 | |
| 296 | auto opt(impl_->store->GetStoreOption()); |
| 297 | |
| 298 | const int ngroup{opt->ngroup}; |
| 299 | const int nqueue{opt->nqueue}; |
| 300 | const int nsub{opt->nsub}; |
| 301 | |
| 302 | shared_ptr<const config::TopicConfig> topic_config; |
| 303 | if (comm::RetCode::RET_OK != |
| 304 | (ret = config::GlobalConfig::GetThreadInstance()-> |
| 305 | GetTopicConfigByTopicID(topic_id, topic_config))) { |
| 306 | QLErr("GetTopicConfigByTopicID ret %d topic_id %d", ret, topic_id); |
| 307 | return; |
| 308 | } |
| 309 | |
| 310 | int store_sync_interval_s = topic_config->GetProto().topic().store_sync_interval_s(); |
| 311 | if (!store_sync_interval_s) store_sync_interval_s = 1; |
| 312 | |
| 313 | |
| 314 | for (int paxos_group_id{0}; paxos_group_id < ngroup; ++paxos_group_id) { |
| 315 | if (!node->IsIMMaster(paxos_group_id)) continue; |
| 316 | |
| 317 | time_t now{time(nullptr)}; |
| 318 | if (now < impl_->next_sync_times[paxos_group_id]) continue; |
| 319 | impl_->next_sync_times[paxos_group_id] = now + store_sync_interval_s + |
| 320 | comm::utils::OtherUtils::FastRand() % store_sync_interval_s; |
| 321 | |
| 322 | proto::SyncCtrlInfo sync_ctrl_info; |
| 323 | uint64_t prev_cursor_id; |
| 324 | for (int queue_id{paxos_group_id}; queue_id < nqueue; queue_id += ngroup) { |
| 325 | proto::SyncCtrlInfo::QueueDetail queue_detail; |
| 326 | queue_detail.set_queue_id(queue_id); |
| 327 | for (int sub_id{1}; sub_id <= nsub; ++sub_id) { |
| 328 | if (0 > as_integer(ret = sync->GetCursorID(sub_id, queue_id, prev_cursor_id))) { |
| 329 | QLErr("sync->GetCursorID ret %d sub_id %d queue %d", |
| 330 | as_integer(ret), sub_id, queue_id); |
| 331 | } else if (!as_integer(ret)) { |
| 332 | proto::SyncCtrlInfo::QueueDetail::SubDetail *sub_detail = |
| 333 | queue_detail.add_sub_details(); |
| 334 | sub_detail->set_sub_id(sub_id); |
| 335 | sub_detail->set_prev_cursor_id(prev_cursor_id); |
| 336 | } |
| 337 | } |
| 338 | if (queue_detail.sub_details_size()) { |
| 339 | sync_ctrl_info.add_queue_details()->CopyFrom(queue_detail); |
| 340 | } |
| 341 | } |
| 342 | |
| 343 | // 1. serialize paxos value |
nothing calls this directly
no test coverage detected