From 0d954823cec5d594898db698e36fb16baa22e791 Mon Sep 17 00:00:00 2001
From: qgymib <qgymib@gmail.com>
Date: Sat, 7 Oct 2023 11:56:15 +0800
Subject: [PATCH] remove active_events in pipe/unix

---
 ev.c                 | 47 ++++++++++++++++++++++++++++++++++++--------
 src/unix/pipe_unix.c | 43 ++++++++++++++++++++++++++++++++++------
 2 files changed, 76 insertions(+), 14 deletions(-)

diff --git a/ev.c b/ev.c
index 1453b2cd4..78277d1cc 100644
--- a/ev.c
+++ b/ev.c
@@ -14155,8 +14155,8 @@ void ev_once_execute(ev_once_t* guard, ev_once_cb cb)
 
 ////////////////////////////////////////////////////////////////////////////////
 // FILE:    src/unix/pipe_unix.c
-// SIZE:    24988
-// SHA-256: 6b43efbeac03930e4f88370f79aff8c4702a668d35a726589363d75501214cf7
+// SIZE:    25751
+// SHA-256: 09de7727043193ce6c9887885a548b7d3ce43c15690b943e73852e2215905cd5
 ////////////////////////////////////////////////////////////////////////////////
 #line 1 "src/unix/pipe_unix.c"
 #define _GNU_SOURCE
@@ -14194,17 +14194,44 @@ static void _ev_pipe_on_close_unix(ev_handle_t* handle)
     }
 }
 
+static void _ev_pipe_smart_deactive(ev_pipe_t* pipe)
+{
+    size_t io_sz = 0;
+
+    if (!ev__handle_is_active(&pipe->base))
+    {
+        return;
+    }
+
+    if (pipe->base.data.flags & EV_HANDLE_PIPE_IPC)
+    {
+        io_sz += ev_list_size(&pipe->backend.ipc_mode.rio.rqueue);
+        io_sz += pipe->backend.ipc_mode.rio.curr.reading != NULL ? 1 : 0;
+        io_sz += ev_list_size(&pipe->backend.ipc_mode.wio.wqueue);
+        io_sz += pipe->backend.ipc_mode.wio.curr.writing != NULL ? 1 : 0;
+    }
+    else
+    {
+        io_sz = ev__nonblock_stream_size(pipe->backend.data_mode.stream, EV_IO_IN | EV_IO_OUT);
+    }
+
+	if (io_sz == 0)
+	{
+		ev__handle_deactive(&pipe->base);
+	}
+}
+
 static void _ev_pipe_w_user_callback_unix(ev_pipe_t* pipe,
     ev_pipe_write_req_t* req, ssize_t size)
 {
-    ev__handle_event_dec(&pipe->base);
+    _ev_pipe_smart_deactive(pipe);
     ev__write_exit(&req->base);
     req->ucb(req, size);
 }
 
 static void _ev_pipe_r_user_callback_unix(ev_pipe_t* pipe, ev_pipe_read_req_t* req, ssize_t size)
 {
-    ev__handle_event_dec(&pipe->base);
+    _ev_pipe_smart_deactive(pipe);
     ev__read_exit(&req->base);
     req->ucb(req, size);
 }
@@ -14998,7 +15025,7 @@ int ev_pipe_write_ex(ev_pipe_t* pipe, ev_pipe_write_req_t* req,
         return ret;
     }
 
