MS-376 Add loader and executor enable flag in Resource avoid diskresource execute task

Former-commit-id: 2ad10b9af0f131d33a10893c6bdc956e175089dd

cpp/CHANGELOG.md

@@ -27,6 +27,7 @@ Please mark all change in change log and use the ticket from JIRA.
 - MS-373 - Add resource test
 - MS-374 - Add action definition
 - MS-375 - Add Dump implementation for Event
+- MS-376 - Add loader and executor enable flag in Resource avoid diskresource execute task
 
 ## New Feature
 - MS-343 - Implement ResourceMgr

cpp/src/scheduler/resource/CpuResource.cpp

@@ -17,17 +17,6 @@ CpuResource::CpuResource(std::string name)
 
 void CpuResource::LoadFile(TaskPtr task) {
     task->Load(LoadType::DISK2CPU, 0);
-    //if (src.type == DISK) {
-    //    fd = open(filename);
-    //    content = fd.read();
-    //    close(fd);
-    //} else if (src.type == CPU) {
-    //    memcpy(src, dest, len);
-    //} else if (src.type == GPU) {
-    //    cudaMemcpyD2H(src, dest);
-    //} else {
-    //    // unknown type, exception
-    //}
 }
 
 void CpuResource::Process(TaskPtr task) {

cpp/src/scheduler/resource/DiskResource.cpp

@@ -12,7 +12,8 @@ namespace engine {
 
 
 DiskResource::DiskResource(std::string name)
-    : Resource(std::move(name), ResourceType::DISK) {}
+    : Resource(std::move(name), ResourceType::DISK, true, false) {
+}
 
 void DiskResource::LoadFile(TaskPtr task) {
 

cpp/src/scheduler/resource/Resource.cpp

@@ -10,10 +10,15 @@ namespace zilliz {
 namespace milvus {
 namespace engine {
 
-Resource::Resource(std::string name, ResourceType type)
+Resource::Resource(std::string name,
+                   ResourceType type,
+                   bool enable_loader,
+                   bool enable_executor)
     : name_(std::move(name)),
       type_(type),
       running_(false),
+      enable_loader_(enable_loader),
+      enable_executor_(enable_executor),
       load_flag_(false),
       exec_flag_(false) {
     task_table_.RegisterSubscriber([&] {
@@ -26,16 +31,24 @@ Resource::Resource(std::string name, ResourceType type)
 
 void Resource::Start() {
     running_ = true;
-    loader_thread_ = std::thread(&Resource::loader_function, this);
-    executor_thread_ = std::thread(&Resource::executor_function, this);
+    if (enable_loader_) {
+        loader_thread_ = std::thread(&Resource::loader_function, this);
+    }
+    if (enable_executor_) {
+        executor_thread_ = std::thread(&Resource::executor_function, this);
+    }
 }
 
 void Resource::Stop() {
     running_ = false;
-    WakeupLoader();
-    WakeupExecutor();
-    loader_thread_.join();
-    executor_thread_.join();
+    if (enable_loader_) {
+        WakeupLoader();
+        loader_thread_.join();
+    }
+    if (enable_executor_) {
+        WakeupExecutor();
+        executor_thread_.join();
+    }
 }
 
 TaskTable &Resource::task_table() {
@@ -106,6 +119,7 @@ void Resource::executor_function() {
         auto task_item = pick_task_execute();
         if (task_item) {
             Process(task_item->task);
+            task_item->state = TaskTableItemState::EXECUTED;
             if (subscriber_) {
                 auto event = std::make_shared<FinishTaskEvent>(shared_from_this(), task_item);
                 subscriber_(std::static_pointer_cast<Event>(event));

cpp/src/scheduler/resource/Resource.h

@@ -88,7 +88,10 @@ public:
     WakeupExecutor();
 
 protected:
-    Resource(std::string name, ResourceType type);
+    Resource(std::string name,
+             ResourceType type,
+             bool enable_loader = true,
+             bool enable_executor = true);
 
     // TODO: SearchContextPtr to TaskPtr
     /*
@@ -148,8 +151,8 @@ private:
     std::function<void(EventPtr)> subscriber_ = nullptr;
 
     bool running_;
-    bool loader_running_ = false;
-    bool executor_running_ = false;
+    bool enable_loader_ = true;
+    bool enable_executor_ = true;
     std::thread loader_thread_;
     std::thread executor_thread_;