diff --git a/mindspore/ccsrc/common/api_register.cc b/mindspore/ccsrc/common/api_register.cc index c16aeba3221..69763ab56d3 100644 --- a/mindspore/ccsrc/common/api_register.cc +++ b/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; } diff --git a/mindspore/ccsrc/common/duplex_pipe.cc b/mindspore/ccsrc/common/duplex_pipe.cc index f355ba2804c..889d4b67775 100644 --- a/mindspore/ccsrc/common/duplex_pipe.cc +++ b/mindspore/ccsrc/common/duplex_pipe.cc @@ -22,68 +22,104 @@ #include #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 &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 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(&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(&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(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 &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 \ No newline at end of file diff --git a/mindspore/ccsrc/common/thread_pool.cc b/mindspore/ccsrc/common/thread_pool.cc index 054a9741be3..4f2a3e311fe 100644 --- a/mindspore/ccsrc/common/thread_pool.cc +++ b/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 &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 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 &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 lock(pool_mtx_); - exit_run_ = false; - size_t task_num = tasks.size(); - size_t thread_num = sync_run_threads_.size(); + + std::unique_lock 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(); + contexts_[i] = std::make_shared(); // 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 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 sync_run_lock(pool_mtx_); - if (exit_run_) { + std::lock_guard 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 \ No newline at end of file