TaskTable.h 4.3 KB
Newer Older
W
wxyu 已提交
1 2 3 4 5 6 7 8 9 10 11
/*******************************************************************************
 * Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
 * Unauthorized copying of this file, via any medium is strictly prohibited.
 * Proprietary and confidential.
 ******************************************************************************/
#pragma once

#include <vector>
#include <deque>
#include <mutex>

12
#include "task/SearchTask.h"
W
wxyu 已提交
13
#include "event/Event.h"
W
wxyu 已提交
14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30


namespace zilliz {
namespace milvus {
namespace engine {

enum class TaskTableItemState {
    INVALID,
    START, // idle
    LOADING, // loading data from other resource
    LOADED, // ready to exec or move
    EXECUTING, // executing, locking util executed or failed
    EXECUTED, // executed, termination state
    MOVING, // moving to another resource, locking util executed or failed
    MOVED, // moved, termination state
};

31 32 33 34 35 36 37 38 39 40
struct TaskTimestamp {
    uint64_t start = 0;
    uint64_t move = 0;
    uint64_t moved = 0;
    uint64_t load = 0;
    uint64_t loaded = 0;
    uint64_t execute = 0;
    uint64_t executed = 0;
};

W
wxyu 已提交
41 42 43 44
struct TaskTableItem {
    TaskTableItem() : id(0), state(TaskTableItemState::INVALID), mutex(), priority(0) {}

    TaskTableItem(const TaskTableItem &src)
W
wxyu 已提交
45
        : id(src.id), state(src.state), mutex(), priority(src.priority) {}
W
wxyu 已提交
46 47 48 49 50 51

    uint64_t id; // auto increment from 0;
    // TODO: add tag into task
    TaskPtr task; // the task;
    TaskTableItemState state; // the state;
    std::mutex mutex;
52
    TaskTimestamp timestamp;
W
wxyu 已提交
53 54

    uint8_t priority; // just a number, meaningless;
55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72

    bool
    Load();

    bool
    Loaded();

    bool
    Execute();

    bool
    Executed();

    bool
    Move();

    bool
    Moved();
W
wxyu 已提交
73 74
};

W
wxyu 已提交
75 76
using TaskTableItemPtr = std::shared_ptr<TaskTableItem>;

W
wxyu 已提交
77 78 79 80
class TaskTable {
public:
    TaskTable() = default;

81 82 83
    TaskTable(const TaskTable &) = delete;
    TaskTable(TaskTable &&) = delete;

W
wxyu 已提交
84 85 86 87 88
    inline void
    RegisterSubscriber(std::function<void(void)> subscriber) {
        subscriber_ = std::move(subscriber);
    }

W
wxyu 已提交
89 90 91 92
    /*
     * Put one task;
     */
    void
W
wxyu 已提交
93
    Put(TaskPtr task);
W
wxyu 已提交
94 95 96 97 98 99

    /*
     * Put tasks back of task table;
     * Called by DBImpl;
     */
    void
W
wxyu 已提交
100
    Put(std::vector<TaskPtr> &tasks);
W
wxyu 已提交
101 102 103 104

    /*
     * Return task table item reference;
     */
W
wxyu 已提交
105
    TaskTableItemPtr
W
wxyu 已提交
106
    Get(uint64_t index);
W
wxyu 已提交
107 108 109 110 111 112 113

    /*
     * TODO
     * Remove sequence task which is DONE or MOVED from front;
     * Called by ?
     */
    void
W
wxyu 已提交
114
    Clear();
W
wxyu 已提交
115

W
wxyu 已提交
116 117 118 119 120 121 122
    /*
     * Return true if task table empty, otherwise false;
     */
    inline bool
    Empty() {
        return table_.empty();
    }
W
wxyu 已提交
123

W
wxyu 已提交
124 125 126
    /*
     * Return size of task table;
     */
W
wxyu 已提交
127
    inline size_t
W
wxyu 已提交
128 129 130
    Size() {
        return table_.size();
    }
W
wxyu 已提交
131 132 133 134 135 136 137 138
public:
    TaskTableItemPtr &
    operator[](uint64_t index) {
        return table_[index];
    }

    std::deque<TaskTableItemPtr>::iterator begin() { return table_.begin(); }
    std::deque<TaskTableItemPtr>::iterator end() { return table_.end(); }
W
wxyu 已提交
139 140 141 142 143 144 145 146 147 148 149

public:

    /******** Action ********/

    // TODO: bool to Status
    /*
     * Load a task;
     * Set state loading;
     * Called by loader;
     */
150 151 152 153
    inline bool
    Load(uint64_t index) {
        return table_[index]->Load();
    }
W
wxyu 已提交
154 155 156 157 158 159

    /*
     * Load task finished;
     * Set state loaded;
     * Called by loader;
     */
160 161 162 163
    inline bool
    Loaded(uint64_t index) {
        return table_[index]->Loaded();
    }
W
wxyu 已提交
164 165 166 167 168 169

    /*
     * Execute a task;
     * Set state executing;
     * Called by executor;
     */
170 171 172 173
    inline bool
    Execute(uint64_t index) {
        return table_[index]->Execute();
    }
W
wxyu 已提交
174 175 176 177 178 179

    /*
     * Execute task finished;
     * Set state executed;
     * Called by executor;
     */
180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204
    inline bool
    Executed(uint64_t index){
        return table_[index]->Executed();
    }

    /*
     * Move a task;
     * Set state moving;
     * Called by scheduler;
     */

    inline bool
    Move(uint64_t index){
        return table_[index]->Move();
    }

    /*
     * Move task finished;
     * Set state moved;
     * Called by scheduler;
     */
    inline bool
    Moved(uint64_t index){
        return table_[index]->Moved();
    }
W
wxyu 已提交
205 206 207 208 209 210 211 212 213 214

public:
    /*
     * Dump;
     */
    std::string
    Dump();

private:
    // TODO: map better ?
215 216
    std::uint64_t id_ = 0;
    mutable std::mutex id_mutex_;
W
wxyu 已提交
217
    std::deque<TaskTableItemPtr> table_;
W
wxyu 已提交
218
    std::function<void(void)> subscriber_ = nullptr;
W
wxyu 已提交
219 220 221 222 223 224
};


}
}
}