Line data    Source code

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