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             : #ifndef __WORK_PROCESSOR_INL_H__
       7             : #define __WORK_PROCESSOR_INL_H__
       8             : 
       9             : #include <vector>
      10             : #include <boost/function.hpp>
      11             : #include <boost/bind/bind.hpp>
      12             : #include <boost/shared_ptr.hpp>
      13             : #include <boost/scoped_ptr.hpp>
      14             : #include <boost/assign/list_of.hpp>
      15             : #include <boost/type_traits.hpp>
      16             : #include <boost/mpl/assert.hpp>
      17             : #include <boost/tuple/tuple.hpp>
      18             : #include <limits>
      19             : #include <sstream>
      20             : #include "base/task.h"
      21             : 
      22             : using namespace boost::placeholders;
      23             : 
      24             : struct PipelineWorker : public Task {
      25      222143 :     PipelineWorker(int tid, int tinst, boost::function<bool(void)> runner) :
      26      222143 :         Task(tid,tinst), runner_(runner) {}
      27             : 
      28      221742 :     bool Run() {
      29      221742 :         if (!(runner_)()) return false;
      30      221596 :         return true;
      31             :     }
      32           0 :     std::string Description() const { return "PipelineWorker"; }
      33             : private:
      34             :     const boost::function<bool(void)> runner_;
      35             : };
      36             : 
      37             : template<typename InputT, typename SubResultT, typename ExternalT>
      38             : class WorkProcessor : public ExternalProcIf<ExternalT> {
      39             : public:
      40             :     typedef boost::function<ExternalBase::Efn(
      41             :         uint32_t inst,
      42             :         const std::vector<ExternalT*> & exts, // Info for previous steps of this stage
      43             :         const InputT & inp, // Info from previous stage
      44             :         SubResultT & subRes // Access to final result of this instance
      45             :         )> ExecuteFn;
      46             : 
      47      109931 :     WorkProcessor(uint32_t stage, ExecuteFn efn, FinFn finFn, const InputT & inp,
      48             :         uint32_t inst, int tid, int tinst) :
      49      109920 :             stage_(stage),
      50      109920 :             finFn_(finFn),
      51      109878 :             efn_(efn),
      52      109843 :             inp_(inp),
      53      109835 :             finished_(false),
      54      109835 :             running_(false),
      55      109835 :             inst_(inst),
      56      109835 :             tid_(tid),
      57      109835 :             tinst_(tinst),
      58      109835 :             w_(new PipelineWorker(tid_, tinst_,
      59             :                 boost::bind(&WorkProcessor<InputT,SubResultT,ExternalT>::Runner,
      60      219774 :                 this))) {
      61      109652 :         res_.reset(new SubResultT);
      62      109786 :     }
      63             : 
      64      110046 :     void Start() {
      65      110046 :         TaskScheduler *scheduler = TaskScheduler::GetInstance();
      66      110042 :         scheduler->Enqueue(w_);
      67      110050 :     }   
      68             : 
      69      109517 :     boost::shared_ptr<SubResultT> Result() const {
      70      109517 :         if (!finished_) return boost::shared_ptr<SubResultT>();
      71      109517 :         return res_;
      72             :     }
      73             : 
      74      108823 :     void Release() {
      75      108823 :         assert(finished_);
      76      108823 :         res_.reset();
      77      109754 :     }
      78             :     
      79         252 :     std::string Key() const {
      80         252 :         std::stringstream keystr;
      81         252 :         keystr << "PROC-STAGE:" << stage_ << "-INST:" << inst_ <<
      82         252 :                 "-STEP:" << externals_.size();
      83         504 :         return keystr.str();
      84         252 :     }
      85             : 
      86             : private:
      87             :     uint32_t stage_;
      88             :     FinFn finFn_;
      89             :     const ExecuteFn efn_;
      90             :     const InputT & inp_;
      91             :     boost::shared_ptr<SubResultT> res_;
      92             :     std::vector<ExternalT*> externals_;
      93             :     bool finished_;
      94             :     bool running_;
      95             :     const uint32_t inst_;
      96             :     const int tid_;
      97             :     const int tinst_;
      98             :     PipelineWorker * const w_;
      99             : 
     100      109484 :     void WorkDone(bool ret_code) {
     101      109484 :         for (typename std::vector<ExternalT*>::iterator it = externals_.begin();
     102      219219 :                 it!=externals_.end(); it++) {
     103      109720 :             delete (*it);
     104             :         }
     105      109164 :         finished_ = true;
     106      109164 :         finFn_(ret_code);
     107      109950 :     }
     108             : 
     109      219548 :     bool Runner(void) {
     110      219548 :         running_ = true;
     111      219548 :         ExternalBase::Efn fn = (efn_)(inst_, externals_, inp_, *res_);
     112      219825 :         running_ = false;
     113      219825 :         if (fn.empty()) {
     114      109656 :             WorkDone(true); 
     115             :         } else {
     116      110299 :             if (fn == &ExternalBase::Incomplete) return false;
     117             : 
     118      110336 :             if (!fn(this)) {
     119           0 :                 WorkDone(false);
     120             :             }
     121             :         }
     122      220274 :         return true;
     123      220274 :     }
     124             : 
     125      110358 :     void Response(std::unique_ptr<ExternalT> resp) {
     126      110358 :         ExternalT * msg = resp.get();
     127      110431 :         resp.release();
     128      110425 :         externals_.push_back(msg);
     129      110426 :         assert(!running_);
     130      110426 :         PipelineWorker *w = new PipelineWorker(tid_, tinst_,
     131             :                 boost::bind(&WorkProcessor<InputT,SubResultT,ExternalT>::Runner,
     132             :                 this));
     133      110421 :         TaskScheduler *scheduler = TaskScheduler::GetInstance();
     134      110419 :         scheduler->Enqueue(w);        
     135      110448 :     }
     136             : };
     137             : 
     138             : template <typename InputT, typename ResultT>
     139             : struct WorkStageIf {
     140             :     virtual void Start(uint32_t stage, FinFn finFn, const boost::shared_ptr<InputT> & inp) = 0;
     141             :     virtual boost::shared_ptr<ResultT> Result() const = 0;
     142             :     virtual void Release() = 0;
     143        2054 :     virtual ~WorkStageIf() {}
     144             : };
     145             : #endif