转载

源码解析Mxnet Dependency Engine

Var

var可以看做是一个tag,用来标示每一个对象的,这样Op对对象的依赖可以简化成对var的 依赖,这样就可以构建出一个不依赖于具体的对象的通用的依赖引擎。Var是依赖引擎的关键。

类图

源码解析Mxnet Dependency Engine

声明:下文说到执行时,意思是Op的当前var的依赖已经就绪,因为一个op可以依赖多个 var,如果其他的Var没有就绪,那么这时op可能并没有实际运行

Var只是一个基类,用来统一类型系统的,主要的工作在 ThreadedVar 中,每一个对象都 会有一个由 VersionedVarBlock 所组成的链表,这个链表就是一个FIFO队列。 head_ 指向的是队列的尾部, 实际是一个哨兵(空对象), head_ 这个命名有误导性, pending_write_ 指向的是最"老"的写依赖,如果没有写依赖,那么就指向 nullptr , 根据依赖引擎的特点,它实际上指向的是队列的头部, ThreadedVar 的那四个方法就是 来操作这个队列的。

  1. num_pending_reads_: 代表当前正在执行(还没有执行完)的读依赖的个数
  2. pending_write_: 代表队列中最“老”的写依赖, 它一直指向队列的头部。
  3. head_: 队列的尾部。

需要注意的是,正在执行的读依赖是不在队列中的,但是正在执行的写依赖是在队列中的。

理解Var的队列

var的队列是依赖引擎的核心,下面我们来分析下各种情况下,如何修改队列的状态。

  1. 添加读依赖: 如果前面没有写依赖,那么直接运行, 否则就插入队列的尾部(head_那一端)
  2. 添加写依赖: 直接将依赖插入队列的尾部,并检查是不是写就绪(既没有读依赖也没有 写依赖在运行),如果是写就绪,那么就运行该依赖。
  3. 读依赖完成
  4. 写依赖完成

源码解析Mxnet Dependency Engine

上图中w1写依赖正在执行。

源码解析Mxnet Dependency Engine 写依赖w1完成将自己移出队列,并执行写依赖w2

源码解析Mxnet Dependency Engine

写依赖w2完成后将自己移出队列,接着并行的执行读依赖r1,r2,记住正在执行的读依赖是被移出队列的, 它们的数目使用 num_pending_reads_ 跟踪的

源码解析Mxnet Dependency Engine

每一个读依赖完成都会将 num_pending_reads_ 减一,如果减为了0,那么就意味着所有 的读依赖都完成了,当r1,r2都完成后,接着执行w3写依赖。

添加读依赖

代码主要在 src/engine/Threaded_engine.ccAppendReadDependency 中。

