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             : //
       7             : // request_pipeline.cc
       8             : //
       9             : // Implementation of RequestPipeline
      10             : // 
      11             : // - The PipeImpl class implements the Pipeline
      12             : //   It contains multiple stages inside it.
      13             : //
      14             : // - The StageImpl class hold a given stage of the Pipeline
      15             : //   It contains multiple instances inside it.
      16             : //
      17             : // - The StageWorker class is a Task which runs the Client's callback functions
      18             : //
      19             : 
      20             : #include <boost/utility.hpp>
      21             : #include <atomic>
      22             : #include <utility>
      23             : #include "base/logging.h"
      24             : #include "base/task.h"
      25             : #include <queue>
      26             : #include <boost/assign/list_of.hpp>
      27             : 
      28             : #include "sandesh/sandesh_types.h"
      29             : #include "sandesh.h"
      30             : #include "request_pipeline.h"
      31             : 
      32             : using namespace std;
      33             : 
      34        1928 : RequestPipeline::PipeSpec::PipeSpec(const SandeshRequest * sr) {
      35        1928 :     snhRequest_ = sr->SharedPtr();
      36        1928 : }
      37             : 
      38             : class RequestPipeline::PipeImpl {
      39             : public:
      40             :     PipeImpl(const PipeSpec& spec);
      41        1928 :     ~PipeImpl() {}
      42             :     
      43             :     const StageData* GetStageData(int stage) const;
      44             : 
      45             :     bool RunInstance(int instNum);
      46             : 
      47             :     void DoneInstance(void);
      48             :     
      49             : private:
      50             :     bool NextStage(void);
      51             : 
      52             :     const PipeSpec spec_;
      53             :     boost::ptr_vector<StageImpl> stageImpls_;
      54             :     int currentStage_;
      55             : 
      56             :     static std::mutex mutex_;
      57             :     static int activePipes_;
      58             :     static queue<PipeImpl*> pendPipes_;
      59             : };
      60             : 
      61             : class RequestPipeline::StageImpl {
      62             : public:
      63             :     StageImpl(const StageSpec& spec, int stage);
      64        2159 :     ~StageImpl() {}
      65             : 
      66             :     // Decrement the number of outstanding instances.
      67             :     // Returns "true" when there are no instances left.
      68        2933 :     bool DecrInst() {
      69        2933 :         int prev = remainingInst_.fetch_sub(1);
      70        2933 :         return (prev == 1 ? true : false);
      71             :     }
      72             : 
      73             :     StageData data_;
      74             :     std::atomic<int> remainingInst_;
      75             : };
      76             : 
      77             : class RequestPipeline::StageWorker : public Task {
      78             : public:
      79        2935 :     StageWorker(PipeImpl& pImpl, int taskId, int instId, int instNum) :
      80        2935 :         Task(taskId, instId) , pImpl_(pImpl), instNum_(instNum) {}
      81             : 
      82        5018 :     virtual bool Run() {
      83             :        
      84        5018 :         if (!pImpl_.RunInstance(instNum_)) return false;
      85             : 
      86        2934 :         pImpl_.DoneInstance();
      87             : 
      88        2935 :         return true;
      89             :     }
      90           0 :     std::string Description() const { return "RequestPipeline::StageWorker"; }
      91             : private:
      92             :     PipeImpl& pImpl_;
      93             :     const int instNum_;
      94             : };
      95             : 
      96             : std::mutex  RequestPipeline::PipeImpl::mutex_;
      97             : int RequestPipeline::PipeImpl::activePipes_ = 0;
      98             : queue<RequestPipeline::PipeImpl*> RequestPipeline::PipeImpl::pendPipes_;
      99             : 
     100             : // This is the interface for the client callback function to look into
     101             : // the Client Data generated during earlier stages of the pipeline.
     102             : //
     103             : // Returns the StageData for a previous stage of the pipeline
     104             : const RequestPipeline::StageData*
     105         233 : RequestPipeline::PipeSpec::GetStageData(int stage) const {
     106         233 :     if (impl_)
     107         233 :         return impl_->GetStageData(stage);
     108             :     else
     109           0 :         return NULL;
     110             : }
     111             : 
     112             : // Constructor of PipeImpl.
     113             : // If there are no active Pipelines, start it's first stage.
     114             : // If there are active Pipelines, queue this up for later
     115        1928 : RequestPipeline::PipeImpl::PipeImpl(const PipeSpec &spec) :
     116        1928 :         spec_(spec, this), currentStage_(-1) {
     117             : 
     118        1928 :     std::scoped_lock lock(mutex_);
     119        1928 :     if (activePipes_ < 1) {
     120        1921 :         activePipes_++;
     121        1921 :         NextStage();
     122             :     }
     123             :     else {
     124           7 :         pendPipes_.push(this);
     125             :     }
     126        1928 : }
     127             : 
     128             : // This function moves the Pipeline to it's next stage.
     129             : // If we are the last stage,
     130             : //  - delete the pipeline 
     131             : //  - release the Sandesh
     132             : //  - if any Pipelines are queue up, start one pipeline from the queue
     133             : //
     134             : //  Return true if we moved to the next stage of the given pipeline.
     135             : //         false if reached the last stage.
     136             : bool
     137        4087 : RequestPipeline::PipeImpl::NextStage(void) {
     138        4087 :     currentStage_++;
     139        4087 :     if (currentStage_ == static_cast<int>(spec_.stages_.size())) {
     140        1928 :         delete this;
     141             :         {
     142        1928 :             std::scoped_lock lock(mutex_);
     143        1928 :             if (!pendPipes_.empty()) {
     144           7 :                 PipeImpl * pipe = pendPipes_.front();
     145           7 :                 pendPipes_.pop();
     146           7 :                 pipe->NextStage();
     147             :             } else {
     148        1921 :                 activePipes_--;
     149             :             }
     150        1928 :         }
     151        1928 :         return false;
     152             :     }
     153             :     else {
     154        2159 :         stageImpls_.push_back(new StageImpl(
     155        2159 :             spec_.stages_[currentStage_],currentStage_));
     156             :         
     157        2159 :         const StageSpec& stageSpec = spec_.stages_[currentStage_];
     158        2159 :         int insts = stageSpec.instances_.size();
     159        5094 :         for (int i=0; i < insts; i++) {
     160        2935 :             StageWorker * sw = new StageWorker(*this, stageSpec.taskId_,
     161        2935 :                                                stageSpec.instances_[i], i);
     162        2935 :             TaskScheduler *scheduler = TaskScheduler::GetInstance();
     163        2935 :             scheduler->Enqueue(sw);
     164             :         } 
     165        2159 :         return true;
     166             :     }
     167             : }
     168             : 
     169             : // The StageWorker calls this function when an Instance completes.
     170             : // If all instances of the current stage have completed execution,
     171             : // we can move to the next stage of the pipeline.
     172             : void
     173        2934 : RequestPipeline::PipeImpl::DoneInstance(void) {
     174        2934 :     if (stageImpls_[currentStage_].DecrInst()) {
     175        2159 :         NextStage();
     176             :     }
     177        2935 : }
     178             : 
     179             : // The StageWorker calls this function to drive execution
     180             : // of the client callback function.
     181             : //
     182             : // Returns true if this instance's execution is complete
     183             : //         false if the callback needs to be scheduled again.
     184             : bool
     185        5018 : RequestPipeline::PipeImpl::RunInstance(int instNum) {
     186        5018 :     const StageSpec &ss = spec_.stages_[currentStage_];
     187        9840 :     return ss.cbFn_(spec_.snhRequest_.get(), spec_,
     188        5014 :                 currentStage_, instNum, (ss.allocFn_.empty() ?
     189        9845 :                     NULL : &(stageImpls_[currentStage_].data_[instNum])));
     190             : }
     191             : 
     192             : // This function allows the client callback function to look into
     193             : // the Client Data generated during earlier stages of the pipeline.
     194             : //
     195             : // Returns the StageData for a previous stage of the pipeline
     196             : const RequestPipeline::StageData*
     197         233 : RequestPipeline::PipeImpl::GetStageData(int stage) const {
     198         233 :     if (stage >= currentStage_)
     199           0 :         return NULL;
     200             :     else
     201         233 :         return &(stageImpls_[stage].data_); 
     202             : }
     203             : 
     204             : // Contructor for StageImpl
     205             : // Creates the StageWorker for each Instance of this Stage
     206             : // Also Creates the Client Data for each Instance.
     207        2159 : RequestPipeline::StageImpl::StageImpl(const StageSpec& spec, int stage) {
     208        2159 :     remainingInst_ = spec.instances_.size(); 
     209        5094 :     for (int i=0; i<remainingInst_; i++) {
     210        2935 :         if (!spec.allocFn_.empty()) data_.push_back(spec.allocFn_(stage));
     211             :     }
     212        2159 : }
     213             : 
     214        1928 : RequestPipeline::RequestPipeline(const PipeSpec& spec) {
     215        1928 :     impl_ = new PipeImpl(spec);
     216        1928 : }
     217             :