标签：bool   cer   gops   dex   简单   键值   state   ted   other   

目录

1. 核心概念
2. executor.h
   1. Executor
   2. NewLocalExecutor
   3. ExecutorBarrier
3. executor.cc
   1. structs
   2. GraphView
   3. ExecutorImpl
   4. ExecutorState
   5. details

3.4 ExecutorState

在执行器的执行图计算的时候，需要一个结构来保存当前计算的即时信息，TF为此设计了类ExecutorState，它被用来保存每一个对ExecutorImpl::Run调用的状态信息。它会在一个节点已经准备好之后调度这个节点，并且保存每个节点尚未完成的输入信息。
下面让我们先来看一下这个类的结构：

class ExecutorState {
  public:
    ExecutorState(const Executor::Args& args, ExecutorImpl* impl);
    void RunAsync(Executor::DoneCallback done);
  private:
    DeviceContextMap device_context_map_;
    
    typedef gtl::InlinedVector<TaggedNode, 8> TaggedNodeSeq;
    typedef gtl::InlinedVector<Entry, 4> EntryVector;
    
    const bool vlog_;
    const bool log_memory_;
    int64 step_id_;
    
    //未拥有
    Rendezvous* rendezvous;
    SessionState* session_state_;
    TensorStore* tensor_store_;
    //每个执行步级别的容器
    ScopedStepContainer* step_container_;
    StepStatesCollector* stats_collector_;
    
    checkpoint::TensorSliceReaderCacheWrapper* slice_reader_cache_;
    FunctionCallFrame* call_frame;
    const ExecutorImpl* impl_;
    CancellationManager* cancellation_manager_;
    Executor::Args::Runner runner_;
    bool sync_on_finish_;
    
    //拥有
    bool dumped_on_error_ = false;
    //当前执行步骤开始的帧
    FrameState* root_frame_;
    //执行器结束时需要调用的回调函数
    Executor::DoneCallback done_cb_;
    std::atomic_int_fast32_t num_outstanding_ops_;
    mutex mu_;
    Status status_ GUARDED_BY(mu_);
    
    //从帧名称到实际帧的映射。在当前帧的某个迭代周期内，可能会产生一个新的帧。新的子帧的唯一键值必须由父帧的名称、迭代编号、以及由nodedef推断出来的新帧的名称组合而成
    gtl::FlatMap<string, FrameState*> outstanding_frames_ GUARDED_BY(mu_);
    
    //一个帧的名称
    inline string MakeFrameName(FrameState* frame, int64 iter_id, const string& name);
    
    //找到一个现存的帧，或者创建一个新帧，在帧frame的iter迭代周期
    void FindOrCreateChildFrame(FrameState* frame, int64 iter, const Node* node, FrameState** child);
    
    //删除一个帧，当帧调用结束时使用
    void DeleteFrame(FrameState* frame, TaggedNodeSeq* ready);
    
    //清除那些起源于帧frame和迭代iter的帧，当一个子帧结束时调用
    void CleanupFramesIterations(FrameState* frame, int64 iter, TaggedNodeSeq* ready);
    
    //在当前的线程中处理一个已准备好的节点
    void Process(TaggedNode node, int64 scheduled_usec);
    
    //在调用item->kernel之前，先填入其输入
    Status PrepareInputs(const NodeItem& item, Entry* first_input, TensorValueVec* inputs, DeviceContextVec* input_device_contexts, AllocatorAttributeVec* input_alloc_attrs, bool* is_input_dead);
    
    //在item->kernel计算结束之后，处理输出
    Status ProcessOutputs(const NodeItem& item, OpKernelContext* ctx, EntryVector* outputs, NodeExecStats* stats);
    
    //在处理完输出之后，将输入传递给下一个输入
    void PropagateOutputs(const TaggedNode& tagged_node, const NodeItem* item, EntryVector* outputs, TaggedNodeSeq* ready);
    
    //节点计算结束后，接管stats，如果执行完成则返回true
    bool NodeDone(const Status& s, const Node* node, const TaggedNodeSeq& ready, NodeExecStats* stats, TaggedNodeReadyQueue* inline_ready);
    
    //调度ready中的所有复杂节点，然后将ready中的非复杂节点放入inline_ready
    void ScheduleReady(const TaggedNodeSeq& ready, TaggedNodeReadyQueue* inline_ready);
    
    //仅用作调试或记录
    inline void MaybeMarkCompleted(FrameState* frame, int64 iter, int64 id);
    
    //输出一个未完成或者活跃节点的信息
    void DumpPendingNodeState(const int node_id, const Entry* input_vector, bool show_nodes_with_no_ready_inputs);
    void DumpActiveNodeState(const FrameState* frame, IterationState* iteration);
    
    //提供执行器的状态信息
    void DumpState();
    const Tensor* GetTensorValueForDump(const Entry& input);
    
    //当执行器结束执行时，清理
    void Finish();
};

