From 75e59ebbe058ad2dc2ee8bb2c0d1abd2ffc047ed Mon Sep 17 00:00:00 2001 From: lanzhineng Date: Tue, 22 Jun 2021 20:19:22 +0800 Subject: [PATCH] Infer exception handle --- .../jit/static_analysis/async_eval_result.cc | 73 +++++++------- .../jit/static_analysis/async_eval_result.h | 64 +++++++++++- .../pipeline/jit/static_analysis/evaluator.cc | 11 +-- .../jit/static_analysis/static_analysis.cc | 98 ++++++++++--------- .../jit/static_analysis/static_analysis.h | 2 +- mindspore/core/utils/ms_exception.h | 14 ++- 6 files changed, 165 insertions(+), 97 deletions(-) diff --git a/mindspore/ccsrc/pipeline/jit/static_analysis/async_eval_result.cc b/mindspore/ccsrc/pipeline/jit/static_analysis/async_eval_result.cc index 5987e629f16..93f4c2da881 100644 --- a/mindspore/ccsrc/pipeline/jit/static_analysis/async_eval_result.cc +++ b/mindspore/ccsrc/pipeline/jit/static_analysis/async_eval_result.cc @@ -24,11 +24,13 @@ namespace mindspore { namespace abstract { +std::mutex AnalysisResultCacheMgr::tiggerToken_; + EvalResultPtr AsyncEvalResult::TryGetResult(int ms) { - if (result_ != nullptr || ms == 0) { + std::unique_lock lock(lock_); + if (ms == 0) { return result_; } - std::unique_lock lock(lock_); auto time = std::chrono::microseconds(ms); // Wait for ms. (void)condition_var_.wait_for(lock, time, [this] { return result_ != nullptr; }); @@ -36,13 +38,18 @@ EvalResultPtr AsyncEvalResult::TryGetResult(int ms) { } EvalResultPtr AsyncEvalResult::GetResult() { + std::unique_lock lock(lock_); if (result_ != nullptr) { return result_; } - std::unique_lock lock(lock_); auto time = std::chrono::seconds(kInferTimeout); - (void)condition_var_.wait_for(lock, time, [this] { return result_ != nullptr; }); - return result_; + auto cond = condition_var_.wait_for(lock, time, [this] { return result_ != nullptr; }); + if (cond) { + return result_; + } else { + MS_LOG(ERROR) << "Timeout!"; + return std::make_shared(std::make_shared(), nullptr); + } } std::string AsyncEvalResult::ToString() { @@ -62,6 +69,7 @@ void AsyncEvalResult::JoinResult(const EvalResultPtr &result) { } void AnalysisResultCacheMgr::Clear() { + std::lock_guard lock(lock_); cache_.clear(); switch_cache_.clear(); todo_.clear(); @@ -74,7 +82,6 @@ AnalysisResultCacheMgr &AnalysisResultCacheMgr::GetInstance() { void AnalysisResultCacheMgr::DumpCache(const std::string &filename) { auto path = pipeline::GetSaveGraphsPathName(Common::AddId(filename, ".cache")); - auto realpath = Common::GetRealPath(path); if (!realpath.has_value()) { MS_LOG(ERROR) << "Get real path failed. path=" << path; @@ -98,23 +105,22 @@ void AnalysisResultCacheMgr::UpdateCaller(const std::string &caller) { buffer << caller << "." << std::this_thread::get_id(); local_threadid = buffer.str(); } -std::mutex AnalysisResultCacheMgr::tiggerToken_; + std::string &AnalysisResultCacheMgr::GetThreadid() { return local_threadid; } -void AnalysisResultCacheMgr::PushTowait(const std::shared_future &future0, - const std::shared_future &future1) { - std::lock_guard lock(lock_); - waiting_.push_back(future0); - waiting_.push_back(future1); +void AnalysisResultCacheMgr::PushTowait(std::future &&future0, std::future &&future1) { + std::lock_guard lock(lock_); + waiting_.emplace_back(std::move(future0)); + waiting_.emplace_back(std::move(future1)); } void AnalysisResultCacheMgr::PushTodo(const AnfNodeConfigPtr &conf) { - std::lock_guard lock(lock_); + std::lock_guard lock(todo_lock_); todo_.push_back(conf); } void AnalysisResultCacheMgr::InitSwitchValue(const AnfNodeConfigPtr &conf) { - std::lock_guard lock(lock_); + std::lock_guard lock(lock_); AsyncEvalResultPtr async_eval_result = switch_cache_.get(conf); if (async_eval_result == nullptr) { async_eval_result = std::make_shared(); @@ -123,6 +129,7 @@ void AnalysisResultCacheMgr::InitSwitchValue(const AnfNodeConfigPtr &conf) { } EvalResultPtr AnalysisResultCacheMgr::GetSwitchValue(const AnfNodeConfigPtr &conf) { + // don't call lock_.lock(). switch_cache is protected. and it waits for result. AsyncEvalResultPtr async_eval_result = switch_cache_.get(conf); // Conf has been visited and set value. if (async_eval_result != nullptr) { @@ -130,7 +137,8 @@ EvalResultPtr AnalysisResultCacheMgr::GetSwitchValue(const AnfNodeConfigPtr &con auto result = async_eval_result->GetResult(); if (result == nullptr) { result = std::make_shared(std::make_shared(), nullptr); - MS_LOG(ERROR) << "AsyncEvalResult for NodeConfig " << conf->ToString() << " is nullptr, maybe timeout."; + MS_LOG(ERROR) << "AsyncEvalResult for NodeConfig " << conf->node()->ToString() << " is nullptr, maybe timeout."; + MS_LOG(ERROR) << "detail:" << conf->ToString(); } return result; } @@ -140,12 +148,11 @@ EvalResultPtr AnalysisResultCacheMgr::GetSwitchValue(const AnfNodeConfigPtr &con void AnalysisResultCacheMgr::SetSwitchValue(const AnfNodeConfigPtr &conf, const EvalResultPtr arg) { MS_EXCEPTION_IF_NULL(conf); if (arg == nullptr || arg->abstract() == nullptr) { - MS_LOG(WARNING) << conf->ToString() << " value is nullptr"; + MS_LOG(EXCEPTION) << conf->ToString() << " value is nullptr"; } - std::lock_guard lock(lock_); + std::lock_guard lock(lock_); AsyncEvalResultPtr async_eval_result = switch_cache_.get(conf); if (async_eval_result == nullptr) { - MS_LOG(EXCEPTION) << conf->ToString() << " Not key."; async_eval_result = std::make_shared(); async_eval_result->JoinResult(arg); switch_cache_.set(conf, async_eval_result); @@ -156,12 +163,9 @@ void AnalysisResultCacheMgr::SetSwitchValue(const AnfNodeConfigPtr &conf, const absList.push_back(arg->abstract()); absList.push_back(ab1->abstract()); // Join two branches's result - auto joined_spec = AbstractJoin(absList); - MS_EXCEPTION_IF_NULL(joined_spec); - MS_LOG(DEBUG) << "Multiple evaluators joined: " << joined_spec->ToString(); - auto joined_result = std::make_shared(joined_spec, std::make_shared()); + auto joined_result = AnalysisEngine::ProcessEvalResults(absList, conf->node()); async_eval_result->JoinResult(joined_result); - if (joined_result != ab1) { + if (!(*joined_result == *ab1)) { PushTodo(conf); } } else { @@ -171,17 +175,10 @@ void AnalysisResultCacheMgr::SetSwitchValue(const AnfNodeConfigPtr &conf, const } void AnalysisResultCacheMgr::Todo() { - while (true) { - AnfNodeConfigPtr conf; - lock_.lock(); - if (!todo_.empty()) { - conf = todo_.front(); - } else { - lock_.unlock(); - break; - } + std::lock_guard lock(todo_lock_); + while (!todo_.empty()) { + AnfNodeConfigPtr conf = todo_.front(); todo_.pop_front(); - lock_.unlock(); if (!(*GetValue(conf)->abstract() == *GetSwitchValue(conf)->abstract())) { MS_LOG(WARNING) << " Switch Value is not eq. " << " switchCache: " << GetSwitchValue(conf)->abstract()->ToString() @@ -192,17 +189,17 @@ void AnalysisResultCacheMgr::Todo() { void AnalysisResultCacheMgr::Wait() { while (true) { - std::shared_future future; + StaticAnalysisException::Instance().CheckException(); lock_.lock(); - if (!waiting_.empty()) { - future = std::move(waiting_.front()); - } else { + if (waiting_.empty()) { lock_.unlock(); break; } - + auto future = std::move(waiting_.front()); waiting_.pop_front(); lock_.unlock(); + + // must be unlock future.wait(); } if (IS_OUTPUT_ON(DEBUG)) { diff --git a/mindspore/ccsrc/pipeline/jit/static_analysis/async_eval_result.h b/mindspore/ccsrc/pipeline/jit/static_analysis/async_eval_result.h index 525d89a5c94..958d27b5dcd 100644 --- a/mindspore/ccsrc/pipeline/jit/static_analysis/async_eval_result.h +++ b/mindspore/ccsrc/pipeline/jit/static_analysis/async_eval_result.h @@ -33,7 +33,7 @@ namespace mindspore { namespace abstract { -constexpr size_t kInferTimeout = 60; +constexpr size_t kInferTimeout = 1800; // 60*30 30min, next pr will change the solution of endless. template class MultiThreadCache { @@ -110,6 +110,61 @@ class AsyncEvalResult { std::condition_variable condition_var_; }; +template +class AsyncResult { + public: + AsyncResult() = default; + ~AsyncResult() = default; + // wait + Type GetResult() { + std::unique_lock lock(lock_); + if (result_ != nullptr) { + return result_; + } + auto time = std::chrono::seconds(kInferTimeout); + auto cond = condition_var_.wait_for(lock, time, [this] { return result_ != nullptr; }); + if (cond) { + return result_; + } else { + MS_LOG(ERROR) << "Timeout!"; + return nullptr; + } + } + // not wait + Type TryGetResult(int ms = 0) { + std::unique_lock lock(lock_); + if (ms == 0) { + return result_; + } + auto time = std::chrono::microseconds(ms); + // Wait for ms. + (void)condition_var_.wait_for(lock, time, [this] { return result_ != nullptr; }); + return result_; + } + void JoinResult(const Type &result) { + MS_EXCEPTION_IF_NULL(result); + { + std::lock_guard lock(lock_); + result_ = result; + } + condition_var_.notify_all(); + } + std::string ToString() { + std::ostringstream buffer; + std::lock_guard lock(lock_); + buffer << (result_ == nullptr ? "NOT SET" : result_->ToString()); + return buffer.str(); + } + + private: + Type result_{nullptr}; + std::mutex lock_; + std::condition_variable condition_var_; +}; + +using AsyncAbstractResult = AsyncResult; +using AsyncAbstractResultPtr = std::shared_ptr; + class EvaluatorCacheMgr { public: EvaluatorCacheMgr() = default; @@ -151,7 +206,7 @@ class AnalysisResultCacheMgr { void DumpCache(const std::string &filename); // Wait for async Eval(conf) to finish. void Wait(); - void PushTowait(const std::shared_future &future0, const std::shared_future &future1); + void PushTowait(std::future &&future0, std::future &&future1); void PushTodo(const AnfNodeConfigPtr &conf); void Todo(); static void UpdateCaller(const std::string &caller); @@ -165,8 +220,9 @@ class AnalysisResultCacheMgr { AnalysisResultCacheMgr() = default; static std::mutex tiggerToken_; - std::recursive_mutex lock_; - std::list> waiting_; + std::mutex lock_; + std::list> waiting_; + std::mutex todo_lock_; std::list todo_; AnalysisConfigResultCache cache_; diff --git a/mindspore/ccsrc/pipeline/jit/static_analysis/evaluator.cc b/mindspore/ccsrc/pipeline/jit/static_analysis/evaluator.cc index 7a72d5b7de8..9056ed2142d 100644 --- a/mindspore/ccsrc/pipeline/jit/static_analysis/evaluator.cc +++ b/mindspore/ccsrc/pipeline/jit/static_analysis/evaluator.cc @@ -247,7 +247,7 @@ EvalResultPtr BaseFuncGraphEvaluator::Eval(AnalysisEnginePtr engine, const Abstr << "), leave, function call depth: " << engine->function_call_depth() << " - " << engine->stack_frame_depth(); auto res = std::make_shared(res_base, nullptr); - evaluator_cache_mgr_->SetValue(args_abs_list, res); + // evaluator_cache_mgr_->SetValue(args_abs_list, res); return res; } @@ -389,11 +389,12 @@ FuncGraphPtr MetaFuncGraphEvaluator::GetFuncGraph(AnalysisEnginePtr engine, cons EvalResultPtr Evaluator::Run(AnalysisEnginePtr engine, const ConfigPtrList &args_conf_list, const AnfNodeConfigPtr &out_conf) { + // The evaluator can't reenter at sametime. std::unique_lock eval_lock(eval_lock_, std::try_to_lock); if (!eval_lock.owns_lock()) { - auto py_tstate = PyEval_SaveThread(); + auto py_tstate = PyEval_SaveThread(); // release GIL eval_lock.try_lock_for(std::chrono::seconds(kInferTimeout)); - PyEval_RestoreThread(py_tstate); + PyEval_RestoreThread(py_tstate); // acquire GIL if (!eval_lock.owns_lock()) { MS_LOG(EXCEPTION) << "It is timeout to run " << ToString(); } @@ -405,6 +406,7 @@ EvalResultPtr Evaluator::Run(AnalysisEnginePtr engine, const ConfigPtrList &args MS_EXCEPTION_IF_NULL(conf); return conf->ObtainEvalResult()->abstract(); }); + args_spec_list = NormalizeArgs(args_spec_list); args_spec_list = BroadenUndeterminedArgs(args_spec_list); trace::TraceGraphEvalEnter(shared_from_base(), out_conf); @@ -570,11 +572,8 @@ EvalResultPtr Evaluator::SingleRun(AnalysisEnginePtr engine, const ConfigPtrList const AnfNodeConfigPtr &out_conf) { auto result = this->Run(engine, args_conf_list, out_conf); - StaticAnalysisException::Instance().CheckException(); pybind11::gil_scoped_release release; AnalysisResultCacheMgr::GetInstance().Wait(); - StaticAnalysisException::Instance().CheckException(); - return result; } } // namespace abstract diff --git a/mindspore/ccsrc/pipeline/jit/static_analysis/static_analysis.cc b/mindspore/ccsrc/pipeline/jit/static_analysis/static_analysis.cc index 5680d6b2f0a..b9276d77ee1 100644 --- a/mindspore/ccsrc/pipeline/jit/static_analysis/static_analysis.cc +++ b/mindspore/ccsrc/pipeline/jit/static_analysis/static_analysis.cc @@ -97,10 +97,8 @@ AnalysisResult AnalysisEngine::Run(const FuncGraphPtr &func_graph, const Abstrac result.inferred = output_conf->ObtainEvalResult(); result.context = root_context; - StaticAnalysisException::Instance().CheckException(); pybind11::gil_scoped_release release; AnalysisResultCacheMgr::GetInstance().Wait(); - StaticAnalysisException::Instance().CheckException(); return result; } @@ -732,25 +730,27 @@ bool NeedWaitForTwoBranches(const AbstractBasePtr &abstract) { return false; } -EvalResultPtr ExecEvaluetor(EvaluatorPtr eval, AnalysisEnginePtr engine, ConfigPtrList args_conf_list, - AnfNodeConfigPtr out_conf, std::string caller, AsyncEvalResultPtr async_result_branch, - AsyncEvalResultPtr async_result_main) { - // TiggerToken_scoped_acquire tigger_token_acquire; - py::gil_scoped_acquire pyGuard; - - EvalResultPtr result = nullptr; +void ExecEvaluator(EvaluatorPtr eval, AnalysisEnginePtr engine, ConfigPtrList args_conf_list, AnfNodeConfigPtr out_conf, + std::string caller, AsyncAbstractResultPtr async_result_branch, + AsyncAbstractResultPtr async_result_main) { + AnalysisResultCacheMgr::UpdateCaller(caller); try { - AnalysisResultCacheMgr::UpdateCaller(caller); - result = eval->Run(engine, args_conf_list, out_conf); + // only one GIL + py::gil_scoped_acquire pyGuard; + auto result = eval->Run(engine, args_conf_list, out_conf); MS_EXCEPTION_IF_NULL(result); - async_result_branch->JoinResult(result); - async_result_main->JoinResult(result); + MS_EXCEPTION_IF_NULL(result->abstract()); + + // broaden the result of switch(c,t,f)() + auto broadAbstract = result->abstract()->Broaden(); + // let main thread to continue. + AnalysisResultCacheMgr::GetInstance().SetSwitchValue(out_conf, + std::make_shared(broadAbstract, nullptr)); + async_result_branch->JoinResult(broadAbstract); + async_result_main->JoinResult(broadAbstract); MS_LOG(DEBUG) << GetInferThread() << "async :" << eval->ToString() << " asyncResult address = " << async_result_branch.get() - << " value = " << async_result_branch->TryGetResult()->abstract()->ToString(); - auto broadAbstract = result->abstract()->Broaden(); - auto broadEvalResult = std::make_shared(broadAbstract, nullptr); - AnalysisResultCacheMgr::GetInstance().SetSwitchValue(out_conf, broadEvalResult); + << " value = " << async_result_branch->TryGetResult()->ToString(); } catch (const std::exception &e) { std::ostringstream oss; trace::GetEvalStackInfo(oss); @@ -758,10 +758,11 @@ EvalResultPtr ExecEvaluetor(EvaluatorPtr eval, AnalysisEnginePtr engine, ConfigP MS_LOG(ERROR) << oss.str(); } auto abstractErrPtr = std::make_shared(std::make_shared(oss.str()), out_conf->node()); - async_result_main->JoinResult(std::make_shared(abstractErrPtr, nullptr)); + AnalysisResultCacheMgr::GetInstance().SetSwitchValue(out_conf, + std::make_shared(abstractErrPtr, nullptr)); + async_result_main->JoinResult(abstractErrPtr); StaticAnalysisException::Instance().SetException(); } - return result; } EvalResultPtr AnalysisEngine::ExecuteMultipleEvaluatorsMultiThread(const std::vector &evaluators, @@ -769,34 +770,40 @@ EvalResultPtr AnalysisEngine::ExecuteMultipleEvaluatorsMultiThread(const std::ve const ConfigPtrList &args_conf_list) { // TiggerToken_scoped_release tigger_token_release; pybind11::gil_scoped_release release; + // Wait for the switch node to finish. MS_LOG(DEBUG) << GetInferThread() << "async : entry switch " << out_conf->ToString(); auto eval_result = AnalysisResultCacheMgr::GetInstance().GetSwitchValue(out_conf); if (eval_result == nullptr) { - MS_LOG(DEBUG) << GetInferThread() << "async : Init switch " << out_conf->ToString(); + MS_LOG(INFO) << GetInferThread() << "async : Init switch " << out_conf->node()->ToString(); AnalysisResultCacheMgr::GetInstance().InitSwitchValue(out_conf); } else { - if (eval_result->isa()) { + if (eval_result->abstract()->isa()) { MS_LOG(EXCEPTION) << "Eval " << out_conf->node()->ToString() << " time out." - << "please check the code if there are recursive functions."; + << " Please check the code if there are recursive functions."; + } + if (eval_result->abstract()->isa()) { + MS_LOG(ERROR) << "Eval " << out_conf->node()->ToString() << " threw exception."; + StaticAnalysisException::Instance().CheckException(); } return eval_result; } // Eval result of the branches and main. - AsyncEvalResultPtr asyncResult0 = std::make_shared(); - AsyncEvalResultPtr asyncResult1 = std::make_shared(); - AsyncEvalResultPtr asyncResult_main = std::make_shared(); + AsyncAbstractResultPtr asyncResult_main = std::make_shared(); + AsyncAbstractResultPtr asyncResult0 = std::make_shared(); + AsyncAbstractResultPtr asyncResult1 = std::make_shared(); + SetUndeterminedFlag(evaluators[0]); SetUndeterminedFlag(evaluators[1]); std::string threadId = AnalysisResultCacheMgr::GetThreadid(); MS_LOG(DEBUG) << GetInferThread() << "async : " << evaluators[0]->ToString(); - auto future0 = std::async(std::launch::async, ExecEvaluetor, evaluators[0], shared_from_this(), args_conf_list, + auto future0 = std::async(std::launch::async, ExecEvaluator, evaluators[0], shared_from_this(), args_conf_list, out_conf, threadId, asyncResult0, asyncResult_main); MS_LOG(DEBUG) << GetInferThread() << "async : " << evaluators[1]->ToString(); - auto future1 = std::async(std::launch::async, ExecEvaluetor, evaluators[1], shared_from_this(), args_conf_list, + auto future1 = std::async(std::launch::async, ExecEvaluator, evaluators[1], shared_from_this(), args_conf_list, out_conf, threadId, asyncResult1, asyncResult_main); // Wait for async threads to finish. @@ -807,43 +814,42 @@ EvalResultPtr AnalysisEngine::ExecuteMultipleEvaluatorsMultiThread(const std::ve auto branchResult = asyncResult_main->GetResult(); if (branchResult == nullptr || branchResult->isa()) { MS_LOG(EXCEPTION) << "Can't finish " << evaluators[0]->ToString() << " or " << evaluators[1]->ToString() - << "please check the code if there are recursive functions."; + << " Please check the code if there are recursive functions."; } if (branchResult->isa()) { - MS_LOG(EXCEPTION) << "async " << out_conf->node()->ToString() << " threw exception."; + MS_LOG(ERROR) << "async " << out_conf->node()->ToString() << " threw exception."; + StaticAnalysisException::Instance().CheckException(); } AbstractBasePtrList out_specs; - if (NeedWaitForTwoBranches(branchResult->abstract())) { - MS_LOG(DEBUG) << GetInferThread() << "async . waiting for " << evaluators[0]->ToString(); + if (NeedWaitForTwoBranches(branchResult)) { + MS_LOG(DEBUG) << GetInferThread() << "async waiting for " << evaluators[0]->ToString(); auto result0 = asyncResult0->GetResult(); - if (result0->isa()) { - MS_LOG(EXCEPTION) << "Eval " << evaluators[0]->ToString() << "is time out." + if (result0 == nullptr || result0->isa()) { + MS_LOG(EXCEPTION) << "Eval " << evaluators[0]->ToString() << " is time out." << " Please check the code if there is recursive function."; } - out_specs.push_back(result0->abstract()); + out_specs.push_back(result0); - MS_LOG(DEBUG) << GetInferThread() << "async . waiting for " << evaluators[1]->ToString(); + MS_LOG(DEBUG) << GetInferThread() << "async waiting for " << evaluators[1]->ToString(); auto result1 = asyncResult1->GetResult(); - if (result1->isa()) { - MS_LOG(EXCEPTION) << "Eval " << evaluators[1]->ToString() << "is time out." + if (result1 == nullptr || result1->isa()) { + MS_LOG(EXCEPTION) << "Eval " << evaluators[1]->ToString() << " is time out." << " Please check the code if there is recursive function."; } - out_specs.push_back(result1->abstract()); + out_specs.push_back(result1); } else { if (asyncResult0->TryGetResult()) { - MS_LOG(DEBUG) << GetInferThread() << "async . waiting for " << evaluators[0]->ToString() - << " value0=" << asyncResult0->GetResult()->abstract()->ToString(); - out_specs.push_back(asyncResult0->GetResult()->abstract()); + MS_LOG(DEBUG) << GetInferThread() << "async waiting for " << evaluators[0]->ToString() + << " value0=" << asyncResult0->GetResult()->ToString(); + out_specs.push_back(asyncResult0->GetResult()); } - if (asyncResult1->TryGetResult()) { - MS_LOG(DEBUG) << GetInferThread() << "async . waiting for " << evaluators[1]->ToString() - << " value1=" << asyncResult1->GetResult()->abstract()->ToString(); - out_specs.push_back(asyncResult1->GetResult()->abstract()); + MS_LOG(DEBUG) << GetInferThread() << "async waiting for " << evaluators[1]->ToString() + << " value1=" << asyncResult1->GetResult()->ToString(); + out_specs.push_back(asyncResult1->GetResult()); } } - return ProcessEvalResults(out_specs, out_conf->node()); } diff --git a/mindspore/ccsrc/pipeline/jit/static_analysis/static_analysis.h b/mindspore/ccsrc/pipeline/jit/static_analysis/static_analysis.h index 02e58d125ad..520011510fb 100644 --- a/mindspore/ccsrc/pipeline/jit/static_analysis/static_analysis.h +++ b/mindspore/ccsrc/pipeline/jit/static_analysis/static_analysis.h @@ -289,13 +289,13 @@ class AnalysisEngine : public std::enable_shared_from_this { size_t stack_frame_max_depth() const { return stack_frame_max_depth_; } void CheckNoStackInSameFuncGraph(const AnfNodeConfigPtr &conf); bool enable_recursive_eval() const { return enable_recursive_eval_; } + static EvalResultPtr ProcessEvalResults(const AbstractBasePtrList &out_specs, const AnfNodePtr &node); private: void SetUndeterminedFlag(const EvaluatorPtr &evaluator); EvaluatorPtr HandleNestedRecursion(const std::vector &evaluators, const EvaluatorPtr &eval, const AbstractBasePtrList &args_spec_list, const EvalTraceRevIter &it, bool *continue_flag); - EvalResultPtr ProcessEvalResults(const AbstractBasePtrList &out_specs, const AnfNodePtr &node); const PrimEvaluatorMap &prim_constructors_; FuncGraphManagerPtr func_graph_manager_; diff --git a/mindspore/core/utils/ms_exception.h b/mindspore/core/utils/ms_exception.h index 76168012fa1..03177b8c49a 100644 --- a/mindspore/core/utils/ms_exception.h +++ b/mindspore/core/utils/ms_exception.h @@ -66,11 +66,18 @@ class StaticAnalysisException { return instance; } - void ClearException() { exception_ptr_ = nullptr; } + void ClearException() { + std::lock_guard lock(lock_); + exception_ptr_ = nullptr; + } - bool HasException() { return exception_ptr_ != nullptr; } + bool HasException() { + std::lock_guard lock(lock_); + return exception_ptr_ != nullptr; + } void SetException() { + std::lock_guard lock(lock_); if (exception_ptr_ != nullptr) { return; } @@ -78,11 +85,13 @@ class StaticAnalysisException { } void SetAndRethrowException() { + std::lock_guard lock(lock_); SetException(); std::rethrow_exception(std::current_exception()); } void CheckException() { + std::lock_guard lock(lock_); if (exception_ptr_ != nullptr) { auto tmp_exception_ptr = exception_ptr_; exception_ptr_ = nullptr; @@ -96,6 +105,7 @@ class StaticAnalysisException { DISABLE_COPY_AND_ASSIGN(StaticAnalysisException) std::exception_ptr exception_ptr_{nullptr}; + std::mutex lock_; }; } // namespace mindspore