| 576 | |
| 577 | |
| 578 | void KeepSyncThread::ReportBacklog() { |
| 579 | QLVerb("KeepSyncThread::ReportBacklog"); |
| 580 | |
| 581 | comm::RetCode ret; |
| 582 | |
| 583 | vector<consumer::Queue_t> queues; |
| 584 | |
| 585 | GetAllLocalQueue(queues); |
| 586 | |
| 587 | const int topic_id = impl_->store->GetTopicID(); |
| 588 | auto opt = impl_->store->GetStoreOption(); |
| 589 | auto sync = impl_->store->GetSyncCtrl(); |
| 590 | |
| 591 | uint64_t cursor_id; |
| 592 | for (auto &&queue : queues) { |
| 593 | int paxos_group_id = queue.queue_id % opt->ngroup; |
| 594 | int immaster = impl_->store->GetNode()->IsIMMaster(paxos_group_id); |
| 595 | |
| 596 | if (as_integer(comm::RetCode::RET_OK) > |
| 597 | as_integer(ret = sync->GetCursorID(queue.sub_id, queue.queue_id, cursor_id))) { |
| 598 | QLErr("GetCursorID ret %d sub_id %d queue_id %d", ret, queue.sub_id, queue.queue_id); |
| 599 | continue; |
| 600 | } |
| 601 | |
| 602 | int backlog = 0; |
| 603 | if (comm::RetCode::RET_OK != |
| 604 | (ret = sync->GetBackLogByCursorID(queue.queue_id, cursor_id, backlog))) { |
| 605 | QLErr("GetBackLogByCursorID ret %d", as_integer(ret)); |
| 606 | continue; |
| 607 | } |
| 608 | |
| 609 | QLInfo("BACKLOG: store_id %d queue_id %d sub_id %d backlog %d cursor_id %" |
| 610 | PRIu64 " immaster %d", |
| 611 | queue.store_id, queue.queue_id, queue.sub_id, backlog, cursor_id, immaster); |
| 612 | if (immaster) |
| 613 | comm::StoreBacklogBP::GetThreadInstance()-> |
| 614 | OnBackLogReport(topic_id, queue.sub_id, queue.queue_id, backlog); |
| 615 | } |
| 616 | } |
| 617 | |
| 618 | void KeepSyncThread::ReportDelayStat() { |
| 619 | QLVerb("KeepSyncThread::ReportDelayStat"); |
nothing calls this directly
no test coverage detected