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_PIPELINE_INL_H__
       7             : #define __WORK_PIPELINE_INL_H__
       8             : 
       9             : 
      10             : template<typename InputT, typename ResultT, typename ExternalT, typename SubResultT>
      11        2054 : WorkStage<InputT, ResultT, ExternalT, SubResultT>::WorkStage(
      12             :     std::vector<std::pair<int,int> > tinfo,
      13             :     ExecuteFn efn, MergeFn mfn, int tid, int tinst) :
      14        4108 :         stage_(std::numeric_limits<uint32_t>::max()),
      15        2053 :         merger_(mfn),
      16        2053 :         efn_(efn),
      17        2053 :         finished_(false),
      18        2053 :         running_(false),
      19        2053 :         tid_(tid),
      20        2053 :         tinst_(tinst),
      21        4107 :         tinfo_(tinfo) {}
      22             : 
      23             : 
      24             : template<typename InputT, typename ResultT, typename ExternalT, typename SubResultT>
      25             : void
      26        2054 : WorkStage<InputT, ResultT, ExternalT, SubResultT>::Start(uint32_t stage, FinFn finFn,
      27             :         const boost::shared_ptr<InputT> & inp) {
      28        2054 :     assert(!running_);
      29        2054 :     assert(!finished_);
      30        2054 :     stage_ = stage;
      31        2054 :     inp_ = inp;
      32        2054 :     finFn_ = finFn;
      33        2054 :     remainingInst_ = tinfo_.size();
      34      111844 :     for (uint32_t tk = 0 ; tk < tinfo_.size(); tk++) {
      35      219552 :         workers_.push_back(boost::shared_ptr<WorkProcessor<InputT,SubResultT,ExternalT> >(
      36      219715 :                 new WorkProcessor<InputT,SubResultT,ExternalT>(stage_, efn_,
      37             :                     boost::bind(&WorkStage<InputT,ResultT,ExternalT,SubResultT>::WorkProcCb,
      38             :                                 this, tk, _1), 
      39      109951 :                     *inp_, tk, tinfo_[tk].first, tinfo_[tk].second)));
      40             :     }
      41      112103 :     for (uint32_t tk = 0 ; tk < tinfo_.size(); tk++) {
      42      110048 :         workers_[tk]->Start();
      43             :     }
      44             : 
      45        2054 : }
      46             : 
      47             : template<typename InputT, typename ResultT, typename ExternalT, typename SubResultT>
      48             : boost::shared_ptr<ResultT>
      49        3981 : WorkStage<InputT, ResultT, ExternalT, SubResultT>::Result() const {
      50        3981 :     if (!finished_) return boost::shared_ptr<ResultT>();
      51        3981 :     return res_;
      52             : }
      53             : 
      54             : template<typename InputT, typename ResultT, typename ExternalT, typename SubResultT>
      55             : void
      56        2054 : WorkStage<InputT, ResultT, ExternalT, SubResultT>::Release() {
      57        2054 :     assert(finished_);
      58        2054 :     assert(subRes_.size() == tinfo_.size());
      59        2054 :     inp_.reset();
      60      111810 :     for (uint32_t i = 0; i < subRes_.size(); i++) {
      61      109714 :         subRes_[i].reset();
      62             :     }
      63        2046 :     res_.reset();
      64        2054 : }
      65             : 
      66             : 
      67             : template<typename InputT, typename ResultT, typename ExternalT, typename SubResultT>
      68             : bool
      69        1927 : WorkStage<InputT, ResultT, ExternalT, SubResultT>::Runner(void) {
      70        1927 :     running_ = true;
      71        1927 :     if (!(merger_)(subRes_, inp_, *res_)) {
      72           0 :        return false; 
      73             :     }
      74        1927 :     running_ = false;
      75        1927 :     finished_ = true;
      76        1927 :     finFn_(true);
      77        1928 :     return true;
      78             : }
      79             : 
      80             : 
      81             : template<typename InputT, typename ResultT, typename ExternalT, typename SubResultT>
      82             : void
      83      108954 : WorkStage<InputT, ResultT, ExternalT, SubResultT>::WorkProcCb(uint32_t inst, bool ret_code) {
      84      108954 :     uint32_t prev = remainingInst_.fetch_sub(1);
      85      108954 :     if (prev == 1) {
      86        2053 :         assert(workers_.size() == tinfo_.size());
      87      111774 :         for (uint32_t i = 0; i < workers_.size(); i++) {
      88      109628 :             subRes_.push_back(workers_[i]->Result());
      89      108868 :             workers_[i]->Release();
      90             :         }
      91        2007 :         StageProceed(boost::is_same<ResultT,SubResultT>());
      92        2054 :         if (merger_.empty()) {
      93         126 :             finished_ = true;
      94         126 :             finFn_(true);
      95             :         } else {
      96        1928 :             res_.reset(new ResultT);
      97        1927 :             PipelineWorker *w = new PipelineWorker(tid_, tinst_,
      98             :                     boost::bind(&WorkStage<InputT,ResultT,ExternalT,SubResultT>::Runner,
      99             :                     this));
     100        1927 :             TaskScheduler *scheduler = TaskScheduler::GetInstance();
     101        1927 :             scheduler->Enqueue(w);                  
     102             :         }
     103             :     }
     104      108955 : }
     105             : 
     106             : 
     107             : template<typename T0, typename T1, typename T2, typename T3, 
     108             :     typename T4, typename T5, typename T6>
     109        1928 : WorkPipeline<T0,T1,T2,T3,T4,T5,T6>::WorkPipeline(
     110             :     WorkStageIf<T0,T1> * s0,
     111             :     WorkStageIf<T1,T2> * s1,
     112             :     WorkStageIf<T2,T3> * s2,
     113             :     WorkStageIf<T3,T4> * s3,
     114             :     WorkStageIf<T4,T5> * s4,
     115             :     WorkStageIf<T5,T6> * s5) :
     116        1928 :         finished_(false),
     117        1927 :         sg_(
     118        3856 :             boost::shared_ptr<WorkStageIf<T0,T1> >(s0),
     119        3856 :             boost::shared_ptr<WorkStageIf<T1,T2> >(s1),
     120        3855 :             boost::shared_ptr<WorkStageIf<T2,T3> >(s2),
     121        3855 :             boost::shared_ptr<WorkStageIf<T3,T4> >(s3),
     122        3855 :             boost::shared_ptr<WorkStageIf<T4,T5> >(s4),
     123        3856 :             boost::shared_ptr<WorkStageIf<T5,T6> >(s5)) {}
     124             : 
     125             : template<typename T0, typename T1, typename T2, typename T3, 
     126             :     typename T4, typename T5, typename T6>
     127             : void
     128        1928 : WorkPipeline<T0,T1,T2,T3,T4,T5,T6>::Start(FinFn finFn, const boost::shared_ptr<T0> & inp) {
     129        1928 :     inp_ = inp;
     130        1928 :     finFn_ = finFn;
     131        1928 :     typename boost::tuples::element<0, sg_type>::type g(boost::get<0>(sg_));
     132        3856 :     g->Start(0, boost::bind(&SelfT::WorkStageCb,
     133        1927 :         this, 0, _1),inp_);
     134        1928 : }
     135             : 
     136             : template<typename T0, typename T1, typename T2, typename T3, 
     137             :     typename T4, typename T5, typename T6>
     138             : boost::shared_ptr<typename WorkPipeline<T0,T1,T2,T3,T4,T5,T6>::ResT>
     139        1928 : WorkPipeline<T0,T1,T2,T3,T4,T5,T6>::Result() const {
     140        1928 :     if (!finished_) return boost::shared_ptr<ResT>();
     141        1928 :     return res_;
     142             : }
     143             : 
     144             : template<typename T0, typename T1, typename T2, typename T3, 
     145             :     typename T4, typename T5, typename T6>
     146             : void
     147        2053 : WorkPipeline<T0,T1,T2,T3,T4,T5,T6>::WorkStageCb(uint32_t stage, bool ret_code) {
     148        2053 :     switch(stage) {
     149        1927 :     case 0: {
     150        1927 :             NextStage<0,T1>();
     151             :         }
     152        1928 :         break;
     153         126 :     case 1: {
     154         126 :             NextStage<1,T2>();
     155             :         }
     156         126 :         break;
     157           0 :     case 2: {
     158           0 :             NextStage<2,T3>();
     159             :         }
     160           0 :         break;
     161           0 :     case 3: {
     162           0 :             NextStage<3,T4>();
     163             :         }
     164           0 :         break;
     165           0 :     case 4: {
     166           0 :             NextStage<4,T5>();
     167             :         }
     168           0 :         break;            
     169           0 :     case 5: {
     170           0 :             typename boost::tuples::element<5, sg_type>::type g(boost::get<5>(sg_));
     171           0 :             res_ = g->Result();
     172           0 :             g->Release();
     173           0 :             finished_ = true;
     174           0 :             finFn_(true);
     175           0 :         }
     176           0 :     break;
     177             :     }
     178        2054 : }
     179             : 
     180             : template<typename T0, typename T1, typename T2, typename T3, 
     181             :     typename T4, typename T5, typename T6>
     182             : template<int kS,typename NextT>
     183             : void
     184        2054 : WorkPipeline<T0,T1,T2,T3,T4,T5,T6>::NextStage() {
     185        2054 :     boost::shared_ptr<NextT> res = boost::get<kS>(sg_)->Result();
     186        2054 :     PipeProceed<kS,boost::is_same<NextT,ResT>::value>::Do(this);
     187        2054 :     boost::get<kS>(sg_)->Release();
     188        2054 :     if (boost::get<kS+1>(sg_)) {
     189         126 :         res_.reset();
     190         126 :         typename boost::tuples::element<kS+1, sg_type>::type g(boost::get<kS+1>(sg_));
     191         126 :         g->Start(kS+1, boost::bind(&SelfT::WorkStageCb,
     192             :             this, kS+1, _1),res);
     193         126 :     } else {
     194        1928 :         finished_ = true;
     195        1928 :         finFn_(true);            
     196             :     }
     197        2053 : }
     198             : 
     199             : #endif