TaskTable.h 4.4 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();
73 74 75

    std::string
    Dump();
W
wxyu 已提交
76 77
};

W
wxyu 已提交
78 79
using TaskTableItemPtr = std::shared_ptr<TaskTableItem>;

W
wxyu 已提交
80 81 82 83
class TaskTable {
public:
    TaskTable() = default;

84 85 86
    TaskTable(const TaskTable &) = delete;
    TaskTable(TaskTable &&) = delete;

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

W
wxyu 已提交
92 93 94 95
    /*
     * Put one task;
     */
    void
W
wxyu 已提交
96
    Put(TaskPtr task);
W
wxyu 已提交
97 98 99 100 101 102

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

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

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

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

W
wxyu 已提交
127 128 129
    /*
     * Return size of task table;
     */
W
wxyu 已提交
130
    inline size_t
W
wxyu 已提交
131 132 133
    Size() {
        return table_.size();
    }
W
wxyu 已提交
134 135 136 137 138 139 140 141
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 已提交
142 143 144 145 146 147 148 149 150 151 152

public:

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

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

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

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

    /*
     * Execute task finished;
     * Set state executed;
     * Called by executor;
     */
183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207
    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 已提交
208 209 210 211 212 213 214 215 216 217

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

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


}
}
}