从API上来看，ExecutorState几乎担当了执行器的职责，从后面的介绍也可以看出，实际上确实如此。执行器内部实际调用的就是ExecutorState内部的API。从类的结构中，我们还是看到了许多未曾相识的结构，下面我们先一一分析这些类的意义和结构。

首先来看Entry，Entry要么是一个张量指针，要么是一个张量值，为计算图中的节点的输入或输出提供了一种统一的类型。

struct Entry {
    Entry(const Entry& other);
    Entry& operator=(const Entry& other);
    
    void ClearVal();//清除val字段
    ManualConstructor<Tensor> val;//一个张量的值，如果val_filed_is_set是true的话
    Tensor* ref = nullptr;//一个张量引用
    mutext* ref_mu = nullptr;//为上述张量引用的互斥量
    bool has_value = false;//值是否存在，不论是val或者ref
    bool val_filed_is_set = false;//val字段是否被设置
    
    AllocatorAttributes alloc_attr;//为当前的张量分配内存的内存分配器的属性
    
    DeviceContext* device_context = nullptr;//包含了关于这个张量如何创建的设备相关的信息
};

接下来看看IterationState，它代表了一轮迭代的状态。

struct IterationState {
  public:
    //一轮迭代的状态，每个迭代轮次都由一个单独的拷贝。对于第k轮迭代，第i个节点的第j个输入在input_tensors[k][impl_->nodes[i].input_start+j]。注意，没有必要对input_tensors做互斥锁，其中的内容只会被边的前一个节点写入，被边的后一个节点擦除，而每条边的前后两个节点是不可能同时运行的
    Entry* input_tensors;
    
    //每一轮迭代中未完成的op数量
    size_t outstanding_ops;
    
    //每一轮迭代中未完成的帧数量
    int outstanding_frame_count;
    int pending(PendingCounts::Handle h);
    int decrement_pending(PendingCounts::Handle int v);
    
    //标记一个merge节点为live
    void mark_live(PendingCounts::Handle h);
    //标记一个节点为处理开始
    void mark_started(PendingCounts::Handle h);
    //标记一个节点为处理结束
    void mark_completed(PendingCounts::Handle h);
    //获取节点状态
    PendingCounts::NodeState node_state(PendingCounts::Handle h);
    int dead_count(PendingCounts::Handle h);
    void increment_dead_count(PendingCounts::Handle h);
    void adjust_for_activation(PendingCounts::Handle h, bool increment_dead, int* pending_result, int* dead_result);
  
  private:
    PendingCounts counts_;
};

接下来是FrameState，代表了一个帧的状态。对于帧和迭代轮次，有以下几点需要说明：

• 对于计算图中的循环来说，每个循环都需要创建一个新的帧。执行从第0个迭代开始。当第0个迭代的某个数值通过了一个NextIteration节点时，第1轮迭代就被创建并开始运行了。注意这时第0轮迭代可能仍在进行，所以多轮迭代可能会同时在运行。帧保持了多种数据结构来保存每轮迭代的状态。当第0轮迭代结束后，我们对其对应的状态进行垃圾回收。
• 一个帧，当它的所有输入都已经被传入，所有的迭代都被计算完成时，这个帧就被认为是完成了，可以被进行垃圾回收了。
• 一个帧保存了其中每一轮迭代的状态。如果以下三个条件都被满足，那么第i轮迭代就会被认为是已经完成了，第一，第i轮迭代已经没有未完成的节点了，第二，所有该轮的接收操作都已经完成了，第三，第i-1轮已完成。对于第0轮迭代，当帧的所有输入都已完成，我们就认为它已经结束了。
• 帧和迭代轮次在结束后，都会进行垃圾回收。我们需要保存的状态量，跟调度器允许的并行度高度相关。我们希望调度器能够动态的控制未完成的并行帧和迭代的数量。为了减少内存消耗，调度器可能需要优先调度内层的帧和较低的迭代轮次。
• 帧的状态一般总是在需要的时候才会被初始化，因此我们没有引入额外的损耗。

下面我们来具体看下FrameState的结构：

struct FrameState {
    const ExecutorImpl* executor = nullptr;//帧所在的执行器
    string frame_name;//当前帧的名称，是父帧，迭代轮次，和frame_name字段拼合起来得到的
    uint64 frame_id;//当前帧的唯一标识
    int64 parent_iter = -1;//父帧的迭代轮次，frame_name和parent_iter共同标识了当前的FrameState
    FrameState* parent_frame = nullptr;//父帧的FrameState
    const int 
    
    max_parallel_iterations;//最大允许的并行迭代数量
    int num_pending_inputs = 0;//当前帧仍然在等待的输入数量
    int64 iteration_count GUARDED_BY(mu) = 0;//当前帧中到达过的最大的迭代数量
    int num_outstanding_iterations GUARDED_BY(mu) = 1;//未完成的迭代数量
    
