reactor: Always retry waitpid

A possible issue with how Docker Desktop v4.34+ (particulary on macOS)
handles `pidfd` functionality causes an assertion failure in Seastar's
`reactor::waitpid` method.

The issue:

- Seastar creates a `pidfd` using `pidfd_open`
- It polls this file descriptor until it becomes readable
- When readable, call `waitpid` with `WNOHANG`.  According to
  [pidfd_open](https://man7.org/linux/man-pages/man2/pidfd_open.2.html),
  the file descriptor becomes readable when the process terminates
- Seastar expects `waitpid` to either return a positive integer
  representing the child process that has ended or a negative value to
  indicate an error.  However if `waitpid` returns `0`, then Seastar
  asserts, crashing the using application.

This change introduces retry logic when using a `pidfd` and removes the
assertion.

Signed-off-by: Michael Boquard <[email protected]>

include/seastar/core/reactor.hh

@@ -642,6 +642,7 @@ private:
     void add_timer(timer<manual_clock>*) noexcept;
     bool queue_timer(timer<manual_clock>*) noexcept;
     void del_timer(timer<manual_clock>*) noexcept;
+    future<int> do_waitpid(pid_t pid);
 
     future<> run_exit_tasks();
     void stop();

src/core/reactor.cc

@@ -2296,6 +2296,32 @@ static auto next_waitpid_timeout(std::chrono::milliseconds this_timeout) {
 
 #endif
 
+future<int> reactor::do_waitpid(pid_t pid) {
+    return do_with(int{}, std::chrono::milliseconds(0), [pid, this](int& wstatus,
+                                                                    std::chrono::milliseconds& wait_timeout) {
+        return repeat_until_value([this,
+                                   pid,
+                                   &wstatus,
+                                   &wait_timeout] {
+            return _thread_pool->submit<syscall_result<pid_t>>([pid, &wstatus] {
+                return wrap_syscall<pid_t>(::waitpid(pid, &wstatus, WNOHANG));
+            }).then([&wstatus, &wait_timeout](syscall_result<pid_t> ret) mutable {
+                if (ret.result == 0) {
+                    wait_timeout = next_waitpid_timeout(wait_timeout);
+                    return ::seastar::sleep(wait_timeout).then([] {
+                        return make_ready_future<std::optional<int>>();
+                    });
+                } else if (ret.result > 0) {
+                    return make_ready_future<std::optional<int>>(wstatus);
+                } else {
+                    ret.throw_if_error();
+                    return make_ready_future<std::optional<int>>(-1);
+                }
+            });
+        });
+    });
+}
+
 future<int> reactor::waitpid(pid_t pid) {
     return _thread_pool->submit<syscall_result<int>>([pid] {
         return wrap_syscall<int>(syscall(__NR_pidfd_open, pid, O_NONBLOCK));
@@ -2304,39 +2330,11 @@ future<int> reactor::waitpid(pid_t pid) {
             // pidfd_open() was introduced in linux 5.3, so the pidfd.error could be ENOSYS on
             // older kernels. But it could be other error like EMFILE or ENFILE. anyway, we
             // should always waitpid().
-            return do_with(int{}, std::chrono::milliseconds(0), [pid, this](int& wstatus,
-                                                                            std::chrono::milliseconds& wait_timeout) {
-                return repeat_until_value([this,
-                                           pid,
-                                           &wstatus,
-                                           &wait_timeout] {
-                    return _thread_pool->submit<syscall_result<pid_t>>([pid, &wstatus] {
-                        return wrap_syscall<pid_t>(::waitpid(pid, &wstatus, WNOHANG));
-                    }).then([&wstatus, &wait_timeout] (syscall_result<pid_t> ret) mutable {
-                        if (ret.result == 0) {
-                            wait_timeout = next_waitpid_timeout(wait_timeout);
-                            return ::seastar::sleep(wait_timeout).then([] {
-                                return make_ready_future<std::optional<int>>();
-                            });
-                        } else if (ret.result > 0) {
-                            return make_ready_future<std::optional<int>>(wstatus);
-                        } else {
-                            ret.throw_if_error();
-                            return make_ready_future<std::optional<int>>(-1);
-                        }
-                    });
-                });
-            });
+            return do_waitpid(pid);
         } else {
-            return do_with(pollable_fd(file_desc::from_fd(pidfd.result)), int{}, [pid, this](auto& pidfd, int& wstatus) {
-                return pidfd.readable().then([pid, &wstatus, this] {
-                    return _thread_pool->submit<syscall_result<pid_t>>([pid, &wstatus] {
-                        return wrap_syscall<pid_t>(::waitpid(pid, &wstatus, WNOHANG));
-                    });
-                }).then([&wstatus] (syscall_result<pid_t> ret) {
-                    ret.throw_if_error();
-                    assert(ret.result > 0);
-                    return make_ready_future<int>(wstatus);
+            return do_with(pollable_fd(file_desc::from_fd(pidfd.result)), [pid, this](auto& pidfd) {
+                return pidfd.readable().then([pid, this] {
+                    return do_waitpid(pid);
                 });
             });
         }