Line data    Source code

       1             : /*
       2             :  * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
       3             :  */
       4             : 
       5             : #include "db/db_partition.h"
       6             : 
       7             : #include <list>
       8             : #include <mutex>
       9             : #include <atomic>
      10             : 
      11             : #include <tbb/concurrent_queue.h>
      12             : 
      13             : #include "base/task.h"
      14             : #include "db/db.h"
      15             : #include "db/db_client.h"
      16             : #include "db/db_entry.h"
      17             : 
      18             : using tbb::concurrent_queue;
      19             : 
      20             : struct RequestQueueEntry {
      21             :     // Constructor takes ownership of DBRequest key, data.
      22     1144261 :     RequestQueueEntry(DBTablePartBase *tpart, DBClient *client, DBRequest *req)
      23     1144261 :         : tpart(tpart), client(client) {
      24     1144253 :         request.Swap(req);
      25     1144229 :     }
      26             :     DBTablePartBase *tpart;
      27             :     DBClient *client;
      28             :     DBRequest request;
      29             : };
      30             : 
      31             : struct RemoveQueueEntry {
      32      514144 :     RemoveQueueEntry(DBTablePartBase *tpart, DBEntryBase *db_entry)
      33      514144 :         : tpart(tpart), db_entry(db_entry) {
      34      514144 :     }
      35             :     DBTablePartBase *tpart;
      36             :     DBEntryBase *db_entry;
      37             : };
      38             : 
      39             : class DBPartition::WorkQueue {
      40             : public:
      41             :     static const int kThreshold = 1024;
      42             :     typedef concurrent_queue<RequestQueueEntry *> RequestQueue;
      43             :     typedef concurrent_queue<RemoveQueueEntry *> RemoveQueue;
      44             :     typedef std::list<DBTablePartBase *> TablePartList;
      45             : 
      46      109350 :     explicit WorkQueue(DBPartition *partition, int partition_id)
      47      109350 :         : db_partition_(partition),
      48      109350 :           db_partition_id_(partition_id),
      49      109350 :           disable_(false),
      50      109350 :           running_(false) {
      51      109350 :         request_count_ = 0;
      52      109350 :         max_request_queue_len_ = 0;
      53      109350 :         total_request_count_ = 0;
      54      109350 :     }
      55      109286 :     ~WorkQueue() {
      56      109286 :         for (RequestQueue::iterator iter = request_queue_.unsafe_begin();
      57      109286 :              iter != request_queue_.unsafe_end();) {
      58           0 :             RequestQueueEntry *req_entry = *iter;
      59           0 :             ++iter;
      60           0 :             delete req_entry;
      61             :         }
      62      109286 :         request_queue_.clear();
      63      109286 :     }
      64             : 
      65     1144226 :     bool EnqueueRequest(RequestQueueEntry *req_entry) {
      66     1144226 :         request_queue_.push(req_entry);
      67     1144279 :         MaybeStartRunner();
      68     1144310 :         uint32_t max = request_count_.fetch_add(1);
      69     1144310 :         if (max > max_request_queue_len_)
      70      161745 :             max_request_queue_len_ = max;
      71     1144310 :         total_request_count_++;
      72     1144310 :         return max < (kThreshold - 1);
      73             : 
      74             :     }
      75             : 
      76     1984999 :     bool DequeueRequest(RequestQueueEntry **req_entry) {
      77     1984999 :         bool success = request_queue_.try_pop(*req_entry);
      78     1984987 :         if (success) {
      79     1144294 :             request_count_.fetch_sub(1);
      80             :         }
      81     1984987 :         return success;
      82             :     }
      83             : 
      84      514256 :     void EnqueueRemove(RemoveQueueEntry *rm_entry) {
      85      514256 :         remove_queue_.push(rm_entry);
      86      514259 :         MaybeStartRunner();
      87      514279 :     }
      88             : 
      89     1359480 :     bool DequeueRemove(RemoveQueueEntry **rm_entry) {
      90     1359480 :         bool success = remove_queue_.try_pop(*rm_entry);
      91     1359510 :         return success;
      92             :     }
      93             : 
      94             :     void MaybeStartRunnerUnlocked();
      95             :     void MaybeStartRunner();
      96             :     bool RunnerDone();
      97             : 
      98             :     // Normally called from single task that either runs in DB context or is
      99             :     // exclusive with DB task, but can be called concurrently from multiple
     100             :     // bgp::ConfigHelper tasks.
     101     1688803 :     void SetActive(DBTablePartBase *tpart) {
     102     1688803 :         std::scoped_lock lock(mutex_);
     103     1689027 :         change_list_.push_back(tpart);
     104     1688662 :         MaybeStartRunnerUnlocked();
     105     1688778 :     }
     106             : 
     107     2529578 :     DBTablePartBase *GetActiveTable() {
     108     2529578 :         DBTablePartBase *tpart = NULL;
     109     2529578 :         if (!change_list_.empty()) {
     110     1688964 :             tpart = change_list_.front();
     111     1688918 :             change_list_.pop_front();
     112             :         }
     113     2529552 :         return tpart;
     114             :     }
     115             : 
     116      794242 :     int db_partition_id() {
     117      794242 :         return db_partition_id_;
     118             :     }
     119             : 
     120      794242 :     int db_task_id() const { return db_partition_->task_id(); }
     121             : 
     122     6250516 :     bool IsDBQueueEmpty() const {
     123     6250516 :         return (request_queue_.empty() && change_list_.empty());
     124             :     }
     125             : 
     126      848428 :     bool disable() { return disable_; }
     127          90 :     void set_disable(bool disable) { disable_ = disable; }
     128             : 
     129           0 :     long request_queue_len() const {
     130           0 :         return request_count_;
     131             :     }
     132             : 
     133           0 :     uint64_t total_request_count() const {
     134           0 :         return total_request_count_;
     135             :     }
     136             : 
     137           0 :     uint64_t max_request_queue_len() const {
     138           0 :         return max_request_queue_len_;
     139             :     }
     140             : 
     141             : private:
     142             :     DBPartition *db_partition_;
     143             :     RequestQueue request_queue_;
     144             :     TablePartList change_list_;
     145             :     std::atomic<long> request_count_;
     146             :     uint64_t total_request_count_;
     147             :     uint64_t max_request_queue_len_;
     148             :     RemoveQueue remove_queue_;
     149             :     std::mutex mutex_;
     150             :     int db_partition_id_;
     151             :     bool disable_;
     152             :     bool running_;
     153             : 
     154             :     DISALLOW_COPY_AND_ASSIGN(WorkQueue);
     155             : };
     156             : 
     157     6250516 : bool DBPartition::IsDBQueueEmpty() const {
     158     6250516 :     return work_queue_->IsDBQueueEmpty();
     159             : }
     160             : 
     161          90 : void DBPartition::SetQueueDisable(bool disable) {
     162          90 :     if (disable) {
     163          45 :         work_queue_->set_disable(true);
     164             :     } else {
     165          45 :         work_queue_->set_disable(false);
     166          45 :         work_queue_->MaybeStartRunner();
     167             :     }
     168          90 : }
     169             : 
     170             : class DBPartition::QueueRunner : public Task {
     171             : public:
     172             :     static const int kMaxIterations = 32;
     173      794246 :     QueueRunner(WorkQueue *queue)
     174      794246 :         : Task(queue->db_task_id(), queue->db_partition_id()),
     175      794246 :           queue_(queue) {
     176      794231 :     }
     177             : 
     178      848425 :     virtual bool Run() {
     179      848425 :         int count = 0;
     180             : 
     181             :         // Skip if the queue is disabled.
     182      848425 :         if (queue_->disable())
     183        1081 :             return queue_->RunnerDone();
     184             : 
     185      847348 :         RemoveQueueEntry *rm_entry = NULL;
     186     1359511 :         while (queue_->DequeueRemove(&rm_entry)) {
     187      514282 :             DBEntryBase *db_entry = rm_entry->db_entry;
     188             :             {
     189             :                 tbb::spin_rw_mutex::scoped_lock
     190      514282 :                     lock(rm_entry->tpart->dbstate_mutex(), false);
     191     1028427 :                 if (!db_entry->IsDeleted() || db_entry->is_onlist() ||
     192      514188 :                     !db_entry->is_state_empty_unlocked(rm_entry->tpart)) {
     193          42 :                     db_entry->ClearOnRemoveQ();
     194          42 :                     db_entry = NULL;
     195             :                 }
     196      514224 :             }
     197      514315 :             if (db_entry) {
     198      514273 :                 rm_entry->tpart->Remove(db_entry);
     199             :             }
     200      514293 :             delete rm_entry;
     201      514310 :             if (++count == kMaxIterations) {
     202        2147 :                 return false;
     203             :             }
     204             :         }
     205             : 
     206      845219 :         RequestQueueEntry *req_entry = NULL;
     207     1985005 :         while (queue_->DequeueRequest(&req_entry)) {
     208     1144278 :             req_entry->tpart->Process(req_entry->client, &req_entry->request);
     209     1144235 :             delete req_entry;
     210     1144288 :             if (++count == kMaxIterations) {
     211        4502 :                 return false;
     212             :             }
     213             :         }
     214             : 
     215             :         while (true) {
     216     2529518 :             DBTablePartBase *tpart = queue_->GetActiveTable();
     217     2529545 :             if (tpart == NULL) {
     218      840657 :                 break;
     219             :             }
     220     1688888 :             bool done = tpart->RunNotify();
     221     1688869 :             if (!done) {
     222          61 :                 return false;
     223             :             }
     224     1688808 :         }
     225             : 
     226             :         // Running is done only if queue_ is empty. It's possible that more
     227             :         // entries are added into in the input or remove queues during the
     228             :         // time we were processing those queues.
     229      840657 :         return queue_->RunnerDone();
     230             :     }
     231             : 
     232           0 :     std::string Description() const {
     233           0 :         return "DBPartition QueueRunner";
     234             :     }
     235             : private:
     236             :     WorkQueue *queue_;
     237             : };
     238             : 
     239     3347189 : void DBPartition::WorkQueue::MaybeStartRunnerUnlocked() {
     240     3347189 :     if (running_) {
     241     2553082 :         return;
     242             :     }
     243      794107 :     running_ = true;
     244      794107 :     QueueRunner *runner = new QueueRunner(this);
     245      794227 :     TaskScheduler *scheduler = TaskScheduler::GetInstance();
     246      794220 :     scheduler->Enqueue(runner);
     247             : }
     248             : 
     249     1658550 : void DBPartition::WorkQueue::MaybeStartRunner() {
     250     1658550 :     std::scoped_lock lock(mutex_);
     251     1658609 :     MaybeStartRunnerUnlocked();
     252     1658637 : }
     253             : 
     254      841728 : bool DBPartition::WorkQueue::RunnerDone() {
     255      841728 :     std::scoped_lock lock(mutex_);
     256      841761 :     if (disable_ || (request_queue_.empty() && remove_queue_.empty())) {
     257      794274 :         running_ = false;
     258      794274 :         return true;
     259             :     }
     260             : 
     261       47437 :     running_ = true;
     262       47437 :     return false;
     263      841711 : }
     264             : 
     265      109350 : DBPartition::DBPartition(DB *db, int partition_id)
     266      109350 :     : db_(db), work_queue_(new WorkQueue(this, partition_id)) {
     267      109350 : }
     268             : 
     269             : // The DBPartition destructor needs to be defined after WorkQueue has
     270             : // been declared.
     271      109286 : DBPartition::~DBPartition() {
     272      109286 : }
     273             : 
     274     1144256 : bool DBPartition::EnqueueRequest(DBTablePartBase *tpart, DBClient *client,
     275             :                                  DBRequest *req) {
     276     1144256 :     RequestQueueEntry *entry = new RequestQueueEntry(tpart, client, req);
     277     1144228 :     return work_queue_->EnqueueRequest(entry);
     278             : }
     279             : 
     280      514140 : void DBPartition::EnqueueRemove(DBTablePartBase *tpart, DBEntryBase *db_entry) {
     281      514140 :     RemoveQueueEntry *entry = new RemoveQueueEntry(tpart, db_entry);
     282      514149 :     db_entry->SetOnRemoveQ();
     283      514324 :     work_queue_->EnqueueRemove(entry);
     284      514271 : }
     285             : 
     286             : // concurrency: called from DBPartition task.
     287     1688867 : void DBPartition::OnTableChange(DBTablePartBase *tablepart) {
     288     1688867 :     work_queue_->SetActive(tablepart);
     289     1688995 : }
     290             : 
     291           0 : long DBPartition::request_queue_len() const {
     292           0 :     return work_queue_->request_queue_len();
     293             : }
     294             : 
     295           0 : uint64_t DBPartition::total_request_count() const {
     296           0 :     return work_queue_->total_request_count();
     297             : }
     298             : 
     299           0 : uint64_t DBPartition::max_request_queue_len() const {
     300           0 :     return work_queue_->max_request_queue_len();
     301             : }
     302             : 
     303      794240 : int DBPartition::task_id() const {
     304      794240 :     return db_->task_id();
     305             : }