    gtl::InlinedVecotr<IterationState*,12> iterations;//当前帧活跃的迭代状态
    std::vector<std::pair<const Node*, Entry>> next_iter_roots GUARDED_BY(mu);
    std::vector<std::pair<const Node*, Entry>> inv_values GUARDED_BY(mu);
    std::vector<const Node*> dead_exits GUARDED_BY(mu);
    
    //属于当前帧的静态信息
    PendingCounts* pending_counts = nullptr;
    int total_input_tensors = 0;
    std::vector<const Node*>* nodes = nullptr;
    
    void InitializeFrameInfo(const string& enter_name);
    inline IterationState* GetInteration(int64 iter);
    inline void SetIteration(int64 iter, IterationState* state);
    
    //减少未完成的操作数量，清理帧中的迭代信息。如果帧执行结束则返回true
    inline bool DecrementOutputstandingOps(const GraphView* gview, int64 iter, TaggedNodeSeq* ready);
    inline bool DecrementOutstandingOpsLocked(const GraphView* gview, int64 iter, TaggedNodeSeq* ready);
    
    //如果帧中的计算都已经完成则返回true
    inline bool IsFrameDone();
    //如果迭代的计算已经结束则返回true
    bool IsIterationDone(int64 iter);
    //增加迭代的编号，如果是一个新迭代，就初始化它
    void IncrementIteration(const GraphView* gview, TaggedNodeSeq* ready);
    //激活一个新的迭代轮次中所有的NextIteration节点
    void ActivateNexts(const GraphView* gview, int64 iter, TaggedNodeSeq* ready);
    void ActivateLoopInvs(const GraphView* gview, int64 iter, TaggedNodeSeq* ready);
    void AddLoopInv(const NodeItem* item, const Entry& value, TaggedNodeSeq* ready);
    void ActivateNodes(const NodeItem* item, const bool is_dead, int64 iter, EntryVector* outputs, TaggedNodeSeq* ready);
    bool CleanupIterations(const GraphView* gview, int64 iter, TaggedNodeSeq* ready);
};

最后让我们来看下最后的两个结构体，TaggedNode和TaggedNodeReadyQueue。其中TaggedNode非常简单，就是一个<frame, iter, node>的结构体，而后者就是前者的一个Queue，用来表示已经准备好的节点的队列。

struct TaggedNode {
    const Node* node = nullptr;
    FrameState* input_frame = nullptr;
    int64 input_iter = -1;
    bool is_dead = false;
    
    TaggedNode(const Node* t_node, FrameState* in_frame, int64 in_iter, bool dead);
};
class TaggedNodeReadyQueue {
  public:
    void push_back(TaggedNode node);
    void pop_front();
    bool empty();
    const TaggedNode* begin();
    const TaggedNode* end();
  private:
    gtl::InlinedVector<TaggedNode, 16> ready_;
    int front_index_;
};

关于TaggedNodeReadyQueue，我们要说明一下，本来这里很自然的可以使用std::deque

3.5 details

终于快要接近终点了。在前文中我们讲了那么多结构，最终计算图的执行过程究竟是怎样的，我们仍然不得而知。因为具体的实现细节都隐藏在函数的实现中，而我们上文中全部都在探讨接口。现在我们就来看下，具体的实现方法。
首先，执行器的入口是Run函数，先来看下ExecutorImpl中的Run函数是如何实现的吧。

void ExecutorImpl::RunAsync(const Args& args, DoneCallback done){
    (new ExecutorState(args,this))->RunAsync(std::move(done));
}

这验证了我们上文中提到的，ExecutorImpl仍然只是一个接口，真正的执行是被推到ExecutorState类中完成的。在上述函数中，我们首先定义了一个ExecutorState对象，然后调用了它的RunAsync函数。在构造函数中，首先初始化了root_frame和iteration 0，我们具体看看RunAsync是如何实现的：

void ExecutorState::RunAsync(Executor::DoneCallback done){
    const Graph* graph = impl_->graph_;//获取计算图指针
    TaggedNodeSeq ready;//构建ready节点序列
    
    //让设备填充设备上下文映射
    Device* device = impl_->params_.device;
    Status fill_status = device->FillContextMap(graph, &device_context_map_);
    if(!fill_status.ok()){
        done(fill_status);
        return;
    }
    
    //初始化ready队列
    for(const Node* n : impl_->root_nodes){
        DCHECK_EQ(n->in_edges().size(),0);
        ready.push_back(TaggedNode{n,root_frame_,0,false});
    }
    if(ready.empty()){
        done(Status::OK());
    } else {
        num_outstanding_ops = ready.size();
        root_frame_->iterations[0]->outstanding_ops = ready.size();
        done_cb_ = std::move(done);
        ScheduleReady(ready,nullptr);
    }
}

可见，主要做了两件事，第一是初始化了ready queue，第二是启动了ScheduleReady函数。
下面我们再来看一下SheduleReady函数的运行机制：

tensorflow源码解析之common_runtime-executor-下

标签：bool   cer   gops   dex   简单   键值   state   ted   other   

原文地址：https://www.cnblogs.com/jicanghai/p/9572217.html