-    ev__handle_event_add(&pipe->base);
+    ev__handle_active(&pipe->base);
 
     if (pipe->base.data.flags & EV_HANDLE_PIPE_IPC)
     {
@@ -15011,8 +15038,10 @@ int ev_pipe_write_ex(ev_pipe_t* pipe, ev_pipe_write_req_t* req,
 
     if (ret != 0)
     {
-        ev__handle_event_dec(&pipe->base);
         _ev_pipe_abort_unix(pipe, ret);
+
+        /* The final state must be non-active. */
+        ev__handle_deactive(&pipe->base);
     }
 
     return ret;
@@ -15032,7 +15061,7 @@ int ev_pipe_read(ev_pipe_t* pipe, ev_pipe_read_req_t* req, ev_buf_t* bufs,
         return ret;
     }
 
-    ev__handle_event_add(&pipe->base);
+    ev__handle_active(&pipe->base);
 
     if (pipe->base.data.flags & EV_HANDLE_PIPE_IPC)
     {
@@ -15045,8 +15074,10 @@ int ev_pipe_read(ev_pipe_t* pipe, ev_pipe_read_req_t* req, ev_buf_t* bufs,
 
     if (ret != 0)
     {
-        ev__handle_event_dec(&pipe->base);
         _ev_pipe_abort_unix(pipe, ret);
+
+        /* The final state must be non-active. */
+        ev__handle_deactive(&pipe->base);
     }
 
     return ret;
diff --git a/src/unix/pipe_unix.c b/src/unix/pipe_unix.c
index 295d8ad7a..bc9c812a1 100644
--- a/src/unix/pipe_unix.c
+++ b/src/unix/pipe_unix.c
@@ -33,17 +33,44 @@ static void _ev_pipe_on_close_unix(ev_handle_t* handle)
     }
 }
 
+static void _ev_pipe_smart_deactive(ev_pipe_t* pipe)
+{
+    size_t io_sz = 0;
+
+    if (!ev__handle_is_active(&pipe->base))
+    {
+        return;
+    }
+
+    if (pipe->base.data.flags & EV_HANDLE_PIPE_IPC)
+    {
+        io_sz += ev_list_size(&pipe->backend.ipc_mode.rio.rqueue);
+        io_sz += pipe->backend.ipc_mode.rio.curr.reading != NULL ? 1 : 0;
+        io_sz += ev_list_size(&pipe->backend.ipc_mode.wio.wqueue);
+        io_sz += pipe->backend.ipc_mode.wio.curr.writing != NULL ? 1 : 0;
+    }
+    else
+    {
+        io_sz = ev__nonblock_stream_size(pipe->backend.data_mode.stream, EV_IO_IN | EV_IO_OUT);
+    }
+
+    if (io_sz == 0)
+    {
+        ev__handle_deactive(&pipe->base);
+    }
+}
+
 static void _ev_pipe_w_user_callback_unix(ev_pipe_t* pipe,
     ev_pipe_write_req_t* req, ssize_t size)
 {
-    ev__handle_event_dec(&pipe->base);
+    _ev_pipe_smart_deactive(pipe);
     ev__write_exit(&req->base);
     req->ucb(req, size);
 }
 
 static void _ev_pipe_r_user_callback_unix(ev_pipe_t* pipe, ev_pipe_read_req_t* req, ssize_t size)
 {
-    ev__handle_event_dec(&pipe->base);
+    _ev_pipe_smart_deactive(pipe);
     ev__read_exit(&req->base);
     req->ucb(req, size);
 }
@@ -837,7 +864,7 @@ int ev_pipe_write_ex(ev_pipe_t* pipe, ev_pipe_write_req_t* req,
         return ret;
     }
 
-    ev__handle_event_add(&pipe->base);
+    ev__handle_active(&pipe->base);
 
     if (pipe->base.data.flags & EV_HANDLE_PIPE_IPC)
     {
@@ -850,8 +877,10 @@ int ev_pipe_write_ex(ev_pipe_t* pipe, ev_pipe_write_req_t* req,
 
     if (ret != 0)
     {
-        ev__handle_event_dec(&pipe->base);
         _ev_pipe_abort_unix(pipe, ret);
+
+        /* The final state must be non-active. */
+        ev__handle_deactive(&pipe->base);
     }
 
     return ret;
@@ -871,7 +900,7 @@ int ev_pipe_read(ev_pipe_t* pipe, ev_pipe_read_req_t* req, ev_buf_t* bufs,
         return ret;
     }
 
-    ev__handle_event_add(&pipe->base);
+    ev__handle_active(&pipe->base);
 
     if (pipe->base.data.flags & EV_HANDLE_PIPE_IPC)
     {
@@ -884,8 +913,10 @@ int ev_pipe_read(ev_pipe_t* pipe, ev_pipe_read_req_t* req, ev_buf_t* bufs,
 
     if (ret != 0)
     {
-        ev__handle_event_dec(&pipe->base);
         _ev_pipe_abort_unix(pipe, ret);
+
+        /* The final state must be non-active. */
+        ev__handle_deactive(&pipe->base);
     }
 
     return ret;