inline void ThreadedVar::AppendReadDependency(OprBlock* opr_block) {     std::lock_guard<std::mutex> lock{m_};     if (pending_write_ == nullptr) {         // invariant: is_ready_to_read()         CHECK_GE(num_pending_reads_, 0);         // STATE CHANGE         ++num_pending_reads_;         // decrease wait counter         opr_block->decr_wait();     } else {         auto&& new_var_block = VersionedVarBlock::New();         assert(head_->next == nullptr);         assert(head_->trigger == nullptr);         assert(head_->write == false);         // append things to next.         head_->next = new_var_block;         head_->trigger = opr_block;         head_ = new_var_block;     } }

代码的基本思路是这样的:检查队列中有没有写依赖,这分两种情况:

  1. 如果没有写依赖,那么意味着,目前该Var没有依赖在执行,或者说只有读依赖在执行, 所以这个新的读依赖可以直接执行,那么它没有必要添加到队列中,只需要更新 num_pending_reads_ 就好,当然因为该op可能还依赖别的var,所以你只能调用 decr_wait ,只有当wait减为0的时候,才能开始运行。这部分代码在engine的push中。
  2. 如果有写依赖,那么读依赖必须在写依赖的后面执行,所以需要把读依赖添加到队列的 尾部。记住 head_ 永远指向一个空的哨兵对象。

添加写依赖

代码主要在 src/engine/Threaded_engine.ccAppendWriteDependency 中。

inline void ThreadedVar::AppendWriteDependency(OprBlock* opr_block) {     auto&& new_var_block = VersionedVarBlock::New();     std::lock_guard<std::mutex> lock{m_};     // invariant.     assert(head_->next == nullptr);     assert(head_->trigger == nullptr);     assert(head_->write == false);     // attach to head.     head_->next = new_var_block;     head_->trigger = opr_block;     head_->write = true;      // check if it is ready to write     if (pending_write_ == nullptr) {         // invariant: is_ready_to_read()         pending_write_ = head_;         CHECK_GE(num_pending_reads_, 0);         if (num_pending_reads_ == 0) {             // STATE CHANGE             opr_block->decr_wait();             num_pending_reads_ = kWriteTriggered;         }     } else {         CHECK_NE(num_pending_reads_, 0);     }     head_ = new_var_block; }

代码的基本思路是这样的: 将该Op放入队列的尾部,接着检查该Op的依赖有没有就绪,这 要检查Var有没有写依赖(pending_read_==nullptr)和读依赖(num_pending_read_==0)的Op 正在执行,只有二者都没有时,才能开始运行,当然你依然要检查该Op对其他的Var的依赖 有没有就绪。需要注意的一点是,即便Op的Var写依赖就绪,该Op也不会从队列中移除,只 有该Op执行完成后才会被移除,这在CompleteWriteDependency中实现。

读依赖完成

代码主要在 src/engine/Threaded_engine.ccCompleteReadDependency 中。

template <typename Dispatcher> inline void ThreadedVar::CompleteReadDependency(Dispatcher dispatcher) {     OprBlock *trigger = nullptr;     {         // this is lock scope         std::lock_guard<std::mutex> lock{m_};         CHECK_GT(num_pending_reads_, 0);          if (--num_pending_reads_ == 0) {             if (pending_write_ != nullptr) {                 // STATE CHANGE                 trigger = pending_write_->trigger;                 num_pending_reads_ = kWriteTriggered;             }         }     }     if (trigger != nullptr && trigger->decr_wait() == 0) {         dispatcher(trigger);     } }

该部分代码会在一个op运算完成后调用,代码逻辑是比较简单的,先更新 num_pending_read_ , 更新后如果该值为0,那么就意味着,所有的读依赖都已经执行完成, 这样就检查队列,若是存在写依赖,那么该写依赖就就绪了,那么Op就可以执行了(前提是 依赖的其他var也都就绪了, wait为0)。上面的dispatcher实际就是用来将Op丢入执行引擎 的,它一般是PushToExecute,这个后文会看到。

写依赖完成

代码主要在 src/engine/Threaded_engine.ccCompleteWriteDependency 中。

template <typename Dispatcher> inline bool ThreadedVar::CompleteWriteDependency(Dispatcher dispatcher) {   // this is lock scope   VersionedVarBlock *old_pending_write, *end_of_read_chain;   OprBlock* trigger_write = nullptr;   {     std::lock_guard<std::mutex> lock{m_};     // invariants     assert(head_->next == nullptr);     assert(pending_write_ != nullptr);     CHECK_EQ(num_pending_reads_, kWriteTriggered);      // really delete     if (to_delete_) {       VersionedVarBlock *head = pending_write_->next;       VersionedVarBlock::Delete(pending_write_);       assert(head_ == head);       VersionedVarBlock::Delete(head);       return true;     }     // detach pending write     old_pending_write = pending_write_;     // search for chains to trigger     end_of_read_chain = old_pending_write->next;     // reset to 0 pending reads     num_pending_reads_ = 0;     while (end_of_read_chain != head_ &&            end_of_read_chain->write == false) {       ++num_pending_reads_;       end_of_read_chain = end_of_read_chain->next;     }     if (end_of_read_chain == head_) {       pending_write_ = nullptr;     } else {       // check if there is pending reads, if not trigger write       assert(end_of_read_chain->write == true);       pending_write_ = end_of_read_chain;       if (num_pending_reads_ == 0) {         // mark write as already actived in this var         num_pending_reads_ = kWriteTriggered;         trigger_write = end_of_read_chain->trigger;       }     }   }   // This is outside of lock scope   // Be very carful, pending_write_ and num_pending_reads_   // can change now, do not reply ont the two variables.   // The linked list /in [old_pending_write, end_of_read_chain)   // is already detached from this Var.   // So it is safe to modify these   VersionedVarBlock *cur_head = old_pending_write->next;   VersionedVarBlock::Delete(old_pending_write);   // dispatch all the events   while (cur_head != end_of_read_chain) {     if (cur_head->trigger->decr_wait() == 0) {       dispatcher(cur_head->trigger);     }     auto prev = cur_head;     cur_head = cur_head->next;     assert(cur_head != nullptr);     VersionedVarBlock::Delete(prev);   }   if (trigger_write != nullptr && trigger_write->decr_wait() == 0) {     dispatcher(trigger_write);   }   return false; }

和读依赖完成类似,只是写依赖的后面可能跟着多个读依赖,所以需要遍历链表直到发现下 一个写依赖,找到之后如果是读依赖,那么直接并行的运行,如果是写依赖直接运行就好。

原文  http://yuyang0.github.io/notes/mxnet-engine.html
正文到此结束
Loading...