mindspore/ccsrc/pipeline/jit/static_analysis/async_eval_result.cc

@@ -25,6 +25,8 @@ namespace mindspore {
 namespace abstract {
 thread_local std::string AnalysisSchedule::thread_id_ = "m";
 
+// The role of this code in the MindSpore project is to control the scheduling and execution of threads,
+// and realize the dynamic management and control of threads by constantly checking the conditions and executing the corresponding actions through the loop.
 void AnalysisSchedule::Schedule() {
   const auto checkPeriod = std::chrono::seconds(3);
   while (run_ || infer_thread_count_.load() > 0) {
@@ -38,6 +40,7 @@ void AnalysisSchedule::Schedule() {
   MS_LOG(DEBUG) << "Success to exit.";
 }
 
+// The thread that performs an asynchronous task frees up CPU resources so that other threads can continue executing.
 void AnalysisSchedule::Yield(const AsyncInferTask *async_infer_task) {
   MS_EXCEPTION_IF_NULL(async_infer_task);
   {
@@ -51,6 +54,12 @@ void AnalysisSchedule::Yield(const AsyncInferTask *async_infer_task) {
   activate_thread_cv_.notify_one();
 }
 
+  
+// Analyze a member function of the scheduling class. 
+// Its role is to handle anomalies that occur during analysis. 
+// Specifically, it logs the first exception and, if the incoming exception is a Python exception, gets the exception stack and logs it; 
+// Then release all locks so that other threads can continue running; 
+// Clear the list of ongoing tasks. Finally, the global raw evaluation cache is cleared to avoid the cache containing invalid results.
 void AnalysisSchedule::HandleException(const std::exception &ex) {
   // Just record the first exception information.
   if (!StaticAnalysisException::Instance().HasException()) {
@@ -84,12 +93,16 @@ void AnalysisSchedule::HandleException(const std::exception &ex) {
   }
 }
 
+// Stop the analysis task in progress. It stops a task by creating an asynchronous inference task in a stopped state and adding it to the scheduler.
 void AnalysisSchedule::Stop() {
   AsyncInferTaskPtr stop_task = AsyncInferTask::MakeShared(std::make_shared<AsyncAbstract>(), kStateStop);
   Add2Schedule(stop_task);
   MS_LOG(DEBUG) << "Set analysis schedule to stop";
 }
 
+// Wait for the analysis task to complete.
+// It waits for the task by waiting for the condition variable and checking the number of threads,
+// and outputs the relevant information and checks the exception after the task is completed.
 void AnalysisSchedule::Wait() {
   EnterWaiting();
   if (infer_thread_count_.load() > 0) {
@@ -104,6 +117,7 @@ void AnalysisSchedule::Wait() {
   StaticAnalysisException::Instance().CheckException();
 }
 
+// Adds asynchronous inference tasks to the scheduling list and updates related statistics.
 void AnalysisSchedule::Add2Schedule(const AsyncInferTaskPtr &async_infer_task_ptr) {
   std::lock_guard<std::mutex> lock(activate_thread_lock_);
   MS_EXCEPTION_IF_NULL(async_infer_task_ptr);
@@ -115,6 +129,9 @@ void AnalysisSchedule::Add2Schedule(const AsyncInferTaskPtr &async_infer_task_pt
                 << " schedule list size: " << schedule_list_.size();
 }
 
+// Set up the next executable analysis task.
+// It determines whether to continue waiting or trigger an infinite loop exception by judging the status of the task and the number of threads in the thread pool, 
+// and marks the task as ready when it finds a result.
 void AnalysisSchedule::SetNextReady() {
   if (schedule_list_.empty()) {
     return;
@@ -154,6 +171,8 @@ void AnalysisSchedule::SetNextReady() {
                 << "  address: " << async_task.get();
 }
 
+// Gets the result of an asynchronous task.
+// It determines whether to wait and schedule by judging whether the result is a null pointer, and outputs relevant information after obtaining the result.
 AbstractBasePtr AsyncAbstract::GetResult() {
   auto ret = TryGetResult();
   if (ret != nullptr) {
@@ -195,6 +214,8 @@ AbstractFunctionPtr GetAbstractFuncRecursively(const AbstractBasePtr &abs, const
 }
 }  // namespace
 
+// Gets a unique asynchronous abstract function pointer, 
+// returned directly if it has already been parsed, otherwise retrieved and parsed by a recursive call.
 AbstractFunctionPtr AsyncAbstractFuncAtom::GetUnique() {
   if (resolved_ != nullptr) {
     return resolved_;
@@ -208,6 +229,9 @@ AbstractFunctionPtr AsyncAbstractFuncAtom::GetUnique() {
   return resolved_;
 }
 
+// Converts the AsyncAbstractFuncAtom object to a string representation. 
+// It determines the content of the returned string by determining whether the member variable resolved_ is a null pointer, 
+// and calls resolved_'s ToString() method to get more information if needed.
 std::string AsyncAbstractFuncAtom::ToString() const {
   if (resolved_ == nullptr) {
     return "AsyncAbstractFuncAtom(Not Resolved)";
@@ -221,6 +245,7 @@ std::string AsyncAbstractFuncAtom::ToString() const {
   return buffer.str();
 }
 
+// Clear the cache of analysis results, including the original evaluation cache and three different types of cache objects.
 void AnalysisResultCacheMgr::Clear() {
   prim_eval_cache_->Clear();
   std::lock_guard<std::mutex> lock(lock_);
@@ -229,6 +254,7 @@ void AnalysisResultCacheMgr::Clear() {
   switch_cache_for_check_.clear();
 }
 
+// Initializes the switch value by fetching or creating a new asynchronous abstract result object from the cache.
 void AnalysisResultCacheMgr::InitSwitchValue(const AnfNodeConfigPtr &conf) {
   std::lock_guard<std::mutex> lock(lock_);
   AsyncAbstractPtr async_eval_result = switch_cache_.get(conf);
@@ -238,6 +264,7 @@ void AnalysisResultCacheMgr::InitSwitchValue(const AnfNodeConfigPtr &conf) {
   }
 }
 
+// According to the given configuration information, the corresponding switch value is obtained from the analysis result cache.
 AbstractBasePtr AnalysisResultCacheMgr::GetSwitchValue(const AnfNodeConfigPtr &conf) {
   // don't call lock_.lock(). switch_cache is protected. and it waits for result.
   AsyncAbstractPtr async_eval_result = switch_cache_.get(conf);
@@ -247,6 +274,7 @@ AbstractBasePtr AnalysisResultCacheMgr::GetSwitchValue(const AnfNodeConfigPtr &c
   return async_eval_result->GetResult();
 }
 
+// Cache the analysis results and update the asynchronous abstract result objects in the cache by merging the current abstract result with the previous abstract result.
 void AnalysisResultCacheMgr::SetCacheValue(const AnfNodeConfigPtr &conf, const AbstractBasePtr &current_abs,
                                            AnalysisConfigAsyncResultCache *cache) {
   MS_EXCEPTION_IF_NULL(conf);
@@ -277,6 +305,7 @@ void AnalysisResultCacheMgr::SetCacheValue(const AnfNodeConfigPtr &conf, const A
   }
 }
 
+// Set and check the cache of switch values in the analysis results cache manager.
 void AnalysisResultCacheMgr::CheckSwitchValueJoinable(const AnfNodeConfigPtr &conf, const AbstractBasePtr &arg) {
   SetCacheValue(conf, arg, &switch_cache_for_check_);
 }