Line data    Source code

       1             : /*
       2             :  * Copyright (c) 2016 Juniper Networks, Inc. All rights reserved.
       3             :  */
       4             : 
       5             : #include "db/db_table_walk_mgr.h"
       6             : 
       7             : #include <boost/bind/bind.hpp>
       8             : #include <boost/foreach.hpp>
       9             : 
      10             : #include "base/logging.h"
      11             : #include "base/task.h"
      12             : #include "base/task_annotations.h"
      13             : #include "db/db.h"
      14             : #include "db/db_partition.h"
      15             : #include "db/db_table.h"
      16             : #include "db/db_table_partition.h"
      17             : 
      18             : using namespace boost::placeholders;
      19             : 
      20       27400 : DBTableWalkMgr::DBTableWalkMgr()
      21       54800 :     : walk_request_trigger_(new TaskTrigger(
      22             :         boost::bind(&DBTableWalkMgr::ProcessWalkRequestList, this),
      23       54800 :         TaskScheduler::GetInstance()->GetTaskId("db::Walker"), 0)),
      24       54800 :       walk_done_trigger_(new TaskTrigger(
      25             :         boost::bind(&DBTableWalkMgr::ProcessWalkDone, this),
      26       54800 :         TaskScheduler::GetInstance()->GetTaskId("db::Walker"), 0)) {
      27       27400 : }
      28             : 
      29      325775 : bool DBTableWalkMgr::ProcessWalkRequestList() {
      30      325775 :     CHECK_CONCURRENCY("db::Walker");
      31      325775 :     std::scoped_lock lock(mutex_);
      32      325775 :     if (!current_table_walk_.empty()) return true;
      33             :     while (true) {
      34      309380 :         if (walk_request_list_.empty()) break;
      35      226018 :         WalkRequestInfoPtr info = walk_request_list_.front();
      36      226018 :         walk_request_set_.erase(info.get());
      37      226018 :         walk_request_list_.pop_front();
      38      226018 :         current_table_walk_.swap(info->pending_requests);
      39      226018 :         DBTable *table = info->table;
      40      226018 :         bool walk_table = false;
      41      453754 :         for (auto walker : current_table_walk_) {
      42      227736 :             if (walker->stopped()) continue;
      43      227734 :             walker->set_in_progress();
      44      227734 :             walker->reset_walk_again();
      45      227734 :             walk_table = true;
      46      227736 :         }
      47      226018 :         if (walk_table) {
      48             :             // start the walk
      49      226017 :             table->StartWalk();
      50      226017 :             break;
      51             :         } else {
      52           1 :             current_table_walk_.clear();
      53             :         }
      54      226019 :     }
      55      309379 :     return true;
      56      325775 : }
      57             : 
      58      226017 : bool DBTableWalkMgr::ProcessWalkDone() {
      59      226017 :     CHECK_CONCURRENCY("db::Walker");
      60      226017 :     assert(!current_table_walk_.empty());
      61      453752 :     for (auto walker : current_table_walk_) {
      62      227735 :         if (walker->walk_again())
      63         218 :             walker->set_walk_requested();
      64      227517 :         else if (!walker->stopped())
      65      227514 :             walker->set_walk_done();
      66      227735 :         if (walker->stopped() || walker->walk_again()) continue;
      67      227514 :         walker->walk_complete()(walker, walker->table());
      68      227735 :     }
      69      226017 :     current_table_walk_.clear();
      70      226017 :     walk_request_trigger_->Set();
      71      226017 :     return true;
      72             : }
      73             : 
      74      219374 : DBTable::DBTableWalkRef DBTableWalkMgr::AllocWalker(DBTable *table,
      75             :                DBTable::WalkFn walk_fn, DBTable::WalkCompleteFn walk_complete) {
      76      219374 :     table->incr_walker_count();
      77      219374 :     DBTableWalk *walker = new DBTableWalk(table, walk_fn, walk_complete);
      78      219373 :     return DBTable::DBTableWalkRef(walker);
      79             : }
      80             : 
      81      157411 : void DBTableWalkMgr::ReleaseWalker(DBTable::DBTableWalkRef &ref) {
      82      157411 :     ref->set_walk_stopped();
      83      157411 :     ref.reset();
      84      157411 : }
      85             : 
      86       29714 : void DBTableWalkMgr::WalkAgain(DBTable::DBTableWalkRef ref) {
      87       29714 :     WalkTable(ref);
      88       29714 : }
      89             : 
      90      255992 : void DBTableWalkMgr::WalkTable(DBTable::DBTableWalkRef walk) {
      91      255992 :     std::scoped_lock lock(mutex_);
      92      255992 :     DBTable *table = walk->table();
      93             : 
      94      255992 :     if (walk->in_progress()) {
      95         412 :         table->incr_walk_again_count();
      96         412 :         walk->set_walk_again();
      97             :     } else {
      98      255580 :         table->incr_walk_request_count();
      99      255580 :         walk->set_walk_requested();
     100             :     }
     101             : 
     102      255992 :     WalkRequestInfo tmp_info = WalkRequestInfo(table);
     103      255992 :     WalkRequestInfoSet::iterator it = walk_request_set_.find(&tmp_info);
     104      255992 :     if (it != walk_request_set_.end()) {
     105       29974 :         (*it)->AppendWalkReq(walk);
     106       29974 :         return;
     107             :     }
     108             : 
     109      226018 :     WalkRequestInfo *new_info = new WalkRequestInfo(table);
     110      226018 :     new_info->AppendWalkReq(walk);
     111      226018 :     walk_request_list_.push_back(WalkRequestInfoPtr(new_info));
     112      226018 :     walk_request_set_.insert(new_info);
     113      226018 :     walk_request_trigger_->Set();
     114      285966 : }
     115             : 
     116      226017 : void DBTableWalkMgr::WalkDone() {
     117      226017 :     walk_done_trigger_->Set();
     118      226017 : }
     119             : 
     120      788591 : bool DBTableWalkMgr::InvokeWalkCb(DBTablePartBase *part, DBEntryBase *entry) {
     121      788591 :     uint32_t skip_walk_count = 0;
     122     1579781 :     for (auto walker : current_table_walk_) {
     123      791070 :         if (walker->done() || walker->stopped() || walker->walk_again()) {
     124         366 :             skip_walk_count++;
     125         366 :             continue;
     126             :         }
     127      790276 :         bool more = walker->walk_fn()(part, entry);
     128      790395 :         if (!more) {
     129           3 :             skip_walk_count++;
     130           3 :             if (!walker->stopped()) walker->set_walk_done();
     131             :         }
     132      790758 :     }
     133      788649 :     return (skip_walk_count < current_table_walk_.size());
     134             : }