[feature] non-blocking interfaces are provided to check whether the transmission task is completed

unifiedcache/csrc/ucmnfsstore/cc/api/ucmnfsstore/ucmnfsstore.cc

@@ -88,6 +88,11 @@ size_t Submit(std::list<TsfTask>& tasks, const size_t size, const size_t number,
 
 int32_t Wait(const size_t taskId) { return Singleton<TsfTaskManager>::Instance()->Wait(taskId).Underlying(); }
 
+int32_t Check(const size_t taskId, bool& finish)
+{
+    return Singleton<TsfTaskManager>::Instance()->Check(taskId, finish).Underlying();
+}
+
 void Commit(const std::string& blockId, const bool success)
 {
     auto s = Singleton<SpaceManager>::Instance()->CommitBlock(blockId, success);

unifiedcache/csrc/ucmnfsstore/cc/api/ucmnfsstore/ucmnfsstore.h

@@ -57,6 +57,7 @@ int32_t Alloc(const std::string& blockId);
 bool Lookup(const std::string& blockId);
 size_t Submit(std::list<TsfTask>& tasks, const size_t size, const size_t number, const std::string& brief);
 int32_t Wait(const size_t taskId);
+int32_t Check(const size_t taskId, bool& finish);
 void Commit(const std::string& blockId, const bool success);
 
 } // namespace UC

unifiedcache/csrc/ucmnfsstore/cc/domain/tsf_task/tsf_task_manager.cc

@@ -75,6 +75,15 @@ Status TsfTaskManager::Wait(const size_t taskId)
     return failure ? Status::Error() : Status::OK();
 }
 
+Status TsfTaskManager::Check(const size_t taskId, bool& finish)
+{
+    std::lock_guard<std::mutex> lk(this->_mutex);
+    auto iter = this->_waiters.find(taskId);
+    if (iter == this->_waiters.end()) { return Status::NotFound(); }
+    finish = iter->second->Finish();
+    return Status::OK();
+}
+
 void TsfTaskManager::Dispatch(std::list<TsfTask>& tasks, std::vector<std::list<TsfTask>>& targets, const size_t taskId,
                               std::shared_ptr<TsfTaskWaiter> waiter) const
 {

unifiedcache/csrc/ucmnfsstore/cc/domain/tsf_task/tsf_task_manager.h

@@ -38,6 +38,7 @@ class TsfTaskManager {
     Status Submit(std::list<TsfTask>& tasks, const size_t size, const size_t number, const std::string& brief,
                   size_t& taskId);
     Status Wait(const size_t taskId);
+    Status Check(const size_t taskId, bool& finish);
 
 private:
     void Dispatch(std::list<TsfTask>& tasks, std::vector<std::list<TsfTask>>& targets, const size_t taskId,

unifiedcache/csrc/ucmnfsstore/cc/domain/tsf_task/tsf_task_waiter.h

@@ -68,6 +68,7 @@ class TsfTaskWaiter : public Latch {
         }
         return finish;
     }
+    bool Finish() { return this->_counter == 0; }
 
 private:
     size_t _id;

unifiedcache/csrc/ucmnfsstore/cpy/ucmnfsstore.py.cc

@@ -131,6 +131,11 @@ PYBIND11_MODULE(ucmnfsstore, module)
     module.def("DumpFromDevice", &UC::DumpFromDevice);
     module.def("DumpFromHost", &UC::DumpFromHost);
     module.def("Wait", &UC::Wait);
+    module.def("Check", [](const size_t taskId) {
+        auto finish = false;
+        auto ret = UC::Check(taskId, finish);
+        return py::make_tuple(ret, finish);
+    });
     module.def("Commit", &UC::Commit);
     module.def("CommitBatch", &UC::CommitBatch);
 }

unifiedcache/csrc/ucmnfsstore/test/case/tsf_task_waiter_test.cc

@@ -36,13 +36,15 @@ TEST_F(UCTsfTaskWaiterTest, TaskTimeout)
     });
     ASSERT_FALSE(waiter.Wait(1));
     fut.get();
+    ASSERT_TRUE(waiter.Finish());
 }
 
 TEST_F(UCTsfTaskWaiterTest, TaskSuccess)
 {
     UC::TsfTaskWaiter waiter{2, 1024, 1, "xxx"};
     auto fut = std::async([&] { waiter.Done(); });
     ASSERT_TRUE(waiter.Wait(1000));
+    ASSERT_TRUE(waiter.Finish());
     fut.get();
 }
 
@@ -54,5 +56,6 @@ TEST_F(UCTsfTaskWaiterTest, TaskTimeoutButSuccess)
         waiter.Done();
     });
     fut.get();
+    ASSERT_TRUE(waiter.Finish());
     ASSERT_TRUE(waiter.Wait(1));
 }