bjutsecurity22
/
mindspore2022
forked from huawei/mindspore2022
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Infer exception handle

This commit is contained in:
lanzhineng 2021-06-22 20:19:22 +08:00
parent 9b5893a8cf
commit 75e59ebbe0
6 changed files with 165 additions and 97 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

73
mindspore/ccsrc/pipeline/jit/static_analysis/async_eval_result.cc
View File

@ -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<std::mutex> lock(lock_);
if (ms == 0) {
return result_;
}
std::unique_lock<std::mutex> 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<std::mutex> lock(lock_);
if (result_ != nullptr) {
return result_;
}
std::unique_lock<std::mutex> 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<EvalResult>(std::make_shared<AbstractTimeOut>(), nullptr);
}
}
std::string AsyncEvalResult::ToString() {
@ -62,6 +69,7 @@ void AsyncEvalResult::JoinResult(const EvalResultPtr &result) {
}
void AnalysisResultCacheMgr::Clear() {
std::lock_guard<std::mutex> 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<EvalResultPtr> &future0,
const std::shared_future<EvalResultPtr> &future1) {
std::lock_guard<std::recursive_mutex> lock(lock_);
waiting_.push_back(future0);
waiting_.push_back(future1);
void AnalysisResultCacheMgr::PushTowait(std::future<void> &&future0, std::future<void> &&future1) {
std::lock_guard<std::mutex> lock(lock_);
waiting_.emplace_back(std::move(future0));
waiting_.emplace_back(std::move(future1));
}
void AnalysisResultCacheMgr::PushTodo(const AnfNodeConfigPtr &conf) {
std::lock_guard<std::recursive_mutex> lock(lock_);
std::lock_guard<std::mutex> lock(todo_lock_);
todo_.push_back(conf);
}
void AnalysisResultCacheMgr::InitSwitchValue(const AnfNodeConfigPtr &conf) {
std::lock_guard<std::recursive_mutex> lock(lock_);
std::lock_guard<std::mutex> lock(lock_);
AsyncEvalResultPtr async_eval_result = switch_cache_.get(conf);
if (async_eval_result == nullptr) {
async_eval_result = std::make_shared<AsyncEvalResult>();
@ -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<EvalResult>(std::make_shared<AbstractTimeOut>(), 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<std::recursive_mutex> lock(lock_);
std::lock_guard<std::mutex> 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<AsyncEvalResult>();
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<EvalResult>(joined_spec, std::make_shared<AttrValueMap>());
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<std::mutex> 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<EvalResultPtr> 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)) {

64
mindspore/ccsrc/pipeline/jit/static_analysis/async_eval_result.h
View File

@ -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 <typename KeyType, typename ValueType, typename CacheType>
class MultiThreadCache {
@ -110,6 +110,61 @@ class AsyncEvalResult {
std::condition_variable condition_var_;
};
template <typename Type>
class AsyncResult {
public:
AsyncResult() = default;
~AsyncResult() = default;
// wait
Type GetResult() {
std::unique_lock<std::mutex> 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<std::mutex> 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<std::mutex> lock(lock_);
result_ = result;
}
condition_var_.notify_all();
}
std::string ToString() {
std::ostringstream buffer;
std::lock_guard<std::mutex> 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<AbstractBasePtr>;
using AsyncAbstractResultPtr = std::shared_ptr<AsyncAbstractResult>;
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<EvalResultPtr> &future0, const std::shared_future<EvalResultPtr> &future1);
void PushTowait(std::future<void> &&future0, std::future<void> &&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<std::shared_future<EvalResultPtr>> waiting_;
std::mutex lock_;
std::list<std::future<void>> waiting_;
std::mutex todo_lock_;
std::list<AnfNodeConfigPtr> todo_;
AnalysisConfigResultCache cache_;

11
mindspore/ccsrc/pipeline/jit/static_analysis/evaluator.cc
View File

@ -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<EvalResult>(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<std::recursive_timed_mutex> 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<Evaluator>(), 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

98
mindspore/ccsrc/pipeline/jit/static_analysis/static_analysis.cc
View File

@ -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<EvalResult>(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<EvalResult>(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<AbstractError>(std::make_shared<StringImm>(oss.str()), out_conf->node());
async_result_main->JoinResult(std::make_shared<EvalResult>(abstractErrPtr, nullptr));
AnalysisResultCacheMgr::GetInstance().SetSwitchValue(out_conf,
std::make_shared<EvalResult>(abstractErrPtr, nullptr));
async_result_main->JoinResult(abstractErrPtr);
StaticAnalysisException::Instance().SetException();
}
return result;
}
EvalResultPtr AnalysisEngine::ExecuteMultipleEvaluatorsMultiThread(const std::vector<EvaluatorPtr> &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<AbstractTimeOut>()) {
if (eval_result->abstract()->isa<AbstractTimeOut>()) {
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<AbstractError>()) {
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<AsyncEvalResult>();
AsyncEvalResultPtr asyncResult1 = std::make_shared<AsyncEvalResult>();
AsyncEvalResultPtr asyncResult_main = std::make_shared<AsyncEvalResult>();
AsyncAbstractResultPtr asyncResult_main = std::make_shared<AsyncAbstractResult>();
AsyncAbstractResultPtr asyncResult0 = std::make_shared<AsyncAbstractResult>();
AsyncAbstractResultPtr asyncResult1 = std::make_shared<AsyncAbstractResult>();
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<AbstractTimeOut>()) {
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<AbstractError>()) {
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<AbstractTimeOut>()) {
MS_LOG(EXCEPTION) << "Eval " << evaluators[0]->ToString() << "is time out."
if (result0 == nullptr || result0->isa<AbstractTimeOut>()) {
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<AbstractTimeOut>()) {
MS_LOG(EXCEPTION) << "Eval " << evaluators[1]->ToString() << "is time out."
if (result1 == nullptr || result1->isa<AbstractTimeOut>()) {
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());
}

2
mindspore/ccsrc/pipeline/jit/static_analysis/static_analysis.h
View File

@ -289,13 +289,13 @@ class AnalysisEngine : public std::enable_shared_from_this<AnalysisEngine> {
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<EvaluatorPtr> &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_;

14
mindspore/core/utils/ms_exception.h
View File

@ -66,11 +66,18 @@ class StaticAnalysisException {
return instance;
}
void ClearException() { exception_ptr_ = nullptr; }
void ClearException() {
std::lock_guard<std::mutex> lock(lock_);
exception_ptr_ = nullptr;
}
bool HasException() { return exception_ptr_ != nullptr; }
bool HasException() {
std::lock_guard<std::mutex> lock(lock_);
return exception_ptr_ != nullptr;
}
void SetException() {
std::lock_guard<std::mutex> lock(lock_);
if (exception_ptr_ != nullptr) {
return;
}
@ -78,11 +85,13 @@ class StaticAnalysisException {
}
void SetAndRethrowException() {
std::lock_guard<std::mutex> lock(lock_);
SetException();
std::rethrow_exception(std::current_exception());
}
void CheckException() {
std::lock_guard<std::mutex> 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