Line data    Source code

       1             : /*
       2             :  * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
       3             :  * Copyright Contributors to the OpenSDN project
       4             :  */
       5             : 
       6             : #include "db/db_table_walker.h"
       7             : 
       8             : #include <atomic>
       9             : 
      10             : #include "base/logging.h"
      11             : #include "base/task.h"
      12             : #include "db/db.h"
      13             : #include "db/db_partition.h"
      14             : #include "db/db_table.h"
      15             : #include "db/db_table_partition.h"
      16             : 
      17             : int DBTableWalker::max_iteration_to_yield_ = kIterationToYield;
      18             : 
      19             : class DBTableWalker::Walker {
      20             : public:
      21             :     Walker(WalkId id, DBTableWalker *wkmgr, DBTable *table,
      22             :            const DBRequestKey *key, WalkFn walker,
      23             :            WalkCompleteFn walk_done, bool postpone_walk);
      24             : 
      25           0 :     void StopWalk() {
      26           0 :         assert(workers_.empty());
      27           0 :         should_stop_.exchange(true);
      28           0 :     }
      29             : 
      30             :     // Test only - resume walk that was postponed at creation.
      31           0 :     void ResumeWalk() {
      32           0 :         assert(!workers_.empty());
      33           0 :         TaskScheduler *scheduler = TaskScheduler::GetInstance();
      34           0 :         for (std::vector<Task *>::iterator it = workers_.begin();
      35           0 :              it != workers_.end(); ++it) {
      36           0 :             scheduler->Enqueue(*it);
      37             :         }
      38           0 :         workers_.clear();
      39           0 :     }
      40             : 
      41             :     WalkId  id_;
      42             : 
      43             :     // Parent walker manager
      44             :     DBTableWalker *wkmgr_;
      45             : 
      46             :     // Table on which walk is done
      47             :     DBTable *table_;
      48             : 
      49             :     // Take the ownership of key passed
      50             :     std::unique_ptr<DBRequestKey> key_start_;
      51             : 
      52             :     WalkFn walker_fn_;
      53             :     WalkCompleteFn done_fn_;
      54             : 
      55             :     // Will be true if Table walk is cancelled
      56             :     std::atomic<bool> should_stop_;
      57             : 
      58             :     // check whether iteration is completed on all Table Partition
      59             :     std::atomic<long> status_;
      60             : 
      61             :     std::vector<Task *> workers_;
      62        1800 :     int task_id() const { return wkmgr_->task_id(); }
      63             : };
      64             : 
      65             : class DBTableWalker::Worker : public Task {
      66             : public:
      67        1800 :     Worker(Walker *walker, int db_partition_id, const DBRequestKey *key)
      68        3600 :         : Task(walker->task_id(), db_partition_id), walker_(walker),
      69        1800 :           key_start_(key) {
      70        1800 :         tbl_partition_ = static_cast<DBTablePartition *>(
      71        1800 :             walker_->table_->GetTablePartition(db_partition_id));
      72        1800 :     }
      73             : 
      74             :     virtual bool Run();
      75           0 :     std::string Description() const { return "DBTableWalker::Worker"; }
      76             : 
      77             : private:
      78             :     DBTableWalker::Walker *walker_;
      79             : 
      80             :     // Store the last visited node to continue walk
      81             :     std::unique_ptr<DBRequestKey> walk_ctx_;
      82             : 
      83             :     // This is where the walk started
      84             :     const DBRequestKey *key_start_;
      85             : 
      86             :     // Table partition for which this worker was created
      87             :     DBTablePartition *tbl_partition_;
      88             : };
      89             : 
      90         162 : static void db_walker_wait() {
      91             :     static unsigned int walk_sleep_usecs_;
      92             :     static bool once;
      93             : 
      94         162 :     if (!once) {
      95           1 :         once = true;
      96             : 
      97           1 :         char *wait = getenv("DB_WALKER_WAIT_USECS");
      98           1 :         if (wait) {
      99           0 :           walk_sleep_usecs_ = (unsigned int) strtoul(wait, NULL, 0);
     100           0 :           if (walk_sleep_usecs_ > 1000000)
     101           0 :             walk_sleep_usecs_ = 1000000;
     102             :         }
     103             :     }
     104             : 
     105         162 :     if (walk_sleep_usecs_) {
     106           0 :         usleep(walk_sleep_usecs_);
     107             :     }
     108         162 : }
     109             : 
     110        1841 : bool DBTableWalker::Worker::Run() {
     111        1841 :     int count = 0;
     112             :     DBRequestKey *key_resume;
     113             : 
     114             :     // Check whether Walker was requested to be cancelled
     115        1841 :     if (walker_->should_stop_) {
     116           0 :         goto walk_done;
     117             :     }
     118             : 
     119             :     // Check where we left in last iteration
     120        1841 :     if ((key_resume = walk_ctx_.get()) == NULL) {
     121             :         // First time invoke of worker thread, start from key_start_
     122        1796 :         key_resume = const_cast <DBRequestKey *>(key_start_);
     123             :     }
     124             : 
     125             :     DBEntry *entry;
     126        1841 :     if (key_resume != NULL) {
     127          45 :         DBTable *table = walker_->table_;
     128          45 :         std::unique_ptr<const DBEntryBase> start;
     129          45 :         start = table->AllocEntry(key_resume);
     130             :         // Find matching or next in sort order
     131          45 :         entry = tbl_partition_->lower_bound(start.get());
     132          45 :     } else {
     133        1796 :         entry = tbl_partition_->GetFirst();
     134             :     }
     135        1845 :     if (entry == NULL) {
     136        1683 :         goto walk_done;
     137             :     }
     138             : 
     139         324 :     for (DBEntry *next = NULL; entry; entry = next) {
     140         207 :         next = tbl_partition_->GetNext(entry);
     141             :         // Check whether Walker was requested to be cancelled
     142         207 :         if (walker_->should_stop_) {
     143           0 :             break;
     144             :         }
     145         207 :         if (count == GetIterationToYield()) {
     146             :             // store the context
     147          45 :             walk_ctx_ = entry->GetDBRequestKey();
     148          45 :             return false;
     149             :         }
     150             : 
     151             :         // Invoke walker function
     152         162 :         bool more = walker_->walker_fn_(tbl_partition_, entry);
     153         162 :         if (!more) {
     154           0 :             break;
     155             :         }
     156             : 
     157         162 :         db_walker_wait();
     158         162 :         count++;
     159             :     }
     160             : 
     161         117 : walk_done:
     162             :     // Check whether all other walks on the table is completed
     163        1800 :     long num_walkers_on_tpart = walker_->status_.fetch_sub(1);
     164        1800 :     if (num_walkers_on_tpart == 1) {
     165         450 :         if (walker_->should_stop_) {
     166           0 :             walker_->table_->incr_walk_cancel_count();
     167             :         } else {
     168         450 :             walker_->table_->incr_walk_complete_count();
     169             :             // Invoke Walker_Complete callback
     170         450 :             if (walker_->done_fn_ != NULL) {
     171         450 :                 walker_->done_fn_(walker_->table_);
     172             :             }
     173             :         }
     174             : 
     175             :         // Release the memory for walker and bitmap
     176         450 :         walker_->wkmgr_->PurgeWalker(walker_->id_);
     177             :     }
     178        1800 :     return true;
     179             : }
     180             : 
     181         450 : DBTableWalker::Walker::Walker(WalkId id, DBTableWalker *wkmgr,
     182             :                               DBTable *table, const DBRequestKey *key,
     183             :                               WalkFn walker, WalkCompleteFn walk_done,
     184         450 :                               bool postpone_walk)
     185         450 :     : id_(id), wkmgr_(wkmgr), table_(table),
     186         450 :       key_start_(const_cast<DBRequestKey *>(key)),
     187         450 :       walker_fn_(walker), done_fn_(walk_done) {
     188         450 :     int num_worker = table->PartitionCount();
     189         450 :     should_stop_ = false;
     190         450 :     status_ = num_worker;
     191        2250 :     for (int i = 0; i < num_worker; i++) {
     192        1800 :         Worker *task = new Worker(this, i, key);
     193        1800 :         if (postpone_walk) {
     194           0 :             workers_.push_back(task);
     195             :         } else {
     196        1800 :             TaskScheduler *scheduler = TaskScheduler::GetInstance();
     197        1800 :             scheduler->Enqueue(task);
     198             :         }
     199             :     }
     200         450 : }
     201             : 
     202       27406 : DBTableWalker::DBTableWalker(int task_id) : task_id_(task_id) {
     203       27406 :     if (task_id == -1) {
     204           0 :         TaskScheduler *scheduler = TaskScheduler::GetInstance();
     205           0 :         task_id_ = scheduler->GetTaskId("db::DBTable");
     206             :     }
     207       27406 : }
     208             : 
     209         450 : DBTableWalker::WalkId DBTableWalker::WalkTable(DBTable *table,
     210             :                                                const DBRequestKey *key_start,
     211             :                                                WalkFn walkerfn ,
     212             :                                                WalkCompleteFn walk_complete,
     213             :                                                bool postpone_walk) {
     214         450 :     table->incr_walk_request_count();
     215         450 :     std::scoped_lock lock(walkers_mutex_);
     216         450 :     size_t i = walker_map_.find_first();
     217         450 :     if (i == walker_map_.npos) {
     218         443 :         i = walkers_.size();
     219             :         Walker *walker = new Walker(i, this, table, key_start,
     220         443 :                                     walkerfn, walk_complete, postpone_walk);
     221         443 :         walkers_.push_back(walker);
     222             :     } else {
     223           7 :         walker_map_.reset(i);
     224           7 :         if (walker_map_.none()) {
     225           7 :             walker_map_.clear();
     226             :         }
     227             :         Walker *walker = new Walker(i, this, table, key_start,
     228           7 :                                     walkerfn, walk_complete, postpone_walk);
     229           7 :         walkers_[i] = walker;
     230             :     }
     231         450 :     table->incr_walker_count();
     232         450 :     return i;
     233         450 : }
     234             : 
     235           0 : void DBTableWalker::WalkCancel(WalkId id) {
     236           0 :     std::scoped_lock lock(walkers_mutex_);
     237           0 :     walkers_[id]->StopWalk();
     238             :     // Purge to be called after task has stopped
     239           0 : }
     240             : 
     241           0 : void DBTableWalker::WalkResume(WalkId id) {
     242           0 :     std::scoped_lock lock(walkers_mutex_);
     243           0 :     walkers_[id]->ResumeWalk();
     244           0 : }
     245             : 
     246         450 : void DBTableWalker::PurgeWalker(WalkId id) {
     247         450 :     std::scoped_lock lock(walkers_mutex_);
     248         450 :     Walker *walker = walkers_[id];
     249         450 :     DBTable *table = walker->table_;
     250         450 :     delete walker;
     251         450 :     walkers_[id] = NULL;
     252         450 :     if ((size_t) id == walkers_.size() - 1) {
     253         459 :         while (!walkers_.empty() && walkers_.back() == NULL) {
     254         443 :             walkers_.pop_back();
     255             :         }
     256          16 :         if (walker_map_.size() > walkers_.size()) {
     257          16 :             walker_map_.resize(walkers_.size());
     258             :         }
     259             :     } else {
     260         434 :         if ((size_t) id >= walker_map_.size()) {
     261         429 :             walker_map_.resize(id + 1);
     262             :         }
     263         434 :         walker_map_.set(id);
     264             :     }
     265             : 
     266             :     // Retry table deletion when the last walker is purged.
     267         450 :     if (table->decr_walker_count() == 0) {
     268         317 :         table->RetryDelete();
     269             :     }
     270         450 : }