From 5dbcd05bacfb0762d7d793c756e1a46d0d5613b9 Mon Sep 17 00:00:00 2001 From: Hevake Date: Tue, 1 Mar 2022 08:33:51 +0800 Subject: [PATCH 01/13] =?UTF-8?q?=E5=AE=8C=E6=88=90=E4=BA=86=E5=B0=8F?= =?UTF-8?q?=E9=83=A8=E5=88=86=E4=BB=A3=E7=A0=81=EF=BC=8C=E6=9C=AA=E5=AE=8C?= =?UTF-8?q?=E6=88=90=EF=BC=8C=E9=80=9A=E8=BF=87=E7=BC=96=E8=AF=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- event/Makefile | 1 + event/common_loop.cpp | 66 +++++++++++++++++++++++++++++++++ event/common_loop.h | 17 +++++++++ event/signal_event_impl.cpp | 49 ++++++++++++++++++++++++ event/signal_event_impl.h | 34 +++++++++++++++++ event/signal_event_test.cpp | 74 +++++++++++++++++++++++++++++++++++++ 6 files changed, 241 insertions(+) create mode 100644 event/signal_event_impl.cpp create mode 100644 event/signal_event_impl.h diff --git a/event/Makefile b/event/Makefile index 0eced0e..0ec627c 100644 --- a/event/Makefile +++ b/event/Makefile @@ -13,6 +13,7 @@ HEAD_FILES = \ CPP_SRC_FILES = \ loop.cpp common_loop.cpp \ + signal_event_impl.cpp \ version.cpp ifeq ($(ENABLE_LIBEVENT), yes) diff --git a/event/common_loop.cpp b/event/common_loop.cpp index 99324b3..054c361 100644 --- a/event/common_loop.cpp +++ b/event/common_loop.cpp @@ -5,6 +5,7 @@ #include #include #include +#include #include @@ -16,6 +17,9 @@ namespace event { using namespace std::chrono; +std::map> CommonLoop::_signal_read_fds_; +std::mutex CommonLoop::_signal_lock_; + CommonLoop::CommonLoop() : has_unhandle_req_(false), read_fd_(-1), write_fd_(-1), @@ -149,6 +153,68 @@ void CommonLoop::endEventProcess() #endif } +bool CommonLoop::subscribeSignal(int signo, SignalEventImpl *who) +{ + if (signal_read_fd_ == -1) { //! 如果还没有创建对应的信号 + int fds[2] = { 0 }; + if (pipe2(fds, O_CLOEXEC | O_NONBLOCK) != 0) { //!FIXME + LogErr("pip2() fail, ret:%d", errno); + return false; + } + int read_fd(fds[0]); + int write_fd(fds[1]); + + auto read_fd_event = newFdEvent(); + read_fd_event->initialize(read_fd, FdEvent::kReadEvent, Event::Mode::kPersist); + read_fd_event->setCallback(std::bind(&CommonLoop::onSignal, this)); + read_fd_event->enable(); + + { + std::unique_lock _g(_signal_lock_); + auto iter = _signal_read_fds_.find(signo); + if (iter != _signal_read_fds_.end()) { + iter->second.insert(read_fd); + } else { + std::set tmp = { read_fd }; + _signal_read_fds_[signo] = tmp; + signal(signo, CommonLoop::HandleSignal); + } + } + + signal_read_fd_ = read_fd; + signal_write_fd_ = write_fd; + sp_signal_read_event_ = read_fd_event; + } + + LogUndo(); //!TODO + + return false; +} + +bool CommonLoop::unsubscribeSignal(int signal_num, SignalEventImpl *who) +{ + LogUndo(); //!TODO + return false; +} + +void CommonLoop::HandleSignal(int signo) +{ + std::unique_lock _g(_signal_lock_); + auto iter = _signal_read_fds_.find(signo); + if (iter != _signal_read_fds_.end()) { + const auto &fd_set = iter->second; + for (int fd : fd_set) + write(fd, &signo, sizeof(signo)); + } else { + LogWarn("uncatch signal: %d", signo); + } +} + +void CommonLoop::onSignal() +{ + LogUndo(); //!TODO +} + void CommonLoop::handleNextFunc() { while (!run_next_func_queue_.empty()) { diff --git a/event/common_loop.h b/event/common_loop.h index b80064f..39ff740 100644 --- a/event/common_loop.h +++ b/event/common_loop.h @@ -4,8 +4,11 @@ #include #include #include +#include +#include #include "loop.h" +#include "signal_event_impl.h" #ifdef ENABLE_STAT #include @@ -36,6 +39,12 @@ class CommonLoop : public Loop { void beginEventProcess(); void endEventProcess(); + //! 信号 + bool subscribeSignal(int signal_num, SignalEventImpl *who); + bool unsubscribeSignal(int signal_num, SignalEventImpl *who); + static void HandleSignal(int signo); + void onSignal(); + protected: void runThisBeforeLoop(); void runThisAfterLoop(); @@ -70,6 +79,14 @@ class CommonLoop : public Loop { uint32_t max_cost_us_ = 0; #endif //ENABLE_STAT + static std::map> _signal_read_fds_; //! 通知 Loop 的 fd，每个 Loop 注册一个 + static std::mutex _signal_lock_; //! 保护 _signal_read_fds_ 用 + + int signal_read_fd_ = -1; + int signal_write_fd_ = -1; + FdEvent *sp_signal_read_event_ = nullptr; + std::map signal_subscriber_; //! signo -> SignalEventImpl*，信号的订阅者 + int cb_level_ = 0; }; diff --git a/event/signal_event_impl.cpp b/event/signal_event_impl.cpp new file mode 100644 index 0000000..704acb9 --- /dev/null +++ b/event/signal_event_impl.cpp @@ -0,0 +1,49 @@ +#include "signal_event_impl.h" +#include "common_loop.h" +#include + +namespace tbox { +namespace event { + +SignalEventImpl::SignalEventImpl(CommonLoop *wp_loop) : + wp_loop_(wp_loop) +{ + LogUndo(); +} + +SignalEventImpl::~SignalEventImpl() +{ + LogUndo(); +} + +bool SignalEventImpl::initialize(int signum, Mode mode) +{ + LogUndo(); + return false; +} + +bool SignalEventImpl::isEnabled() const +{ + LogUndo(); + return false; +} + +bool SignalEventImpl::enable() +{ + LogUndo(); + return false; +} + +bool SignalEventImpl::disable() +{ + LogUndo(); + return false; +} + +Loop* SignalEventImpl::getLoop() const +{ + return wp_loop_; +} + +} +} diff --git a/event/signal_event_impl.h b/event/signal_event_impl.h new file mode 100644 index 0000000..a00017a --- /dev/null +++ b/event/signal_event_impl.h @@ -0,0 +1,34 @@ +#ifndef TBOX_EVENT_SIGNAL_EVENT_IMPL_H_20220301 +#define TBOX_EVENT_SIGNAL_EVENT_IMPL_H_20220301 + +#include "signal_event.h" + +namespace tbox { +namespace event { + +class CommonLoop; + +class SignalEventImpl : SignalEvent { + public: + explicit SignalEventImpl(CommonLoop *wp_loop); + virtual ~SignalEventImpl(); + + public: + bool initialize(int signum, Mode mode) override; + void setCallback(const CallbackFunc &cb) override { cb_ = cb; } + + bool isEnabled() const override; + bool enable() override; + bool disable() override; + + Loop* getLoop() const override; + + private: + CommonLoop *wp_loop_; + CallbackFunc cb_; +}; + +} +} + +#endif //TBOX_EVENT_SIGNAL_EVENT_IMPL_H_20220301 diff --git a/event/signal_event_test.cpp b/event/signal_event_test.cpp index a38db6c..287662c 100644 --- a/event/signal_event_test.cpp +++ b/event/signal_event_test.cpp @@ -1,5 +1,8 @@ #include #include +#include +#include +#include #include "loop.h" #include "signal_event.h" @@ -112,3 +115,74 @@ TEST(SignalEvent, IntAndTermSignal) } } +TEST(SignalEvent, MultiThread) +{ + auto engins = Loop::Engines(); + for (auto e : engins) { + cout << "engin: " << e << endl; + auto sp_loop = Loop::New(e); + auto int_signal_event = sp_loop->newSignalEvent(); + EXPECT_TRUE(int_signal_event->initialize(SIGINT, Event::Mode::kOneshot)); + EXPECT_TRUE(int_signal_event->enable()); + + auto term_signal_event = sp_loop->newSignalEvent(); + EXPECT_TRUE(term_signal_event->initialize(SIGTERM, Event::Mode::kOneshot)); + EXPECT_TRUE(term_signal_event->enable()); + + int int_run_time = 0; + int_signal_event->setCallback([&]() { ++int_run_time; }); + int term_run_time = 0; + term_signal_event->setCallback([&]() { ++term_run_time; }); + + sp_loop->run([] + { + raise(SIGINT); + raise(SIGTERM); + } + ); + + bool t1_run = false; + //! t1 线程sleep 200ms + auto t1 = std::thread( + [&] { + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + t1_run = true; + } + ); + + //! t2 线程等待一个信号 + bool exit_thread = false; + std::mutex lock; + std::condition_variable cond_var; + bool t2_run = false; + auto t2 = std::thread( + [&] { + std::unique_lock lk(lock); + cond_var.wait(lk, [&]{ return exit_thread; }); + t2_run = true; + } + ); + + + sp_loop->exitLoop(std::chrono::milliseconds(100)); + sp_loop->runLoop(); + + t1.join(); + { + std::unique_lock lk(lock); + exit_thread = true; + cond_var.notify_one(); + } + t2.join(); + + EXPECT_TRUE(t1_run); + EXPECT_TRUE(t2_run); + EXPECT_EQ(int_run_time, 1); + EXPECT_EQ(term_run_time, 1); + + delete int_signal_event; + delete term_signal_event; + delete sp_loop; + } +} + -- Gitee From e2483fabab6f21a48a68f9809c17bea26fd7bc93 Mon Sep 17 00:00:00 2001 From: Hevake Date: Wed, 2 Mar 2022 08:35:01 +0800 Subject: [PATCH 02/13] =?UTF-8?q?=E5=AE=8C=E6=88=90=E4=BA=86SignalEventImp?= =?UTF-8?q?l=E7=9A=84=E4=BB=A3=E7=A0=81=E7=BC=96=E5=86=99=EF=BC=8C?= =?UTF-8?q?=E7=BC=96=E8=AF=91=E9=80=9A=E8=BF=87=EF=BC=8C=E4=BD=86=E6=B2=A1?= =?UTF-8?q?=E6=9C=89=E9=80=9A=E8=BF=87=E5=8D=95=E5=85=83=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- event/common_loop.cpp | 74 +++++++++++++++++++++++++++++----- event/common_loop.h | 7 ++-- event/engins/epoll/loop.cpp | 6 --- event/engins/epoll/loop.h | 1 - event/engins/libev/loop.cpp | 5 --- event/engins/libev/loop.h | 1 - event/engins/libevent/loop.cpp | 5 --- event/engins/libevent/loop.h | 1 - event/signal_event_impl.cpp | 42 ++++++++++++------- event/signal_event_impl.h | 22 +++++++++- 10 files changed, 117 insertions(+), 47 deletions(-) diff --git a/event/common_loop.cpp b/event/common_loop.cpp index 054c361..f7fd6ed 100644 --- a/event/common_loop.cpp +++ b/event/common_loop.cpp @@ -4,6 +4,7 @@ #include #include #include +#include #include #include @@ -153,7 +154,7 @@ void CommonLoop::endEventProcess() #endif } -bool CommonLoop::subscribeSignal(int signo, SignalEventImpl *who) +bool CommonLoop::subscribeSignal(int signo, SignalSubscribuer *who) { if (signal_read_fd_ == -1) { //! 如果还没有创建对应的信号 int fds[2] = { 0 }; @@ -176,7 +177,7 @@ bool CommonLoop::subscribeSignal(int signo, SignalEventImpl *who) iter->second.insert(read_fd); } else { std::set tmp = { read_fd }; - _signal_read_fds_[signo] = tmp; + _signal_read_fds_[signo] = std::move(tmp); signal(signo, CommonLoop::HandleSignal); } } @@ -186,20 +187,55 @@ bool CommonLoop::subscribeSignal(int signo, SignalEventImpl *who) sp_signal_read_event_ = read_fd_event; } - LogUndo(); //!TODO + auto iter = signal_subscribers_.find(signo); + if (iter != signal_subscribers_.end()) { + iter->second.insert(who); + } else { + std::set tmp = { who }; + signal_subscribers_[signo] = std::move(tmp); + } - return false; + return true; } -bool CommonLoop::unsubscribeSignal(int signal_num, SignalEventImpl *who) +bool CommonLoop::unsubscribeSignal(int signo, SignalSubscribuer *who) { - LogUndo(); //!TODO - return false; + auto iter = signal_subscribers_.find(signo); + if (iter == signal_subscribers_.end()) //! 如果本来就不存在，就直接返回了 + return true; + + auto &subscriber_set = iter->second; + subscriber_set.erase(who); //! 将订阅信息删除 + + if (!subscriber_set.empty()) //! 检查本Loop中是否已经没有SignalSubscribuer订阅该信号了 + return true; //! 如果还有，就到此为止 + + //! 如果本Loop已经没有SignalSubscribuer订阅该信号了 + signal_subscribers_.erase(iter); //! 则将该信号的订阅记录表删除 + + std::unique_lock _g(_signal_lock_); + //! 并将 _signal_read_fds_ 中的记录删除 + auto fd_iter = _signal_read_fds_.find(signo); + assert(fd_iter != _signal_read_fds_.end()); + + auto &fd_set = fd_iter->second; + fd_set.erase(signal_read_fd_); + + //! 检查是否还有其它的Loop订阅该信号 + if (!fd_set.empty()) + return true; + + //! 如果没有了，则删除 _signal_read_fds_ 中该信号的订阅记录表 + _signal_read_fds_.erase(fd_iter); + + //! 并还原信号处理函数 + signal(signo, SIG_DFL); + return true; } void CommonLoop::HandleSignal(int signo) { - std::unique_lock _g(_signal_lock_); + //std::unique_lock _g(_signal_lock_); //!FIXME: 这是在信号中执行的，是否需要加锁？ auto iter = _signal_read_fds_.find(signo); if (iter != _signal_read_fds_.end()) { const auto &fd_set = iter->second; @@ -212,7 +248,27 @@ void CommonLoop::HandleSignal(int signo) void CommonLoop::onSignal() { - LogUndo(); //!TODO + while (true) { + int signo = 0; + auto rsize = read(signal_read_fd_, &signo, sizeof(signo)); + if (rsize > 0) { + auto iter = signal_subscribers_.find(signo); + if (iter != signal_subscribers_.end()) { + for (auto s : iter->second) { + s->onSignal(signo); + } + } + } else { + if (errno != EAGAIN) + LogWarn("read error, rsize:%d, errno:%d, %s", rsize, errno, strerror(errno)); + break; + } + } +} + +SignalEvent* CommonLoop::newSignalEvent() +{ + return new SignalEventImpl(this); } void CommonLoop::handleNextFunc() diff --git a/event/common_loop.h b/event/common_loop.h index 39ff740..8f41c2e 100644 --- a/event/common_loop.h +++ b/event/common_loop.h @@ -40,10 +40,11 @@ class CommonLoop : public Loop { void endEventProcess(); //! 信号 - bool subscribeSignal(int signal_num, SignalEventImpl *who); - bool unsubscribeSignal(int signal_num, SignalEventImpl *who); + bool subscribeSignal(int signal_num, SignalSubscribuer *who); + bool unsubscribeSignal(int signal_num, SignalSubscribuer *who); static void HandleSignal(int signo); void onSignal(); + SignalEvent* newSignalEvent() override; protected: void runThisBeforeLoop(); @@ -85,7 +86,7 @@ class CommonLoop : public Loop { int signal_read_fd_ = -1; int signal_write_fd_ = -1; FdEvent *sp_signal_read_event_ = nullptr; - std::map signal_subscriber_; //! signo -> SignalEventImpl*，信号的订阅者 + std::map> signal_subscribers_; //! signo -> SignalSubscribuer*，信号的订阅者 int cb_level_ = 0; }; diff --git a/event/engins/epoll/loop.cpp b/event/engins/epoll/loop.cpp index 3bfcefc..e021192 100644 --- a/event/engins/epoll/loop.cpp +++ b/event/engins/epoll/loop.cpp @@ -218,11 +218,5 @@ TimerEvent* EpollLoop::newTimerEvent() return new EpollTimerEvent(this); } -SignalEvent* EpollLoop::newSignalEvent() -{ - LogWarn("EpollSignalEvent is not stable in multithread"); - return new EpollSignalEvent(this); -} - } } diff --git a/event/engins/epoll/loop.h b/event/engins/epoll/loop.h index 5b5d146..c62f1e6 100644 --- a/event/engins/epoll/loop.h +++ b/event/engins/epoll/loop.h @@ -28,7 +28,6 @@ class EpollLoop : public CommonLoop { virtual FdEvent* newFdEvent(); virtual TimerEvent* newTimerEvent(); - virtual SignalEvent* newSignalEvent(); public: inline int epollFd() const { return epoll_fd_; } diff --git a/event/engins/libev/loop.cpp b/event/engins/libev/loop.cpp index 37718c2..6636c76 100644 --- a/event/engins/libev/loop.cpp +++ b/event/engins/libev/loop.cpp @@ -59,10 +59,5 @@ TimerEvent* LibevLoop::newTimerEvent() return new LibevTimerEvent(this); } -SignalEvent* LibevLoop::newSignalEvent() -{ - return new LibevSignalEvent(this); -} - } } diff --git a/event/engins/libev/loop.h b/event/engins/libev/loop.h index 09c8919..9087bd2 100644 --- a/event/engins/libev/loop.h +++ b/event/engins/libev/loop.h @@ -19,7 +19,6 @@ class LibevLoop : public CommonLoop { virtual FdEvent* newFdEvent(); virtual TimerEvent* newTimerEvent(); - virtual SignalEvent* newSignalEvent(); public: struct ev_loop* getEvLoopPtr() const { return sp_ev_loop_; } diff --git a/event/engins/libevent/loop.cpp b/event/engins/libevent/loop.cpp index 645225a..aa09ece 100644 --- a/event/engins/libevent/loop.cpp +++ b/event/engins/libevent/loop.cpp @@ -48,10 +48,5 @@ TimerEvent* LibeventLoop::newTimerEvent() return new LibeventTimerEvent(this); } -SignalEvent* LibeventLoop::newSignalEvent() -{ - return new LibeventSignalEvent(this); -} - } } diff --git a/event/engins/libevent/loop.h b/event/engins/libevent/loop.h index 9280beb..58a9fe9 100644 --- a/event/engins/libevent/loop.h +++ b/event/engins/libevent/loop.h @@ -19,7 +19,6 @@ class LibeventLoop : public CommonLoop { virtual FdEvent* newFdEvent(); virtual TimerEvent* newTimerEvent(); - virtual SignalEvent* newSignalEvent(); public: struct event_base* getEventBasePtr() const { return sp_event_base_; } diff --git a/event/signal_event_impl.cpp b/event/signal_event_impl.cpp index 704acb9..759fc9c 100644 --- a/event/signal_event_impl.cpp +++ b/event/signal_event_impl.cpp @@ -7,36 +7,41 @@ namespace event { SignalEventImpl::SignalEventImpl(CommonLoop *wp_loop) : wp_loop_(wp_loop) -{ - LogUndo(); -} +{ } SignalEventImpl::~SignalEventImpl() { - LogUndo(); + disable(); } -bool SignalEventImpl::initialize(int signum, Mode mode) +bool SignalEventImpl::initialize(int signo, Mode mode) { - LogUndo(); - return false; -} + signo_ = signo; + mode_ = mode; -bool SignalEventImpl::isEnabled() const -{ - LogUndo(); - return false; + is_inited_ = true; + return true; } bool SignalEventImpl::enable() { - LogUndo(); + if (is_inited_) { + if (wp_loop_->subscribeSignal(signo_, this)) { + is_enabled_ = true; + return true; + } + } return false; } bool SignalEventImpl::disable() { - LogUndo(); + if (is_enabled_) { + if (wp_loop_->unsubscribeSignal(signo_, this)) { + is_enabled_ = false; + return true; + } + } return false; } @@ -45,5 +50,14 @@ Loop* SignalEventImpl::getLoop() const return wp_loop_; } +void SignalEventImpl::onSignal(int /*signo*/) +{ + if (mode_ == Mode::kOneshot) + disable(); + + if (cb_) + cb_(); +} + } } diff --git a/event/signal_event_impl.h b/event/signal_event_impl.h index a00017a..44cb208 100644 --- a/event/signal_event_impl.h +++ b/event/signal_event_impl.h @@ -8,7 +8,16 @@ namespace event { class CommonLoop; -class SignalEventImpl : SignalEvent { +class SignalSubscribuer { + public: + virtual void onSignal(int signo) = 0; + + protected: + virtual ~SignalSubscribuer() { } +}; + +class SignalEventImpl : public SignalEvent, + public SignalSubscribuer { public: explicit SignalEventImpl(CommonLoop *wp_loop); virtual ~SignalEventImpl(); @@ -17,15 +26,24 @@ class SignalEventImpl : SignalEvent { bool initialize(int signum, Mode mode) override; void setCallback(const CallbackFunc &cb) override { cb_ = cb; } - bool isEnabled() const override; + bool isEnabled() const override { return is_enabled_; } bool enable() override; bool disable() override; Loop* getLoop() const override; + public: + void onSignal(int signo) override; + private: CommonLoop *wp_loop_; CallbackFunc cb_; + + bool is_inited_ = false; + bool is_enabled_ = false; + + int signo_ = 0; + Mode mode_ = Mode::kPersist; }; } -- Gitee From 95ebb9db0b18d9b9492d8e4cc1bad0fed47f5685 Mon Sep 17 00:00:00 2001 From: Hevake Date: Wed, 2 Mar 2022 22:55:25 +0800 Subject: [PATCH 03/13] =?UTF-8?q?=E9=80=9A=E8=BF=87=E8=B0=83=E8=AF=95?= =?UTF-8?q?=EF=BC=8C=E9=80=9A=E8=BF=87=E4=BA=86event=E7=9A=84=E5=8D=95?= =?UTF-8?q?=E5=85=83=E6=B5=8B=E8=AF=95=E3=80=82=E4=BD=86=E6=B5=8B=E8=AF=95?= =?UTF-8?q?=E7=94=A8=E4=BE=8B=E8=BF=98=E4=B8=8D=E5=AE=8C=E5=96=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- event/Makefile | 4 - event/common_loop.cpp | 166 +++++++++++++++----------------- event/common_loop.h | 6 +- event/engins/epoll/fd_event.cpp | 1 + event/engins/epoll/loop.cpp | 2 + event/fd_event_test.cpp | 14 +++ event/loop.cpp | 16 +-- event/signal_event_test.cpp | 45 ++++++--- 8 files changed, 139 insertions(+), 115 deletions(-) diff --git a/event/Makefile b/event/Makefile index 0ec627c..9f3bfa4 100644 --- a/event/Makefile +++ b/event/Makefile @@ -1,7 +1,6 @@ include config.mk LIB_NAME = event -LIB_NAME_EXT = - LIB_VERSION_X = 1 LIB_VERSION_Y = 1 LIB_VERSION_Z = 0 @@ -24,7 +23,6 @@ CPP_SRC_FILES += \ engins/libevent/signal_event.cpp \ engins/libevent/common.cpp CXXFLAGS += -DENABLE_LIBEVENT=1 -LIB_NAME_EXT := $(LIB_NAME_EXT)n endif ifeq ($(ENABLE_LIBEV), yes) @@ -34,7 +32,6 @@ CPP_SRC_FILES += \ engins/libev/signal_event.cpp \ engins/libev/timer_event.cpp CXXFLAGS += -DENABLE_LIBEV=1 -LIB_NAME_EXT := $(LIB_NAME_EXT)e endif ifeq ($(ENABLE_EPOLL), yes) @@ -44,7 +41,6 @@ CPP_SRC_FILES += \ engins/epoll/signal_event.cpp \ engins/epoll/timer_event.cpp CXXFLAGS += -DENABLE_EPOLL=1 -LIB_NAME_EXT := endif TEST_CPP_SRC_FILES = \ diff --git a/event/common_loop.cpp b/event/common_loop.cpp index f7fd6ed..873b0d9 100644 --- a/event/common_loop.cpp +++ b/event/common_loop.cpp @@ -9,6 +9,7 @@ #include #include +#include #include "fd_event.h" #include "stat.h" @@ -18,8 +19,22 @@ namespace event { using namespace std::chrono; -std::map> CommonLoop::_signal_read_fds_; -std::mutex CommonLoop::_signal_lock_; +std::map> CommonLoop::_signal_write_fds_; +std::mutex CommonLoop::_signal_lock_; + +namespace { +bool CreateFdPair(int &read_fd, int &write_fd) +{ + int fds[2] = { 0 }; + if (pipe2(fds, O_CLOEXEC | O_NONBLOCK) != 0) { //!FIXME + LogErr("pip2() fail, ret:%d", errno); + return false; + } + read_fd = fds[0]; + write_fd = fds[1]; + return true; +} +} CommonLoop::CommonLoop() : has_unhandle_req_(false), @@ -32,8 +47,6 @@ CommonLoop::~CommonLoop() { assert(cb_level_ == 0); - std::lock_guard g(lock_); - cleanupDeferredTasks(); } bool CommonLoop::isInLoopThread() @@ -50,14 +63,9 @@ bool CommonLoop::isRunning() const void CommonLoop::runThisBeforeLoop() { - int fds[2] = { 0 }; - if (pipe2(fds, O_CLOEXEC | O_NONBLOCK) != 0) { //!FIXME - LogErr("pip2() fail, ret:%d", errno); + int read_fd = -1, write_fd = -1; + if (!CreateFdPair(read_fd, write_fd)) return; - } - - int read_fd(fds[0]); - int write_fd(fds[1]); FdEvent *sp_read_event = newFdEvent(); if (!sp_read_event->initialize(read_fd, FdEvent::kReadEvent, Event::Mode::kPersist)) { @@ -92,13 +100,9 @@ void CommonLoop::runThisAfterLoop() loop_thread_id_ = std::thread::id(); //! 清空 loop_thread_id_ if (sp_read_event_ != nullptr) { - delete sp_read_event_; - close(write_fd_); - close(read_fd_); - - sp_read_event_ = nullptr; - write_fd_ = -1; - read_fd_ = -1; + CHECK_DELETE_RESET_OBJ(sp_read_event_); + CHECK_CLOSE_RESET_FD(write_fd_); + CHECK_CLOSE_RESET_FD(read_fd_); } } @@ -157,105 +161,89 @@ void CommonLoop::endEventProcess() bool CommonLoop::subscribeSignal(int signo, SignalSubscribuer *who) { if (signal_read_fd_ == -1) { //! 如果还没有创建对应的信号 - int fds[2] = { 0 }; - if (pipe2(fds, O_CLOEXEC | O_NONBLOCK) != 0) { //!FIXME - LogErr("pip2() fail, ret:%d", errno); + if (!CreateFdPair(signal_read_fd_, signal_write_fd_)) return false; - } - int read_fd(fds[0]); - int write_fd(fds[1]); - - auto read_fd_event = newFdEvent(); - read_fd_event->initialize(read_fd, FdEvent::kReadEvent, Event::Mode::kPersist); - read_fd_event->setCallback(std::bind(&CommonLoop::onSignal, this)); - read_fd_event->enable(); - - { - std::unique_lock _g(_signal_lock_); - auto iter = _signal_read_fds_.find(signo); - if (iter != _signal_read_fds_.end()) { - iter->second.insert(read_fd); - } else { - std::set tmp = { read_fd }; - _signal_read_fds_[signo] = std::move(tmp); - signal(signo, CommonLoop::HandleSignal); - } - } - signal_read_fd_ = read_fd; - signal_write_fd_ = write_fd; - sp_signal_read_event_ = read_fd_event; + sp_signal_read_event_ = newFdEvent(); + sp_signal_read_event_->initialize(signal_read_fd_, FdEvent::kReadEvent, Event::Mode::kPersist); + sp_signal_read_event_->setCallback(std::bind(&CommonLoop::onSignal, this)); + sp_signal_read_event_->enable(); } - auto iter = signal_subscribers_.find(signo); - if (iter != signal_subscribers_.end()) { - iter->second.insert(who); - } else { - std::set tmp = { who }; - signal_subscribers_[signo] = std::move(tmp); + auto &this_signal_subscribers = all_signals_subscribers_[signo]; + if (this_signal_subscribers.empty()) { + //! 如果本Loop没有监听该信号，则要去 _signal_write_fds_ 中订阅 + std::unique_lock _g(_signal_lock_); + auto & signo_fds = _signal_write_fds_[signo]; + if (signo_fds.empty()) { + signal(signo, CommonLoop::HandleSignal); + LogTrace("register %d", signo); + } + signo_fds.insert(signal_write_fd_); } + this_signal_subscribers.insert(who); return true; } bool CommonLoop::unsubscribeSignal(int signo, SignalSubscribuer *who) { - auto iter = signal_subscribers_.find(signo); - if (iter == signal_subscribers_.end()) //! 如果本来就不存在，就直接返回了 - return true; - - auto &subscriber_set = iter->second; - subscriber_set.erase(who); //! 将订阅信息删除 - - if (!subscriber_set.empty()) //! 检查本Loop中是否已经没有SignalSubscribuer订阅该信号了 + auto &this_signal_subscribers = all_signals_subscribers_[signo]; + this_signal_subscribers.erase(who); //! 将订阅信息删除 + if (!this_signal_subscribers.empty()) //! 检查本Loop中是否已经没有SignalSubscribuer订阅该信号了 return true; //! 如果还有，就到此为止 //! 如果本Loop已经没有SignalSubscribuer订阅该信号了 - signal_subscribers_.erase(iter); //! 则将该信号的订阅记录表删除 - - std::unique_lock _g(_signal_lock_); - //! 并将 _signal_read_fds_ 中的记录删除 - auto fd_iter = _signal_read_fds_.find(signo); - assert(fd_iter != _signal_read_fds_.end()); - - auto &fd_set = fd_iter->second; - fd_set.erase(signal_read_fd_); + all_signals_subscribers_.erase(signo); //! 则将该信号的订阅记录表删除 + { + std::unique_lock _g(_signal_lock_); + //! 并将 _signal_write_fds_ 中的记录删除 + auto &this_signal_fds = _signal_write_fds_[signo]; + this_signal_fds.erase(signal_write_fd_); + if (this_signal_fds.empty()) { + //! 并还原信号处理函数 + signal(signo, SIG_DFL); + LogTrace("unregister %d", signo); + _signal_write_fds_.erase(signo); + } + } - //! 检查是否还有其它的Loop订阅该信号 - if (!fd_set.empty()) + if (!all_signals_subscribers_.empty()) return true; - //! 如果没有了，则删除 _signal_read_fds_ 中该信号的订阅记录表 - _signal_read_fds_.erase(fd_iter); + //! 已经没有任何SignalSubscribuer订阅任何信号了 + sp_signal_read_event_->disable(); + CHECK_CLOSE_RESET_FD(signal_write_fd_); + CHECK_CLOSE_RESET_FD(signal_read_fd_); + + FdEvent *tobe_delete = nullptr; + std::swap(tobe_delete, sp_signal_read_event_); + run([tobe_delete] { delete tobe_delete; }); - //! 并还原信号处理函数 - signal(signo, SIG_DFL); return true; } void CommonLoop::HandleSignal(int signo) { - //std::unique_lock _g(_signal_lock_); //!FIXME: 这是在信号中执行的，是否需要加锁？ - auto iter = _signal_read_fds_.find(signo); - if (iter != _signal_read_fds_.end()) { - const auto &fd_set = iter->second; - for (int fd : fd_set) - write(fd, &signo, sizeof(signo)); - } else { - LogWarn("uncatch signal: %d", signo); - } + auto &this_signal_fds = _signal_write_fds_[signo]; + for (int fd : this_signal_fds) + write(fd, &signo, sizeof(signo)); } void CommonLoop::onSignal() { - while (true) { - int signo = 0; - auto rsize = read(signal_read_fd_, &signo, sizeof(signo)); + while (signal_read_fd_ != -1) { + int signo_array[10]; //! 一次性读10个 + auto rsize = read(signal_read_fd_, &signo_array, sizeof(signo_array)); if (rsize > 0) { - auto iter = signal_subscribers_.find(signo); - if (iter != signal_subscribers_.end()) { - for (auto s : iter->second) { - s->onSignal(signo); + const auto num = rsize / sizeof(int); + for (size_t i = 0; i < num; ++i) { + int signo = signo_array[i]; + auto iter = all_signals_subscribers_.find(signo); + if (iter != all_signals_subscribers_.end()) { + for (auto s : iter->second) { + s->onSignal(signo); + } } } } else { diff --git a/event/common_loop.h b/event/common_loop.h index 8f41c2e..9fa43fa 100644 --- a/event/common_loop.h +++ b/event/common_loop.h @@ -80,13 +80,13 @@ class CommonLoop : public Loop { uint32_t max_cost_us_ = 0; #endif //ENABLE_STAT - static std::map> _signal_read_fds_; //! 通知 Loop 的 fd，每个 Loop 注册一个 - static std::mutex _signal_lock_; //! 保护 _signal_read_fds_ 用 + static std::map> _signal_write_fds_; //! 通知 Loop 的 fd，每个 Loop 注册一个 + static std::mutex _signal_lock_; //! 保护 _signal_write_fds_ 用 int signal_read_fd_ = -1; int signal_write_fd_ = -1; FdEvent *sp_signal_read_event_ = nullptr; - std::map> signal_subscribers_; //! signo -> SignalSubscribuer*，信号的订阅者 + std::map> all_signals_subscribers_; //! signo -> SignalSubscribuer*，信号的订阅者 int cb_level_ = 0; }; diff --git a/event/engins/epoll/fd_event.cpp b/event/engins/epoll/fd_event.cpp index b38e58a..f341b40 100644 --- a/event/engins/epoll/fd_event.cpp +++ b/event/engins/epoll/fd_event.cpp @@ -5,6 +5,7 @@ #include "fd_event.h" #include "loop.h" +#include namespace tbox { namespace event { diff --git a/event/engins/epoll/loop.cpp b/event/engins/epoll/loop.cpp index e021192..03fbf8d 100644 --- a/event/engins/epoll/loop.cpp +++ b/event/engins/epoll/loop.cpp @@ -34,6 +34,8 @@ EpollLoop::EpollLoop() : EpollLoop::~EpollLoop() { + cleanupDeferredTasks(); + CHECK_CLOSE_RESET_FD(epoll_fd_); CHECK_DELETE_RESET_OBJ(sp_exit_timer_); } diff --git a/event/fd_event_test.cpp b/event/fd_event_test.cpp index 27f0692..7438093 100644 --- a/event/fd_event_test.cpp +++ b/event/fd_event_test.cpp @@ -236,3 +236,17 @@ TEST(FdEvent, MultiWriteOneRead) close(read_fd); } } + + +TEST(FdEvent, DeleteLater) +{ + auto engins = Loop::Engines(); + for (auto e : engins) { + cout << "engin: " << e << endl; + auto sp_loop = Loop::New(e); + auto sp_fd_event = sp_loop->newFdEvent(); + sp_fd_event->initialize(1, FdEvent::kReadEvent, Event::Mode::kPersist); + sp_loop->run([=] { delete sp_fd_event; }); + delete sp_loop; + } +} diff --git a/event/loop.cpp b/event/loop.cpp index a65ab64..47ec08b 100644 --- a/event/loop.cpp +++ b/event/loop.cpp @@ -3,12 +3,14 @@ #if defined(ENABLE_LIBEVENT) #include "engins/libevent/loop.h" -#elif defined(ENABLE_LIBEV) +#endif + +#if defined(ENABLE_LIBEV) #include "engins/libev/loop.h" -#elif defined(ENABLE_EPOLL) +#endif + +#if defined(ENABLE_EPOLL) #include "engins/epoll/loop.h" -#else -#error("no engin specified!!!") #endif namespace tbox { @@ -48,14 +50,14 @@ Loop* Loop::New(const std::string &engine_type) std::vector Loop::Engines() { std::vector types; -#ifdef ENABLE_EPOLL - types.push_back("epoll"); -#endif #ifdef ENABLE_LIBEVENT types.push_back("libevent"); #endif #ifdef ENABLE_LIBEV types.push_back("libev"); +#endif +#ifdef ENABLE_EPOLL + types.push_back("epoll"); #endif return types; } diff --git a/event/signal_event_test.cpp b/event/signal_event_test.cpp index 287662c..94c52fb 100644 --- a/event/signal_event_test.cpp +++ b/event/signal_event_test.cpp @@ -3,6 +3,7 @@ #include #include #include +#include #include "loop.h" #include "signal_event.h" @@ -15,12 +16,14 @@ const int kAcceptableError = 10; TEST(SignalEvent, Oneshot) { + LogOutput_Initialize("test"); + auto engins = Loop::Engines(); for (auto e : engins) { cout << "engin: " << e << endl; auto sp_loop = Loop::New(e); auto signal_event = sp_loop->newSignalEvent(); - EXPECT_TRUE(signal_event->initialize(SIGINT, Event::Mode::kOneshot)); + EXPECT_TRUE(signal_event->initialize(SIGUSR1, Event::Mode::kOneshot)); EXPECT_TRUE(signal_event->enable()); int run_time = 0; @@ -28,7 +31,7 @@ TEST(SignalEvent, Oneshot) sp_loop->run([] { - raise(SIGINT); + raise(SIGUSR1); } ); sp_loop->exitLoop(std::chrono::milliseconds(100)); @@ -48,7 +51,7 @@ TEST(SignalEvent, PersistWithTimerEvent) cout << "engin: " << e << endl; auto sp_loop = Loop::New(e); auto signal_event = sp_loop->newSignalEvent(); - EXPECT_TRUE(signal_event->initialize(SIGINT, Event::Mode::kPersist)); + EXPECT_TRUE(signal_event->initialize(SIGUSR1, Event::Mode::kPersist)); EXPECT_TRUE(signal_event->enable()); auto timer_event = sp_loop->newTimerEvent(); @@ -59,7 +62,7 @@ TEST(SignalEvent, PersistWithTimerEvent) { ++count; if (count <= 5) { - raise(SIGINT); + raise(SIGUSR1); } } ); @@ -85,11 +88,11 @@ TEST(SignalEvent, IntAndTermSignal) cout << "engin: " << e << endl; auto sp_loop = Loop::New(e); auto int_signal_event = sp_loop->newSignalEvent(); - EXPECT_TRUE(int_signal_event->initialize(SIGINT, Event::Mode::kOneshot)); + EXPECT_TRUE(int_signal_event->initialize(SIGUSR1, Event::Mode::kOneshot)); EXPECT_TRUE(int_signal_event->enable()); auto term_signal_event = sp_loop->newSignalEvent(); - EXPECT_TRUE(term_signal_event->initialize(SIGTERM, Event::Mode::kOneshot)); + EXPECT_TRUE(term_signal_event->initialize(SIGUSR2, Event::Mode::kOneshot)); EXPECT_TRUE(term_signal_event->enable()); int int_run_time = 0; @@ -99,8 +102,8 @@ TEST(SignalEvent, IntAndTermSignal) sp_loop->run([] { - raise(SIGINT); - raise(SIGTERM); + raise(SIGUSR1); + raise(SIGUSR2); } ); sp_loop->exitLoop(std::chrono::milliseconds(100)); @@ -109,6 +112,7 @@ TEST(SignalEvent, IntAndTermSignal) EXPECT_EQ(int_run_time, 1); EXPECT_EQ(term_run_time, 1); + delete int_signal_event; delete term_signal_event; delete sp_loop; @@ -122,11 +126,11 @@ TEST(SignalEvent, MultiThread) cout << "engin: " << e << endl; auto sp_loop = Loop::New(e); auto int_signal_event = sp_loop->newSignalEvent(); - EXPECT_TRUE(int_signal_event->initialize(SIGINT, Event::Mode::kOneshot)); + EXPECT_TRUE(int_signal_event->initialize(SIGUSR1, Event::Mode::kOneshot)); EXPECT_TRUE(int_signal_event->enable()); auto term_signal_event = sp_loop->newSignalEvent(); - EXPECT_TRUE(term_signal_event->initialize(SIGTERM, Event::Mode::kOneshot)); + EXPECT_TRUE(term_signal_event->initialize(SIGUSR2, Event::Mode::kOneshot)); EXPECT_TRUE(term_signal_event->enable()); int int_run_time = 0; @@ -136,8 +140,8 @@ TEST(SignalEvent, MultiThread) sp_loop->run([] { - raise(SIGINT); - raise(SIGTERM); + raise(SIGUSR1); + raise(SIGUSR2); } ); @@ -186,3 +190,20 @@ TEST(SignalEvent, MultiThread) } } +//! 同一种事件被多个信号事件监听 +TEST(SignalEvent, OneSignalMultiEvents) +{ + //!TODO +} + +//! 同多种事件被多个信号事件监听 +TEST(SignalEvent, MultiSignalMultiEvents) +{ + //!TODO +} + +//! 多线程下多个Loop的事件监听同一个信号 +TEST(SignalEvent, OneSignalMultiLoop) +{ + //!TODO +} -- Gitee From b3da3e30727c3eccff3dfbf3885e38a6c16d653a Mon Sep 17 00:00:00 2001 From: Hevake Date: Wed, 2 Mar 2022 23:53:24 +0800 Subject: [PATCH 04/13] =?UTF-8?q?=E5=AE=8C=E5=96=84event=E9=92=88=E5=AF=B9?= =?UTF-8?q?=E4=BF=A1=E5=8F=B7=E7=9A=84=E5=8D=95=E5=85=83=E6=B5=8B=E8=AF=95?= =?UTF-8?q?=E7=94=A8=E4=BE=8B=EF=BC=8C=E5=B9=B6=E9=80=9A=E8=BF=87=E6=B5=8B?= =?UTF-8?q?=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- event/common_loop.cpp | 9 +- event/common_loop.h | 4 +- event/config.mk | 4 +- event/engins/libev/loop.cpp | 2 + event/engins/libevent/loop.cpp | 2 + event/signal_event_test.cpp | 197 +++++++++++++++++++++++++-------- 6 files changed, 165 insertions(+), 53 deletions(-) diff --git a/event/common_loop.cpp b/event/common_loop.cpp index 873b0d9..1f3bb30 100644 --- a/event/common_loop.cpp +++ b/event/common_loop.cpp @@ -30,6 +30,7 @@ bool CreateFdPair(int &read_fd, int &write_fd) LogErr("pip2() fail, ret:%d", errno); return false; } + read_fd = fds[0]; write_fd = fds[1]; return true; @@ -46,7 +47,6 @@ CommonLoop::CommonLoop() : CommonLoop::~CommonLoop() { assert(cb_level_ == 0); - } bool CommonLoop::isInLoopThread() @@ -174,11 +174,10 @@ bool CommonLoop::subscribeSignal(int signo, SignalSubscribuer *who) if (this_signal_subscribers.empty()) { //! 如果本Loop没有监听该信号，则要去 _signal_write_fds_ 中订阅 std::unique_lock _g(_signal_lock_); + //!FIXME: 要禁止信号触发 auto & signo_fds = _signal_write_fds_[signo]; - if (signo_fds.empty()) { + if (signo_fds.empty()) signal(signo, CommonLoop::HandleSignal); - LogTrace("register %d", signo); - } signo_fds.insert(signal_write_fd_); } this_signal_subscribers.insert(who); @@ -197,13 +196,13 @@ bool CommonLoop::unsubscribeSignal(int signo, SignalSubscribuer *who) all_signals_subscribers_.erase(signo); //! 则将该信号的订阅记录表删除 { std::unique_lock _g(_signal_lock_); + //!FIXME: 要禁止信号触发 //! 并将 _signal_write_fds_ 中的记录删除 auto &this_signal_fds = _signal_write_fds_[signo]; this_signal_fds.erase(signal_write_fd_); if (this_signal_fds.empty()) { //! 并还原信号处理函数 signal(signo, SIG_DFL); - LogTrace("unregister %d", signo); _signal_write_fds_.erase(signo); } } diff --git a/event/common_loop.h b/event/common_loop.h index 9fa43fa..8d9b2c4 100644 --- a/event/common_loop.h +++ b/event/common_loop.h @@ -39,12 +39,12 @@ class CommonLoop : public Loop { void beginEventProcess(); void endEventProcess(); - //! 信号 + //! 信号处理相关 + SignalEvent* newSignalEvent() override; bool subscribeSignal(int signal_num, SignalSubscribuer *who); bool unsubscribeSignal(int signal_num, SignalSubscribuer *who); static void HandleSignal(int signo); void onSignal(); - SignalEvent* newSignalEvent() override; protected: void runThisBeforeLoop(); diff --git a/event/config.mk b/event/config.mk index 48da8d5..d8045f6 100644 --- a/event/config.mk +++ b/event/config.mk @@ -1,5 +1,5 @@ -#ENABLE_LIBEVENT = yes -#ENABLE_LIBEV = yes +ENABLE_LIBEVENT = yes +ENABLE_LIBEV = yes ENABLE_EPOLL = yes ENABLE_STAT = yes diff --git a/event/engins/libev/loop.cpp b/event/engins/libev/loop.cpp index 6636c76..e0b0928 100644 --- a/event/engins/libev/loop.cpp +++ b/event/engins/libev/loop.cpp @@ -16,6 +16,8 @@ LibevLoop::LibevLoop() : LibevLoop::~LibevLoop() { + cleanupDeferredTasks(); + delete sp_exit_timer_; ev_loop_destroy(sp_ev_loop_); } diff --git a/event/engins/libevent/loop.cpp b/event/engins/libevent/loop.cpp index aa09ece..4768c8e 100644 --- a/event/engins/libevent/loop.cpp +++ b/event/engins/libevent/loop.cpp @@ -17,6 +17,8 @@ LibeventLoop::LibeventLoop() : LibeventLoop::~LibeventLoop() { + cleanupDeferredTasks(); + event_base_free(sp_event_base_); sp_event_base_ = NULL; } diff --git a/event/signal_event_test.cpp b/event/signal_event_test.cpp index 94c52fb..13f1b64 100644 --- a/event/signal_event_test.cpp +++ b/event/signal_event_test.cpp @@ -3,6 +3,7 @@ #include #include #include +#include #include #include "loop.h" @@ -16,8 +17,6 @@ const int kAcceptableError = 10; TEST(SignalEvent, Oneshot) { - LogOutput_Initialize("test"); - auto engins = Loop::Engines(); for (auto e : engins) { cout << "engin: " << e << endl; @@ -29,11 +28,8 @@ TEST(SignalEvent, Oneshot) int run_time = 0; signal_event->setCallback([&]() { ++run_time; }); - sp_loop->run([] - { - raise(SIGUSR1); - } - ); + sp_loop->run([] { raise(SIGUSR1); }); + sp_loop->exitLoop(std::chrono::milliseconds(100)); sp_loop->runLoop(); @@ -50,6 +46,7 @@ TEST(SignalEvent, PersistWithTimerEvent) for (auto e : engins) { cout << "engin: " << e << endl; auto sp_loop = Loop::New(e); + auto signal_event = sp_loop->newSignalEvent(); EXPECT_TRUE(signal_event->initialize(SIGUSR1, Event::Mode::kPersist)); EXPECT_TRUE(signal_event->enable()); @@ -81,24 +78,24 @@ TEST(SignalEvent, PersistWithTimerEvent) } } -TEST(SignalEvent, IntAndTermSignal) +TEST(SignalEvent, MultiSignalMultiEvents) { auto engins = Loop::Engines(); for (auto e : engins) { cout << "engin: " << e << endl; auto sp_loop = Loop::New(e); - auto int_signal_event = sp_loop->newSignalEvent(); - EXPECT_TRUE(int_signal_event->initialize(SIGUSR1, Event::Mode::kOneshot)); - EXPECT_TRUE(int_signal_event->enable()); + auto user1_signal_event = sp_loop->newSignalEvent(); + EXPECT_TRUE(user1_signal_event->initialize(SIGUSR1, Event::Mode::kOneshot)); + EXPECT_TRUE(user1_signal_event->enable()); - auto term_signal_event = sp_loop->newSignalEvent(); - EXPECT_TRUE(term_signal_event->initialize(SIGUSR2, Event::Mode::kOneshot)); - EXPECT_TRUE(term_signal_event->enable()); + auto user2_signal_event = sp_loop->newSignalEvent(); + EXPECT_TRUE(user2_signal_event->initialize(SIGUSR2, Event::Mode::kOneshot)); + EXPECT_TRUE(user2_signal_event->enable()); - int int_run_time = 0; - int_signal_event->setCallback([&]() { ++int_run_time; }); - int term_run_time = 0; - term_signal_event->setCallback([&]() { ++term_run_time; }); + int user1_run_time = 0; + user1_signal_event->setCallback([&]() { ++user1_run_time; }); + int user2_run_time = 0; + user2_signal_event->setCallback([&]() { ++user2_run_time; }); sp_loop->run([] { @@ -109,12 +106,12 @@ TEST(SignalEvent, IntAndTermSignal) sp_loop->exitLoop(std::chrono::milliseconds(100)); sp_loop->runLoop(); - EXPECT_EQ(int_run_time, 1); - EXPECT_EQ(term_run_time, 1); + EXPECT_EQ(user1_run_time, 1); + EXPECT_EQ(user2_run_time, 1); - delete int_signal_event; - delete term_signal_event; + delete user1_signal_event; + delete user2_signal_event; delete sp_loop; } } @@ -125,18 +122,18 @@ TEST(SignalEvent, MultiThread) for (auto e : engins) { cout << "engin: " << e << endl; auto sp_loop = Loop::New(e); - auto int_signal_event = sp_loop->newSignalEvent(); - EXPECT_TRUE(int_signal_event->initialize(SIGUSR1, Event::Mode::kOneshot)); - EXPECT_TRUE(int_signal_event->enable()); + auto user1_signal_event = sp_loop->newSignalEvent(); + EXPECT_TRUE(user1_signal_event->initialize(SIGUSR1, Event::Mode::kOneshot)); + EXPECT_TRUE(user1_signal_event->enable()); - auto term_signal_event = sp_loop->newSignalEvent(); - EXPECT_TRUE(term_signal_event->initialize(SIGUSR2, Event::Mode::kOneshot)); - EXPECT_TRUE(term_signal_event->enable()); + auto user2_signal_event = sp_loop->newSignalEvent(); + EXPECT_TRUE(user2_signal_event->initialize(SIGUSR2, Event::Mode::kOneshot)); + EXPECT_TRUE(user2_signal_event->enable()); - int int_run_time = 0; - int_signal_event->setCallback([&]() { ++int_run_time; }); - int term_run_time = 0; - term_signal_event->setCallback([&]() { ++term_run_time; }); + int user1_run_time = 0; + user1_signal_event->setCallback([&]() { ++user1_run_time; }); + int user2_run_time = 0; + user2_signal_event->setCallback([&]() { ++user2_run_time; }); sp_loop->run([] { @@ -181,29 +178,141 @@ TEST(SignalEvent, MultiThread) EXPECT_TRUE(t1_run); EXPECT_TRUE(t2_run); - EXPECT_EQ(int_run_time, 1); - EXPECT_EQ(term_run_time, 1); + EXPECT_EQ(user1_run_time, 1); + EXPECT_EQ(user2_run_time, 1); - delete int_signal_event; - delete term_signal_event; + delete user1_signal_event; + delete user2_signal_event; delete sp_loop; } } -//! 同一种事件被多个信号事件监听 +//! 同一种信号被多个事件监听 TEST(SignalEvent, OneSignalMultiEvents) { - //!TODO + auto engins = Loop::Engines(); + for (auto e : engins) { + cout << "engin: " << e << endl; + auto sp_loop = Loop::New(e); + + auto signal_event_1 = sp_loop->newSignalEvent(); + auto signal_event_2 = sp_loop->newSignalEvent(); + + EXPECT_TRUE(signal_event_1->initialize(SIGUSR1, Event::Mode::kOneshot)); + EXPECT_TRUE(signal_event_2->initialize(SIGUSR1, Event::Mode::kOneshot)); + + EXPECT_TRUE(signal_event_1->enable()); + EXPECT_TRUE(signal_event_2->enable()); + + int run_time_1 = 0; + int run_time_2 = 0; + + signal_event_1->setCallback([&]() { ++run_time_1; }); + signal_event_2->setCallback([&]() { ++run_time_2; }); + + sp_loop->run([] { raise(SIGUSR1); }); + sp_loop->exitLoop(std::chrono::milliseconds(100)); + sp_loop->runLoop(); + + EXPECT_EQ(run_time_1, 1); + EXPECT_EQ(run_time_2, 1); + + + delete signal_event_1; + delete signal_event_2; + delete sp_loop; + } } -//! 同多种事件被多个信号事件监听 -TEST(SignalEvent, MultiSignalMultiEvents) +//! 多线程下多个Loop的事件监听同一个信号 +TEST(SignalEvent, OneSignalMultiLoopInMultiThread) { - //!TODO + auto engins = Loop::Engines(); + for (auto e : engins) { + cout << "engin: " << e << endl; + + auto thread_func = [e] (int &run_time) { + auto sp_loop = Loop::New(e); + auto signal_event = sp_loop->newSignalEvent(); + signal_event->initialize(SIGUSR1, Event::Mode::kPersist); + signal_event->enable(); + signal_event->setCallback([&]() { ++run_time; }); + + sp_loop->exitLoop(std::chrono::milliseconds(200)); + sp_loop->runLoop(); + + delete signal_event; + delete sp_loop; + }; + + int run_time_1 = 0; + int run_time_2 = 0; + + auto t1 = std::thread(std::bind(thread_func, std::ref(run_time_1))); + auto t2 = std::thread(std::bind(thread_func, std::ref(run_time_2))); + + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + raise(SIGUSR1); + + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + raise(SIGUSR1); + + t1.join(); + t2.join(); + + EXPECT_EQ(run_time_1, 2); + EXPECT_EQ(run_time_2, 2); + } } -//! 多线程下多个Loop的事件监听同一个信号 -TEST(SignalEvent, OneSignalMultiLoop) +//! 多线程下多个Loop的事件监听同多个信号 +TEST(SignalEvent, MultiSignalMultiLoopInMultiThread) { - //!TODO + LogOutput_Initialize("test"); + + auto engins = Loop::Engines(); + for (auto e : engins) { + cout << "engin: " << e << endl; + + auto thread_func = [e] (int &user1_run_time, int &user2_run_time) { + auto sp_loop = Loop::New(e); + + auto user1_signal_event = sp_loop->newSignalEvent(); + user1_signal_event->initialize(SIGUSR1, Event::Mode::kPersist); + user1_signal_event->enable(); + user1_signal_event->setCallback([&]() { ++user1_run_time; }); + + auto user2_signal_event = sp_loop->newSignalEvent(); + user2_signal_event->initialize(SIGUSR2, Event::Mode::kPersist); + user2_signal_event->enable(); + user2_signal_event->setCallback([&]() { ++user2_run_time; }); + + sp_loop->exitLoop(std::chrono::milliseconds(200)); + sp_loop->runLoop(); + + delete user2_signal_event; + delete user1_signal_event; + delete sp_loop; + }; + + int user1_run_time_1 = 0, user2_run_time_1 = 0; + int user1_run_time_2 = 0, user2_run_time_2 = 0; + + auto t1 = std::thread(std::bind(thread_func, std::ref(user1_run_time_1), std::ref(user2_run_time_1))); + auto t2 = std::thread(std::bind(thread_func, std::ref(user1_run_time_2), std::ref(user2_run_time_2))); + + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + raise(SIGUSR1); + + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + raise(SIGUSR2); + + t1.join(); + t2.join(); + + EXPECT_EQ(user1_run_time_1, 1); + EXPECT_EQ(user2_run_time_1, 1); + EXPECT_EQ(user1_run_time_2, 1); + EXPECT_EQ(user2_run_time_2, 1); + } } -- Gitee From 1c9fee0101bbd5d78421955262b4103df53aad69 Mon Sep 17 00:00:00 2001 From: Hevake Date: Wed, 2 Mar 2022 23:57:54 +0800 Subject: [PATCH 05/13] =?UTF-8?q?=E5=88=A0=E9=99=A4=E4=B9=8B=E5=89=8D?= =?UTF-8?q?=E5=90=84engine=E4=B8=AD=E7=9A=84signal=5Fevent=E6=96=87?= =?UTF-8?q?=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- event/Makefile | 3 - event/engins/libev/loop.cpp | 1 - event/engins/libev/signal_event.cpp | 118 ----------------------- event/engins/libev/signal_event.h | 44 --------- event/engins/libevent/loop.cpp | 1 - event/engins/libevent/signal_event.cpp | 124 ------------------------- event/engins/libevent/signal_event.h | 43 --------- 7 files changed, 334 deletions(-) delete mode 100644 event/engins/libev/signal_event.cpp delete mode 100644 event/engins/libev/signal_event.h delete mode 100644 event/engins/libevent/signal_event.cpp delete mode 100644 event/engins/libevent/signal_event.h diff --git a/event/Makefile b/event/Makefile index 9f3bfa4..fc7414e 100644 --- a/event/Makefile +++ b/event/Makefile @@ -20,7 +20,6 @@ CPP_SRC_FILES += \ engins/libevent/loop.cpp \ engins/libevent/fd_event.cpp \ engins/libevent/timer_event.cpp \ - engins/libevent/signal_event.cpp \ engins/libevent/common.cpp CXXFLAGS += -DENABLE_LIBEVENT=1 endif @@ -29,7 +28,6 @@ ifeq ($(ENABLE_LIBEV), yes) CPP_SRC_FILES += \ engins/libev/loop.cpp \ engins/libev/fd_event.cpp \ - engins/libev/signal_event.cpp \ engins/libev/timer_event.cpp CXXFLAGS += -DENABLE_LIBEV=1 endif @@ -38,7 +36,6 @@ ifeq ($(ENABLE_EPOLL), yes) CPP_SRC_FILES += \ engins/epoll/loop.cpp \ engins/epoll/fd_event.cpp \ - engins/epoll/signal_event.cpp \ engins/epoll/timer_event.cpp CXXFLAGS += -DENABLE_EPOLL=1 endif diff --git a/event/engins/libev/loop.cpp b/event/engins/libev/loop.cpp index e0b0928..f7a6e63 100644 --- a/event/engins/libev/loop.cpp +++ b/event/engins/libev/loop.cpp @@ -4,7 +4,6 @@ #include "fd_event.h" #include "timer_event.h" -#include "signal_event.h" namespace tbox { namespace event { diff --git a/event/engins/libev/signal_event.cpp b/event/engins/libev/signal_event.cpp deleted file mode 100644 index 690e70b..0000000 --- a/event/engins/libev/signal_event.cpp +++ /dev/null @@ -1,118 +0,0 @@ -#include "signal_event.h" - -#include - -#include "loop.h" -#include - -namespace tbox { -namespace event { - -LibevSignalEvent::LibevSignalEvent(LibevLoop *wp_loop) : - wp_loop_(wp_loop), - is_inited_(false), - is_stop_after_trigger_(false), - cb_level_(0) -{ - memset(&signal_ev_, 0, sizeof(signal_ev_)); -} - -LibevSignalEvent::~LibevSignalEvent() -{ - assert(cb_level_ == 0); - disable(); -} - -bool LibevSignalEvent::initialize(int signum, Mode mode) -{ - disable(); - - signal_ev_.active = signal_ev_.pending = 0; - signal_ev_.priority = 0; - signal_ev_.cb = LibevSignalEvent::OnEventCallback; - signal_ev_.data = this; - - signal_ev_.signum = signum; - - if (mode == Mode::kOneshot) //! 如果是单次有效的，需要设置标记，使之在触发后停止事件 - is_stop_after_trigger_ = true; - - is_inited_ = true; - return true; -} - -void LibevSignalEvent::setCallback(const CallbackFunc &cb) -{ - cb_ = cb; -} - -bool LibevSignalEvent::isEnabled() const -{ - if (!is_inited_) - return false; - - return signal_ev_.active; -} - -bool LibevSignalEvent::enable() -{ - if (!is_inited_) { - LogErr("can't enable() before initialize()"); - return false; - } - - if (isEnabled()) - return true; - - ev_signal_start(wp_loop_->getEvLoopPtr(), &signal_ev_); - - return true; -} - -bool LibevSignalEvent::disable() -{ - if (!is_inited_) - return false; - - if (!isEnabled()) - return true; - - ev_signal_stop(wp_loop_->getEvLoopPtr(), &signal_ev_); - - return true; -} - -Loop* LibevSignalEvent::getLoop() const -{ - return wp_loop_; -} - -void LibevSignalEvent::OnEventCallback(struct ev_loop*, ev_signal *p_w, int events) -{ - assert(p_w != NULL); - - LibevSignalEvent *pthis = static_cast(p_w->data); - pthis->onEvent(); -} - -void LibevSignalEvent::onEvent() -{ - wp_loop_->beginEventProcess(); - - if (cb_) { - ++cb_level_; - cb_(); - --cb_level_; - - if (is_stop_after_trigger_) - disable(); - - } else { - LogErr("you should specify event callback by setCallback()"); - } - - wp_loop_->endEventProcess(); -} - -} -} diff --git a/event/engins/libev/signal_event.h b/event/engins/libev/signal_event.h deleted file mode 100644 index 4c19f06..0000000 --- a/event/engins/libev/signal_event.h +++ /dev/null @@ -1,44 +0,0 @@ -#ifndef TBOX_EVENT_LIBEV_SINGAL_EVENT_H_20170827 -#define TBOX_EVENT_LIBEV_SINGAL_EVENT_H_20170827 - -#include "../../signal_event.h" - -#include - -namespace tbox { -namespace event { - -class LibevLoop; - -class LibevSignalEvent : public SignalEvent { - public: - explicit LibevSignalEvent(LibevLoop *wp_loop); - virtual ~LibevSignalEvent(); - - public: - virtual bool initialize(int signum, Mode mode); - virtual void setCallback(const CallbackFunc &cb); - - virtual bool isEnabled() const; - virtual bool enable(); - virtual bool disable(); - - virtual Loop* getLoop() const; - - protected: - static void OnEventCallback(struct ev_loop *p_loop, ev_signal *p_w, int events); - void onEvent(); - - private: - LibevLoop *wp_loop_; - ev_signal signal_ev_; - bool is_inited_; - bool is_stop_after_trigger_; - CallbackFunc cb_; - int cb_level_; -}; - -} -} - -#endif //TBOX_EVENT_LIBEV_SINGAL_EVENT_H_20170827 diff --git a/event/engins/libevent/loop.cpp b/event/engins/libevent/loop.cpp index 4768c8e..d48c754 100644 --- a/event/engins/libevent/loop.cpp +++ b/event/engins/libevent/loop.cpp @@ -5,7 +5,6 @@ #include "fd_event.h" #include "timer_event.h" -#include "signal_event.h" #include "common.h" namespace tbox { diff --git a/event/engins/libevent/signal_event.cpp b/event/engins/libevent/signal_event.cpp deleted file mode 100644 index 2649590..0000000 --- a/event/engins/libevent/signal_event.cpp +++ /dev/null @@ -1,124 +0,0 @@ -#include "signal_event.h" - -#include -#include - -#include "loop.h" - -#include - -namespace tbox { -namespace event { - -LibeventSignalEvent::LibeventSignalEvent(LibeventLoop *wp_loop) : - wp_loop_(wp_loop), - is_inited_(false), - cb_level_(0) -{ - event_assign(&event_, NULL, -1, 0, NULL, NULL); -} - -LibeventSignalEvent::~LibeventSignalEvent() -{ - assert(cb_level_ == 0); - disable(); -} - -bool LibeventSignalEvent::initialize(int signum, Mode mode) -{ - disable(); - - short libevent_events = 0; - if (mode == Mode::kPersist) - libevent_events |= EV_PERSIST; - - int ret = event_assign(&event_, wp_loop_->getEventBasePtr(), signum, - libevent_events | EV_SIGNAL, - LibeventSignalEvent::OnEventCallback, - this); - if (ret == 0) { - is_inited_ = true; - return true; - } - - LogErr("event_assign() fail"); - return false; -} - -void LibeventSignalEvent::setCallback(const CallbackFunc &cb) -{ - cb_ = cb; -} - -bool LibeventSignalEvent::isEnabled() const -{ - if (!is_inited_) - return false; - - return event_pending(&event_, EV_SIGNAL, NULL) != 0; -} - -bool LibeventSignalEvent::enable() -{ - if (!is_inited_) { - LogErr("can't enable() before initialize()"); - return false; - } - - if (isEnabled()) - return true; - - int ret = event_add(&event_, NULL); - if (ret != 0) { - LogErr("event_add() fail"); - return false; - } - - return true; -} - -bool LibeventSignalEvent::disable() -{ - if (!is_inited_) - return false; - - if (!isEnabled()) - return true; - - int ret = event_del(&event_); - if (ret != 0) { - LogErr("event_del() fail"); - return false; - } - - return true; -} - -Loop* LibeventSignalEvent::getLoop() const -{ - return wp_loop_; -} - -void LibeventSignalEvent::OnEventCallback(int, short, void *args) -{ - LibeventSignalEvent *pthis = static_cast(args); - pthis->onEvent(); -} - -void LibeventSignalEvent::onEvent() -{ - wp_loop_->beginEventProcess(); - - if (cb_) { - ++cb_level_; - cb_(); - --cb_level_; - } else { - LogWarn("you should specify event callback by setCallback()"); - } - - wp_loop_->endEventProcess(); -} - -} -} diff --git a/event/engins/libevent/signal_event.h b/event/engins/libevent/signal_event.h deleted file mode 100644 index e9c0116..0000000 --- a/event/engins/libevent/signal_event.h +++ /dev/null @@ -1,43 +0,0 @@ -#ifndef TBOX_EVENT_LIBEVENT_SINGAL_EVENT_H_20170715 -#define TBOX_EVENT_LIBEVENT_SINGAL_EVENT_H_20170715 - -#include "../../signal_event.h" -#include - -namespace tbox { -namespace event { - -class LibeventLoop; - -class LibeventSignalEvent : public SignalEvent { - public: - explicit LibeventSignalEvent(LibeventLoop *wp_loop); - virtual ~LibeventSignalEvent(); - - public: - virtual bool initialize(int signum, Mode mode); - virtual void setCallback(const CallbackFunc &cb); - - virtual bool isEnabled() const; - virtual bool enable(); - virtual bool disable(); - - virtual Loop* getLoop() const; - - protected: - static void OnEventCallback(int, short, void *args); - void onEvent(); - - private: - LibeventLoop *wp_loop_; - - struct event event_; - bool is_inited_; - CallbackFunc cb_; - int cb_level_; -}; - -} -} - -#endif //TBOX_EVENT_LIBEVENT_SINGAL_EVENT_H_20170715 -- Gitee From 784415dda226eae689afe13476615b25b463d011 Mon Sep 17 00:00:00 2001 From: Hevake Date: Thu, 3 Mar 2022 08:14:03 +0800 Subject: [PATCH 06/13] =?UTF-8?q?=E4=BF=AE=E6=94=B9SignalEvent=EF=BC=8C?= =?UTF-8?q?=E4=BD=BF=E4=B9=8B=E6=94=AF=E6=8C=81=E5=8D=95=E4=BA=8B=E4=BB=B6?= =?UTF-8?q?=EF=BC=8C=E5=A4=9A=E4=BF=A1=E5=8F=B7=E7=9B=91=E5=90=AC=C2=80?= =?UTF-8?q?=C2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- event/common_loop.cpp | 8 ++- event/signal_event.h | 4 +- event/signal_event_impl.cpp | 35 ++++++++---- event/signal_event_impl.h | 4 +- event/signal_event_test.cpp | 107 ++++++++++++++++++++++++++++++++---- 5 files changed, 133 insertions(+), 25 deletions(-) diff --git a/event/common_loop.cpp b/event/common_loop.cpp index 1f3bb30..2b9f096 100644 --- a/event/common_loop.cpp +++ b/event/common_loop.cpp @@ -176,8 +176,10 @@ bool CommonLoop::subscribeSignal(int signo, SignalSubscribuer *who) std::unique_lock _g(_signal_lock_); //!FIXME: 要禁止信号触发 auto & signo_fds = _signal_write_fds_[signo]; - if (signo_fds.empty()) + if (signo_fds.empty()) { signal(signo, CommonLoop::HandleSignal); + //LogTrace("set signal:%d", signo); + } signo_fds.insert(signal_write_fd_); } this_signal_subscribers.insert(who); @@ -203,6 +205,7 @@ bool CommonLoop::unsubscribeSignal(int signo, SignalSubscribuer *who) if (this_signal_fds.empty()) { //! 并还原信号处理函数 signal(signo, SIG_DFL); + //LogTrace("unset signal:%d", signo); _signal_write_fds_.erase(signo); } } @@ -224,6 +227,7 @@ bool CommonLoop::unsubscribeSignal(int signo, SignalSubscribuer *who) void CommonLoop::HandleSignal(int signo) { + //LogTrace("got signal :%d", signo); auto &this_signal_fds = _signal_write_fds_[signo]; for (int fd : this_signal_fds) write(fd, &signo, sizeof(signo)); @@ -236,8 +240,10 @@ void CommonLoop::onSignal() auto rsize = read(signal_read_fd_, &signo_array, sizeof(signo_array)); if (rsize > 0) { const auto num = rsize / sizeof(int); + //LogTrace("rsize:%d, num:%u", rsize, num); for (size_t i = 0; i < num; ++i) { int signo = signo_array[i]; + //LogTrace("signo:%d", signo); auto iter = all_signals_subscribers_.find(signo); if (iter != all_signals_subscribers_.end()) { for (auto s : iter->second) { diff --git a/event/signal_event.h b/event/signal_event.h index 374866f..e8235cb 100644 --- a/event/signal_event.h +++ b/event/signal_event.h @@ -3,6 +3,7 @@ #include #include +#include #include "event.h" @@ -12,8 +13,9 @@ namespace event { class SignalEvent : public Event { public: virtual bool initialize(int signum, Mode mode) = 0; + virtual bool initialize(const std::set &sigset, Mode mode) = 0; - using CallbackFunc = std::function; + using CallbackFunc = std::function; virtual void setCallback(const CallbackFunc &cb) = 0; public: diff --git a/event/signal_event_impl.cpp b/event/signal_event_impl.cpp index 759fc9c..a5da9da 100644 --- a/event/signal_event_impl.cpp +++ b/event/signal_event_impl.cpp @@ -16,7 +16,16 @@ SignalEventImpl::~SignalEventImpl() bool SignalEventImpl::initialize(int signo, Mode mode) { - signo_ = signo; + sigset_.insert(signo); + mode_ = mode; + + is_inited_ = true; + return true; +} + +bool SignalEventImpl::initialize(const std::set &sigset, Mode mode) +{ + sigset_ = sigset; mode_ = mode; is_inited_ = true; @@ -26,23 +35,27 @@ bool SignalEventImpl::initialize(int signo, Mode mode) bool SignalEventImpl::enable() { if (is_inited_) { - if (wp_loop_->subscribeSignal(signo_, this)) { - is_enabled_ = true; - return true; + for (int signo : sigset_) { + if (!wp_loop_->subscribeSignal(signo, this)) { + return false; + } } } - return false; + is_enabled_ = true; + return true; } bool SignalEventImpl::disable() { if (is_enabled_) { - if (wp_loop_->unsubscribeSignal(signo_, this)) { - is_enabled_ = false; - return true; + for (int signo : sigset_) { + if (!wp_loop_->unsubscribeSignal(signo, this)) { + return false; + } } } - return false; + is_enabled_ = false; + return true; } Loop* SignalEventImpl::getLoop() const @@ -50,13 +63,13 @@ Loop* SignalEventImpl::getLoop() const return wp_loop_; } -void SignalEventImpl::onSignal(int /*signo*/) +void SignalEventImpl::onSignal(int signo) { if (mode_ == Mode::kOneshot) disable(); if (cb_) - cb_(); + cb_(signo); } } diff --git a/event/signal_event_impl.h b/event/signal_event_impl.h index 44cb208..fe6125a 100644 --- a/event/signal_event_impl.h +++ b/event/signal_event_impl.h @@ -24,6 +24,8 @@ class SignalEventImpl : public SignalEvent, public: bool initialize(int signum, Mode mode) override; + bool initialize(const std::set &sigset, Mode mode) override; + void setCallback(const CallbackFunc &cb) override { cb_ = cb; } bool isEnabled() const override { return is_enabled_; } @@ -42,7 +44,7 @@ class SignalEventImpl : public SignalEvent, bool is_inited_ = false; bool is_enabled_ = false; - int signo_ = 0; + std::set sigset_; Mode mode_ = Mode::kPersist; }; diff --git a/event/signal_event_test.cpp b/event/signal_event_test.cpp index 13f1b64..4d1fb10 100644 --- a/event/signal_event_test.cpp +++ b/event/signal_event_test.cpp @@ -26,7 +26,12 @@ TEST(SignalEvent, Oneshot) EXPECT_TRUE(signal_event->enable()); int run_time = 0; - signal_event->setCallback([&]() { ++run_time; }); + signal_event->setCallback( + [&](int signo) { + EXPECT_EQ(signo, SIGUSR1); + ++run_time; + } + ); sp_loop->run([] { raise(SIGUSR1); }); @@ -65,7 +70,12 @@ TEST(SignalEvent, PersistWithTimerEvent) ); int run_time = 0; - signal_event->setCallback([&]() { ++run_time; }); + signal_event->setCallback( + [&](int signo) { + EXPECT_EQ(signo, SIGUSR1); + ++run_time; + } + ); sp_loop->exitLoop(std::chrono::milliseconds(100)); sp_loop->runLoop(); @@ -93,9 +103,19 @@ TEST(SignalEvent, MultiSignalMultiEvents) EXPECT_TRUE(user2_signal_event->enable()); int user1_run_time = 0; - user1_signal_event->setCallback([&]() { ++user1_run_time; }); + user1_signal_event->setCallback( + [&](int signo) { + EXPECT_EQ(signo, SIGUSR1); + ++user1_run_time; + } + ); int user2_run_time = 0; - user2_signal_event->setCallback([&]() { ++user2_run_time; }); + user2_signal_event->setCallback( + [&](int signo) { + EXPECT_EQ(signo, SIGUSR2); + ++user2_run_time; + } + ); sp_loop->run([] { @@ -131,9 +151,19 @@ TEST(SignalEvent, MultiThread) EXPECT_TRUE(user2_signal_event->enable()); int user1_run_time = 0; - user1_signal_event->setCallback([&]() { ++user1_run_time; }); + user1_signal_event->setCallback( + [&](int signo) { + EXPECT_EQ(signo, SIGUSR1); + ++user1_run_time; + } + ); int user2_run_time = 0; - user2_signal_event->setCallback([&]() { ++user2_run_time; }); + user2_signal_event->setCallback( + [&](int signo) { + EXPECT_EQ(signo, SIGUSR2); + ++user2_run_time; + } + ); sp_loop->run([] { @@ -207,8 +237,18 @@ TEST(SignalEvent, OneSignalMultiEvents) int run_time_1 = 0; int run_time_2 = 0; - signal_event_1->setCallback([&]() { ++run_time_1; }); - signal_event_2->setCallback([&]() { ++run_time_2; }); + signal_event_1->setCallback( + [&](int signo) { + EXPECT_EQ(signo, SIGUSR1); + ++run_time_1; + } + ); + signal_event_2->setCallback( + [&](int signo) { + EXPECT_EQ(signo, SIGUSR1); + ++run_time_2; + } + ); sp_loop->run([] { raise(SIGUSR1); }); sp_loop->exitLoop(std::chrono::milliseconds(100)); @@ -236,7 +276,7 @@ TEST(SignalEvent, OneSignalMultiLoopInMultiThread) auto signal_event = sp_loop->newSignalEvent(); signal_event->initialize(SIGUSR1, Event::Mode::kPersist); signal_event->enable(); - signal_event->setCallback([&]() { ++run_time; }); + signal_event->setCallback([&](int) { ++run_time; }); sp_loop->exitLoop(std::chrono::milliseconds(200)); sp_loop->runLoop(); @@ -280,12 +320,12 @@ TEST(SignalEvent, MultiSignalMultiLoopInMultiThread) auto user1_signal_event = sp_loop->newSignalEvent(); user1_signal_event->initialize(SIGUSR1, Event::Mode::kPersist); user1_signal_event->enable(); - user1_signal_event->setCallback([&]() { ++user1_run_time; }); + user1_signal_event->setCallback([&](int) { ++user1_run_time; }); auto user2_signal_event = sp_loop->newSignalEvent(); user2_signal_event->initialize(SIGUSR2, Event::Mode::kPersist); user2_signal_event->enable(); - user2_signal_event->setCallback([&]() { ++user2_run_time; }); + user2_signal_event->setCallback([&](int) { ++user2_run_time; }); sp_loop->exitLoop(std::chrono::milliseconds(200)); sp_loop->runLoop(); @@ -316,3 +356,48 @@ TEST(SignalEvent, MultiSignalMultiLoopInMultiThread) EXPECT_EQ(user2_run_time_2, 1); } } + +//! 同一个事件，监听多个事件 +TEST(SignalEvent, OneEventMultiSignal) +{ + LogOutput_Initialize("test"); + auto engins = Loop::Engines(); + for (auto e : engins) { + cout << "engin: " << e << endl; + auto sp_loop = Loop::New(e); + auto signal_event = sp_loop->newSignalEvent(); + std::set sigset = { SIGUSR1, SIGUSR2 }; + EXPECT_TRUE(signal_event->initialize(sigset, Event::Mode::kPersist)); + EXPECT_TRUE(signal_event->enable()); + + int user1_run_time = 0; + int user2_run_time = 0; + signal_event->setCallback( + [&](int signo) { + LogTrace("signo:%d", signo); + if (signo == SIGUSR1) + ++user1_run_time; + else if (signo == SIGUSR2) + ++user2_run_time; + else {} + } + ); + + sp_loop->run( + [] { + raise(SIGUSR1); + raise(SIGUSR2); + } + ); + + sp_loop->exitLoop(std::chrono::milliseconds(100)); + sp_loop->runLoop(); + + EXPECT_EQ(user1_run_time, 1); + EXPECT_EQ(user2_run_time, 1); + + delete signal_event; + delete sp_loop; + } +} + -- Gitee From fb27817ba8814a88031d9319b6b39c5593f6e317 Mon Sep 17 00:00:00 2001 From: Hevake Date: Thu, 3 Mar 2022 08:24:21 +0800 Subject: [PATCH 07/13] =?UTF-8?q?1.=E5=85=B3=E9=97=ADevent=E4=B8=AD?= =?UTF-8?q?=E7=9A=84libevent=E4=B8=8Elibev;=202.=E5=9C=A8main=E4=B8=AD?= =?UTF-8?q?=E9=87=8D=E6=96=B0=E4=BD=BF=E7=94=A8SignalEvent=E9=80=80?= =?UTF-8?q?=E5=87=BA=E8=BF=9B=E7=A8=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- event/config.mk | 4 ++-- main/main.cpp | 36 ++++++++++++++++++++++-------------- main/signal.cpp | 11 ----------- 3 files changed, 24 insertions(+), 27 deletions(-) diff --git a/event/config.mk b/event/config.mk index d8045f6..48da8d5 100644 --- a/event/config.mk +++ b/event/config.mk @@ -1,5 +1,5 @@ -ENABLE_LIBEVENT = yes -ENABLE_LIBEV = yes +#ENABLE_LIBEVENT = yes +#ENABLE_LIBEV = yes ENABLE_EPOLL = yes ENABLE_STAT = yes diff --git a/main/main.cpp b/main/main.cpp index f1bbde9..8700da1 100644 --- a/main/main.cpp +++ b/main/main.cpp @@ -8,6 +8,7 @@ #include #include +#include #include #include @@ -25,7 +26,6 @@ extern void RegisterApps(Apps &apps); //! 由用户去实现 extern void Run(ContextImp &ctx, AppsImp &apps, int loop_exit_wait); std::function error_exit_func; //!< 出错异常退出前要做的事件 -std::function normal_stop_func; //!< 正常退出前要做的事件 int Main(int argc, char **argv) { @@ -109,20 +109,27 @@ int Main(int argc, char **argv) void Run(ContextImp &ctx, AppsImp &apps, int loop_exit_wait) { - auto feeddog_timer = ctx.loop()->newTimerEvent(); + auto feeddog_timer = ctx.loop()->newTimerEvent(); + auto stop_signal = ctx.loop()->newSignalEvent(); + //! 预定在离开时自动释放对象，确保无内存泄漏 - SetScopeExitAction([feeddog_timer] { delete feeddog_timer; }); - - normal_stop_func = [&] { - ctx.loop()->runInLoop([&] { - LogInfo("Got stop signal"); - apps.stop(); - ctx.stop(); - ctx.loop()->exitLoop(std::chrono::seconds(loop_exit_wait)); - LogInfo("Loop will exit after %d sec", loop_exit_wait); - } - ); - }; + SetScopeExitAction( + [=] { + delete stop_signal; + delete feeddog_timer; + } + ); + + stop_signal->initialize(std::set{SIGINT, SIGTERM}, event::Event::Mode::kOneshot); + stop_signal->setCallback( + [&] (int signo) { + LogInfo("Got signal %d", signo); + apps.stop(); + ctx.stop(); + ctx.loop()->exitLoop(std::chrono::seconds(loop_exit_wait)); + LogInfo("Loop will exit after %d sec", loop_exit_wait); + } + ); //! 创建喂狗定时器 feeddog_timer->initialize(std::chrono::seconds(2), event::Event::Mode::kPersist); @@ -133,6 +140,7 @@ void Run(ContextImp &ctx, AppsImp &apps, int loop_exit_wait) util::ThreadWDog::Register("main", 3); feeddog_timer->enable(); + stop_signal->enable(); LogInfo("Start!"); diff --git a/main/signal.cpp b/main/signal.cpp index 5de2369..ba83d31 100644 --- a/main/signal.cpp +++ b/main/signal.cpp @@ -10,18 +10,9 @@ namespace tbox::main { extern std::function error_exit_func; //!< 出错异常退出前要做的事件 -extern std::function normal_stop_func; //!< 正常退出前要做的事件 namespace { -void OnStopSignal(int) -{ - if (normal_stop_func) - normal_stop_func(); - else - exit(0); -} - //! 打印调用栈 void PrintCallStack() { @@ -68,8 +59,6 @@ void RegisterSignals() signal(SIGABRT, OnErrorSignal); signal(SIGBUS, OnErrorSignal); signal(SIGPIPE, OnWarnSignal); - signal(SIGINT, OnStopSignal); - signal(SIGTERM, OnStopSignal); } } -- Gitee From fd14269e9cedf1c8cc9906f8f35f1abfd4cde1d5 Mon Sep 17 00:00:00 2001 From: Hevake Date: Thu, 3 Mar 2022 20:39:34 +0800 Subject: [PATCH 08/13] =?UTF-8?q?=E5=B0=86common=5Floop.cpp=E6=A0=B9?= =?UTF-8?q?=E6=8D=AE=E5=8A=9F=E8=83=BD=EF=BC=8C=E4=BB=8E=E4=B8=AD=E7=A7=BB?= =?UTF-8?q?=E5=87=BA=E9=83=A8=E5=88=86=E4=BB=A3=E7=A0=81=E5=88=B0common=5F?= =?UTF-8?q?loop=5Fsignal.cpp=E4=B8=8Ecommon=5Floop=5Frun.cpp=E4=B8=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- event/Makefile | 6 +- event/common_loop.cpp | 245 +---------------------------------- event/common_loop_run.cpp | 127 ++++++++++++++++++ event/common_loop_signal.cpp | 124 ++++++++++++++++++ event/misc.cpp | 26 ++++ event/misc.h | 12 ++ 6 files changed, 295 insertions(+), 245 deletions(-) create mode 100644 event/common_loop_run.cpp create mode 100644 event/common_loop_signal.cpp create mode 100644 event/misc.cpp create mode 100644 event/misc.h diff --git a/event/Makefile b/event/Makefile index fc7414e..9bc2265 100644 --- a/event/Makefile +++ b/event/Makefile @@ -11,8 +11,12 @@ HEAD_FILES = \ stat.h version.h CPP_SRC_FILES = \ - loop.cpp common_loop.cpp \ + loop.cpp \ + common_loop.cpp \ + common_loop_signal.cpp \ + common_loop_run.cpp \ signal_event_impl.cpp \ + misc.cpp \ version.cpp ifeq ($(ENABLE_LIBEVENT), yes) diff --git a/event/common_loop.cpp b/event/common_loop.cpp index 2b9f096..6737c52 100644 --- a/event/common_loop.cpp +++ b/event/common_loop.cpp @@ -2,9 +2,6 @@ #include #include -#include -#include -#include #include #include @@ -13,30 +10,13 @@ #include "fd_event.h" #include "stat.h" +#include "misc.h" namespace tbox { namespace event { using namespace std::chrono; -std::map> CommonLoop::_signal_write_fds_; -std::mutex CommonLoop::_signal_lock_; - -namespace { -bool CreateFdPair(int &read_fd, int &write_fd) -{ - int fds[2] = { 0 }; - if (pipe2(fds, O_CLOEXEC | O_NONBLOCK) != 0) { //!FIXME - LogErr("pip2() fail, ret:%d", errno); - return false; - } - - read_fd = fds[0]; - write_fd = fds[1]; - return true; -} -} - CommonLoop::CommonLoop() : has_unhandle_req_(false), read_fd_(-1), write_fd_(-1), @@ -106,35 +86,6 @@ void CommonLoop::runThisAfterLoop() } } -void CommonLoop::runInLoop(const Func &func) -{ - std::lock_guard g(lock_); - run_in_loop_func_queue_.push_back(func); - - if (sp_read_event_ == nullptr) - return; - - commitRequest(); -} - -void CommonLoop::runNext(const Func &func) -{ - if (!isInLoopThread()) { - LogWarn("Fail, use runInLoop() instead."); - return; - } - - run_next_func_queue_.push_back(func); -} - -void CommonLoop::run(const Func &func) -{ - if (isInLoopThread()) - run_next_func_queue_.push_back(func); - else - runInLoop(func); -} - void CommonLoop::beginEventProcess() { #ifdef ENABLE_STAT @@ -158,200 +109,6 @@ void CommonLoop::endEventProcess() #endif } -bool CommonLoop::subscribeSignal(int signo, SignalSubscribuer *who) -{ - if (signal_read_fd_ == -1) { //! 如果还没有创建对应的信号 - if (!CreateFdPair(signal_read_fd_, signal_write_fd_)) - return false; - - sp_signal_read_event_ = newFdEvent(); - sp_signal_read_event_->initialize(signal_read_fd_, FdEvent::kReadEvent, Event::Mode::kPersist); - sp_signal_read_event_->setCallback(std::bind(&CommonLoop::onSignal, this)); - sp_signal_read_event_->enable(); - } - - auto &this_signal_subscribers = all_signals_subscribers_[signo]; - if (this_signal_subscribers.empty()) { - //! 如果本Loop没有监听该信号，则要去 _signal_write_fds_ 中订阅 - std::unique_lock _g(_signal_lock_); - //!FIXME: 要禁止信号触发 - auto & signo_fds = _signal_write_fds_[signo]; - if (signo_fds.empty()) { - signal(signo, CommonLoop::HandleSignal); - //LogTrace("set signal:%d", signo); - } - signo_fds.insert(signal_write_fd_); - } - this_signal_subscribers.insert(who); - - return true; -} - -bool CommonLoop::unsubscribeSignal(int signo, SignalSubscribuer *who) -{ - auto &this_signal_subscribers = all_signals_subscribers_[signo]; - this_signal_subscribers.erase(who); //! 将订阅信息删除 - if (!this_signal_subscribers.empty()) //! 检查本Loop中是否已经没有SignalSubscribuer订阅该信号了 - return true; //! 如果还有，就到此为止 - - //! 如果本Loop已经没有SignalSubscribuer订阅该信号了 - all_signals_subscribers_.erase(signo); //! 则将该信号的订阅记录表删除 - { - std::unique_lock _g(_signal_lock_); - //!FIXME: 要禁止信号触发 - //! 并将 _signal_write_fds_ 中的记录删除 - auto &this_signal_fds = _signal_write_fds_[signo]; - this_signal_fds.erase(signal_write_fd_); - if (this_signal_fds.empty()) { - //! 并还原信号处理函数 - signal(signo, SIG_DFL); - //LogTrace("unset signal:%d", signo); - _signal_write_fds_.erase(signo); - } - } - - if (!all_signals_subscribers_.empty()) - return true; - - //! 已经没有任何SignalSubscribuer订阅任何信号了 - sp_signal_read_event_->disable(); - CHECK_CLOSE_RESET_FD(signal_write_fd_); - CHECK_CLOSE_RESET_FD(signal_read_fd_); - - FdEvent *tobe_delete = nullptr; - std::swap(tobe_delete, sp_signal_read_event_); - run([tobe_delete] { delete tobe_delete; }); - - return true; -} - -void CommonLoop::HandleSignal(int signo) -{ - //LogTrace("got signal :%d", signo); - auto &this_signal_fds = _signal_write_fds_[signo]; - for (int fd : this_signal_fds) - write(fd, &signo, sizeof(signo)); -} - -void CommonLoop::onSignal() -{ - while (signal_read_fd_ != -1) { - int signo_array[10]; //! 一次性读10个 - auto rsize = read(signal_read_fd_, &signo_array, sizeof(signo_array)); - if (rsize > 0) { - const auto num = rsize / sizeof(int); - //LogTrace("rsize:%d, num:%u", rsize, num); - for (size_t i = 0; i < num; ++i) { - int signo = signo_array[i]; - //LogTrace("signo:%d", signo); - auto iter = all_signals_subscribers_.find(signo); - if (iter != all_signals_subscribers_.end()) { - for (auto s : iter->second) { - s->onSignal(signo); - } - } - } - } else { - if (errno != EAGAIN) - LogWarn("read error, rsize:%d, errno:%d, %s", rsize, errno, strerror(errno)); - break; - } - } -} - -SignalEvent* CommonLoop::newSignalEvent() -{ - return new SignalEventImpl(this); -} - -void CommonLoop::handleNextFunc() -{ - while (!run_next_func_queue_.empty()) { - Func &func = run_next_func_queue_.front(); - if (func) { - ++cb_level_; - func(); - --cb_level_; - } - run_next_func_queue_.pop_front(); - } -} - -void CommonLoop::onGotRunInLoopFunc(short) -{ - /** - * NOTICE: - * 这里使用 tmp 将 run_in_loop_func_queue_ 中的内容交换出去。然后再从 tmp 逐一取任务出来执行。 - * 其目的在于腾空 run_in_loop_func_queue_，让新 runInLoop() 的任务则会在下一轮循环中执行。 - * 从而防止无限 runInLoop() 引起的死循环，导致其它事件得不到处理。 - * - * 这点与 runNext() 不同 - */ - std::deque tmp; - { - std::lock_guard g(lock_); - run_in_loop_func_queue_.swap(tmp); - finishRequest(); - } - - while (!tmp.empty()) { - Func &func = tmp.front(); - if (func) { - ++cb_level_; - func(); - --cb_level_; - - handleNextFunc(); - } - tmp.pop_front(); - } -} - -//! 清理 run_in_loop_func_queue_ 与 run_next_func_queue_ 中的任务 -void CommonLoop::cleanupDeferredTasks() -{ - int remain_loop_count = 10; //! 限定次数，防止出现 runNext() 递归导致无法退出循环的问题 - while ((!run_in_loop_func_queue_.empty() || !run_next_func_queue_.empty()) - && remain_loop_count-- > 0) { - std::deque tasks = std::move(run_next_func_queue_); - tasks.insert(tasks.end(), run_in_loop_func_queue_.begin(), run_in_loop_func_queue_.end()); - run_in_loop_func_queue_.clear(); - - while (!tasks.empty()) { - Func &func = tasks.front(); - if (func) { - ++cb_level_; - func(); - --cb_level_; - } - tasks.pop_front(); - } - } - - if (remain_loop_count == 0) - LogWarn("found recursive actions, force quit"); -} - -void CommonLoop::commitRequest() -{ - if (!has_unhandle_req_) { - char ch = 0; - ssize_t wsize = write(write_fd_, &ch, 1); - (void)wsize; - - has_unhandle_req_ = true; - } -} - -void CommonLoop::finishRequest() -{ - char ch = 0; - ssize_t rsize = read(read_fd_, &ch, 1); - (void)rsize; - - has_unhandle_req_ = false; -} - void CommonLoop::setStatEnable(bool enable) { #ifdef ENABLE_STAT diff --git a/event/common_loop_run.cpp b/event/common_loop_run.cpp new file mode 100644 index 0000000..08f9df6 --- /dev/null +++ b/event/common_loop_run.cpp @@ -0,0 +1,127 @@ +#include "common_loop.h" + +#include +#include + +namespace tbox { +namespace event { + +void CommonLoop::runInLoop(const Func &func) +{ + std::lock_guard g(lock_); + run_in_loop_func_queue_.push_back(func); + + if (sp_read_event_ == nullptr) + return; + + commitRequest(); +} + +void CommonLoop::runNext(const Func &func) +{ + if (!isInLoopThread()) { + LogWarn("Fail, use runInLoop() instead."); + return; + } + + run_next_func_queue_.push_back(func); +} + +void CommonLoop::run(const Func &func) +{ + if (isInLoopThread()) + run_next_func_queue_.push_back(func); + else + runInLoop(func); +} + +void CommonLoop::handleNextFunc() +{ + while (!run_next_func_queue_.empty()) { + Func &func = run_next_func_queue_.front(); + if (func) { + ++cb_level_; + func(); + --cb_level_; + } + run_next_func_queue_.pop_front(); + } +} + +void CommonLoop::onGotRunInLoopFunc(short) +{ + /** + * NOTICE: + * 这里使用 tmp 将 run_in_loop_func_queue_ 中的内容交换出去。然后再从 tmp 逐一取任务出来执行。 + * 其目的在于腾空 run_in_loop_func_queue_，让新 runInLoop() 的任务则会在下一轮循环中执行。 + * 从而防止无限 runInLoop() 引起的死循环，导致其它事件得不到处理。 + * + * 这点与 runNext() 不同 + */ + std::deque tmp; + { + std::lock_guard g(lock_); + run_in_loop_func_queue_.swap(tmp); + finishRequest(); + } + + while (!tmp.empty()) { + Func &func = tmp.front(); + if (func) { + ++cb_level_; + func(); + --cb_level_; + + handleNextFunc(); + } + tmp.pop_front(); + } +} + +//! 清理 run_in_loop_func_queue_ 与 run_next_func_queue_ 中的任务 +void CommonLoop::cleanupDeferredTasks() +{ + int remain_loop_count = 10; //! 限定次数，防止出现 runNext() 递归导致无法退出循环的问题 + while ((!run_in_loop_func_queue_.empty() || !run_next_func_queue_.empty()) + && remain_loop_count-- > 0) { + std::deque tasks = std::move(run_next_func_queue_); + tasks.insert(tasks.end(), run_in_loop_func_queue_.begin(), run_in_loop_func_queue_.end()); + run_in_loop_func_queue_.clear(); + + while (!tasks.empty()) { + Func &func = tasks.front(); + if (func) { + ++cb_level_; + func(); + --cb_level_; + } + tasks.pop_front(); + } + } + + if (remain_loop_count == 0) + LogWarn("found recursive actions, force quit"); +} + +void CommonLoop::commitRequest() +{ + if (!has_unhandle_req_) { + char ch = 0; + ssize_t wsize = write(write_fd_, &ch, 1); + (void)wsize; + + has_unhandle_req_ = true; + } +} + +void CommonLoop::finishRequest() +{ + char ch = 0; + ssize_t rsize = read(read_fd_, &ch, 1); + (void)rsize; + + has_unhandle_req_ = false; +} + +} +} diff --git a/event/common_loop_signal.cpp b/event/common_loop_signal.cpp new file mode 100644 index 0000000..e086268 --- /dev/null +++ b/event/common_loop_signal.cpp @@ -0,0 +1,124 @@ +#include "common_loop.h" + +#include +#include +#include +#include + +#include "misc.h" +#include "fd_event.h" + +namespace tbox { +namespace event { + +std::map> CommonLoop::_signal_write_fds_; +std::mutex CommonLoop::_signal_lock_; + +bool CommonLoop::subscribeSignal(int signo, SignalSubscribuer *who) +{ + if (signal_read_fd_ == -1) { //! 如果还没有创建对应的信号 + if (!CreateFdPair(signal_read_fd_, signal_write_fd_)) + return false; + + sp_signal_read_event_ = newFdEvent(); + sp_signal_read_event_->initialize(signal_read_fd_, FdEvent::kReadEvent, Event::Mode::kPersist); + sp_signal_read_event_->setCallback(std::bind(&CommonLoop::onSignal, this)); + sp_signal_read_event_->enable(); + } + + auto &this_signal_subscribers = all_signals_subscribers_[signo]; + if (this_signal_subscribers.empty()) { + //! 如果本Loop没有监听该信号，则要去 _signal_write_fds_ 中订阅 + std::unique_lock _g(_signal_lock_); + //!FIXME: 要禁止信号触发 + auto & signo_fds = _signal_write_fds_[signo]; + if (signo_fds.empty()) { + signal(signo, CommonLoop::HandleSignal); + //LogTrace("set signal:%d", signo); + } + signo_fds.insert(signal_write_fd_); + } + this_signal_subscribers.insert(who); + + return true; +} + +bool CommonLoop::unsubscribeSignal(int signo, SignalSubscribuer *who) +{ + auto &this_signal_subscribers = all_signals_subscribers_[signo]; + this_signal_subscribers.erase(who); //! 将订阅信息删除 + if (!this_signal_subscribers.empty()) //! 检查本Loop中是否已经没有SignalSubscribuer订阅该信号了 + return true; //! 如果还有，就到此为止 + + //! 如果本Loop已经没有SignalSubscribuer订阅该信号了 + all_signals_subscribers_.erase(signo); //! 则将该信号的订阅记录表删除 + { + std::unique_lock _g(_signal_lock_); + //!FIXME: 要禁止信号触发 + //! 并将 _signal_write_fds_ 中的记录删除 + auto &this_signal_fds = _signal_write_fds_[signo]; + this_signal_fds.erase(signal_write_fd_); + if (this_signal_fds.empty()) { + //! 并还原信号处理函数 + signal(signo, SIG_DFL); + //LogTrace("unset signal:%d", signo); + _signal_write_fds_.erase(signo); + } + } + + if (!all_signals_subscribers_.empty()) + return true; + + //! 已经没有任何SignalSubscribuer订阅任何信号了 + sp_signal_read_event_->disable(); + CHECK_CLOSE_RESET_FD(signal_write_fd_); + CHECK_CLOSE_RESET_FD(signal_read_fd_); + + FdEvent *tobe_delete = nullptr; + std::swap(tobe_delete, sp_signal_read_event_); + run([tobe_delete] { delete tobe_delete; }); + + return true; +} + +void CommonLoop::HandleSignal(int signo) +{ + //LogTrace("got signal :%d", signo); + auto &this_signal_fds = _signal_write_fds_[signo]; + for (int fd : this_signal_fds) + write(fd, &signo, sizeof(signo)); +} + +void CommonLoop::onSignal() +{ + while (signal_read_fd_ != -1) { + int signo_array[10]; //! 一次性读10个 + auto rsize = read(signal_read_fd_, &signo_array, sizeof(signo_array)); + if (rsize > 0) { + const auto num = rsize / sizeof(int); + //LogTrace("rsize:%d, num:%u", rsize, num); + for (size_t i = 0; i < num; ++i) { + int signo = signo_array[i]; + //LogTrace("signo:%d", signo); + auto iter = all_signals_subscribers_.find(signo); + if (iter != all_signals_subscribers_.end()) { + for (auto s : iter->second) { + s->onSignal(signo); + } + } + } + } else { + if (errno != EAGAIN) + LogWarn("read error, rsize:%d, errno:%d, %s", rsize, errno, strerror(errno)); + break; + } + } +} + +SignalEvent* CommonLoop::newSignalEvent() +{ + return new SignalEventImpl(this); +} + +} +} diff --git a/event/misc.cpp b/event/misc.cpp new file mode 100644 index 0000000..6ef301e --- /dev/null +++ b/event/misc.cpp @@ -0,0 +1,26 @@ +#include "misc.h" + +#include +#include +#include + +#include + +namespace tbox { +namespace event { + +bool CreateFdPair(int &read_fd, int &write_fd) +{ + int fds[2] = { 0 }; + if (pipe2(fds, O_CLOEXEC | O_NONBLOCK) != 0) { //!FIXME + LogErr("pip2() fail, ret:%d", errno); + return false; + } + + read_fd = fds[0]; + write_fd = fds[1]; + return true; +} + +} +} diff --git a/event/misc.h b/event/misc.h new file mode 100644 index 0000000..8875ce7 --- /dev/null +++ b/event/misc.h @@ -0,0 +1,12 @@ +#ifndef TBOX_EVENT_MISC_H_20220303 +#define TBOX_EVENT_MISC_H_20220303 + +namespace tbox { +namespace event { + +bool CreateFdPair(int &read_fd, int &write_fd); + +} +} + +#endif //TBOX_EVENT_MISC_H_20220303 -- Gitee From d048cec0661cd2b433cb59eade669333fd013dc0 Mon Sep 17 00:00:00 2001 From: Hevake Date: Thu, 3 Mar 2022 20:54:13 +0800 Subject: [PATCH 09/13] =?UTF-8?q?=E4=BF=AE=E6=94=B9event=EF=BC=8C=E4=BC=98?= =?UTF-8?q?=E5=8C=96CommonLoop=E4=B8=AD=E7=9A=84=E5=91=BD=E5=90=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- event/common_loop.cpp | 27 ++++++++++----------------- event/common_loop.h | 20 +++++++------------- event/common_loop_run.cpp | 22 +++++++++++----------- event/signal_event_test.cpp | 4 ++-- 4 files changed, 30 insertions(+), 43 deletions(-) diff --git a/event/common_loop.cpp b/event/common_loop.cpp index 6737c52..6645025 100644 --- a/event/common_loop.cpp +++ b/event/common_loop.cpp @@ -17,13 +17,6 @@ namespace event { using namespace std::chrono; -CommonLoop::CommonLoop() : - has_unhandle_req_(false), - read_fd_(-1), write_fd_(-1), - sp_read_event_(nullptr), - cb_level_(0) -{ } - CommonLoop::~CommonLoop() { assert(cb_level_ == 0); @@ -38,7 +31,7 @@ bool CommonLoop::isInLoopThread() bool CommonLoop::isRunning() const { std::lock_guard g(lock_); - return sp_read_event_ != nullptr; + return sp_run_read_event_ != nullptr; } void CommonLoop::runThisBeforeLoop() @@ -56,17 +49,17 @@ void CommonLoop::runThisBeforeLoop() } using std::placeholders::_1; - sp_read_event->setCallback(std::bind(&CommonLoop::onGotRunInLoopFunc, this, _1)); + sp_read_event->setCallback(std::bind(&CommonLoop::handleRunInLoopRequest, this, _1)); sp_read_event->enable(); std::lock_guard g(lock_); loop_thread_id_ = std::this_thread::get_id(); - read_fd_ = read_fd; - write_fd_ = write_fd; - sp_read_event_ = sp_read_event; + run_read_fd_ = read_fd; + run_write_fd_ = write_fd; + sp_run_read_event_ = sp_read_event; if (!run_in_loop_func_queue_.empty()) - commitRequest(); + commitRunRequest(); #ifdef ENABLE_STAT resetStat(); @@ -79,10 +72,10 @@ void CommonLoop::runThisAfterLoop() cleanupDeferredTasks(); loop_thread_id_ = std::thread::id(); //! 清空 loop_thread_id_ - if (sp_read_event_ != nullptr) { - CHECK_DELETE_RESET_OBJ(sp_read_event_); - CHECK_CLOSE_RESET_FD(write_fd_); - CHECK_CLOSE_RESET_FD(read_fd_); + if (sp_run_read_event_ != nullptr) { + CHECK_DELETE_RESET_OBJ(sp_run_read_event_); + CHECK_CLOSE_RESET_FD(run_write_fd_); + CHECK_CLOSE_RESET_FD(run_read_fd_); } } diff --git a/event/common_loop.h b/event/common_loop.h index 8d9b2c4..c29982a 100644 --- a/event/common_loop.h +++ b/event/common_loop.h @@ -19,7 +19,6 @@ namespace event { class CommonLoop : public Loop { public: - CommonLoop(); ~CommonLoop() override; public: @@ -50,24 +49,19 @@ class CommonLoop : public Loop { void runThisBeforeLoop(); void runThisAfterLoop(); - void onGotRunInLoopFunc(short); - + void handleRunInLoopRequest(short); void cleanupDeferredTasks(); - - void commitRequest(); - void finishRequest(); - + void commitRunRequest(); + void finishRunRequest(); void handleNextFunc(); private: mutable std::recursive_mutex lock_; - std::thread::id loop_thread_id_; - - bool has_unhandle_req_ = false; - int read_fd_ = -1, write_fd_ = -1; - FdEvent *sp_read_event_ = nullptr; - + bool has_commit_run_req_ = false; + int run_read_fd_ = -1; + int run_write_fd_ = -1; + FdEvent *sp_run_read_event_ = nullptr; std::deque run_in_loop_func_queue_; std::deque run_next_func_queue_; diff --git a/event/common_loop_run.cpp b/event/common_loop_run.cpp index 08f9df6..92d72f2 100644 --- a/event/common_loop_run.cpp +++ b/event/common_loop_run.cpp @@ -11,10 +11,10 @@ void CommonLoop::runInLoop(const Func &func) std::lock_guard g(lock_); run_in_loop_func_queue_.push_back(func); - if (sp_read_event_ == nullptr) + if (sp_run_read_event_ == nullptr) return; - commitRequest(); + commitRunRequest(); } void CommonLoop::runNext(const Func &func) @@ -48,7 +48,7 @@ void CommonLoop::handleNextFunc() } } -void CommonLoop::onGotRunInLoopFunc(short) +void CommonLoop::handleRunInLoopRequest(short) { /** * NOTICE: @@ -62,7 +62,7 @@ void CommonLoop::onGotRunInLoopFunc(short) { std::lock_guard g(lock_); run_in_loop_func_queue_.swap(tmp); - finishRequest(); + finishRunRequest(); } while (!tmp.empty()) { @@ -103,24 +103,24 @@ void CommonLoop::cleanupDeferredTasks() LogWarn("found recursive actions, force quit"); } -void CommonLoop::commitRequest() +void CommonLoop::commitRunRequest() { - if (!has_unhandle_req_) { + if (!has_commit_run_req_) { char ch = 0; - ssize_t wsize = write(write_fd_, &ch, 1); + ssize_t wsize = write(run_write_fd_, &ch, 1); (void)wsize; - has_unhandle_req_ = true; + has_commit_run_req_ = true; } } -void CommonLoop::finishRequest() +void CommonLoop::finishRunRequest() { char ch = 0; - ssize_t rsize = read(read_fd_, &ch, 1); + ssize_t rsize = read(run_read_fd_, &ch, 1); (void)rsize; - has_unhandle_req_ = false; + has_commit_run_req_ = false; } } diff --git a/event/signal_event_test.cpp b/event/signal_event_test.cpp index 4d1fb10..96b9e20 100644 --- a/event/signal_event_test.cpp +++ b/event/signal_event_test.cpp @@ -360,7 +360,7 @@ TEST(SignalEvent, MultiSignalMultiLoopInMultiThread) //! 同一个事件，监听多个事件 TEST(SignalEvent, OneEventMultiSignal) { - LogOutput_Initialize("test"); + //LogOutput_Initialize("test"); auto engins = Loop::Engines(); for (auto e : engins) { cout << "engin: " << e << endl; @@ -374,7 +374,7 @@ TEST(SignalEvent, OneEventMultiSignal) int user2_run_time = 0; signal_event->setCallback( [&](int signo) { - LogTrace("signo:%d", signo); + //LogTrace("signo:%d", signo); if (signo == SIGUSR1) ++user1_run_time; else if (signo == SIGUSR2) -- Gitee From d905042940ba44eb40e6fa0bf85c0e65cf50ff13 Mon Sep 17 00:00:00 2001 From: Hevake Date: Thu, 3 Mar 2022 21:07:59 +0800 Subject: [PATCH 10/13] =?UTF-8?q?=E5=AE=8C=E5=96=84SignalEvent=E7=9A=84?= =?UTF-8?q?=E6=B5=8B=E8=AF=95=E7=94=A8=E4=BE=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- event/signal_event_test.cpp | 44 ++++++++++++++++++++++++++++++++++++- 1 file changed, 43 insertions(+), 1 deletion(-) diff --git a/event/signal_event_test.cpp b/event/signal_event_test.cpp index 96b9e20..e30587a 100644 --- a/event/signal_event_test.cpp +++ b/event/signal_event_test.cpp @@ -360,7 +360,6 @@ TEST(SignalEvent, MultiSignalMultiLoopInMultiThread) //! 同一个事件，监听多个事件 TEST(SignalEvent, OneEventMultiSignal) { - //LogOutput_Initialize("test"); auto engins = Loop::Engines(); for (auto e : engins) { cout << "engin: " << e << endl; @@ -401,3 +400,46 @@ TEST(SignalEvent, OneEventMultiSignal) } } +//! 短时间内触发非常多的信号 +TEST(SignalEvent, LargeNumberOfSignals) +{ + auto engins = Loop::Engines(); + for (auto e : engins) { + cout << "engin: " << e << endl; + auto sp_loop = Loop::New(e); + auto signal_event = sp_loop->newSignalEvent(); + EXPECT_TRUE(signal_event->initialize(SIGUSR1, Event::Mode::kPersist)); + EXPECT_TRUE(signal_event->enable()); + + int user1_run_time = 0; + signal_event->setCallback( + [&](int signo) { + if (signo == SIGUSR1) + ++user1_run_time; + } + ); + + int remain = 1000; + std::function func = \ + [&] { + --remain; + if (remain >= 0) { + raise(SIGUSR1); + sp_loop->runInLoop(func); + } else { + sp_loop->exitLoop(std::chrono::milliseconds(10)); + } + }; + sp_loop->runInLoop(func); + + sp_loop->runLoop(); + + EXPECT_EQ(remain, -1); + EXPECT_EQ(user1_run_time, 1000); + + delete signal_event; + delete sp_loop; + } +} + + -- Gitee From 597e7736bb67c0eb7cf88ad2c7a08a388300c181 Mon Sep 17 00:00:00 2001 From: Hevake Date: Thu, 3 Mar 2022 21:25:59 +0800 Subject: [PATCH 11/13] =?UTF-8?q?=E4=BF=AE=E6=94=B9CommonLoop=EF=BC=8C?= =?UTF-8?q?=E5=AE=9E=E7=8E=B0=E5=9C=A8=E4=BF=AE=E6=94=B9=20=5Fsignal=5Fwri?= =?UTF-8?q?te=5Ffds=5F=20=E6=9C=9F=E9=97=B4=E7=A6=81=E6=AD=A2=E4=BF=A1?= =?UTF-8?q?=E5=8F=B7=E8=A7=A6=E5=8F=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- event/common_loop_signal.cpp | 21 +++++++++++++++++++-- 1 file changed, 19 insertions(+), 2 deletions(-) diff --git a/event/common_loop_signal.cpp b/event/common_loop_signal.cpp index e086268..64f01f3 100644 --- a/event/common_loop_signal.cpp +++ b/event/common_loop_signal.cpp @@ -30,13 +30,21 @@ bool CommonLoop::subscribeSignal(int signo, SignalSubscribuer *who) if (this_signal_subscribers.empty()) { //! 如果本Loop没有监听该信号，则要去 _signal_write_fds_ 中订阅 std::unique_lock _g(_signal_lock_); - //!FIXME: 要禁止信号触发 + + //! 要禁止信号触发 + sigset_t new_sigmask, old_sigmask; + sigfillset(&new_sigmask); + sigprocmask(SIG_BLOCK, &new_sigmask, &old_sigmask); + auto & signo_fds = _signal_write_fds_[signo]; if (signo_fds.empty()) { signal(signo, CommonLoop::HandleSignal); //LogTrace("set signal:%d", signo); } signo_fds.insert(signal_write_fd_); + + //! 恢复信号 + sigprocmask(SIG_SETMASK, &old_sigmask, 0); } this_signal_subscribers.insert(who); @@ -54,7 +62,12 @@ bool CommonLoop::unsubscribeSignal(int signo, SignalSubscribuer *who) all_signals_subscribers_.erase(signo); //! 则将该信号的订阅记录表删除 { std::unique_lock _g(_signal_lock_); - //!FIXME: 要禁止信号触发 + + //! 要禁止信号触发 + sigset_t new_sigmask, old_sigmask; + sigfillset(&new_sigmask); + sigprocmask(SIG_BLOCK, &new_sigmask, &old_sigmask); + //! 并将 _signal_write_fds_ 中的记录删除 auto &this_signal_fds = _signal_write_fds_[signo]; this_signal_fds.erase(signal_write_fd_); @@ -64,6 +77,9 @@ bool CommonLoop::unsubscribeSignal(int signo, SignalSubscribuer *who) //LogTrace("unset signal:%d", signo); _signal_write_fds_.erase(signo); } + + //! 恢复信号 + sigprocmask(SIG_SETMASK, &old_sigmask, 0); } if (!all_signals_subscribers_.empty()) @@ -81,6 +97,7 @@ bool CommonLoop::unsubscribeSignal(int signo, SignalSubscribuer *who) return true; } +//! 信号处理函数 void CommonLoop::HandleSignal(int signo) { //LogTrace("got signal :%d", signo); -- Gitee From f39a921670cc47b7ebd56eeaf53291da2a21d3d1 Mon Sep 17 00:00:00 2001 From: Hevake Date: Thu, 3 Mar 2022 21:46:51 +0800 Subject: [PATCH 12/13] =?UTF-8?q?=E6=9B=B4=E6=96=B0README.md?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index b94da2a..1e457ee 100644 --- a/README.md +++ b/README.md @@ -15,8 +15,8 @@ apt -y install g++ make libgtest libevent-dev libev-dev libgtest-dev - base，基础库，含日志打印、常用工具等； - util，工具库，在业务代码中可能会用到的库； -- event，事件库，对 libevent2, libev 库进行了统一的封装，实现Fd,Timer,Signal三种事件驱动； -- eventx，事件扩展库，含 ThreadPool 线程池模块，专用于处理阻塞性事务； +- event，事件库，实现Fd,Timer,Signal三种事件驱动； +- eventx，事件扩展库，含 ThreadPool 线程池模块，专用于处理阻塞性事务；TimerPool 定时器池模块； - network，网络库，实现了串口、终端、UDP、TCP 通信模块； - coroutine，协程库，众所周知，异步框架不方便处理顺序性业务，协程弥补之； - mqtt，MQTT客户端库； @@ -29,10 +29,10 @@ apt -y install g++ make libgtest libevent-dev libev-dev libgtest-dev - base --> None - util --> base - event --> base, [libevent\_core, libev] -- eventx --> base, event, [pthread] +- eventx --> base, event, - network --> base, event - coroutine --> base, event -- mqtt --> base, event, [mosquitto] +- mqtt --> base, event, - terminal --> base, util, event, network - main --> base, util, event, eventx - sample --> main @@ -40,7 +40,6 @@ apt -y install g++ make libgtest libevent-dev libev-dev libgtest-dev #### 未来规化 - 创建 http 模块，实现 Http 相关的 Server 与 Client 端； -- 在 event 中支持 buildin 事件驱动（进行中 feature-epoll）； - 在 network 中支持 TLS； - 实现异步日志输出模块； - 实现使用 CMake 进行工程管理； -- Gitee From 961a42522b84dce77f099053cead7d7b380b9293 Mon Sep 17 00:00:00 2001 From: Hevake Date: Fri, 4 Mar 2022 21:11:55 +0800 Subject: [PATCH 13/13] =?UTF-8?q?=E5=88=A0=E9=99=A4=E4=B8=8D=E5=BF=85?= =?UTF-8?q?=E8=A6=81=E7=9A=84=E6=96=87=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- event/engins/epoll/fd_event.h | 4 +- event/engins/epoll/signal_event.cpp | 131 ---------------------------- event/engins/epoll/signal_event.h | 52 ----------- 3 files changed, 2 insertions(+), 185 deletions(-) delete mode 100644 event/engins/epoll/signal_event.cpp delete mode 100644 event/engins/epoll/signal_event.h diff --git a/event/engins/epoll/fd_event.h b/event/engins/epoll/fd_event.h index e45fc51..ffada80 100644 --- a/event/engins/epoll/fd_event.h +++ b/event/engins/epoll/fd_event.h @@ -8,8 +8,8 @@ namespace tbox { namespace event { -class EpollLoop; -class EpollFdSharedData; +class EpollLoop; +struct EpollFdSharedData; class EpollFdEvent : public FdEvent { public: diff --git a/event/engins/epoll/signal_event.cpp b/event/engins/epoll/signal_event.cpp deleted file mode 100644 index beb80fc..0000000 --- a/event/engins/epoll/signal_event.cpp +++ /dev/null @@ -1,131 +0,0 @@ -#include -#include -#include -#include -#include - -#include "signal_event.h" - -#include -#include -#include "loop.h" - -namespace tbox { -namespace event { - -EpollSignalEvent::EpollSignalEvent(EpollLoop *wp_loop) : - wp_loop_(wp_loop), - signal_fd_event_(wp_loop) -{ - sigemptyset(&sig_mask_); - signal_fd_ = signalfd(-1, &sig_mask_, SFD_NONBLOCK | SFD_CLOEXEC); - assert(signal_fd_ >= 0); -} - -EpollSignalEvent::~EpollSignalEvent() -{ - assert(cb_level_ == 0); - disable(); - - CHECK_CLOSE_RESET_FD(signal_fd_); -} - -bool EpollSignalEvent::initialize(int signum, Mode mode) -{ - disable(); - - if (!signal_fd_event_.initialize(signal_fd_, FdEvent::kReadEvent, mode)) - return false; - - signal_fd_event_.setCallback(std::bind(&EpollSignalEvent::onEvent, this, std::placeholders::_1)); - - sigaddset(&sig_mask_, signum); - - if (mode == Mode::kOneshot) - is_stop_after_trigger_ = true; - - is_inited_ = true; - return true; -} - -void EpollSignalEvent::setCallback(const CallbackFunc &cb) -{ - cb_ = cb; -} - -bool EpollSignalEvent::isEnabled() const -{ - if (!is_inited_) - return false; - - return signal_fd_event_.isEnabled(); -} - -bool EpollSignalEvent::enable() -{ - if (!is_inited_) - return false; - - if (isEnabled()) - return true; - - if (!signal_fd_event_.enable()) - return false; - - signalfd(signal_fd_, &sig_mask_, 0); - sigprocmask(SIG_BLOCK, &sig_mask_, 0); - - return true; -} - -bool EpollSignalEvent::disable() -{ - if (!is_inited_) - return false; - - if (!isEnabled()) - return true; - - sigprocmask(SIG_UNBLOCK, &sig_mask_, 0); - signalfd(signal_fd_, &sig_mask_, 0); - - if (!signal_fd_event_.disable()) - return false; - - return true; -} - -Loop* EpollSignalEvent::getLoop() const -{ - return wp_loop_; -} - -void EpollSignalEvent::onEvent(short events) -{ - wp_loop_->beginEventProcess(); - - if (!(events & FdEvent::kReadEvent)) - return; - - /// We need read the signal_fd_ if got some signals - struct signalfd_siginfo info; - if (read(signal_fd_, &info, sizeof(info)) != sizeof(info)) - return; - - /// signal number = info.ssi_signo - /// pid = info.ssi_pid - /// uid = info.ssi_uid - - if (cb_) { - ++cb_level_; - cb_(); - --cb_level_; - - if (is_stop_after_trigger_) - disable(); - } - - wp_loop_->endEventProcess(); -} -} -} diff --git a/event/engins/epoll/signal_event.h b/event/engins/epoll/signal_event.h deleted file mode 100644 index 61d8a1f..0000000 --- a/event/engins/epoll/signal_event.h +++ /dev/null @@ -1,52 +0,0 @@ -#ifndef TBOX_EVENT_EPOLL_SINGAL_EVENT_H_20220110 -#define TBOX_EVENT_EPOLL_SINGAL_EVENT_H_20220110 - -#include -#include "../../signal_event.h" -#include "fd_event.h" - -struct epoll_event; - -namespace tbox { -namespace event { - -class EpollLoop; -class EpollFdEvent; - -class EpollSignalEvent : public SignalEvent { - public: - explicit EpollSignalEvent(EpollLoop *wp_loop); - virtual ~EpollSignalEvent(); - - public: - virtual bool initialize(int signum, Mode mode); - virtual void setCallback(const CallbackFunc &cb); - - virtual bool isEnabled() const; - virtual bool enable(); - virtual bool disable(); - - virtual Loop* getLoop() const; - - protected: - void onEvent(short events); - - private: - EpollLoop *wp_loop_{ nullptr }; - - bool is_inited_{ false }; - bool is_stop_after_trigger_{ false }; - CallbackFunc cb_{ nullptr }; - - int signal_fd_ = -1; - sigset_t sig_mask_; - EpollFdEvent signal_fd_event_; - - int cb_level_ = 0; -}; - -} -} - -#endif //TBOX_EVENT_EPOLL_SINGAL_EVENT_H_20220110 - -- Gitee