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;
47 struct Opr;
49 typedef Var* VarHandle;
51 typedef Opr* OprHandle;
57  public:
58  // use implicit copy and assign
60  inline void operator()() const {
61  (*callback_)(engine_, param_);
62  }
63 
64  private:
66  friend class ::mxnet::Engine;
68  void (*callback_)(Engine *, void *);
70  Engine* engine_;
72  void* param_;
73 };
74 } // namespace engine
75 
76 #if DMLC_USE_CXX11
77 
78 enum class FnProperty {
80  kNormal,
84  kCopyToGPU,
88  kAsync,
91 }; // enum class FnProperty
92 
97  public:
101  typedef std::function<void(RunContext)> SyncFn;
103  typedef std::function<void(RunContext, CallbackOnComplete)> AsyncFn;
115  virtual void NotifyShutdown() = 0;
119  virtual void Stop() {
120  LOG(FATAL) << "Engine cannot be stopped";
121  }
125  virtual void Start() {
126  LOG(FATAL) << "Engine cannot be restarted";
127  }
134  virtual VarHandle NewVariable() = 0;
146  virtual OprHandle NewOperator(AsyncFn fn,
147  std::vector<VarHandle> const& const_vars,
148  std::vector<VarHandle> const& mutable_vars,
150  const char* opr_name = nullptr) = 0;
158  virtual void DeleteOperator(OprHandle op) = 0;
166  virtual void Push(OprHandle op, Context exec_ctx, int priority = 0, bool profiling = false) = 0;
180  virtual void PushAsync(AsyncFn exec_fun, Context exec_ctx,
181  std::vector<VarHandle> const& const_vars,
182  std::vector<VarHandle> const& mutable_vars,
184  int priority = 0,
185  const char* opr_name = nullptr) = 0;
197  virtual void DeleteVariable(SyncFn delete_fn,
198  Context exec_ctx,
199  VarHandle var) = 0;
205  virtual void WaitForVar(VarHandle var) = 0;
209  virtual void WaitForAll() = 0;
211  virtual ~Engine() noexcept(false) {}
215  static Engine* Get();
224  static std::shared_ptr<Engine> _GetSharedRef();
237  virtual void PushSync(SyncFn exec_fn, Context exec_ctx,
238  std::vector<VarHandle> const& const_vars,
239  std::vector<VarHandle> const& mutable_vars,
241  int priority = 0,
242  const char* opr_name = nullptr) {
243  this->PushAsync([exec_fn](RunContext ctx, CallbackOnComplete on_complete) {
244  exec_fn(ctx);
245  on_complete();
246  }, exec_ctx, const_vars, mutable_vars, prop, priority, opr_name);
247  }
248 
254  inline CallbackOnComplete CreateCallback(
255  void (*callback)(Engine *, void *), void *param) {
256  CallbackOnComplete ret;
257  ret.callback_ = callback;
258  ret.engine_ = this;
259  ret.param_ = param;
260  return ret;
261  }
262  // For each var vector, sort it and remove the duplicated vars.
263  // Also remove vars from read_vars if it also appears in write_vars
264  inline void DeduplicateVarHandle(std::vector<engine::VarHandle> *read_vars,
265  std::vector<engine::VarHandle> *write_vars) {
266  std::sort(write_vars->begin(), write_vars->end());
267  write_vars->resize(std::unique(write_vars->begin(), write_vars->end()) -
268  write_vars->begin());
269  std::sort(read_vars->begin(), read_vars->end());
270  read_vars->resize(std::unique(read_vars->begin(), read_vars->end()) -
271  read_vars->begin());
272  auto wit = write_vars->begin();
273  auto rtop = read_vars->begin();
274  for (auto rit = read_vars->begin(); rit != read_vars->end(); ++rit) {
275  while (wit != write_vars->end() && *wit < *rit) ++wit;
276  if (wit == write_vars->end() || *wit != *rit) {
277  *rtop = *rit;
278  ++rtop;
279  }
280  }
281  read_vars->resize(rtop - read_vars->begin());
282  }
284  virtual int bulk_size() const {
285  return 0;
286  }
288  virtual int set_bulk_size(int) {
289  return 0;
290  }
291 }; // class Engine
292 #endif // DMLC_USE_CXX11
293 } // namespace mxnet
294 #endif // MXNET_ENGINE_H_
void DeduplicateVarHandle(std::vector< engine::VarHandle > *read_vars, std::vector< engine::VarHandle > *write_vars)
Definition: engine.h:264
FnProperty
Function property, used to hint what action is pushed to engine.
Definition: engine.h:78
virtual void Stop()
Stop all workers in the engine.
Definition: engine.h:119
std::function< void(RunContext)> SyncFn
Synchronous operation to pass to engine.
Definition: engine.h:101
std::function< void(RunContext, CallbackOnComplete)> AsyncFn
Asynchronous operation to pass to engine.
Definition: engine.h:103
namespace of mxnet
Definition: base.h:127
virtual int bulk_size() const
query current limit for bulk size
Definition: engine.h:284
Asynchronous function call.
CallbackOnComplete CreateCallback(void(*callback)(Engine *, void *), void *param)
factory function to create OnComplete callback.
Definition: engine.h:254
void operator()() const
involve the callback
Definition: engine.h:60
execution time context. The information needed in runtime for actual execution.
Definition: base.h:253
Delete variable call.
Normal operation.
virtual ~Engine() noexcept(false)
virtual destructor
Definition: engine.h:211
virtual void Start()
Restart all workers in the engine.
Definition: engine.h:125
Copy operation from GPU to other devices.
virtual int set_bulk_size(int)
set maximum limit for bulk size
Definition: engine.h:288
engine::OprHandle OprHandle
Operator pointer.
Definition: engine.h:107
engine::VarHandle VarHandle
Variable pointer.
Definition: engine.h:105
Var * VarHandle
Variable pointer type, usually hold by user used to specify dependencies.
Definition: engine.h:47
Prioritized sync operation on CPU.
engine::CallbackOnComplete CallbackOnComplete
callback on complete
Definition: engine.h:99
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:237
Dependency engine that schedules operations.
Definition: engine.h:96
Symbol sort(const std::string &symbol_name, Symbol data, dmlc::optional< int > axis=dmlc::optional< int >(-1), bool is_ascend=true)
Definition: op.h:2439
OnComplete Callback to the engine, called by AsyncFn when action completes.
Definition: engine.h:56
Context information about the execution environment.
Definition: base.h:142
#define MXNET_API
define compatible keywords in g++ Used to support g++-4.6 and g++4.7
Definition: base.h:92
Copy operation from CPU to other devices.
Opr * OprHandle
Operator pointer type, usually hold by user.
Definition: engine.h:51