feat: change subprocess implementation.

NEED TEST: BufferStream, AutoReadStream, SubProcess.

2 files changed, 241 insertions, 65 deletions

diff --git a/include/cru/common/SubProcess.h b/include/cru/common/SubProcess.h
index 86dd3ebe..9cfe3a8e 100644
--- a/include/cru/common/SubProcess.h
+++ b/include/cru/common/SubProcess.h
@@ -79,28 +79,66 @@ struct SubProcessExitResult {
   }
 };
 
-/**
- * @brief Base class of a platform process. It is one-time, which means it
- * starts and exits and can't start again.
- * @remarks
- * If an object of this class is destructed before the process exits, the
- * process will be killed.
- */
-class PlatformSubProcessBase : public Object {
-  CRU_DEFINE_CLASS_LOG_TAG(u"PlatformSubProcessBase")
-
- public:
-  explicit PlatformSubProcessBase(SubProcessStartInfo start_info);
-
-  ~PlatformSubProcessBase() override;
-
+template <typename T>
+concept PlatformSubProcessImpl =
+    requires(T process, const SubProcessStartInfo& start_info) {
+      /**
+       * @brief Default constructible.
+       */
+      new T();
+
+      /**
+       * @brief Create and start a real process.
+       *
+       * If the program can't be created or start, an exception should be
+       * thrown.
+       *
+       * Implementation should fill internal data of the new process and start
+       * it.
+       *
+       * This method will be called only once in first call of `Start` on this
+       * thread with lock holdDefaultConstructible.
+       */
+      process.PlatformCreateProcess(start_info);
+
+      /**
+       * @brief Wait for the created process forever and return the exit result
+       * when process stops.
+       *
+       * Implementation should wait for the real process forever, after that,
+       * fill internal data and returns exit result.
+       *
+       * This method will be called only once on another thread after
+       * `PlatformCreateProcess` returns successfully
+       */
+      {
+        process.PlatformWaitForProcess()
+      } -> std::same_as<SubProcessExitResult>;
+
+      /**
+       * @brief Kill the process immediately.
+       *
+       * This method will be called only once on this thread given
+       * `PlatformCreateProcess` returns successfully. There will be a window
+       * that the window already exits but the status has not been updated and
+       * this is called. So handle this gracefully and write to internal data
+       * carefully.
+       */
+      process.PlatformKillProcess();
+
+      { process.GetStdinStream() } -> std::convertible_to<io::Stream*>;
+      { process.GetStdoutStream() } -> std::convertible_to<io::Stream*>;
+      { process.GetStderrStream() } -> std::convertible_to<io::Stream*>;
+    };
+
+struct IPlatformSubProcess : virtual Interface {
   /**
    * @brief Create and start a real process. If the process can't be created or
    * start, `SubProcessFailedToStartException` will be thrown. If this function
    * is already called once, `SubProcessException` will be thrown. Ensure only
    * call this method once.
    */
-  void Start();
+  virtual void Start() = 0;
 
   /**
    * @brief Wait for the process to exit optionally for at most `wait_time`. If
@@ -112,7 +150,7 @@ class PlatformSubProcessBase : public Object {
    * or the process exits. But no, even if it is timeout, the process may also
    * have exited due to task schedule.
    */
-  void Wait(std::optional<std::chrono::milliseconds> wait_time);
+  virtual void Wait(std::optional<std::chrono::milliseconds> wait_time) = 0;
 
   /**
    * @brief kill the process if it is running. If the process already exits,
@@ -120,7 +158,7 @@ class PlatformSubProcessBase : public Object {
    * `SubProcessException` will be throw. Ensure `Start` is called and does not
    * throw before calling this.
    */
-  void Kill();
+  virtual void Kill() = 0;
 
   /**
    * @brief Get the status of the process.
@@ -131,66 +169,203 @@ class PlatformSubProcessBase : public Object {
    * actually running. Because there might be a window that the process exits
    * already but status is not updated.
    */
-  SubProcessStatus GetStatus();
+  virtual SubProcessStatus GetStatus() = 0;
 
   /**
    * @brief Get the exit result. If the process is not started, failed to start
    * or running, `SubProcessException` will be thrown.
    */
-  SubProcessExitResult GetExitResult();
+  virtual SubProcessExitResult GetExitResult() = 0;
 
   virtual io::Stream* GetStdinStream() = 0;
   virtual io::Stream* GetStdoutStream() = 0;
   virtual io::Stream* GetStderrStream() = 0;
+};
+
+/**
+ * @brief A wrapper platform process. It is one-time, which means it
+ * starts and exits and can't start again.
+ *
+ * TODO: Current implementation has a problem. If the process does not exit for
+ * a long time, the resource related to it will not be released. It may cause a
+ * leak.
+ */
+template <PlatformSubProcessImpl Impl>
+class PlatformSubProcess : public Object, public virtual IPlatformSubProcess {
+  CRU_DEFINE_CLASS_LOG_TAG(u"PlatformSubProcessBase")
+
+ private:
+  struct State {
+    explicit State(SubProcessStartInfo start_info)
+        : start_info(std::move(start_info)) {}
+
+    std::mutex mutex;
+    std::unique_lock<std::mutex> lock{mutex, std::defer_lock};
+    std::condition_variable condition_variable;
+    SubProcessStartInfo start_info;
+    SubProcessExitResult exit_result;
+    SubProcessStatus status = SubProcessStatus::Prepare;
+    bool killed = false;
+    Impl impl;
+  };
+
+ public:
+  explicit PlatformSubProcess(SubProcessStartInfo start_info)
+      : state_(new State(std::move(start_info))) {}
 
-  void SetDeleteSelfOnExit(bool enable);
+  ~PlatformSubProcess() override {}
 
- protected:
   /**
-   * @brief Create and start a real process. If the program can't be created or
-   * start, an exception should be thrown.
-   *
-   * Implementation should fill internal data of the new process and start it.
-   *
-   * This method will be called only once in first call of `Start` on this
-   * thread with lock hold.
+   * @brief Create and start a real process. If the process can't be created or
+   * start, `SubProcessFailedToStartException` will be thrown. If this function
+   * is already called once, `SubProcessException` will be thrown. Ensure only
+   * call this method once.
    */
-  virtual void PlatformCreateProcess() = 0;
+  void Start() override {
+    std::lock_guard lock_guard(this->state_->lock);
+
+    if (this->state_->status != SubProcessStatus::Prepare) {
+      throw SubProcessException(u"The process has already tried to start.");
+    }
+
+    try {
+      this->state_->impl.PlatformCreateProcess(this->state_->start_info);
+      this->state_->status = SubProcessStatus::Running;
+
+      auto thread = std::thread([state = state_] {
+        std::lock_guard lock_guard(state->lock);
+        state->exit_result = state->impl.PlatformWaitForProcess();
+        state->status = SubProcessStatus::Exited;
+        state->condition_variable.notify_all();
+      });
+
+      thread.detach();
+    } catch (const std::exception& e) {
+      this->state_->status = SubProcessStatus::FailedToStart;
+      throw SubProcessFailedToStartException(u"Sub-process failed to start. " +
+                                             String::FromUtf8(e.what()));
+    }
+  }
 
   /**
-   * @brief Wait for the created process forever.
-   *
-   * Implementation should wait for the real process forever, after that, fill
-   * internal data and returns exit result.
+   * @brief Wait for the process to exit optionally for at most `wait_time`. If
+   * the process already exits, it will return immediately. If the process has
+   * not started or failed to start, `SubProcessException` will be thrown.
+   * Ensure `Start` is called and does not throw before calling this.
    *
-   * This method will be called only once on another thread after
-   * `PlatformCreateProcess` returns successfully
+   * @remarks You may wish this returns bool to indicate whether it is timeout
+   * or the process exits. But no, even if it is timeout, the process may also
+   * have exited due to task schedule.
    */
-  virtual SubProcessExitResult PlatformWaitForProcess() = 0;
+  void Wait(std::optional<std::chrono::milliseconds> wait_time) override {
+    std::lock_guard lock_guard(this->state_->lock);
+
+    if (this->state_->status == SubProcessStatus::Prepare) {
+      throw SubProcessException(
+          u"The process does not start. Can't wait for it.");
+    }
+
+    if (this->state_->status == SubProcessStatus::FailedToStart) {
+      throw SubProcessException(
+          u"The process failed to start. Can't wait for it.");
+    }
+
+    if (this->state_->status == SubProcessStatus::Exited) {
+      return;
+    }
+
+    auto predicate = [this] {
+      return this->state_->status == SubProcessStatus::Exited;
+    };
+
+    if (wait_time) {
+      this->state_->condition_variable.wait_for(this->state_->lock, *wait_time,
+                                                predicate);
+    } else {
+      this->state_->condition_variable.wait(this->state_->lock, predicate);
+    }
+  }
 
   /**
-   * @brief Kill the process immediately.
-   *
-   * This method will be called only once on this thread given
-   * `PlatformCreateProcess` returns successfullyThere will be a window that the
-   * window already exits but the status has not been updated and this is
-   * called. So handle this gracefully and write to internal data carefully.
+   * @brief kill the process if it is running. If the process already exits,
+   * nothing will happen. If the process has not started or failed to start,
+   * `SubProcessException` will be throw. Ensure `Start` is called and does not
+   * throw before calling this.
    */
-  virtual void PlatformKillProcess() = 0;
+  void Kill() override {
+    std::lock_guard lock_guard(this->state_->lock);
 
- protected:
-  SubProcessStartInfo start_info_;
-  SubProcessExitResult exit_result_;
+    if (this->state_->status == SubProcessStatus::Prepare) {
+      throw SubProcessException(u"The process does not start. Can't kill it.");
+    }
 
- private:
-  SubProcessStatus status_;
+    if (this->state_->status == SubProcessStatus::FailedToStart) {
+      throw SubProcessException(u"The process failed to start. Can't kill it.");
+    }
+
+    if (this->state_->status == SubProcessStatus::Exited) {
+      return;
+    }
+
+    if (this->state_->killed) {
+      return;
+    }
+
+    this->state_->impl.PlatformKillProcess();
+    this->state_->killed = true;
+  }
+
+  /**
+   * @brief Get the status of the process.
+   * 1. If the process has tried to start, aka `Start` is called, then this
+   * method will return one of `Running`, `FailedToStart`, `Exited`.
+   * 2. If it returns `Prepare`, `Start` is not called.
+   * 3. It does NOT guarantee that this return `Running` and the process is
+   * actually running. Because there might be a window that the process exits
+   * already but status is not updated.
+   */
+  SubProcessStatus GetStatus() override {
+    std::lock_guard lock_guard(this->state_->lock);
+    return this->state_->status;
+  }
+
+  /**
+   * @brief Get the exit result. If the process is not started, failed to start
+   * or running, `SubProcessException` will be thrown.
+   */
+  SubProcessExitResult GetExitResult() override {
+    std::lock_guard lock_guard(this->state_->lock);
 
-  bool delete_self_;
+    if (this->state_->status == SubProcessStatus::Prepare) {
+      throw SubProcessException(
+          u"The process does not start. Can't get exit result.");
+    }
 
-  std::thread process_thread_;
-  std::recursive_mutex process_mutex_;
-  std::unique_lock<std::recursive_mutex> process_lock_;
-  std::condition_variable_any process_condition_variable_;
+    if (this->state_->status == SubProcessStatus::FailedToStart) {
+      throw SubProcessException(
+          u"The process failed to start. Can't get exit result.");
+    }
+
+    if (this->state_->status == SubProcessStatus::Running) {
+      throw SubProcessException(
+          u"The process is running. Can't get exit result.");
+    }
+
+    return this->state_->exit_result;
+  }
+
+  io::Stream* GetStdinStream() override {
+    return this->state_->impl.GetStdinStream();
+  }
+  io::Stream* GetStdoutStream() override {
+    return this->state_->impl.GetStdoutStream();
+  }
+  io::Stream* GetStderrStream() override {
+    return this->state_->impl.GetStderrStream();
+  }
+
+ private:
+  std::shared_ptr<State> state_;
 };
 
 class CRU_BASE_API SubProcess : public Object {
@@ -231,6 +406,6 @@ class CRU_BASE_API SubProcess : public Object {
   void CheckValid() const;
 
  private:
-  std::unique_ptr<PlatformSubProcessBase> platform_process_;
+  std::unique_ptr<IPlatformSubProcess> platform_process_;
 };
 }  // namespace cru
diff --git a/include/cru/common/platform/unix/PosixSpawnSubProcess.h b/include/cru/common/platform/unix/PosixSpawnSubProcess.h
index 9c303700..d4df284b 100644
--- a/include/cru/common/platform/unix/PosixSpawnSubProcess.h
+++ b/include/cru/common/platform/unix/PosixSpawnSubProcess.h
@@ -16,21 +16,20 @@
 #include <spawn.h>
 
 namespace cru::platform::unix {
-class PosixSpawnSubProcess : public PlatformSubProcessBase {
+class PosixSpawnSubProcessImpl {
   CRU_DEFINE_CLASS_LOG_TAG(u"PosixSpawnSubProcess")
 
  public:
-  explicit PosixSpawnSubProcess(SubProcessStartInfo start_info);
-  ~PosixSpawnSubProcess();
+  explicit PosixSpawnSubProcessImpl();
+  ~PosixSpawnSubProcessImpl();
 
-  io::Stream* GetStdinStream() override;
-  io::Stream* GetStdoutStream() override;
-  io::Stream* GetStderrStream() override;
+  io::Stream* GetStdinStream();
+  io::Stream* GetStdoutStream();
+  io::Stream* GetStderrStream();
 
- protected:
-  void PlatformCreateProcess() override;
-  SubProcessExitResult PlatformWaitForProcess() override;
-  void PlatformKillProcess() override;
+  void PlatformCreateProcess(const SubProcessStartInfo& start_info);
+  SubProcessExitResult PlatformWaitForProcess();
+  void PlatformKillProcess();
 
  private:
   pid_t pid_;
@@ -47,4 +46,6 @@ class PosixSpawnSubProcess : public PlatformSubProcessBase {
   std::unique_ptr<io::AutoReadStream> stdout_buffer_stream_;
   std::unique_ptr<io::AutoReadStream> stderr_buffer_stream_;
 };
+
+using PosixSpawnSubProcess = PlatformSubProcess<PosixSpawnSubProcessImpl>;
 }  // namespace cru::platform::unix