mindspore/ccsrc/common/api_register.cc

@@ -14,12 +14,16 @@
  * limitations under the License.
  */
 
+// Include the header file "api_register.h" from the "common/pybind_api" directory
 #include "include/common/pybind_api/api_register.h"
 
 namespace mindspore {
 
+// Define the member function GetSingleton() of the class PybindDefineRegister
 PybindDefineRegister &PybindDefineRegister::GetSingleton() {
+  // Create a static instance of PybindDefineRegister using the default constructor
   static PybindDefineRegister instance{};
+  // Return the static instance
   return instance;
 }
 

mindspore/ccsrc/common/duplex_pipe.cc

@@ -22,68 +22,104 @@
 #include <algorithm>
 #include "utils/convert_utils_base.h"
 
+// Start of the namespace "mindspore"
 namespace mindspore {
-DuplexPipe::~DuplexPipe() {
-  // pid_ < 0 means the child process is invalid or closed, pid_ == 0 means this process is child
-  if (pid_ > 0) {
-    (void)kill(pid_, SIGKILL);
+  
+  // Destructor for the DuplexPipe class
+  DuplexPipe::~DuplexPipe() {
+    // Check if the child process is valid or closed
+    // pid_ < 0 means the child process is invalid or closed
+    // pid_ == 0 means this process is the child process
+    if (pid_ > 0) {
+      // Send the SIGKILL signal to terminate the child process
+      (void)kill(pid_, SIGKILL);
+    }
   }
-}
+  
+} // End of the namespace "mindspore"
 
+// Open a duplex pipe by creating two pipes, one for each direction of communication
 int DuplexPipe::Open(const std::initializer_list<std::string> &arg_list, bool append_fds) {
+  // Create the first pipe
   if (pipe(fd1_) == -1) {
+    // If the pipe creation fails, log an error message with the error number
     DP_EXCEPTION << "pipe 1 failed, errno: " << errno;
   }
+  // Create the second pipe
   if (pipe(fd2_) == -1) {
+    // If the pipe creation fails, close the first pipe and log an error message with the error number
     close(fd1_[0]);
     close(fd1_[1]);
     DP_EXCEPTION << "pipe 2 failed, errno: " << errno;
   }
+}
 
+// Create a child process using fork() and store the process ID in pid_
   pid_ = fork();
+  
+  // Check if fork() failed
   if (pid_ < 0) {
+    // Close file descriptors to prevent resource leaks
     close(fd1_[0]);
     close(fd1_[1]);
     close(fd2_[0]);
     close(fd2_[1]);
+    
+    // Log an error message with the error code
     DP_EXCEPTION << "fork failed, errno: " << errno;
-  } else if (pid_ == 0) {  // Remote process
-    // Here cannot record log before execvp called, because glog will warn "File Exist" if log to file.
+  } 
+  // Check if the process is the child process
+  else if (pid_ == 0) {  // Remote process
+    // Save the current standard output and standard input file descriptors
     remote_stdout_ = dup(STDOUT_FILENO);
     remote_stdin_ = dup(STDIN_FILENO);
+    
+    // Close unnecessary file descriptors
     close(fd1_[1]);
     close(fd2_[0]);
+    
+    // Check if append_fds is false
     if (!append_fds) {
+      // Redirect the standard input and standard output to the pipe file descriptors
       dup2(fd1_[0], STDIN_FILENO);
       dup2(fd2_[1], STDOUT_FILENO);
     }
+    
+    // Create a vector to store the arguments for execvp
     std::vector<const char *> args;
+    
+    // Convert each argument from a string to a const char * and add it to the vector
     std::transform(arg_list.begin(), arg_list.end(), std::back_inserter(args),
                    [](const std::string &arg) -> const char * { return arg.c_str(); });
-    if (append_fds) {
-      std::string fd10 = std::to_string(fd1_[0]).c_str();
-      args.emplace_back(fd10.c_str());
-      std::string fd21 = std::to_string(fd2_[1]).c_str();
-      args.emplace_back(fd21.c_str());
-    }
-    args.emplace_back(nullptr);
-    if (execvp(args[0], const_cast<char *const *>(&args[0])) == -1) {
-      DP_EXCEPTION << "execute " << args[0] << " failed, errno: " << errno;
-    }
-  } else {  // Local process
-    DP_INFO << "Local process, id: " << getpid() << ", " << fd2_[0] << "/" << fd1_[1];
-    local_stdout_ = dup(STDOUT_FILENO);
-    local_stdin_ = dup(STDIN_FILENO);
-    close(fd1_[0]);
-    close(fd2_[1]);
 
+    if (append_fds) { // If append_fds is true
+      std::string fd10 = std::to_string(fd1_[0]).c_str(); // Convert the integer fd1_[0] to a string
+      args.emplace_back(fd10.c_str()); // Add the converted string to the args vector
+      std::string fd21 = std::to_string(fd2_[1]).c_str(); // Convert the integer fd2_[1] to a string
+      args.emplace_back(fd21.c_str()); // Add the converted string to the args vector
+    }
+    args.emplace_back(nullptr); // Add a nullptr to the end of the args vector
+    if (execvp(args[0], const_cast<char *const *>(&args[0])) == -1) { // Execute the command specified by args[0] with the arguments in args
+      DP_EXCEPTION << "execute " << args[0] << " failed, errno: " << errno; // If execvp returns -1, log an error message with the failed command and the value of errno
+    }
+  } else { // If append_fds is false, i.e., local process
+    DP_INFO << "Local process, id: " << getpid() << ", " << fd2_[0] << "/" << fd1_[1]; // Log an information message with the process ID and the values of fd2_[0] and fd1_[1]
+    local_stdout_ = dup(STDOUT_FILENO); // Duplicate the file descriptor for standard output and store it in local_stdout_
+    local_stdin_ = dup(STDIN_FILENO); // Duplicate the file descriptor for standard input and store it in local_stdin_
+    close(fd1_[0]); // Close the read end of the pipe
+    close(fd2_[1]); // Close the write end of the pipe
+
+// Create a shared pointer to a SignalHandler object using std::make_shared
+// Pass weak_from_this() as the first argument, which returns a weak pointer to the current object
+// Pass &pid_ as the second argument, which is the address of the pid_ variable
     signal_handler_ = std::make_shared<SignalHandler>(weak_from_this(), &pid_);
   }
+// Return 0 to indicate successful program termination
   return 0;
 }
 
 void DuplexPipe::Write(const std::string &buf, bool flush) const {
-  // Write the string into pipe
+  // Write the string into the pipe
   if (write(fd1_[1], buf.data(), buf.size()) == -1) {
     DP_ERROR << "write failed, errno: " << errno;
     return;
@@ -98,110 +134,175 @@ void DuplexPipe::Write(const std::string &buf, bool flush) const {
   DP_DEBUG << "<< [" << buf << "]";
 }
 
+// Function to read a string from a pipe
 std::string DuplexPipe::Read() {
-  // Read the string from pipe
-  std::string buf;
-  // Read one line or multiple lines
-  while (1) {
-    SetTimeOut();
-    ssize_t size = read(fd2_[0], c_buf_, kBufferSize);  // MAYBE BLOCKED, Till reading something
-    if (size <= 0) {
+  std::string buf; // Create an empty string to store the read data
+  while (1) { // Loop indefinitely until a break statement is encountered
+    SetTimeOut(); // Set a timeout for reading from the pipe
+    ssize_t size = read(fd2_[0], c_buf_, kBufferSize); // Read data from the pipe into a buffer
+    if (size <= 0) { // If no data is read or an error occurs, break the loop
       break;
     }
-    CancelTimeOut();
-    bool line_end = c_buf_[size - 1] == '\n';
-    buf.append(c_buf_, LongToSize(line_end ? size - 1 : size));  // Copy without the last '\n'
-    if (line_end) {
+    CancelTimeOut(); // Cancel the timeout as data has been successfully read
+    bool line_end = c_buf_[size - 1] == '\n'; // Check if the last character read is a newline character
+    buf.append(c_buf_, LongToSize(line_end ? size - 1 : size)); // Append the read data to the buffer, excluding the last newline character if present
+    if (line_end) { // If the last character read is a newline character, break the loop
       break;
     }
   }
-  DP_DEBUG << ">> [" << buf << "]";
-  return buf;
+  DP_DEBUG << ">> [" << buf << "]"; // Print the read string to the debug log
+  return buf; // Return the read string
 }
 
+// A member function of the DuplexPipe class that writes a string to stdout with the option to flush the output
 void DuplexPipe::WriteWithStdout(const std::string &buf, bool flush) {
+  // Redirect the standard output to the write end of the pipe
   dup2(fd1_[1], STDOUT_FILENO);
   // Write the string into pipe
   std::cout << buf;
+
+  // If flush is true, flush the output by adding a newline character
   if (flush) {
-    // Flush into the pipe
     std::cout << std::endl;
   }
+
+  // Restore the original stdout by redirecting it back to the local stdout
   dup2(local_stdout_, STDOUT_FILENO);
 }
 
+// Define a member function `ReadWithStdin` of the `DuplexPipe` class that returns a `std::string`
 std::string DuplexPipe::ReadWithStdin() {
-  std::string buf;
+  std::string buf; // Declare a string variable `buf` to store the input
+
+  // Redirect the read end of the pipe (`fd2_[0]`) to the standard input file descriptor (`STDIN_FILENO`)
   dup2(fd2_[0], STDIN_FILENO);
-  // Maybe blocked
+
+  // Set a timeout for reading from `std::cin`
   SetTimeOut();
+
+  // Read a line of input from `std::cin` and store it in `buf`
   std::getline(std::cin, buf);  // Not use 'std::cin >>' to include space
+
+  // Cancel the timeout
   CancelTimeOut();
+
+  // Restore the original standard input file descriptor
   dup2(local_stdin_, STDIN_FILENO);
+
+  // Return the read input stored in `buf`
   return buf;
 }
 
 DuplexPipe &DuplexPipe::operator<<(const std::string &buf) {
   Write(buf);
+  // Return a reference to the current object (this) to allow chaining of << operators
   return *this;
 }
 
 DuplexPipe &DuplexPipe::operator>>(std::string &buf) {
   buf = Read();
+  // Return a reference to the current object (DuplexPipe) to allow chaining of the operator
   return *this;
 }
 
+// This function is a member function of the DuplexPipe class
+// It is used to close the file descriptors and reset the process ID
+
 void DuplexPipe::Close() noexcept {
+  // Close the read end of the first pipe
   close(fd1_[0]);
+  // Close the write end of the first pipe
   close(fd1_[1]);
+  // Close the read end of the second pipe
   close(fd2_[0]);
+  // Close the write end of the second pipe
   close(fd2_[1]);
+  // Reset the process ID to -1
   pid_ = -1;
 }
 
+// Define the constructor for the SignalHandler class, which takes a weak pointer to a DuplexPipe object and a pointer to a pid_t variable as parameters
 DuplexPipe::SignalHandler::SignalHandler(const std::weak_ptr<DuplexPipe> &dp, pid_t *pid) {
+  // Assign the weak pointer to the dp_ member variable
   dp_ = dp;
+  // Assign the pid pointer to the child_pid_ member variable
   child_pid_ = pid;
+  // Register a signal handler for the SIGCHLD signal using the SigChildHandler function
   signal(SIGCHLD, SigChildHandler);
+  // Register a signal handler for the SIGPIPE signal using the SigPipeHandler function
   signal(SIGPIPE, SigPipeHandler);
 }
 
+// Definition of the destructor for the SignalHandler class within the DuplexPipe namespace
 DuplexPipe::SignalHandler::~SignalHandler() {}
 
+// Define the SetAlarm function of the SignalHandler class, which takes an unsigned integer interval_secs as a parameter
 void DuplexPipe::SignalHandler::SetAlarm(unsigned int interval_secs) const {
+  // Set the signal handler for the SIGALRM signal to the SigAlarmHandler function
   signal(SIGALRM, SigAlarmHandler);
+  // Set an alarm to trigger after the specified interval_secs
   alarm(interval_secs);
 }
 
-void DuplexPipe::SignalHandler::CancelAlarm() const { (void)alarm(0); }
+// Define a member function named "CancelAlarm" in the class "SignalHandler" of the "DuplexPipe" class
+void DuplexPipe::SignalHandler::CancelAlarm() const {
+    // Call the alarm function with a parameter of 0 to cancel any previously set alarm
+    (void)alarm(0);
+}
 
+// Define the function `SigAlarmHandler` which is a member function of the `SignalHandler` class inside the `DuplexPipe` class
 void DuplexPipe::SignalHandler::SigAlarmHandler(int sig) {
+  // Print the signal number and the value of the `child_pid_` member variable using the DP_INFO macro
   DP_INFO << "Signal: " << sig << ", child_pid_: " << child_pid_;
+  
+  // Try to lock the weak pointer `dp_` to get a shared pointer to the `DuplexPipe` object
   auto shared_dp = dp_.lock();
+  
+  // Check if the shared pointer is not null
   if (shared_dp != nullptr) {
+    // Call the `NotifyTimeOut` function of the `DuplexPipe` object through the shared pointer
     shared_dp->NotifyTimeOut();
   }
+  
+  // Check if the `child_pid_` member variable is not null
   if (child_pid_ != nullptr) {
+    // Set the value of the `child_pid_` pointer to -1
     *child_pid_ = -1;
   }
 }
 
+// Define the SignalHandler function for handling the SIGPIPE signal
 void DuplexPipe::SignalHandler::SigPipeHandler(int sig) {
+  // Log the signal number and the child process ID
   DP_INFO << "Signal: " << sig << ", child_pid_: " << child_pid_;
+  
+  // Attempt to lock the weak pointer to the DuplexPipe object
   auto shared_dp = dp_.lock();
+  
+  // If the weak pointer is still valid (i.e., the DuplexPipe object still exists)
   if (shared_dp != nullptr) {
+    // Call the NotifyFinalize function of the DuplexPipe object to notify it of the signal
     shared_dp->NotifyFinalize();
   }
+  
+  // If the child process ID is not null
   if (child_pid_ != nullptr) {
+    // Set the value of the child process ID to -1 to indicate that it is no longer valid
     *child_pid_ = -1;
   }
 }
 
+// Define the function `SigChildHandler` belonging to the `SignalHandler` class in the `DuplexPipe` namespace
 void DuplexPipe::SignalHandler::SigChildHandler(int) {
   int status;
+  // Check if the child process ID is not null
   if (child_pid_ != nullptr) {
+    // Use the `waitpid` function to wait for the child process to change state
+    // The `WNOHANG` flag specifies that the function should return immediately if the child process has not changed state
+    // The `WUNTRACED` flag specifies that the function should also return if the child process has been stopped
     (void)waitpid(*child_pid_, &status, WNOHANG | WUNTRACED);
+    // Set the child process ID to -1 to indicate that there is no child process
     *child_pid_ = -1;
   }
 }
-}  // namespace mindspore
+}  // namespace mindspore

mindspore/ccsrc/common/thread_pool.cc

@@ -29,144 +29,236 @@ constexpr size_t kDeviceNum = 8;
 constexpr size_t kMaxThreadNum = 23;
 constexpr size_t kYieldThreshold = 1000;
 
+// Constructor for the ThreadPool class
 ThreadPool::ThreadPool() {
+  // Get the number of hardware cores available on the system and subtract 1
   size_t process_core_num = std::thread::hardware_concurrency() - 1;
+  
+  // If the calculated number of cores is less than 1, set it to 1
   if (process_core_num < 1) {
     process_core_num = 1;
   }
+  
+  // Check if the ENABLE_D or ENABLE_GPU macros are defined
 #if ENABLE_D || ENABLE_GPU
+  // If defined, divide the calculated number of cores by kDeviceNum and assign it to max_thread_num_
   max_thread_num_ = process_core_num / kDeviceNum;
 #else
+  // If not defined, assign the calculated number of cores to max_thread_num_
   max_thread_num_ = process_core_num;
 #endif
+  
+  // If the value of max_thread_num_ is less than 1, set it to 1
   if (max_thread_num_ < 1) {
     max_thread_num_ = 1;
   }
+  
+  // If the value of max_thread_num_ is greater than kMaxThreadNum, set it to kMaxThreadNum
   if (max_thread_num_ > kMaxThreadNum) {
     max_thread_num_ = kMaxThreadNum;
   }
 }
 
+// Synchronize the run loop of the thread pool with the given thread context
 void ThreadPool::SyncRunLoop(const std::shared_ptr<ThreadContext> &context) {
+  // If the thread context is null, return immediately
   if (context == nullptr) {
     return;
   }
+  
+  // Initialize a variable to keep track of the number of times the loop has yielded
   size_t yield_count = 0;
+  
+  // Enter an infinite loop
   while (true) {
+    // If the exit flag is set, return from the function
     if (exit_run_) {
       return;
     }
+    
+    // ... (additional code here)
+  }
+}
 
+// Check if the task pointer in the context is null
     if (!context->task) {
+      // Increment the yield count
       ++yield_count;
+      // Check if the yield count has exceeded the yield threshold
       if (yield_count > kYieldThreshold) {
+        // Reset the yield count
         yield_count = 0;
+        // Acquire a lock on the context's mutex
         std::unique_lock<std::mutex> lock(context->mutex);
+        // Wait on the condition variable until either a new task is assigned or the exit flag is set
         context->cond_var.wait(lock, [&context, this] { return context->task != nullptr || exit_run_; });
       } else {
+        // Yield the current thread's execution to allow other threads to run
         std::this_thread::yield();
+        // Continue to the next iteration of the loop
         continue;
       }
     }
 
+    // Check if the variable "exit_run_" is true
     if (exit_run_) {
       return;
     }
 
     try {
+      // Get a reference to the task object stored in the context
       auto &task = *(context->task);
+      // Call the task function
       task();
     } catch (std::exception &e) {
+      // If an exception is thrown, set the exception flag in the MsException singleton
       MsException::Instance().SetException();
     }
+    // Reset the yield count to 0
     yield_count = 0;
+    // Set the task pointer in the context to nullptr
     context->task = nullptr;
   }
 }
 
 bool ThreadPool::SyncRun(const std::vector<Task> &tasks) {
   if (tasks.empty()) {
-    return true;
+    return true; // If it is empty, return true indicating successful execution
   }
+  
+  // Check if there is only one task in the vector
   if (tasks.size() == 1) {
-    auto ret = tasks[0]();
-    return ret == SUCCESS;
+    auto ret = tasks[0](); // Execute the task and store the return value
+    return ret == SUCCESS; // Return true if the return value is SUCCESS, false otherwise
   }
-  std::unique_lock<std::mutex> lock(pool_mtx_);
-  exit_run_ = false;
-  size_t task_num = tasks.size();
-  size_t thread_num = sync_run_threads_.size();
+  
+  std::unique_lock<std::mutex> lock(pool_mtx_); // Acquire a unique lock on the mutex
+
+  exit_run_ = false; // Set the exit_run_ flag to false
+
+  size_t task_num = tasks.size(); // Get the number of tasks
+  size_t thread_num = sync_run_threads_.size(); // Get the number of threads
+  
+  // Check if the number of threads is less than the maximum thread number and less than the number of tasks
   if (thread_num < max_thread_num_ && thread_num < task_num) {
-    auto new_thread_num = max_thread_num_;
+    auto new_thread_num = max_thread_num_; // Set the new thread number to the maximum thread number
+    
+    // Check if the number of tasks is less than the maximum thread number
     if (task_num < max_thread_num_) {
-      new_thread_num = task_num;
+      new_thread_num = task_num; // Set the new thread number to the number of tasks
     }
-    contexts_.resize(new_thread_num);
+    
+    contexts_.resize(new_thread_num); // Resize the contexts_ vector to the new thread number
+    
+    // Create new thread contexts for the additional threads
     for (size_t i = thread_num; i < new_thread_num; ++i) {
-      contexts_[i] = std::make_shared<ThreadContext>();
+      contexts_[i] = std::make_shared<ThreadContext>(); // Create a new ThreadContext and store it in the contexts_ vector
+    }
+      // Create a new thread and add it to the vector sync_run_threads_ by calling the SyncRunLoop function of the ThreadPool class with the current context as an argument
       sync_run_threads_.emplace_back(std::thread(&ThreadPool::SyncRunLoop, this, contexts_[i]));
     }
   }
+  // If there are no contexts, return true
   if (contexts_.empty()) {
     return true;
   }
+  // Set the used_thread_num to the size of the contexts_ vector
   size_t used_thread_num = contexts_.size();
+  // If the task_num is less than the used_thread_num, set the used_thread_num to the task_num
   if (task_num < used_thread_num) {
     used_thread_num = task_num;
   }
+  // Initialize the running variable to true
   bool running = true;
+  // Initialize the task_index variable to 0
   size_t task_index = 0;
+  // Enter a while loop that continues until running is false
   while (running) {
+    // Set running to false
     running = false;
+    // Iterate over the used_thread_num
     for (size_t i = 0; i < used_thread_num; ++i) {
+      // Check if the current context is not null
       MS_EXCEPTION_IF_NULL(contexts_[i]);
+      // Get the task_run variable from the current context
       auto &task_run = contexts_[i]->task;
+      // If task_run is not null, set running to true
       if (task_run) {
         running = true;
-      } else if (task_index < task_num) {
+      } 
+      // If task_index is less than task_num, continue to the next iteration
+      else if (task_index < task_num) {
+        // Create a lock_guard object to lock the mutex associated with the i-th context
         std::lock_guard<std::mutex> task_lock(contexts_[i]->mutex);
+        
+        // Assign the i-th context's task pointer to the task at task_index in the tasks array
         contexts_[i]->task = &(tasks[task_index]);
+        
+        // Notify one waiting thread that a task is available
         contexts_[i]->cond_var.notify_one();
+        
+        // Set the running flag to true to indicate that a task is being executed
         running = true;
+        
+        // Increment the task_index to move to the next task in the tasks array
         ++task_index;
       }
     }
+    
+    // If there is at least one task running, yield the current thread to allow other threads to run
     if (running) {
       std::this_thread::yield();
     }
   }
+  
+  // Return true to indicate that all tasks have been assigned and executed
   return true;
 }
 
+// Define a member function named "GetInstance" for the ThreadPool class
 ThreadPool &ThreadPool::GetInstance() {
+  // Declare a static instance of the ThreadPool class, initialized with default constructor
   static ThreadPool instance{};
+  // Return the static instance of the ThreadPool class
   return instance;
 }
 
+// A function to clear the thread pool
 void ThreadPool::ClearThreadPool() {
-  std::lock_guard<std::mutex> sync_run_lock(pool_mtx_);
-  if (exit_run_) {
+  std::lock_guard<std::mutex> sync_run_lock(pool_mtx_); // Lock the mutex to ensure thread safety
+
+  if (exit_run_) { // If the exit flag is already set, return
     return;
   }
-  exit_run_ = true;
-  for (auto &context : contexts_) {
-    MS_EXCEPTION_IF_NULL(context);
-    context->cond_var.notify_one();
+
+  exit_run_ = true; // Set the exit flag to true
+
+  for (auto &context : contexts_) { // Iterate through all the contexts in the thread pool
+    MS_EXCEPTION_IF_NULL(context); // Check if the context is null
+    context->cond_var.notify_one(); // Notify the context's condition variable
   }
-  for (auto &it : sync_run_threads_) {
-    if (it.joinable()) {
-      it.join();
+
+  for (auto &it : sync_run_threads_) { // Iterate through all the threads in the thread pool
+    if (it.joinable()) { // If the thread is joinable
+      it.join(); // Join the thread
     }
   }
-  sync_run_threads_.clear();
+
+  sync_run_threads_.clear(); // Clear the vector of threads
 }
 
+// Destructor for the ThreadPool class
 ThreadPool::~ThreadPool() {
   try {
-    ClearThreadPool();
+    ClearThreadPool(); // Call the ClearThreadPool function to clean up the thread pool
   } catch (...) {
+    // If an exception is thrown during the cleanup process, catch it here
+    // and do nothing (since we are in a destructor and cannot throw exceptions)
+    // This is not ideal, but it prevents the program from crashing
+    // It would be better to handle the exception properly, but that is not done here
+    // Handle the exception properly
     // exit
   }
 }
 }  // namespace common
-}  // namespace mindspore
+}  // namespace mindspore