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_partition.h"
       7             : 
       8             : #include <list>
       9             : #include <mutex>
      10             : #include <atomic>
      11             : 
      12             : #include <tbb/concurrent_queue.h>
      13             : 
      14             : #include "base/task.h"
      15             : #include "db/db.h"
      16             : #include "db/db_client.h"
      17             : #include "db/db_entry.h"
      18             : 
      19             : using tbb::concurrent_queue;
      20             : 
      21             : struct RequestQueueEntry {
      22             :     // Constructor takes ownership of DBRequest key, data.
      23     1166277 :     RequestQueueEntry(DBTablePartBase *tpart, DBClient *client, DBRequest *req)
      24     1166277 :         : tpart(tpart), client(client) {
      25     1166271 :         request.Swap(req);
      26     1166220 :     }
      27             :     DBTablePartBase *tpart;
      28             :     DBClient *client;
      29             :     DBRequest request;
      30             : };
      31             : 
      32             : struct RemoveQueueEntry {
      33      524740 :     RemoveQueueEntry(DBTablePartBase *tpart, DBEntryBase *db_entry)
      34      524740 :         : tpart(tpart), db_entry(db_entry) {
      35      524740 :     }
      36             :     DBTablePartBase *tpart;
      37             :     DBEntryBase *db_entry;
      38             : };
      39             : 
      40             : class DBPartition::WorkQueue {
      41             : public:
      42             :     static const int kThreshold = 1024;
      43             :     typedef concurrent_queue<RequestQueueEntry *> RequestQueue;
      44             :     typedef concurrent_queue<RemoveQueueEntry *> RemoveQueue;
      45             :     typedef std::list<DBTablePartBase *> TablePartList;
      46             : 
      47      109374 :     explicit WorkQueue(DBPartition *partition, int partition_id)
      48      109374 :         : db_partition_(partition),
      49      109374 :           db_partition_id_(partition_id),
      50      109374 :           disable_(false),
      51      109374 :           running_(false) {
      52      109374 :         request_count_ = 0;
      53      109374 :         max_request_queue_len_ = 0;
      54      109374 :         total_request_count_ = 0;
      55      109374 :     }
      56      109310 :     ~WorkQueue() {
      57      109310 :         for (RequestQueue::iterator iter = request_queue_.unsafe_begin();
      58      109310 :              iter != request_queue_.unsafe_end();) {
      59           0 :             RequestQueueEntry *req_entry = *iter;
      60           0 :             ++iter;
      61           0 :             delete req_entry;
      62             :         }
      63      109310 :         request_queue_.clear();
      64      109310 :     }
      65             : 
      66     1166234 :     bool EnqueueRequest(RequestQueueEntry *req_entry) {
      67     1166234 :         request_queue_.push(req_entry);
      68     1166303 :         MaybeStartRunner();
      69     1166331 :         uint32_t max = request_count_.fetch_add(1);
      70     1166331 :         if (max > max_request_queue_len_)
      71      160521 :             max_request_queue_len_ = max;
      72     1166331 :         total_request_count_++;
      73     1166331 :         return max < (kThreshold - 1);
      74             : 
      75             :     }
      76             : 
      77     2014940 :     bool DequeueRequest(RequestQueueEntry **req_entry) {
      78     2014940 :         bool success = request_queue_.try_pop(*req_entry);
      79     2014942 :         if (success) {
      80     1166318 :             request_count_.fetch_sub(1);
      81             :         }
      82     2014942 :         return success;
      83             :     }
      84             : 
      85      524858 :     void EnqueueRemove(RemoveQueueEntry *rm_entry) {
      86      524858 :         remove_queue_.push(rm_entry);
      87      524829 :         MaybeStartRunner();
      88      524869 :     }
      89             : 
      90     1378090 :     bool DequeueRemove(RemoveQueueEntry **rm_entry) {
      91     1378090 :         bool success = remove_queue_.try_pop(*rm_entry);
      92     1378119 :         return success;
      93             :     }
      94             : 
      95             :     void MaybeStartRunnerUnlocked();
      96             :     void MaybeStartRunner();
      97             :     bool RunnerDone();
      98             : 
      99             :     // Normally called from single task that either runs in DB context or is
     100             :     // exclusive with DB task, but can be called concurrently from multiple
     101             :     // bgp::ConfigHelper tasks.
     102     1688681 :     void SetActive(DBTablePartBase *tpart) {
     103     1688681 :         std::scoped_lock lock(mutex_);
     104     1688920 :         change_list_.push_back(tpart);
     105     1688558 :         MaybeStartRunnerUnlocked();
     106     1688672 :     }
     107             : 
     108     2537381 :     DBTablePartBase *GetActiveTable() {
     109     2537381 :         DBTablePartBase *tpart = NULL;
     110     2537381 :         if (!change_list_.empty()) {
     111     1688838 :             tpart = change_list_.front();
     112     1688781 :             change_list_.pop_front();
     113             :         }
     114     2537381 :         return tpart;
     115             :     }
     116             : 
     117      801582 :     int db_partition_id() {
     118      801582 :         return db_partition_id_;
     119             :     }
     120             : 
     121      801577 :     int db_task_id() const { return db_partition_->task_id(); }
     122             : 
     123     6254074 :     bool IsDBQueueEmpty() const {
     124     6254074 :         return (request_queue_.empty() && change_list_.empty());
     125             :     }
     126             : 
     127      856721 :     bool disable() { return disable_; }
     128          90 :     void set_disable(bool disable) { disable_ = disable; }
     129             : 
     130           0 :     long request_queue_len() const {
     131           0 :         return request_count_;
     132             :     }
     133             : 
     134           0 :     uint64_t total_request_count() const {
     135           0 :         return total_request_count_;
     136             :     }
     137             : 
     138           0 :     uint64_t max_request_queue_len() const {
     139           0 :         return max_request_queue_len_;
     140             :     }
     141             : 
     142             : private:
     143             :     DBPartition *db_partition_;
     144             :     RequestQueue request_queue_;
     145             :     TablePartList change_list_;
     146             :     std::atomic<long> request_count_;
     147             :     uint64_t total_request_count_;
     148             :     uint64_t max_request_queue_len_;
     149             :     RemoveQueue remove_queue_;
     150             :     std::mutex mutex_;
     151             :     int db_partition_id_;
     152             :     bool disable_;
     153             :     bool running_;
     154             : 
     155             :     DISALLOW_COPY_AND_ASSIGN(WorkQueue);
     156             : };
     157             : 
     158     6254074 : bool DBPartition::IsDBQueueEmpty() const {
     159     6254074 :     return work_queue_->IsDBQueueEmpty();
     160             : }
     161             : 
     162          90 : void DBPartition::SetQueueDisable(bool disable) {
     163          90 :     if (disable) {
     164          45 :         work_queue_->set_disable(true);
     165             :     } else {
     166          45 :         work_queue_->set_disable(false);
     167          45 :         work_queue_->MaybeStartRunner();
     168             :     }
     169          90 : }
     170             : 
     171             : class DBPartition::QueueRunner : public Task {
     172             : public:
     173             :     static const int kMaxIterations = 32;
     174      801586 :     QueueRunner(WorkQueue *queue)
     175      801586 :         : Task(queue->db_task_id(), queue->db_partition_id()),
     176      801586 :           queue_(queue) {
     177      801563 :     }
     178             : 
     179      856719 :     virtual bool Run() {
     180      856719 :         int count = 0;
     181             : 
     182             :         // Skip if the queue is disabled.
     183      856719 :         if (queue_->disable())
     184        1138 :             return queue_->RunnerDone();
     185             : 
     186      855583 :         RemoveQueueEntry *rm_entry = NULL;
     187     1378132 :         while (queue_->DequeueRemove(&rm_entry)) {
     188      524889 :             DBEntryBase *db_entry = rm_entry->db_entry;
     189             :             {
     190             :                 tbb::spin_rw_mutex::scoped_lock
     191      524889 :                     lock(rm_entry->tpart->dbstate_mutex(), false);
     192     1049631 :                 if (!db_entry->IsDeleted() || db_entry->is_onlist() ||
     193      524793 :                     !db_entry->is_state_empty_unlocked(rm_entry->tpart)) {
     194          41 :                     db_entry->ClearOnRemoveQ();
     195          42 :                     db_entry = NULL;
     196             :                 }
     197      524828 :             }
     198      524911 :             if (db_entry) {
     199      524869 :                 rm_entry->tpart->Remove(db_entry);
     200             :             }
     201      524898 :             delete rm_entry;
     202      524907 :             if (++count == kMaxIterations) {
     203        2358 :                 return false;
     204             :             }
     205             :         }
     206             : 
     207      853226 :         RequestQueueEntry *req_entry = NULL;
     208     2014939 :         while (queue_->DequeueRequest(&req_entry)) {
     209     1166287 :             req_entry->tpart->Process(req_entry->client, &req_entry->request);
     210     1166269 :             delete req_entry;
     211     1166287 :             if (++count == kMaxIterations) {
     212        4574 :                 return false;
     213             :             }
     214             :         }
     215             : 
     216             :         while (true) {
     217     2537193 :             DBTablePartBase *tpart = queue_->GetActiveTable();
     218     2537377 :             if (tpart == NULL) {
     219      848613 :                 break;
     220             :             }
     221     1688764 :             bool done = tpart->RunNotify();
     222     1688608 :             if (!done) {
     223          58 :                 return false;
     224             :             }
     225     1688550 :         }
     226             : 
     227             :         // Running is done only if queue_ is empty. It's possible that more
     228             :         // entries are added into in the input or remove queues during the
     229             :         // time we were processing those queues.
     230      848613 :         return queue_->RunnerDone();
     231             :     }
     232             : 
     233           0 :     std::string Description() const {
     234           0 :         return "DBPartition QueueRunner";
     235             :     }
     236             : private:
     237             :     WorkQueue *queue_;
     238             : };
     239             : 
     240     3379707 : void DBPartition::WorkQueue::MaybeStartRunnerUnlocked() {
     241     3379707 :     if (running_) {
     242     2578250 :         return;
     243             :     }
     244      801457 :     running_ = true;
     245      801457 :     QueueRunner *runner = new QueueRunner(this);
     246      801554 :     TaskScheduler *scheduler = TaskScheduler::GetInstance();
     247      801539 :     scheduler->Enqueue(runner);
     248             : }
     249             : 
     250     1691148 : void DBPartition::WorkQueue::MaybeStartRunner() {
     251     1691148 :     std::scoped_lock lock(mutex_);
     252     1691234 :     MaybeStartRunnerUnlocked();
     253     1691254 : }
     254             : 
     255      849739 : bool DBPartition::WorkQueue::RunnerDone() {
     256      849739 :     std::scoped_lock lock(mutex_);
     257      849773 :     if (disable_ || (request_queue_.empty() && remove_queue_.empty())) {
     258      801605 :         running_ = false;
     259      801605 :         return true;
     260             :     }
     261             : 
     262       48102 :     running_ = true;
     263       48102 :     return false;
     264      849707 : }
     265             : 
     266      109374 : DBPartition::DBPartition(DB *db, int partition_id)
     267      109374 :     : db_(db), work_queue_(new WorkQueue(this, partition_id)) {
     268      109374 : }
     269             : 
     270             : // The DBPartition destructor needs to be defined after WorkQueue has
     271             : // been declared.
     272      109310 : DBPartition::~DBPartition() {
     273      109310 : }
     274             : 
     275     1166272 : bool DBPartition::EnqueueRequest(DBTablePartBase *tpart, DBClient *client,
     276             :                                  DBRequest *req) {
     277     1166272 :     RequestQueueEntry *entry = new RequestQueueEntry(tpart, client, req);
     278     1166220 :     return work_queue_->EnqueueRequest(entry);
     279             : }
     280             : 
     281      524715 : void DBPartition::EnqueueRemove(DBTablePartBase *tpart, DBEntryBase *db_entry) {
     282      524715 :     RemoveQueueEntry *entry = new RemoveQueueEntry(tpart, db_entry);
     283      524740 :     db_entry->SetOnRemoveQ();
     284      524924 :     work_queue_->EnqueueRemove(entry);
     285      524867 : }
     286             : 
     287             : // concurrency: called from DBPartition task.
     288     1688729 : void DBPartition::OnTableChange(DBTablePartBase *tablepart) {
     289     1688729 :     work_queue_->SetActive(tablepart);
     290     1688897 : }
     291             : 
     292           0 : long DBPartition::request_queue_len() const {
     293           0 :     return work_queue_->request_queue_len();
     294             : }
     295             : 
     296           0 : uint64_t DBPartition::total_request_count() const {
     297           0 :     return work_queue_->total_request_count();
     298             : }
     299             : 
     300           0 : uint64_t DBPartition::max_request_queue_len() const {
     301           0 :     return work_queue_->max_request_queue_len();
     302             : }
     303             : 
     304      801576 : int DBPartition::task_id() const {
     305      801576 :     return db_->task_id();
     306             : }