Line data    Source code

       1             : /*
       2             :  * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
       3             :  */
       4             : 
       5             : #ifndef __WORK_PROCESSOR_INL_H__
       6             : #define __WORK_PROCESSOR_INL_H__
       7             : 
       8             : #include <vector>
       9             : #include <boost/function.hpp>
      10             : #include <boost/bind/bind.hpp>
      11             : #include <boost/shared_ptr.hpp>
      12             : #include <boost/scoped_ptr.hpp>
      13             : #include <boost/assign/list_of.hpp>
      14             : #include <boost/type_traits.hpp>
      15             : #include <boost/mpl/assert.hpp>
      16             : #include <boost/tuple/tuple.hpp>
      17             : #include <limits>
      18             : #include <sstream>
      19             : #include "base/task.h"
      20             : 
      21             : using namespace boost::placeholders;
      22             : 
      23             : struct PipelineWorker : public Task {
      24      222668 :     PipelineWorker(int tid, int tinst, boost::function<bool(void)> runner) :
      25      222668 :         Task(tid,tinst), runner_(runner) {}
      26             : 
      27      221883 :     bool Run() {
      28      221883 :         if (!(runner_)()) return false;
      29      221484 :         return true;
      30             :     }
      31           0 :     std::string Description() const { return "PipelineWorker"; }
      32             : private:
      33             :     const boost::function<bool(void)> runner_;
      34             : };
      35             : 
      36             : template<typename InputT, typename SubResultT, typename ExternalT>
      37             : class WorkProcessor : public ExternalProcIf<ExternalT> {
      38             : public:
      39             :     typedef boost::function<ExternalBase::Efn(
      40             :         uint32_t inst,
      41             :         const std::vector<ExternalT*> & exts, // Info for previous steps of this stage
      42             :         const InputT & inp, // Info from previous stage
      43             :         SubResultT & subRes // Access to final result of this instance
      44             :         )> ExecuteFn;
      45             : 
      46      110124 :     WorkProcessor(uint32_t stage, ExecuteFn efn, FinFn finFn, const InputT & inp,
      47             :         uint32_t inst, int tid, int tinst) :
      48      110142 :             stage_(stage),
      49      110142 :             finFn_(finFn),
      50      110108 :             efn_(efn),
      51      110074 :             inp_(inp),
      52      110051 :             finished_(false),
      53      110051 :             running_(false),
      54      110051 :             inst_(inst),
      55      110051 :             tid_(tid),
      56      110051 :             tinst_(tinst),
      57      110051 :             w_(new PipelineWorker(tid_, tinst_,
      58             :                 boost::bind(&WorkProcessor<InputT,SubResultT,ExternalT>::Runner,
      59      220198 :                 this))) {
      60      109991 :         res_.reset(new SubResultT);
      61      109959 :     }
      62             : 
      63      110377 :     void Start() {
      64      110377 :         TaskScheduler *scheduler = TaskScheduler::GetInstance();
      65      110374 :         scheduler->Enqueue(w_);
      66      110367 :     }   
      67             : 
      68      109512 :     boost::shared_ptr<SubResultT> Result() const {
      69      109512 :         if (!finished_) return boost::shared_ptr<SubResultT>();
      70      109512 :         return res_;
      71             :     }
      72             : 
      73      108914 :     void Release() {
      74      108914 :         assert(finished_);
      75      108914 :         res_.reset();
      76      109934 :     }
      77             :     
      78         254 :     std::string Key() const {
      79         254 :         std::stringstream keystr;
      80         254 :         keystr << "PROC-STAGE:" << stage_ << "-INST:" << inst_ <<
      81         254 :                 "-STEP:" << externals_.size();
      82         508 :         return keystr.str();
      83         254 :     }
      84             : 
      85             : private:
      86             :     uint32_t stage_;
      87             :     FinFn finFn_;
      88             :     const ExecuteFn efn_;
      89             :     const InputT & inp_;
      90             :     boost::shared_ptr<SubResultT> res_;
      91             :     std::vector<ExternalT*> externals_;
      92             :     bool finished_;
      93             :     bool running_;
      94             :     const uint32_t inst_;
      95             :     const int tid_;
      96             :     const int tinst_;
      97             :     PipelineWorker * const w_;
      98             : 
      99      109160 :     void WorkDone(bool ret_code) {
     100      109160 :         for (typename std::vector<ExternalT*>::iterator it = externals_.begin();
     101      218334 :                 it!=externals_.end(); it++) {
     102      109330 :             delete (*it);
     103             :         }
     104      108801 :         finished_ = true;
     105      108801 :         finFn_(ret_code);
     106      109943 :     }
     107             : 
     108      219607 :     bool Runner(void) {
     109      219607 :         running_ = true;
     110      219607 :         ExternalBase::Efn fn = (efn_)(inst_, externals_, inp_, *res_);
     111      219278 :         running_ = false;
     112      219278 :         if (fn.empty()) {
     113      109221 :             WorkDone(true); 
     114             :         } else {
     115      110578 :             if (fn == &ExternalBase::Incomplete) return false;
     116             : 
     117      110613 :             if (!fn(this)) {
     118           0 :                 WorkDone(false);
     119             :             }
     120             :         }
     121      220228 :         return true;
     122      220228 :     }
     123             : 
     124      110754 :     void Response(std::unique_ptr<ExternalT> resp) {
     125      110754 :         ExternalT * msg = resp.get();
     126      110763 :         resp.release();
     127      110759 :         externals_.push_back(msg);
     128      110756 :         assert(!running_);
     129      110756 :         PipelineWorker *w = new PipelineWorker(tid_, tinst_,
     130             :                 boost::bind(&WorkProcessor<InputT,SubResultT,ExternalT>::Runner,
     131             :                 this));
     132      110757 :         TaskScheduler *scheduler = TaskScheduler::GetInstance();
     133      110756 :         scheduler->Enqueue(w);        
     134      110791 :     }
     135             : };
     136             : 
     137             : template <typename InputT, typename ResultT>
     138             : struct WorkStageIf {
     139             :     virtual void Start(uint32_t stage, FinFn finFn, const boost::shared_ptr<InputT> & inp) = 0;
     140             :     virtual boost::shared_ptr<ResultT> Result() const = 0;
     141             :     virtual void Release() = 0;
     142        2063 :     virtual ~WorkStageIf() {}
     143             : };
     144             : #endif