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             : // queue_task.h
       7             : //
       8             : // Task based queue processor implementing thread safe enqueue and dequeue
       9             : // using concurrent queues. If queue is empty, enqueue creates a dequeue task
      10             : // that drains the queue. The dequeue task runs a maximum of kMaxIterations
      11             : // before yielding.
      12             : //
      13             : #ifndef __QUEUE_TASK_H__
      14             : #define __QUEUE_TASK_H__
      15             : 
      16             : #include <iostream>
      17             : #include <sstream>
      18             : #include <algorithm>
      19             : #include <vector>
      20             : #include <set>
      21             : #include <mutex>
      22             : #include <atomic>
      23             : 
      24             : #include <tbb/concurrent_queue.h>
      25             : #include <tbb/spin_rw_mutex.h>
      26             : 
      27             : #include <base/task.h>
      28             : #include <base/time_util.h>
      29             : #include <base/watermark.h>
      30             : 
      31             : template <typename QueueEntryT, typename QueueT>
      32             : class QueueTaskRunner : public Task {
      33             : public:
      34     1873117 :     QueueTaskRunner(QueueT *queue)
      35     1873117 :         : Task(queue->GetTaskId(), queue->GetTaskInstance()), queue_(queue) {
      36     1873097 :     }
      37             : 
      38     2481326 :     bool Run() {
      39             :         // Check if this run needs to be deferred
      40     2481326 :         if (!queue_->OnEntry()) {
      41       18690 :             return false;
      42             :         }
      43     2462676 :         return RunQueue();
      44             :         // No more client callbacks after updating
      45             :         // queue running_ and current_runner_ in RunQueue to
      46             :         // avoid client callbacks running concurrently
      47             :     }
      48             : 
      49           0 :     virtual std::string Description() const {
      50           0 :         return queue_->Description();
      51             :     }
      52             : 
      53             : private:
      54     2462650 :     bool RunQueue() {
      55             :         // Check if we need to abort
      56     2462650 :         if (queue_->RunnerAbort()) {
      57          19 :             return queue_->RunnerDone();
      58             :         }
      59             : 
      60     2462756 :         uint64_t start = 0;
      61     2462756 :         if (queue_->measure_busy_time_)
      62           0 :             start = ClockMonotonicUsec();
      63             : 
      64     2462752 :         QueueEntryT entry = QueueEntryT();
      65     2462709 :         size_t count = 0;
      66     9691695 :         while (queue_->Dequeue(&entry)) {
      67             :             // Process the entry
      68     7825200 :             if (!queue_->GetCallback()(entry)) {
      69      542957 :                 break;
      70             :             }
      71     7282344 :             if (++count == queue_->max_iterations_) {
      72       53358 :                 if (start)
      73           0 :                     queue_->add_busy_time(ClockMonotonicUsec() - start);
      74       53358 :                 return queue_->RunnerDone();
      75             :             }
      76             :         }
      77             : 
      78     2409306 :         if (start)
      79           0 :             queue_->add_busy_time(ClockMonotonicUsec() - start);
      80             : 
      81             :         // Running is done if queue_ is empty
      82             :         // While notification is being run, its possible that more entries
      83             :         // are added into queue_
      84     2409306 :         return queue_->RunnerDone();
      85     1370339 :     }
      86             : 
      87             :     QueueT *queue_;
      88             : };
      89             : 
      90             : template <typename QueueEntryT>
      91             : struct WorkQueueDelete {
      92             :     template <typename QueueT>
      93       56415 :     void operator()(QueueT &, bool) {}
      94             : };
      95             : 
      96             : template <typename QueueEntryT>
      97             : struct WorkQueueDelete<QueueEntryT *> {
      98             :     template <typename QueueT>
      99      162982 :     void operator()(QueueT &q, bool delete_entry) {
     100             :         QueueEntryT *entry;
     101      162982 :         while (q.try_pop(entry)) {
     102           0 :             if (delete_entry) {
     103           0 :                 delete entry;
     104             :             }
     105             :         }
     106      162982 :     }
     107             : };
     108             : 
     109             : template <typename QueueEntryT>
     110             : class WorkQueue {
     111             : public:
     112             :     static const int kMaxSize = 1024;
     113             :     static const int kMaxIterations = 32;
     114             :     typedef tbb::concurrent_queue<QueueEntryT> Queue;
     115             :     typedef boost::function<bool (QueueEntryT)> Callback;
     116             :     typedef boost::function<bool (void)> StartRunnerFunc;
     117             :     typedef boost::function<void (bool)> TaskExitCallback;
     118             :     typedef boost::function<bool ()> TaskEntryCallback;
     119             : 
     120      240427 :     WorkQueue(int taskId, int taskInstance, Callback callback,
     121             :               size_t size = kMaxSize,
     122             :               size_t max_iterations = kMaxIterations) :
     123      240425 :         running_(false),
     124      240425 :         taskId_(taskId),
     125      240425 :         taskInstance_(taskInstance),
     126      240425 :         name_(""),
     127      240425 :         callback_(callback),
     128      240425 :         on_entry_cb_(0),
     129      240425 :         on_exit_cb_(0),
     130      240425 :         start_runner_(0),
     131      240425 :         current_runner_(NULL),
     132      240425 :         on_entry_defer_count_(0),
     133      240425 :         deleted_(false),
     134      240425 :         enqueues_(0),
     135      240425 :         dequeues_(0),
     136      240425 :         drops_(0),
     137      240425 :         max_iterations_(max_iterations),
     138      240425 :         size_(size),
     139      240425 :         bounded_(false),
     140      240425 :         shutdown_scheduled_(false),
     141      240425 :         delete_entries_on_shutdown_(true),
     142      240425 :         task_starts_(0),
     143      240425 :         max_queue_len_(0),
     144      240425 :         busy_time_(0),
     145      480852 :         measure_busy_time_(false) {
     146      240425 :         count_ = 0;
     147      240427 :         disabled_ = false;
     148      240427 :     }
     149             : 
     150             :     // Concurrency - should be called from a task whose policy
     151             :     // assures that the dequeue task - QueueTaskRunner is not running
     152             :     // concurrently
     153      219621 :     void Shutdown(bool delete_entries = true) {
     154      219621 :         std::scoped_lock lock(mutex_);
     155      219621 :         ShutdownLocked(delete_entries);
     156      219621 :     }
     157             : 
     158             :     // Concurrency - can be called from any context
     159             :     // Schedule shutdown of the WorkQueue, shutdown may happen asynchronously
     160             :     // or in the caller's context also
     161           5 :     void ScheduleShutdown(bool delete_entries = true) {
     162           5 :         std::scoped_lock lock(mutex_);
     163           5 :         if (shutdown_scheduled_) {
     164           0 :             return;
     165             :         }
     166           5 :         shutdown_scheduled_ = true;
     167           5 :         delete_entries_on_shutdown_ = delete_entries;
     168             : 
     169             :         // Cancel QueueTaskRunner
     170           5 :         if (running_) {
     171           4 :             assert(current_runner_);
     172           4 :             TaskScheduler *scheduler = TaskScheduler::GetInstance();
     173             :             TaskScheduler::CancelReturnCode cancel_code =
     174           4 :                 scheduler->Cancel(current_runner_);
     175           4 :             if (cancel_code == TaskScheduler::CANCELLED) {
     176           1 :                 running_ = false;
     177           1 :                 current_runner_ = NULL;
     178           1 :                 ShutdownLocked(delete_entries);
     179             :             } else {
     180           3 :                 assert(cancel_code == TaskScheduler::QUEUED);
     181             :             }
     182             :         } else {
     183           1 :             ShutdownLocked(delete_entries);
     184             :         }
     185           5 :     }
     186             : 
     187      240399 :     ~WorkQueue() {
     188      240399 :         std::scoped_lock lock(mutex_);
     189             :         // Shutdown() needs to be called before deleting
     190             :         //assert(!running_ && deleted_);
     191      240399 :     }
     192             : 
     193         245 :     void SetStartRunnerFunc(StartRunnerFunc start_runner_fn) {
     194         245 :         start_runner_ = start_runner_fn;
     195         245 :     }
     196             : 
     197          21 :     void SetSize(size_t size) {
     198          21 :         size_ = size;
     199          21 :     }
     200             : 
     201          27 :     void SetBounded(bool bounded) {
     202          27 :         bounded_ = bounded;
     203          27 :     }
     204             : 
     205             :     bool GetBounded() const {
     206             :         return bounded_;
     207             :     }
     208             : 
     209           1 :     void SetHighWaterMark(const WaterMarkInfos &high_water) {
     210           1 :         std::scoped_lock lock(water_mutex_);
     211           1 :         watermarks_.SetHighWaterMark(high_water);
     212           1 :     }
     213             : 
     214         435 :     void SetHighWaterMark(const WaterMarkInfo& hwm_info) {
     215         435 :         std::scoped_lock lock(water_mutex_);
     216         435 :         watermarks_.SetHighWaterMark(hwm_info);
     217         435 :     }
     218             : 
     219      219627 :     void ResetHighWaterMark() {
     220      219627 :         std::scoped_lock lock(water_mutex_);
     221      219627 :         watermarks_.ResetHighWaterMark();
     222      219627 :     }
     223             : 
     224           1 :     WaterMarkInfos GetHighWaterMark() const {
     225           1 :         std::scoped_lock lock(water_mutex_);
     226           2 :         return watermarks_.GetHighWaterMark();
     227           1 :     }
     228             : 
     229           4 :     void SetLowWaterMark(const WaterMarkInfos &low_water) {
     230           4 :         std::scoped_lock lock(water_mutex_);
     231           4 :         watermarks_.SetLowWaterMark(low_water);
     232           4 :      }
     233             : 
     234         420 :     void SetLowWaterMark(const WaterMarkInfo& lwm_info) {
     235         420 :         std::scoped_lock lock(water_mutex_);
     236         420 :         watermarks_.SetLowWaterMark(lwm_info);
     237         420 :      }
     238             : 
     239      219626 :     void ResetLowWaterMark() {
     240      219626 :         std::scoped_lock lock(water_mutex_);
     241      219626 :         watermarks_.ResetLowWaterMark();
     242      219626 :     }
     243             : 
     244           1 :     WaterMarkInfos GetLowWaterMark() const {
     245           1 :         std::scoped_lock lock(water_mutex_);
     246           2 :         return watermarks_.GetLowWaterMark();
     247           1 :     }
     248             : 
     249     7819594 :     bool Enqueue(QueueEntryT entry) {
     250     7819594 :         if (bounded_) {
     251        2205 :             if (AreWaterMarksSet()) {
     252           0 :                 return EnqueueBoundedLocked(entry);
     253             :             } else {
     254        2205 :                 return EnqueueBounded(entry);
     255             :             }
     256             :         } else {
     257     7817389 :             if (AreWaterMarksSet()) {
     258      113012 :                 return EnqueueInternalLocked(entry);
     259             :             } else {
     260     7704288 :                 return EnqueueInternal(entry);
     261             :             }
     262             :         }
     263             :     }
     264             : 
     265             :     // Returns true if pop is successful.
     266     9691588 :     bool Dequeue(QueueEntryT *entry) {
     267     9691588 :         if (AreWaterMarksSet()) {
     268      119092 :             return DequeueInternalLocked(entry);
     269             :         } else {
     270     9572372 :             return DequeueInternal(entry);
     271             :         }
     272             :     }
     273             : 
     274     1873103 :     int GetTaskId() const {
     275     1873103 :         return taskId_;
     276             :     }
     277             : 
     278     1873111 :     int GetTaskInstance() const {
     279     1873111 :         return taskInstance_;
     280             :     }
     281             : 
     282     7843007 :     void MayBeStartRunner() {
     283     7843007 :         std::scoped_lock lock(mutex_);
     284     7843124 :         if (running_ || queue_.empty() || deleted_ || RunnerAbortLocked()) {
     285     5969796 :             return;
     286             :         }
     287     1873076 :         task_starts_++;
     288     1873076 :         running_ = true;
     289     1873076 :         assert(current_runner_ == NULL);
     290     1873090 :         current_runner_ =
     291     1873076 :             new QueueTaskRunner<QueueEntryT, WorkQueue<QueueEntryT> >(this);
     292     1873090 :         TaskScheduler *scheduler = TaskScheduler::GetInstance();
     293     1873080 :         scheduler->Enqueue(current_runner_);
     294     7843208 :     }
     295             : 
     296     7825176 :     Callback GetCallback() const {
     297     7825176 :         return callback_;
     298             :     }
     299             : 
     300        2090 :     void SetEntryCallback(TaskEntryCallback on_entry) {
     301        2090 :         on_entry_cb_ = on_entry;
     302        2090 :     }
     303             : 
     304       36072 :     void SetExitCallback(TaskExitCallback on_exit) {
     305       36072 :         on_exit_cb_ = on_exit;
     306       36072 :     }
     307             : 
     308       17525 :     void set_name(const std::string &name) {
     309       17525 :         name_ = name;
     310       17525 :     }
     311          40 :     std::string Description() const {
     312          40 :         if (name_.empty() == false)
     313          40 :             return name_;
     314             : 
     315           0 :         std::ostringstream str;
     316           0 :         str << "Function " << callback_;
     317           0 :         return str.str();
     318           0 :     }
     319             : 
     320       34560 :     void set_disable(bool disabled) {
     321       34560 :         if (disabled_ != disabled) {
     322       34560 :             disabled_ = disabled;
     323       34560 :             if (!disabled_) {
     324       17280 :                 MayBeStartRunner();
     325             :             }
     326             :         }
     327       34560 :     }
     328             : 
     329             :     bool IsDisabled() const {
     330             :         return disabled_;
     331             :     }
     332             : 
     333             :     size_t on_entry_defer_count() const {
     334             :         return on_entry_defer_count_;
     335             :     }
     336             : 
     337     2481363 :     bool OnEntry() {
     338     2481363 :         bool run = (on_entry_cb_.empty() || on_entry_cb_());
     339             : 
     340             :         // Track number of times this queue run is deferred
     341     2481373 :         if (!run) {
     342       18690 :             on_entry_defer_count_++;
     343             :         }
     344     2481373 :         return run;
     345             :     }
     346             : 
     347     2462708 :     void OnExit(bool done) {
     348     2462708 :         if (!on_exit_cb_.empty()) {
     349       21035 :             on_exit_cb_(done);
     350             :         }
     351     2462706 :     }
     352             : 
     353     8455791 :     bool IsQueueEmpty() const {
     354     8455791 :         return queue_.empty();
     355             :     }
     356             : 
     357        7999 :     size_t Length() const {
     358        7999 :         return count_;
     359             :     }
     360             : 
     361          65 :     size_t NumEnqueues() const {
     362          65 :         return enqueues_;
     363             :     }
     364             : 
     365          66 :     size_t NumDequeues() const {
     366          66 :         return dequeues_;
     367             :     }
     368             : 
     369             :     size_t NumDrops() const {
     370             :         return drops_;
     371             :     }
     372             : 
     373           6 :     bool deleted() const {
     374           6 :         return deleted_;
     375             :     }
     376             : 
     377          45 :     uint32_t task_starts() const { return task_starts_; }
     378          61 :     size_t max_queue_len() const { return max_queue_len_; }
     379             :     bool measure_busy_time() const { return measure_busy_time_; }
     380          73 :     void set_measure_busy_time(bool val) const { measure_busy_time_ = val; }
     381          45 :     uint64_t busy_time() const { return busy_time_; }
     382           0 :     void add_busy_time(uint64_t t) { busy_time_ += t; }
     383           0 :     void ClearStats() const {
     384           0 :         max_queue_len_ = 0;
     385           0 :         enqueues_ = 0;
     386           0 :         dequeues_ = 0;
     387           0 :         busy_time_ = 0;
     388           0 :         task_starts_ = 0;
     389           0 :     }
     390             : private:
     391             :     // Returns true if pop is successful.
     392     9691441 :     bool DequeueInternal(QueueEntryT *entry) {
     393     9691441 :         bool success = queue_.try_pop(*entry);
     394     9691669 :         if (success) {
     395     7825281 :             dequeues_++;
     396     7825281 :             size_t ncount(AtomicDecrementQueueCount(entry));
     397     7825351 :             ProcessLowWaterMarks(ncount);
     398             :         }
     399     9691591 :         return success;
     400             :     }
     401             : 
     402      119092 :     bool DequeueInternalLocked(QueueEntryT *entry) {
     403      119092 :         std::scoped_lock lock(water_mutex_);
     404      238184 :         return DequeueInternal(entry);
     405      119092 :     }
     406             : 
     407    17517059 :     bool AreWaterMarksSet() const {
     408    17517059 :         return watermarks_.AreWaterMarksSet();
     409             :     }
     410             : 
     411      219626 :     void ShutdownLocked(bool delete_entries) {
     412             :         // Cancel QueueTaskRunner from the scheduler
     413      219626 :         assert(!deleted_);
     414      219626 :         if (running_) {
     415         427 :             running_ = false;
     416         427 :             assert(current_runner_);
     417         427 :             TaskScheduler *scheduler = TaskScheduler::GetInstance();
     418             :             TaskScheduler::CancelReturnCode cancel_code =
     419         427 :                 scheduler->Cancel(current_runner_);
     420         427 :             assert(cancel_code == TaskScheduler::CANCELLED);
     421         427 :             current_runner_ = NULL;
     422             :         }
     423      219626 :         ResetHighWaterMark();
     424      219626 :         ResetLowWaterMark();
     425             :         WorkQueueDelete<QueueEntryT> deleter;
     426      219626 :         deleter(queue_, delete_entries);
     427      219626 :         queue_.clear();
     428      219626 :         count_ = 0;
     429      219626 :         deleted_ = true;
     430      219626 :     }
     431             : 
     432     7793947 :     size_t AtomicIncrementQueueCount(QueueEntryT *entry) {
     433    15587894 :         return count_.fetch_add(1) + 1;
     434             :     }
     435             : 
     436     7799656 :     size_t AtomicDecrementQueueCount(QueueEntryT *entry) {
     437    15599312 :         return count_.fetch_sub(1) - 1;
     438             :     }
     439             : 
     440     7819785 :     void ProcessHighWaterMarks(size_t count) {
     441     7819785 :         watermarks_.ProcessHighWaterMarks(count);
     442     7819640 :     }
     443             : 
     444     7825331 :     void ProcessLowWaterMarks(size_t count) {
     445     7825331 :         watermarks_.ProcessLowWaterMarks(count);
     446     7825219 :     }
     447             : 
     448     7817353 :     bool EnqueueInternal(QueueEntryT entry) {
     449     7817353 :         enqueues_++;
     450     7817353 :         size_t ncount(AtomicIncrementQueueCount(&entry));
     451     7817606 :         if (ncount > max_queue_len_)
     452      876595 :             max_queue_len_ = ncount;
     453     7817606 :         ProcessHighWaterMarks(ncount);
     454     7817439 :         queue_.push(entry);
     455     7817452 :         MayBeStartRunner();
     456     7817622 :         return ncount < size_;
     457             :     }
     458             : 
     459      113012 :     bool EnqueueInternalLocked(QueueEntryT entry) {
     460      113012 :         std::scoped_lock lock(water_mutex_);
     461      226024 :         return EnqueueInternal(entry);
     462      113012 :     }
     463             : 
     464        2205 :     bool EnqueueBounded(QueueEntryT entry) {
     465        2205 :         size_t ncount(AtomicIncrementQueueCount(&entry));
     466        2205 :         if (ncount > max_queue_len_)
     467         108 :             max_queue_len_ = ncount;
     468        2205 :         if (ncount < size_) {
     469        2200 :             enqueues_++;
     470        2200 :             ProcessHighWaterMarks(ncount);
     471        2200 :             queue_.push(entry);
     472        2200 :             MayBeStartRunner();
     473        2200 :             return true;
     474             :         }
     475           5 :         AtomicDecrementQueueCount(&entry);
     476           5 :         drops_++;
     477           5 :         max_queue_len_ = count_;
     478           5 :         return false;
     479             :     }
     480             : 
     481           0 :     bool EnqueueBoundedLocked(QueueEntryT entry) {
     482           0 :         std::scoped_lock lock(water_mutex_);
     483           0 :         return EnqueueBounded(entry);
     484           0 :     }
     485             : 
     486     4958558 :     bool RunnerAbortLocked() {
     487     9884227 :         return (disabled_ || shutdown_scheduled_ ||
     488     9884130 :                 (!start_runner_.empty() && !start_runner_()));
     489             :     }
     490             : 
     491     2462648 :     bool RunnerAbort() {
     492     2462648 :         std::scoped_lock lock(mutex_);
     493     4925553 :         return RunnerAbortLocked();
     494     2462711 :     }
     495             : 
     496     2462783 :     bool RunnerDone() {
     497     2462783 :         std::scoped_lock lock(mutex_);
     498     2462785 :         bool done = false;
     499     2462785 :         if (queue_.empty() || RunnerAbortLocked()) {
     500     1872881 :             done = true;
     501     1872881 :             OnExit(done);
     502     1872878 :             current_runner_ = NULL;
     503     1872878 :             running_ = false;
     504     1872878 :             if (shutdown_scheduled_) {
     505           3 :                 ShutdownLocked(delete_entries_on_shutdown_);
     506             :             }
     507             :         } else {
     508      589836 :             OnExit(done);
     509      589833 :             running_ = true;
     510             :         }
     511     2462789 :         return done;
     512     2462711 :     }
     513             : 
     514             :     Queue queue_;
     515             :     std::atomic<size_t> count_;
     516             :     std::mutex mutex_;
     517             :     bool running_;
     518             :     int taskId_;
     519             :     int taskInstance_;
     520             :     std::string name_;
     521             :     Callback callback_;
     522             :     TaskEntryCallback on_entry_cb_;
     523             :     TaskExitCallback on_exit_cb_;
     524             :     StartRunnerFunc start_runner_;
     525             :     QueueTaskRunner<QueueEntryT, WorkQueue<QueueEntryT> > *current_runner_;
     526             :     size_t on_entry_defer_count_;
     527             :     std::atomic<bool> disabled_;
     528             :     bool deleted_;
     529             :     mutable size_t enqueues_;
     530             :     mutable size_t dequeues_;
     531             :     size_t drops_;
     532             :     size_t max_iterations_;
     533             :     size_t size_;
     534             :     bool bounded_;
     535             :     bool shutdown_scheduled_;
     536             :     bool delete_entries_on_shutdown_;
     537             :     WaterMarkTuple watermarks_;
     538             :     mutable std::mutex water_mutex_;
     539             :     mutable uint32_t task_starts_;
     540             :     mutable size_t max_queue_len_;
     541             :     mutable uint64_t busy_time_;
     542             :     mutable bool measure_busy_time_;
     543             : 
     544             :     friend class QueueTaskTest;
     545             :     friend class QueueTaskShutdownTest;
     546             :     friend class QueueTaskWaterMarkTest;
     547             :     friend class QueueTaskRunner<QueueEntryT, WorkQueue<QueueEntryT> >;
     548             : 
     549             :     DISALLOW_COPY_AND_ASSIGN(WorkQueue);
     550             : };
     551             : 
     552             : #endif /* __QUEUE_TASK_H__ */