mxnet
engine.h
Go to the documentation of this file.
1 /*
2  * Licensed to the Apache Software Foundation (ASF) under one
3  * or more contributor license agreements. See the NOTICE file
4  * distributed with this work for additional information
5  * regarding copyright ownership. The ASF licenses this file
6  * to you under the Apache License, Version 2.0 (the
7  * "License"); you may not use this file except in compliance
8  * with the License. You may obtain a copy of the License at
9  *
10  * http://www.apache.org/licenses/LICENSE-2.0
11  *
12  * Unless required by applicable law or agreed to in writing,
13  * software distributed under the License is distributed on an
14  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15  * KIND, either express or implied. See the License for the
16  * specific language governing permissions and limitations
17  * under the License.
18  */
19 
25 #ifndef MXNET_ENGINE_H_
26 #define MXNET_ENGINE_H_
27 
28 #include <dmlc/base.h>
29 #if DMLC_USE_CXX11
30 #include <algorithm>
31 #include <memory>
32 #include <functional>
33 #endif
34 #include <vector>
35 #include "./base.h"
36 
37 namespace mxnet {
38 
39 // forward declare engine
40 class Engine;
41 
43 namespace engine {
45 struct Var {
46  virtual size_t version() {
47  return version_;
48  }
49  virtual ~Var() = default;
55  template <typename T>
56  inline T* Cast();
61  size_t version_{0};
62 }; // struct Var
63 
65 struct Opr;
67 typedef Var* VarHandle;
69 typedef Opr* OprHandle;
75  public:
76  // use implicit copy and assign
78  inline void operator()() const {
79  (*callback_)(engine_, param_);
80  }
81 
82  private:
84  friend class ::mxnet::Engine;
86  void (*callback_)(Engine *, void *);
88  Engine* engine_;
90  void* param_;
91 };
92 } // namespace engine
93 
94 #if DMLC_USE_CXX11
95 
96 enum class FnProperty {
98  kNormal,
100  kCopyFromGPU,
102  kCopyToGPU,
106  kAsync,
108  kDeleteVar,
111 }; // enum class FnProperty
112 
117  public:
121  typedef std::function<void(RunContext)> SyncFn;
123  typedef std::function<void(RunContext, CallbackOnComplete)> AsyncFn;
135  virtual void NotifyShutdown() = 0;
139  virtual void Stop() {
140  LOG(FATAL) << "Engine cannot be stopped";
141  }
145  virtual void Start() {
146  LOG(FATAL) << "Engine cannot be restarted";
147  }
154  virtual VarHandle NewVariable() = 0;
167  virtual OprHandle NewOperator(AsyncFn fn,
168  std::vector<VarHandle> const& const_vars,
169  std::vector<VarHandle> const& mutable_vars,
171  const char* opr_name = nullptr,
172  bool wait = false) = 0;
180  virtual void DeleteOperator(OprHandle op) = 0;
188  virtual void Push(OprHandle op, Context exec_ctx, int priority = 0, bool profiling = false) = 0;
203  virtual void PushAsync(AsyncFn exec_fun, Context exec_ctx,
204  std::vector<VarHandle> const& const_vars,
205  std::vector<VarHandle> const& mutable_vars,
207  int priority = 0,
208  const char* opr_name = nullptr,
209  bool wait = false) = 0;
221  virtual void DeleteVariable(SyncFn delete_fn,
222  Context exec_ctx,
223  VarHandle var) = 0;
229  virtual void WaitForVar(VarHandle var) = 0;
233  virtual void WaitForAll() = 0;
235  virtual ~Engine() noexcept(false) {}
239  static Engine* Get();
248  static std::shared_ptr<Engine> _GetSharedRef();
261  virtual void PushSync(SyncFn exec_fn, Context exec_ctx,
262  std::vector<VarHandle> const& const_vars,
263  std::vector<VarHandle> const& mutable_vars,
265  int priority = 0,
266  const char* opr_name = nullptr) {
267  this->PushAsync([exec_fn](RunContext ctx, CallbackOnComplete on_complete) {
268  exec_fn(ctx);
269  on_complete();
270  }, exec_ctx, const_vars, mutable_vars, prop, priority, opr_name);
271  }
272 
278  inline CallbackOnComplete CreateCallback(
279  void (*callback)(Engine *, void *), void *param) {
280  CallbackOnComplete ret;
281  ret.callback_ = callback;
282  ret.engine_ = this;
283  ret.param_ = param;
284  return ret;
285  }
286  // For each var vector, sort it and remove the duplicated vars.
287  // Also remove vars from read_vars if it also appears in write_vars
288  inline void DeduplicateVarHandle(std::vector<engine::VarHandle> *read_vars,
289  std::vector<engine::VarHandle> *write_vars) {
290  std::sort(write_vars->begin(), write_vars->end());
291  write_vars->resize(std::unique(write_vars->begin(), write_vars->end()) -
292  write_vars->begin());
293  std::sort(read_vars->begin(), read_vars->end());
294  read_vars->resize(std::unique(read_vars->begin(), read_vars->end()) -
295  read_vars->begin());
296  auto wit = write_vars->begin();
297  auto rtop = read_vars->begin();
298  for (auto rit = read_vars->begin(); rit != read_vars->end(); ++rit) {
299  while (wit != write_vars->end() && *wit < *rit) ++wit;
300  if (wit == write_vars->end() || *wit != *rit) {
301  *rtop = *rit;
302  ++rtop;
303  }
304  }
305  read_vars->resize(rtop - read_vars->begin());
306  }
308  virtual int bulk_size() const {
309  return 0;
310  }
312  virtual int set_bulk_size(int) {
313  return 0;
314  }
315 }; // class Engine
316 #endif // DMLC_USE_CXX11
317 } // namespace mxnet
318 #endif // MXNET_ENGINE_H_
void DeduplicateVarHandle(std::vector< engine::VarHandle > *read_vars, std::vector< engine::VarHandle > *write_vars)
Definition: engine.h:288
FnProperty
Function property, used to hint what action is pushed to engine.
Definition: engine.h:96
virtual void Stop()
Stop all workers in the engine.
Definition: engine.h:139
std::function< void(RunContext)> SyncFn
Synchronous operation to pass to engine.
Definition: engine.h:121
std::function< void(RunContext, CallbackOnComplete)> AsyncFn
Asynchronous operation to pass to engine.
Definition: engine.h:123
size_t version_
version number of the var. Every time the object it is associated with is modified, the version number is incremented by 1.
Definition: engine.h:61
namespace of mxnet
Definition: base.h:118
virtual int bulk_size() const
query current limit for bulk size
Definition: engine.h:308
Asynchronous function call.
CallbackOnComplete CreateCallback(void(*callback)(Engine *, void *), void *param)
factory function to create OnComplete callback.
Definition: engine.h:278
void operator()() const
involve the callback
Definition: engine.h:78
execution time context. The information needed in runtime for actual execution.
Definition: base.h:257
base class of engine variables.
Definition: engine.h:45
Delete variable call.
Normal operation.
virtual ~Var()=default
virtual ~Engine() noexcept(false)
virtual destructor
Definition: engine.h:235
Prioritized sync operation on GPU.
#define MXNET_API
define compatible keywords in g++ Used to support g++-4.6 and g++4.7
Definition: base.h:92
virtual void Start()
Restart all workers in the engine.
Definition: engine.h:145
Copy operation from GPU to other devices.
virtual int set_bulk_size(int)
set maximum limit for bulk size
Definition: engine.h:312
engine::OprHandle OprHandle
Operator pointer.
Definition: engine.h:127
engine::VarHandle VarHandle
Variable pointer.
Definition: engine.h:125
Var * VarHandle
Variable pointer type, usually hold by user used to specify dependencies.
Definition: engine.h:65
Prioritized sync operation on CPU.
engine::CallbackOnComplete CallbackOnComplete
callback on complete
Definition: engine.h:119
virtual void PushSync(SyncFn exec_fn, Context exec_ctx, std::vector< VarHandle > const &const_vars, std::vector< VarHandle > const &mutable_vars, FnProperty prop=FnProperty::kNormal, int priority=0, const char *opr_name=nullptr)
Push an synchronous operation to the engine.
Definition: engine.h:261
Dependency engine that schedules operations.
Definition: engine.h:116
OnComplete Callback to the engine, called by AsyncFn when action completes.
Definition: engine.h:74
configuration of MXNet as well as basic data structure.
T * Cast()
cast variable to derived type T
Context information about the execution environment.
Definition: base.h:133
Copy operation from CPU to other devices.
Opr * OprHandle
Operator pointer type, usually hold by user.
Definition: engine.h:69
virtual size_t version()
Definition: engine.h:46