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,
90  kDeleteVar,
93 }; // enum class FnProperty
94 
99  public:
103  typedef std::function<void(RunContext)> SyncFn;
105  typedef std::function<void(RunContext, CallbackOnComplete)> AsyncFn;
117  virtual void NotifyShutdown() = 0;
121  virtual void Stop() {
122  LOG(FATAL) << "Engine cannot be stopped";
123  }
127  virtual void Start() {
128  LOG(FATAL) << "Engine cannot be restarted";
129  }
136  virtual VarHandle NewVariable() = 0;
149  virtual OprHandle NewOperator(AsyncFn fn,
150  std::vector<VarHandle> const& const_vars,
151  std::vector<VarHandle> const& mutable_vars,
153  const char* opr_name = nullptr,
154  bool wait = false) = 0;
162  virtual void DeleteOperator(OprHandle op) = 0;
170  virtual void Push(OprHandle op, Context exec_ctx, int priority = 0, bool profiling = false) = 0;
185  virtual void PushAsync(AsyncFn exec_fun, Context exec_ctx,
186  std::vector<VarHandle> const& const_vars,
187  std::vector<VarHandle> const& mutable_vars,
189  int priority = 0,
190  const char* opr_name = nullptr,
191  bool wait = false) = 0;
203  virtual void DeleteVariable(SyncFn delete_fn,
204  Context exec_ctx,
205  VarHandle var) = 0;
211  virtual void WaitForVar(VarHandle var) = 0;
215  virtual void WaitForAll() = 0;
217  virtual ~Engine() noexcept(false) {}
221  static Engine* Get();
230  static std::shared_ptr<Engine> _GetSharedRef();
243  virtual void PushSync(SyncFn exec_fn, Context exec_ctx,
244  std::vector<VarHandle> const& const_vars,
245  std::vector<VarHandle> const& mutable_vars,
247  int priority = 0,
248  const char* opr_name = nullptr) {
249  this->PushAsync([exec_fn](RunContext ctx, CallbackOnComplete on_complete) {
250  exec_fn(ctx);
251  on_complete();
252  }, exec_ctx, const_vars, mutable_vars, prop, priority, opr_name);
253  }
254 
260  inline CallbackOnComplete CreateCallback(
261  void (*callback)(Engine *, void *), void *param) {
262  CallbackOnComplete ret;
263  ret.callback_ = callback;
264  ret.engine_ = this;
265  ret.param_ = param;
266  return ret;
267  }
268  // For each var vector, sort it and remove the duplicated vars.
269  // Also remove vars from read_vars if it also appears in write_vars
270  inline void DeduplicateVarHandle(std::vector<engine::VarHandle> *read_vars,
271  std::vector<engine::VarHandle> *write_vars) {
272  std::sort(write_vars->begin(), write_vars->end());
273  write_vars->resize(std::unique(write_vars->begin(), write_vars->end()) -
274  write_vars->begin());
275  std::sort(read_vars->begin(), read_vars->end());
276  read_vars->resize(std::unique(read_vars->begin(), read_vars->end()) -
277  read_vars->begin());
278  auto wit = write_vars->begin();
279  auto rtop = read_vars->begin();
280  for (auto rit = read_vars->begin(); rit != read_vars->end(); ++rit) {
281  while (wit != write_vars->end() && *wit < *rit) ++wit;
282  if (wit == write_vars->end() || *wit != *rit) {
283  *rtop = *rit;
284  ++rtop;
285  }
286  }
287  read_vars->resize(rtop - read_vars->begin());
288  }
290  virtual int bulk_size() const {
291  return 0;
292  }
294  virtual int set_bulk_size(int) {
295  return 0;
296  }
297 }; // class Engine
298 #endif // DMLC_USE_CXX11
299 } // namespace mxnet
300 #endif // MXNET_ENGINE_H_
void DeduplicateVarHandle(std::vector< engine::VarHandle > *read_vars, std::vector< engine::VarHandle > *write_vars)
Definition: engine.h:270
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:121
std::function< void(RunContext)> SyncFn
Synchronous operation to pass to engine.
Definition: engine.h:103
std::function< void(RunContext, CallbackOnComplete)> AsyncFn
Asynchronous operation to pass to engine.
Definition: engine.h:105
namespace of mxnet
Definition: base.h:118
virtual int bulk_size() const
query current limit for bulk size
Definition: engine.h:290
Asynchronous function call.
CallbackOnComplete CreateCallback(void(*callback)(Engine *, void *), void *param)
factory function to create OnComplete callback.
Definition: engine.h:260
void operator()() const
involve the callback
Definition: engine.h:60
execution time context. The information needed in runtime for actual execution.
Definition: base.h:249
Delete variable call.
Normal operation.
virtual ~Engine() noexcept(false)
virtual destructor
Definition: engine.h:217
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:127
Copy operation from GPU to other devices.
virtual int set_bulk_size(int)
set maximum limit for bulk size
Definition: engine.h:294
engine::OprHandle OprHandle
Operator pointer.
Definition: engine.h:109
engine::VarHandle VarHandle
Variable pointer.
Definition: engine.h:107
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:101
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:243
Dependency engine that schedules operations.
Definition: engine.h:98
OnComplete Callback to the engine, called by AsyncFn when action completes.
Definition: engine.h:56
configuration of MXNet as well as basic data structure.
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:51