Line data    Source code

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