io: support reset photon at fork

In photon context, event engine based module need reset after
fork, if exec will not be called after fork.
This is implicitly done by pthread_atfork hook.

Signed-off-by: yuchen.cc <yuchen.cc@alibaba-inc.com>

CMakeLists.txt

@@ -166,6 +166,7 @@ file(GLOB PHOTON_SRC
         fs/xfile.cpp
         fs/httpfs/*.cpp
         io/signal.cpp
+        io/resettable_ee.cpp
         net/*.cpp
         net/http/*.cpp
         net/security-context/tls-stream.cpp

io/epoll.cpp

@@ -26,6 +26,7 @@ limitations under the License.
 #include <photon/thread/thread.h>
 #include <photon/io/fd-events.h>
 #include "events_map.h"
+#include "resettable_ee.h"
 
 namespace photon {
 #ifndef EPOLLRDHUP
@@ -48,7 +49,7 @@ struct InFlightEvent {
     void* error_data;
 };
 
-class EventEngineEPoll : public MasterEventEngine, public CascadingEventEngine {
+class EventEngineEPoll : public MasterEventEngine, public CascadingEventEngine, public ResettableEventEngine {
 public:
     int _evfd = -1;
     int _engine_fd = -1;
@@ -71,6 +72,13 @@ class EventEngineEPoll : public MasterEventEngine, public CascadingEventEngine {
         epfd = evfd = -1;
         return 0;
     }
+    int reset() override {
+        if_close_fd(_engine_fd);    // close original fd
+        if_close_fd(_evfd);
+        _inflight_events.clear();   // reset members
+        _events_remain = 0;
+        return init();              // re-init
+    }
     virtual ~EventEngineEPoll() override {
         LOG_INFO("Finish event engine: epoll");
         if_close_fd(_engine_fd);

io/fstack-dpdk.cpp

@@ -27,6 +27,7 @@ limitations under the License.
 #include "../thread/thread11.h"
 #include "../common/alog.h"
 #include "../net/basic_socket.h"
+#include "resettable_ee.h"
 
 #ifndef EVFILT_EXCEPT
 #define EVFILT_EXCEPT (-15)
@@ -37,7 +38,7 @@ namespace photon {
 constexpr static EventsMap<EVUnderlay<EVFILT_READ, EVFILT_WRITE, EVFILT_EXCEPT>>
     evmap;
 
-class FstackDpdkEngine : public MasterEventEngine, public CascadingEventEngine {
+class FstackDpdkEngine : public MasterEventEngine, public CascadingEventEngine, public ResettableEventEngine {
 public:
     struct InFlightEvent {
         uint32_t interests = 0;
@@ -76,6 +77,10 @@ class FstackDpdkEngine : public MasterEventEngine, public CascadingEventEngine {
         return 0;
     }
 
+    int reset() override {
+        assert(false);
+    }
+
     ~FstackDpdkEngine() override {
         LOG_INFO("Finish f-stack dpdk engine");
         // if (_n > 0) LOG_INFO(VALUE(_events[0].ident), VALUE(_events[0].filter), VALUE(_events[0].flags));

io/iouring-wrapper.cpp

@@ -30,6 +30,7 @@ limitations under the License.
 #include <photon/thread/thread11.h>
 #include <photon/io/fd-events.h>
 #include "events_map.h"
+#include "resettable_ee.h"
 
 #ifndef _GNU_SOURCE
 #define _GNU_SOURCE
@@ -42,12 +43,22 @@ namespace photon {
 
 constexpr static EventsMap<EVUnderlay<POLLIN | POLLRDHUP, POLLOUT, POLLERR>> evmap;
 
-class iouringEngine : public MasterEventEngine, public CascadingEventEngine {
+class iouringEngine : public MasterEventEngine, public CascadingEventEngine, public ResettableEventEngine {
 public:
     explicit iouringEngine(bool master) : m_master(master) {}
 
     ~iouringEngine() {
         LOG_INFO("Finish event engine: iouring ", VALUE(m_master));
+        fini();
+    }
+
+    int reset() override {
+        fini();
+        m_event_contexts.clear();
+        return init();
+    }
+
+    int fini() {
         if (m_eventfd >= 0 && !m_master) {
             if (io_uring_unregister_eventfd(m_ring) != 0) {
                 LOG_ERROR("iouring: failed to unregister cascading event fd");
@@ -61,6 +72,7 @@ class iouringEngine : public MasterEventEngine, public CascadingEventEngine {
         }
         delete m_ring;
         m_ring = nullptr;
+        return 0;
     }
 
     int init() {

io/kqueue.cpp

@@ -21,13 +21,14 @@ limitations under the License.
 #include <sys/event.h>
 #include <photon/common/alog.h>
 #include "events_map.h"
+#include "resettable_ee.h"
 
 namespace photon {
 
 constexpr static EventsMap<EVUnderlay<EVFILT_READ, EVFILT_WRITE, EVFILT_EXCEPT>>
     evmap;
 
-class KQueue : public MasterEventEngine, public CascadingEventEngine {
+class KQueue : public MasterEventEngine, public CascadingEventEngine, public ResettableEventEngine {
 public:
     struct InFlightEvent {
         uint32_t interests = 0;
@@ -55,6 +56,15 @@ class KQueue : public MasterEventEngine, public CascadingEventEngine {
         return 0;
     }
 
+    int reset() override {
+        LOG_INFO("Reset event engine: kqueue");
+        _kq = -1;                   // kqueue fd is not inherited from the parent process
+        _inflight_events.clear();   // reset members
+        _n = 0;
+        _tm = {0, 0};
+        return init();              // re-init
+    }
+
     ~KQueue() override {
         LOG_INFO("Finish event engine: kqueue");
         // if (_n > 0) LOG_INFO(VALUE(_events[0].ident), VALUE(_events[0].filter), VALUE(_events[0].flags));

io/resettable_ee.cpp

@@ -0,0 +1,54 @@
+/*
+Copyright 2022 The Photon Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+#include "resettable_ee.h"
+#include <photon/common/alog.h>
+#include <pthread.h>
+
+namespace photon {
+
+static thread_local bool registed = false;
+
+static intrusive_list<ResettableEventEngine> &ree_list() {
+    static thread_local intrusive_list<ResettableEventEngine> list;
+    return list;
+}
+
+void fork_hook_event_engine() {
+    if (!registed || ree_list().empty())
+        return;
+    LOG_INFO("reset event engine at fork");
+    for (auto ree : ree_list()) {
+        LOG_DEBUG("reset event engine ", VALUE(ree));
+        ree->reset();
+    }
+}
+
+ResettableEventEngine::ResettableEventEngine() {
+    if (!registed) {
+        pthread_atfork(nullptr, nullptr, &fork_hook_event_engine);
+        registed = true;
+    }
+    LOG_DEBUG("push ", VALUE(this));
+    ree_list().push_back(this);
+}
+
+ResettableEventEngine::~ResettableEventEngine() {
+    LOG_DEBUG("erase ", VALUE(this));
+    ree_list().erase(this);
+}
+
+} // namespace photon

io/resettable_ee.h

@@ -0,0 +1,30 @@
+/*
+Copyright 2022 The Photon Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+#pragma once
+#include <photon/thread/list.h>
+
+namespace photon {
+
+class ResettableEventEngine : public intrusive_list_node<ResettableEventEngine> {
+public:
+    ResettableEventEngine();
+    virtual ~ResettableEventEngine();
+
+    virtual int reset() = 0;
+};
+
+} // namespace photon

io/signal.cpp

@@ -227,13 +227,30 @@ namespace photon
 #endif
     }
 
+    static thread_local bool registed = false;
     // should be invoked in child process after forked, to clear signal mask
-    static void fork_hook_child(void)
+    static void fork_hook_signalfd(void)
     {
-        LOG_DEBUG("Fork hook");
+        if (!registed || sgfd < 0)
+            return;
+        LOG_INFO("reset signalfd at fork");
         sigset_t sigset0;       // can NOT use photon::clear_signal_mask(),
         sigemptyset(&sigset0);  // as memory may be shared with parent, when vfork()ed
         sigprocmask(SIG_SETMASK, &sigset0, nullptr);
+        // reset sgfd
+        memset(sighandlers, 0, sizeof(sighandlers));
+#ifdef __APPLE__
+        sgfd = kqueue();        // kqueue fd is not inherited from the parent process
+#else
+        close(sgfd);
+        sigfillset(&sigset);
+        sgfd = signalfd(-1, &sigset, SFD_CLOEXEC | SFD_NONBLOCK);
+#endif
+        if (sgfd == -1)
+            LOG_ERROR("failed to create signalfd() or kqueue()");
+        // interrupt event loop by ETIMEDOUT to replace sgfd
+        if (eloop)
+            thread_interrupt(eloop->loop_thread(), ETIMEDOUT);
     }
 
     int sync_signal_init()
@@ -264,7 +281,10 @@ namespace photon
         eloop->async_run();
         LOG_INFO("signalfd initialized");
         thread_yield(); // give a chance let eloop to execute do_wait
-        pthread_atfork(nullptr, nullptr, &fork_hook_child);
+        if (!registed) {
+            pthread_atfork(nullptr, nullptr, &fork_hook_signalfd);
+            registed = true;
+        }
         return clear_signal_mask();
     }
 

io/test/CMakeLists.txt

@@ -8,6 +8,13 @@ add_executable(signalfdboom signalfdboom.cpp)
 target_link_libraries(signalfdboom PRIVATE photon_shared)
 add_test(NAME signalfdboom COMMAND $<TARGET_FILE:signalfdboom>)
 
+add_executable(test-fork test-fork.cpp)
+target_link_libraries(test-fork PRIVATE photon_shared)
+if (PHOTON_ENABLE_URING)
+    target_compile_definitions(test-fork PRIVATE PHOTON_URING=on)
+endif()
+add_test(NAME test-fork COMMAND $<TARGET_FILE:test-fork>)
+
 if (NOT APPLE)
 add_executable(test-syncio test-syncio.cpp)
 target_link_libraries(test-syncio PRIVATE photon_shared)