From 2d8ea9ed99116bf034e447307f8360dd24e9449c Mon Sep 17 00:00:00 2001
From: jiangheng <jiangheng14@huawei.com>
Date: Tue, 10 Dec 2024 16:35:44 +0800
Subject: [PATCH] fix epoll and recv threads blocked on the same semaphore.
 data cannot be read in recv thread.

---
 src/lstack/api/lstack_epoll.c     | 24 +++++++++++++----
 src/lstack/core/lstack_lwip.c     | 43 ++++++++++++++++++++-----------
 src/lstack/include/lstack_epoll.h |  1 +
 3 files changed, 48 insertions(+), 20 deletions(-)

diff --git a/src/lstack/api/lstack_epoll.c b/src/lstack/api/lstack_epoll.c
index ff9cccf..644efc0 100644
--- a/src/lstack/api/lstack_epoll.c
+++ b/src/lstack/api/lstack_epoll.c
@@ -73,11 +73,18 @@ void add_sock_event_nolock(struct lwip_sock *sock, uint32_t event)
     return;
 }
 
-void add_sock_event(struct lwip_sock *sock, uint32_t event)
+static void _add_sock_event(struct lwip_sock *sock, struct wakeup_poll *wakeup, uint32_t event)
 {
-    struct wakeup_poll *wakeup = sock->wakeup;
     struct protocol_stack *stack = sock->stack;
-    if (wakeup == NULL || wakeup->type == WAKEUP_CLOSE || (event & sock->epoll_events) == 0) {
+    if (wakeup == NULL || wakeup->type == WAKEUP_CLOSE) {
+        return;
+    }
+
+    if (wakeup->type == WAKEUP_BLOCK) {
+        if (!(event & (EPOLLIN | EPOLLERR))) {
+            return;
+        }
+    } else if (!(event & sock->epoll_events)) {
         return;
     }
 
@@ -91,6 +98,12 @@ void add_sock_event(struct lwip_sock *sock, uint32_t event)
     return;
 }
 
+void add_sock_event(struct lwip_sock *sock, uint32_t event)
+{
+    _add_sock_event(sock, sock->wakeup, event);
+    _add_sock_event(sock, sock->recv_block, event);
+}
+
 void del_sock_event_nolock(struct lwip_sock *sock, uint32_t event)
 {
     if (get_global_cfg_params()->stack_mode_rtc) {
@@ -99,7 +112,7 @@ void del_sock_event_nolock(struct lwip_sock *sock, uint32_t event)
         if ((event & EPOLLOUT) && !NETCONN_IS_OUTIDLE(sock)) {
             sock->events &= ~EPOLLOUT;
         }
-	if ((event & EPOLLIN) && !NETCONN_IS_DATAIN(sock) && !NETCONN_IS_ACCEPTIN(sock)) {
+        if ((event & EPOLLIN) && !NETCONN_IS_DATAIN(sock) && !NETCONN_IS_ACCEPTIN(sock)) {
             sock->events &= ~EPOLLIN;
         }
     }
@@ -212,7 +225,8 @@ static void raise_pending_events(struct wakeup_poll *wakeup, struct lwip_sock *s
         if (wakeup->type == WAKEUP_EPOLL && (sock->events & sock->epoll_events) &&
             list_node_null(&sock->event_list)) {
             list_add_node(&sock->event_list, &wakeup->event_list);
-	    sem_post(&wakeup->wait);
+            rte_mb();
+            sem_post(&wakeup->wait);
         }
     }
     pthread_spin_unlock(&wakeup->event_list_lock);
diff --git a/src/lstack/core/lstack_lwip.c b/src/lstack/core/lstack_lwip.c
index 648da58..c1338aa 100644
--- a/src/lstack/core/lstack_lwip.c
+++ b/src/lstack/core/lstack_lwip.c
@@ -1002,30 +1002,43 @@ static int recv_ring_get_one(struct lwip_sock *sock, bool noblock, struct pbuf *
         return 0;
     }
 
-    while (gazelle_ring_read(sock->recv_ring, (void **)pbuf, expect) != expect) {
-        if (noblock) {
+    if (noblock) {
+        if (gazelle_ring_read(sock->recv_ring, (void **)pbuf, expect) != expect) {
             GAZELLE_RETURN(EAGAIN);
         }
+        goto END;
+    }
+
+    if (sock->recv_block == NULL) {
+        sock->recv_block = poll_construct_wakeup();
+        if (sock->recv_block == NULL) {
+            GAZELLE_RETURN(ENOMEM);
+        }
+        sock->recv_block->type = WAKEUP_BLOCK;
+    }
+
+    do {
+        __atomic_store_n(&sock->recv_block->in_wait, true, __ATOMIC_RELEASE);
+        if (gazelle_ring_read(sock->recv_ring, (void **)pbuf, expect) == expect) {
+            break;
+        }
         if (recv_break_for_err(sock)) {
+            sock->recv_block = NULL;
             return -1;
         }
-        if (unlikely(sock->wakeup == NULL)) {
-            sock->wakeup = poll_construct_wakeup();
-            if (sock->wakeup == NULL) {
-                return -1;
+        ret = lstack_block_wait(sock->recv_block, sock->conn->recv_timeout);
+        if (ret != 0) {
+            if (errno = ETIMEDOUT) {
+                errno = EAGAIN;
             }
-            sock->epoll_events = POLLIN | POLLERR;
-        }
-
-        ret = lstack_block_wait(sock->wakeup, sock->conn->recv_timeout);
-        if (ret == ETIMEDOUT) {
-            noblock = true;
-        } else if (ret != 0 && errno == EINTR) {
-            /* SIGALRM signal may interrupt blocking */
+            sock->recv_block = NULL;
             return ret;
         }
-    }
+    } while (1);
+    __atomic_store_n(&sock->recv_block->in_wait, false, __ATOMIC_RELEASE);
+    sock->recv_block = NULL;
 
+END:
     if (get_protocol_stack_group()->latency_start) {
         calculate_lstack_latency(&sock->stack->latency, *pbuf, GAZELLE_LATENCY_READ_APP_CALL, time_stamp);
     }
diff --git a/src/lstack/include/lstack_epoll.h b/src/lstack/include/lstack_epoll.h
index cad9aed..83eace7 100644
--- a/src/lstack/include/lstack_epoll.h
+++ b/src/lstack/include/lstack_epoll.h
@@ -28,6 +28,7 @@ enum wakeup_type {
     WAKEUP_EPOLL = 0,
     WAKEUP_POLL,
     WAKEUP_CLOSE,
+    WAKEUP_BLOCK,
 };
 
 struct protocol_stack;
-- 
2.33.0