Line data    Source code

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