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 : * work_pipeline.h
8 : *
9 : * This file provides some templates that can be used to manage processing.
10 : *
11 : * The user creates a WorkPipeline, with multiple WorkStage's inside it.
12 : *
13 : * The stages will execute in sequence. A stage has an execution phase, where
14 : * multiple threads (instances) can do processing in parallel, and an optional
15 : * merge phase where the results of the parallel execution can be consolidated.
16 : * For each stage, the user provides the execution function (of type ExecuteFn)
17 : * (optionally) a merge function (of type MergeFn).
18 : *
19 : * The execution function runs in parallel over a number of instances. Each
20 : * instance executes in steps. During a step, the ExecuteFn is expected to
21 : * provide the external command that needs to be run to get information (e.g.
22 : * cassandra or redis, or another WorkPipleline) . ExecuteFn will called again
23 : * (as a next step) when the results of the external command are available. In
24 : * that next step, ExecuteFn can consolidate the external result into the
25 : * instances' subresult. The user can run multiple such steps within the
26 : * instance.
27 : *
28 : * After the subresults from all instances of that stage are available, MergeFn
29 : * will run which can consolidate the subresults into the final result for
30 : * that stage. If a stage has a single instance, then MergeFn is not needed
31 : * for that stage.
32 : *
33 : * After all stages have been executed, the user's callback function is called.
34 : * The user can then access the final result of the WorkPipeline.
35 : *
36 : * Here are some of the features:
37 : *
38 : * - Buffers are shared and transferred wherever possible (instead of copy)
39 : *
40 : * - A generic mechanism is provided to get information from any source.
41 : * Both Async and Sync processing is allowed for the external sources.
42 : *
43 : * - Strong typechecking for Input Type, Subresult Type and Result type
44 : * across stages.
45 : *
46 : * - TaskId and TaskInstance can be customized for ExecuteFn (for every
47 : * instance of every stage) and MergeFn (for every stage)
48 : *
49 : *
50 : * This file has the interfaces provided by WorkPipeline and WorkStage.
51 : * Most of the implementation is split across two other files -
52 : * work_processor-inl.h and work_pipeline-inl.h
53 : *
54 : */
55 :
56 : #ifndef __WORK_PIPELINE_H__
57 : #define __WORK_PIPELINE_H__
58 :
59 : #include <vector>
60 : #include <limits>
61 : #include <sstream>
62 : #include <atomic>
63 :
64 : #include <boost/function.hpp>
65 : #include <boost/bind/bind.hpp>
66 : #include <boost/shared_ptr.hpp>
67 : #include <boost/scoped_ptr.hpp>
68 : #include <boost/assign/list_of.hpp>
69 : #include <boost/type_traits.hpp>
70 : #include <boost/mpl/assert.hpp>
71 : #include <boost/tuple/tuple.hpp>
72 : #include "base/task.h"
73 :
74 : using namespace boost::placeholders;
75 :
76 : // When the WorkPipeline is done, it will call this function back
77 : // to indicate that the result is ready.
78 : // The bool indicates success or failure.
79 : typedef boost::function<void(bool)> FinFn;
80 :
81 :
82 : // This is the interface to the external command execution that
83 : // can be requested by the user in ExecuteFn
84 : //
85 : // The user provides the function ExternalBase::Efn that the
86 : // WorkStage can call with a single argument (ExternalBase *)
87 : //
88 : // The user should ensure that when the command is done, the
89 : // argument is cast into ExternalProcIf *, and the
90 : // Response function is called with the result of the the
91 : // external command
92 : struct ExternalBase {
93 : typedef boost::function<bool(ExternalBase *)> Efn;
94 0 : static bool Incomplete(void *) { assert(0); }
95 : };
96 : template<typename ExternalT>
97 : struct ExternalProcIf : public ExternalBase {
98 : virtual std::string Key() const = 0;
99 : virtual void Response(std::unique_ptr<ExternalT>) = 0;
100 109309 : virtual ~ExternalProcIf() {}
101 : };
102 :
103 : #include "work_processor-inl.h"
104 :
105 : // When creating the WorkPipeline, the user needs to supply it
106 : // constructed WorkStage objects.
107 : // The Constructor of this class is the only function
108 : // that the client needs to call.
109 : template<typename InputT, typename ResultT, typename ExternalT, typename SubResultT = ResultT>
110 : class WorkStage : public WorkStageIf<InputT, ResultT> {
111 : public:
112 :
113 : // A WorkStage executes multiple instances in parallel by calling ExecuteFn
114 : // Each execution run of ExecuteFn is called a "step".
115 : // - If the client returns ExternalBase::Incomplete, the step is incomplete.
116 : // After yielding, ExecuteFn will be called again to continue the step.
117 : // - If the client returns NULL, the step ends, and so does the instance.
118 : // - Otherwise, WorkStage will call the returned callback function
119 : // and move to the next step when the client calls ExternalProcIf's
120 : // Response function. The buffer passed in the Response function is added
121 : // to the exts vector before calling ExecuteFn for the next step.
122 : //
123 : // The client should fill in the final result in subRes before the
124 : // instance ends.
125 : typedef boost::function<
126 : ExternalBase::Efn( // Client returns External fn to call and
127 : // whether processing is incomplete.
128 : uint32_t inst, // Instance num. (the instances execute in parallel)
129 : const std::vector<ExternalT*>
130 : & exts, // Info for previous steps of this stage
131 : const InputT & inp, // Info from previous stage
132 : SubResultT & subRes // Access to final result of this instance
133 : )> ExecuteFn;
134 :
135 : // When all the ExecuteFn instances end, the Execute phase of the stage
136 : // is over. At that time, WorkStage calls MergeFn. MergeFn is expected to
137 : // aggregate the SubResults from the ExecuteFn instances into the final
138 : // Result for the stage.
139 : // If the SubResult type and Result type is the same, MergeFn is optional.
140 : // If no MergeFn is supplied, the SubResult of the 1st instance will be
141 : // used as the final result of the stage.
142 : typedef boost::function<
143 : bool( // Client returns whether processing is incomplete.
144 : const std::vector<boost::shared_ptr<SubResultT> >
145 : & subs, // Subresults from this stage's instances
146 : const boost::shared_ptr<InputT> & inp, // Info from previous stage
147 : ResultT & res // Final result to be reported by this stage
148 : )> MergeFn;
149 :
150 : // The client needs to constuct WorkStages.
151 : // tinfo is a vector of TaskId,TaskInstance pairs for ExecuteFn
152 : // ExecuteFn and MergeFn were explained above
153 : // tid and tinst are the TaskId and TaskInstance used for MergeFn
154 : WorkStage(std::vector<std::pair<int,int> > tinfo,
155 : ExecuteFn efn, MergeFn mfn = 0, int tid = 0, int tinst = -1);
156 :
157 : void Start(uint32_t stage, FinFn finFn, const boost::shared_ptr<InputT> & inp);
158 :
159 : boost::shared_ptr<ResultT> Result() const;
160 :
161 : void Release();
162 :
163 : private:
164 : uint32_t stage_;
165 : std::atomic<uint32_t> remainingInst_;
166 : FinFn finFn_;
167 : boost::shared_ptr<InputT> inp_;
168 : boost::shared_ptr<ResultT> res_;
169 : std::vector<boost::shared_ptr<WorkProcessor<InputT,SubResultT,ExternalT> > > workers_;
170 : std::vector<boost::shared_ptr<SubResultT> > subRes_;
171 : const MergeFn merger_;
172 : const ExecuteFn efn_;
173 : bool finished_;
174 : bool running_;
175 : const int tid_;
176 : const int tinst_;
177 : const std::vector<std::pair<int,int> > tinfo_;
178 :
179 : bool Runner(void);
180 : void WorkProcCb(uint32_t inst, bool ret_code);
181 126 : void StageProceed(boost::true_type) { res_ = subRes_[0]; }
182 1928 : void StageProceed(boost::false_type) { assert(!merger_.empty()); }
183 :
184 : };
185 :
186 :
187 : template<typename T0, typename T1, typename T2 = T1,
188 : typename T3 = T2, typename T4 = T3,
189 : typename T5 = T4, typename T6 = T5>
190 : class WorkPipeline {
191 : public:
192 : typedef T6 ResT;
193 : typedef WorkPipeline<T0,T1,T2,T3,T4,T5,T6> SelfT;
194 :
195 : // The client should instantiate the WorkPipeline with instances
196 : // of WorkStage. The WorkPipeline accomodates a minimum of 1 and
197 : // maximum of 6 stages. Each stage has an Input and Output type.
198 : // The Output type of a stage must match the Input type of the
199 : // next stage. This sequence of types must be used to instantiate
200 : // the WorkPipeline template.
201 : WorkPipeline(
202 : WorkStageIf<T0,T1> * s0,
203 : WorkStageIf<T1,T2> * s1 = NULL,
204 : WorkStageIf<T2,T3> * s2 = NULL,
205 : WorkStageIf<T3,T4> * s3 = NULL,
206 : WorkStageIf<T4,T5> * s4 = NULL,
207 : WorkStageIf<T5,T6> * s5 = NULL);
208 :
209 : // The client should call this function to start the WorkPipeline
210 : // "finFn" is the callback function that WorkPipeline will call when
211 : // the last stage ends. "inp" is the Input to the 1st stage.
212 : // The WorkPipeline will retain a shared_ptr to the initial
213 : // input for the lifetime of the WorkPipeline object.
214 : void Start(FinFn finFn, const boost::shared_ptr<T0> & inp);
215 :
216 : // After "finFn" has been called by WorkPipeline, the client can
217 : // access the final result by calling this function
218 : // The WorkPipeline will retain a shared_ptr to the final result
219 : // for the lifetime of the WorkPipeline object.
220 : boost::shared_ptr<ResT> Result() const;
221 :
222 : private:
223 : typedef boost::tuple<
224 : const boost::shared_ptr<WorkStageIf<T0,T1> >,
225 : const boost::shared_ptr<WorkStageIf<T1,T2> >,
226 : const boost::shared_ptr<WorkStageIf<T2,T3> >,
227 : const boost::shared_ptr<WorkStageIf<T3,T4> >,
228 : const boost::shared_ptr<WorkStageIf<T4,T5> >,
229 : const boost::shared_ptr<WorkStageIf<T5,T6> > > sg_type;
230 :
231 : bool finished_;
232 : FinFn finFn_;
233 : boost::shared_ptr<T0> inp_;
234 : boost::shared_ptr<ResT> res_;
235 : sg_type sg_;
236 :
237 : template<int kS, bool same> struct PipeProceed {};
238 :
239 : template<int kS> struct PipeProceed<kS, true> {
240 1928 : static void Do(SelfT * wp) { wp->res_ = boost::get<kS>(wp->sg_)->Result(); }
241 : };
242 :
243 : template<int kS> struct PipeProceed<kS, false> {
244 126 : static void Do(SelfT * wp) { assert(boost::get<kS+1>(wp->sg_)); }
245 : };
246 :
247 : void WorkStageCb(uint32_t stage, bool ret_code);
248 : template<int kS,typename NextT> void NextStage();
249 :
250 : };
251 :
252 : #include "work_pipeline-inl.h"
253 :
254 :
255 : #endif
|
