fix sink mode pyfunc raise error issue and profiling file exception scenario

mindspore/ccsrc/minddata/dataset/engine/perf/connector_size.cc

@@ -68,8 +68,13 @@ Status ConnectorSize::SaveToFile() {
   json output;
   if (path.Exists()) {
     MS_LOG(DEBUG) << file_path_ << " exists";
-    std::ifstream file(file_path_);
-    file >> output;
+    try {
+      std::ifstream file(file_path_);
+      file >> output;
+    } catch (const std::exception &err) {
+      RETURN_STATUS_UNEXPECTED("Invalid file, failed to open json file: " + file_path_ +
+                               ", please delete it and try again!");
+    }
   } else {
     output["sampling_interval"] = GlobalContext::config_manager()->monitor_sampling_interval();
   }

mindspore/ccsrc/minddata/dataset/engine/perf/connector_throughput.cc

@@ -97,8 +97,13 @@ Status ConnectorThroughput::SaveToFile() {
   json output;
   if (path.Exists()) {
     MS_LOG(DEBUG) << file_path_ << " exists";
-    std::ifstream file(file_path_);
-    file >> output;
+    try {
+      std::ifstream file(file_path_);
+      file >> output;
+    } catch (const std::exception &err) {
+      RETURN_STATUS_UNEXPECTED("Invalid file, failed to open json file: " + file_path_ +
+                               ", please delete it and try again!");
+    }
   } else {
     output["sampling_interval"] = GlobalContext::config_manager()->monitor_sampling_interval();
   }

mindspore/ccsrc/minddata/dataset/engine/perf/monitor.cc

@@ -37,7 +37,7 @@ Status Monitor::operator()() {
   // 2) Iterator has not received EOF
   while (!this_thread::is_interrupted() && !(tree_->isFinished())) {
     if (tree_->IsEpochEnd()) {
-      tree_->GetProfilingManager()->SaveProfilingData();
+      RETURN_IF_NOT_OK(tree_->GetProfilingManager()->SaveProfilingData());
       tree_->SetExecuting();
     }
     for (auto &node : tree_->GetProfilingManager()->GetSamplingNodes()) {
@@ -47,8 +47,8 @@ Status Monitor::operator()() {
   }
 
   // Output all profiling data upon request.
-  tree_->GetProfilingManager()->SaveProfilingData();
-  tree_->GetProfilingManager()->ChangeFileMode();
+  RETURN_IF_NOT_OK(tree_->GetProfilingManager()->SaveProfilingData());
+  RETURN_IF_NOT_OK(tree_->GetProfilingManager()->ChangeFileMode());
   return Status::OK();
 }
 

mindspore/ccsrc/minddata/dataset/kernels/py_func_op.cc

@@ -81,6 +81,8 @@ Status PyFuncOp::Compute(const TensorRow &input, TensorRow *output) {
         }
       }
     } catch (const py::error_already_set &e) {
+      MS_LOG(ERROR) << "Pyfunc error, " << e.what() << ". Under sink mode, progress will late exit after 30s "
+                    << "for resource release and thread safe";
       ret = Status(StatusCode::kPyFuncException, e.what());
     }
   }