!831 Check return code from WaitPost::Register

Merge pull request !831 from JesseKLee/check_rc

mindspore/ccsrc/dataset/engine/datasetops/source/celeba_op.cc

@@ -94,7 +94,7 @@ Status CelebAOp::LaunchThreadsAndInitOp() {
 
   RETURN_IF_NOT_OK(io_block_queues_.Register(tree_->AllTasks()));
   RETURN_IF_NOT_OK(attr_info_queue_->Register(tree_->AllTasks()));
-  wp_.Register(tree_->AllTasks());
+  RETURN_IF_NOT_OK(wp_.Register(tree_->AllTasks()));
 
   RETURN_IF_NOT_OK(tree_->AllTasks()->CreateAsyncTask("Walking attr file", std::bind(&CelebAOp::ParseAttrFile, this)));
   RETURN_IF_NOT_OK(tree_->LaunchWorkers(num_workers_, std::bind(&CelebAOp::WorkerEntry, this, std::placeholders::_1)));

mindspore/ccsrc/dataset/engine/datasetops/source/cifar_op.cc

@@ -149,7 +149,7 @@ Status CifarOp::LaunchThreadsAndInitOp() {
     RETURN_STATUS_UNEXPECTED("tree_ not set");
   }
   RETURN_IF_NOT_OK(io_block_queues_.Register(tree_->AllTasks()));
-  wp_.Register(tree_->AllTasks());
+  RETURN_IF_NOT_OK(wp_.Register(tree_->AllTasks()));
   RETURN_IF_NOT_OK(
     tree_->AllTasks()->CreateAsyncTask("Get cifar data block", std::bind(&CifarOp::ReadCifarBlockDataAsync, this)));
   RETURN_IF_NOT_OK(tree_->LaunchWorkers(num_workers_, std::bind(&CifarOp::WorkerEntry, this, std::placeholders::_1)));

mindspore/ccsrc/dataset/engine/datasetops/source/generator_op.cc

@@ -168,7 +168,7 @@ Status GeneratorOp::FillBuffer(TensorQTable *tt) {
 Status GeneratorOp::operator()() {
   // Handshake with TaskManager to synchronize thread creation
   TaskManager::FindMe()->Post();
-  wp_.Register(tree_->AllTasks());
+  RETURN_IF_NOT_OK(wp_.Register(tree_->AllTasks()));
   std::unique_ptr<DataBuffer> fetched_buffer;
   bool eof = false;
   while (!eof) {

mindspore/ccsrc/dataset/engine/datasetops/source/image_folder_op.cc

@@ -386,7 +386,7 @@ Status ImageFolderOp::LaunchThreadsAndInitOp() {
   RETURN_IF_NOT_OK(io_block_queues_.Register(tree_->AllTasks()));
   RETURN_IF_NOT_OK(folder_name_queue_->Register(tree_->AllTasks()));
   RETURN_IF_NOT_OK(image_name_queue_->Register(tree_->AllTasks()));
-  wp_.Register(tree_->AllTasks());
+  RETURN_IF_NOT_OK(wp_.Register(tree_->AllTasks()));
   // The following code launch 3 threads group
   // 1) A thread that walks all folders and push the folder names to a util:Queue mFoldernameQueue.
   // 2) Workers that pull foldername from mFoldernameQueue, walk it and return the sorted images to mImagenameQueue

mindspore/ccsrc/dataset/engine/datasetops/source/manifest_op.cc

@@ -140,7 +140,7 @@ Status ManifestOp::LaunchThreadsAndInitOp() {
     RETURN_STATUS_UNEXPECTED("tree_ not set");
   }
   RETURN_IF_NOT_OK(io_block_queues_.Register(tree_->AllTasks()));
-  wp_.Register(tree_->AllTasks());
+  RETURN_IF_NOT_OK(wp_.Register(tree_->AllTasks()));
 
   RETURN_IF_NOT_OK(
     tree_->LaunchWorkers(num_workers_, std::bind(&ManifestOp::WorkerEntry, this, std::placeholders::_1)));

mindspore/ccsrc/dataset/engine/datasetops/source/mindrecord_op.cc

@@ -644,7 +644,7 @@ Status MindRecordOp::LaunchThreadAndInitOp() {
   }
 
   RETURN_IF_NOT_OK(io_blk_queues_.Register(tree_->AllTasks()));
-  shard_reader_wait_post_.Register(tree_->AllTasks());
+  RETURN_IF_NOT_OK(shard_reader_wait_post_.Register(tree_->AllTasks()));
   if (shard_reader_->Launch(!block_reader_) == MSRStatus::FAILED) {
     RETURN_STATUS_UNEXPECTED("MindRecordOp launch failed.");
   }

mindspore/ccsrc/dataset/engine/datasetops/source/mnist_op.cc

@@ -395,7 +395,7 @@ Status MnistOp::LaunchThreadsAndInitOp() {
     RETURN_STATUS_UNEXPECTED("tree_ not set");
   }
   RETURN_IF_NOT_OK(io_block_queues_.Register(tree_->AllTasks()));
-  wp_.Register(tree_->AllTasks());
+  RETURN_IF_NOT_OK(wp_.Register(tree_->AllTasks()));
   RETURN_IF_NOT_OK(tree_->LaunchWorkers(num_workers_, std::bind(&MnistOp::WorkerEntry, this, std::placeholders::_1)));
   TaskManager::FindMe()->Post();
   RETURN_IF_NOT_OK(this->WalkAllFiles());

mindspore/ccsrc/dataset/engine/datasetops/source/text_file_op.cc

@@ -370,7 +370,7 @@ Status TextFileOp::operator()() {
   // must be called after launching workers.
   TaskManager::FindMe()->Post();
 
-  io_block_queue_wait_post_.Register(tree_->AllTasks());
+  RETURN_IF_NOT_OK(io_block_queue_wait_post_.Register(tree_->AllTasks()));
   NotifyToFillIOBlockQueue();
   while (!finished_reading_dataset_) {
     int64_t buffer_id = 0;

mindspore/ccsrc/dataset/engine/datasetops/source/tf_reader_op.cc

@@ -222,7 +222,7 @@ Status TFReaderOp::operator()() {
   // so workers have to be kept alive until the end of the program
   TaskManager::FindMe()->Post();
 
-  io_block_queue_wait_post_.Register(tree_->AllTasks());
+  RETURN_IF_NOT_OK(io_block_queue_wait_post_.Register(tree_->AllTasks()));
 
   NotifyToFillIOBlockQueue();
   while (!finished_reading_dataset_) {

mindspore/ccsrc/dataset/engine/datasetops/source/voc_op.cc

@@ -231,7 +231,7 @@ Status VOCOp::LaunchThreadsAndInitOp() {
     RETURN_STATUS_UNEXPECTED("tree_ not set");
   }
   RETURN_IF_NOT_OK(io_block_queues_.Register(tree_->AllTasks()));
-  wp_.Register(tree_->AllTasks());
+  RETURN_IF_NOT_OK(wp_.Register(tree_->AllTasks()));
   RETURN_IF_NOT_OK(tree_->LaunchWorkers(num_workers_, std::bind(&VOCOp::WorkerEntry, this, std::placeholders::_1)));
   TaskManager::FindMe()->Post();
   RETURN_IF_NOT_OK(this->ParseImageIds());

mindspore/ccsrc/dataset/util/CMakeLists.txt

@@ -3,7 +3,6 @@ add_library(utils OBJECT
     circular_pool.cc
     memory_pool.cc
     cond_var.cc
-    semaphore.cc
     intrp_service.cc
     task.cc
     task_manager.cc

mindspore/ccsrc/dataset/util/semaphore.cc

@@ -1,38 +0,0 @@
-/**
- * Copyright 2019 Huawei Technologies Co., Ltd
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-#include "dataset/util/semaphore.h"
-#include "dataset/util/task_manager.h"
-
-namespace mindspore {
-namespace dataset {
-Status Semaphore::P() {
-  std::unique_lock<std::mutex> lck(mutex_);
-  return (wait_cond_.Wait(&lck, [this]() { return value_ != 0; }));
-}
-
-void Semaphore::V() {
-  std::unique_lock<std::mutex> lck(mutex_);
-  ++value_;
-  wait_cond_.NotifyOne();
-}
-
-void Semaphore::Register(TaskGroup *vg) { (void)wait_cond_.Register(vg->GetIntrpService()); }
-
-Status Semaphore::Deregister() { return (wait_cond_.Deregister()); }
-
-void Semaphore::ResetIntrpState() { wait_cond_.ResetIntrpState(); }
-}  // namespace dataset
-}  // namespace mindspore

mindspore/ccsrc/dataset/util/semaphore.h

@@ -1,49 +0,0 @@
-/**
- * Copyright 2019 Huawei Technologies Co., Ltd
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-#ifndef DATASET_UTIL_SEMAPHORE_H_
-#define DATASET_UTIL_SEMAPHORE_H_
-
-#include "dataset/util/cond_var.h"
-
-namespace mindspore {
-namespace dataset {
-class TaskGroup;
-
-class Semaphore {
- public:
-  explicit Semaphore(int init) : value_(init) {}
-
-  virtual ~Semaphore() {}
-
-  Status P();
-
-  void V();
-
-  void Register(TaskGroup *vg);
-
-  Status Deregister();
-
-  void ResetIntrpState();
-
- private:
-  int value_;
-
-  std::mutex mutex_;
-  CondVar wait_cond_;
-};
-}  // namespace dataset
-}  // namespace mindspore
-#endif  // DATASET_UTIL_SEMAPHORE_H_

mindspore/ccsrc/dataset/util/task_manager.cc

@@ -53,7 +53,7 @@ Status TaskManager::CreateAsyncTask(const std::string &my_name, const std::funct
     LockGuard lck(&tg_lock_);
     this->grp_list_.insert(vg);
   }
-  (*task)->wp_.Register(vg);
+  RETURN_IF_NOT_OK((*task)->wp_.Register(vg));
   RETURN_IF_NOT_OK((*task)->Run());
   // Wait for the thread to initialize successfully.
   RETURN_IF_NOT_OK((*task)->Wait());

mindspore/ccsrc/dataset/util/wait_post.cc

@@ -36,7 +36,7 @@ void WaitPost::Clear() {
   value_ = 0;
 }
 
-void WaitPost::Register(TaskGroup *vg) { (void)wait_cond_.Register(vg->GetIntrpService()); }
+Status WaitPost::Register(TaskGroup *vg) { return wait_cond_.Register(vg->GetIntrpService()); }
 
 void WaitPost::ResetIntrpState() { wait_cond_.ResetIntrpState(); }
 

mindspore/ccsrc/dataset/util/wait_post.h

@@ -36,7 +36,7 @@ class WaitPost {
 
   void Clear();
 
-  void Register(TaskGroup *vg);
+  Status Register(TaskGroup *vg);
 
   Status Deregister();
 

tests/ut/cpp/dataset/interrupt_test.cc

@@ -20,7 +20,6 @@
 #include "dataset/util/intrp_service.h"
 #include "dataset/util/task_manager.h"
 #include "dataset/util/queue.h"
-#include "dataset/util/semaphore.h"
 
 using namespace mindspore::dataset;
 using mindspore::MsLogLevel::INFO;
@@ -55,11 +54,12 @@ TEST_F(MindDataTestIntrpService, Test1) {
 TEST_F(MindDataTestIntrpService, Test2) {
   MS_LOG(INFO) << "Test Semaphore";
   Status rc;
-  Semaphore sem(0);
+  WaitPost wp;
-  sem.Register(&vg_);
+  rc = wp.Register(&vg_);
+  EXPECT_TRUE(rc.IsOk());
   vg_.CreateAsyncTask("Test1", [&]() -> Status {
     TaskManager::FindMe()->Post();
-      Status rc = sem.P();
+      Status rc = wp.Wait();
       EXPECT_TRUE(rc.IsInterrupted());
       return rc;
   });