[Kernel] Add flag in wait queue and fix wakeup issue.

components/dfs/filesystems/net/net_sockets.c

@@ -141,7 +141,7 @@ int accept(int s, struct sockaddr *addr, socklen_t *addrlen)
         d->fops = dfs_net_get_fops();
         /* initialize wait head */
         lwsock = lwip_tryget_socket(new_client);
-        rt_list_init(&(lwsock->wait_head));
+        rt_wqueue_init(&(lwsock->wait_head));
 
         d->flags = O_RDWR; /* set flags as read and write */
         d->size = 0;
@@ -317,7 +317,7 @@ int socket(int domain, int type, int protocol)
         d->data = (void *) sock;
 
         lwsock = lwip_tryget_socket(sock);
-        rt_list_init(&(lwsock->wait_head));
+        rt_wqueue_init(&(lwsock->wait_head));
         lwsock->conn->callback = event_callback;
     }
     else

components/drivers/include/ipc/waitqueue.h

@@ -3,9 +3,10 @@
 
 #include <rtthread.h>
 
-struct rt_wqueue_node;
+#define RT_WQ_FLAG_CLEAN    0x00
+#define RT_WQ_FLAG_WAKEUP   0x01
 
-typedef rt_list_t rt_wqueue_t;
+struct rt_wqueue_node;
 typedef int (*rt_wqueue_func_t)(struct rt_wqueue_node *wait, void *key);
 
 struct rt_wqueue_node
@@ -20,6 +21,14 @@ typedef struct rt_wqueue_node rt_wqueue_node_t;
 
 int __wqueue_default_wake(struct rt_wqueue_node *wait, void *key);
 
+rt_inline void rt_wqueue_init(rt_wqueue_t *queue)
+{
+    RT_ASSERT(queue != RT_NULL);
+
+    queue->flag = RT_WQ_FLAG_CLEAN;
+    rt_list_init(&(queue->waiting_list));
+}
+
 void rt_wqueue_add(rt_wqueue_t *queue, struct rt_wqueue_node *node);
 void rt_wqueue_remove(struct rt_wqueue_node *node);
 int  rt_wqueue_wait(rt_wqueue_t *queue, int condition, int timeout);
@@ -37,3 +46,4 @@ void rt_wqueue_wakeup(rt_wqueue_t *queue, void *key);
 #define DEFINE_WAIT(name) DEFINE_WAIT_FUNC(name, __wqueue_default_wake)
 
 #endif
+

components/drivers/src/pipe.c

@@ -438,8 +438,8 @@ rt_pipe_t *rt_pipe_create(const char *name, int bufsz)
 
     rt_memset(pipe, 0, sizeof(rt_pipe_t));
     rt_mutex_init(&(pipe->lock), name, RT_IPC_FLAG_FIFO);
-    rt_list_init(&(pipe->reader_queue));
-    rt_list_init(&(pipe->writer_queue));
+	rt_wqueue_init(&(pipe->reader_queue));
+	rt_wqueue_init(&(pipe->writer_queue));
 
     RT_ASSERT(bufsz < 0xFFFF);
     pipe->bufsz = bufsz;

components/drivers/src/waitqueue.c

@@ -4,14 +4,12 @@
 #include <rtdevice.h>
 #include <rtservice.h>
 
-extern struct rt_thread *rt_current_thread;
-
 void rt_wqueue_add(rt_wqueue_t *queue, struct rt_wqueue_node *node)
 {
     rt_base_t level;
 
     level = rt_hw_interrupt_disable();
-    rt_list_insert_before(queue, &(node->list));
+    rt_list_insert_before(&(queue->waiting_list), &(node->list));
     rt_hw_interrupt_enable(level);
 }
 
@@ -34,24 +32,30 @@ void rt_wqueue_wakeup(rt_wqueue_t *queue, void *key)
     rt_base_t level;
     register int need_schedule = 0;
 
+	rt_list_t *queue_list;
     struct rt_list_node *node;
     struct rt_wqueue_node *entry;
 
-    if (rt_list_isempty(queue))
-        return;
+	queue_list = &(queue->waiting_list);
 
     level = rt_hw_interrupt_disable();
-    for (node = queue->next; node != queue; node = node->next)
+	/* set wakeup flag in the queue */
+	queue->flag = RT_WQ_FLAG_WAKEUP;
+
+    if (!(rt_list_isempty(queue_list)))
     {
-        entry = rt_list_entry(node, struct rt_wqueue_node, list);
-        if (entry->wakeup(entry, key) == 0)
-        {
-            rt_thread_resume(entry->polling_thread);
-            need_schedule = 1;
-
-            rt_wqueue_remove(entry);
-            break;
-        }
+	    for (node = queue_list->next; node != queue_list; node = node->next)
+	    {
+	        entry = rt_list_entry(node, struct rt_wqueue_node, list);
+	        if (entry->wakeup(entry, key) == 0)
+	        {
+	            rt_thread_resume(entry->polling_thread);
+	            need_schedule = 1;
+
+	            rt_wqueue_remove(entry);
+	            break;
+	        }
+	    }
     }
     rt_hw_interrupt_enable(level);
 
@@ -62,7 +66,7 @@ void rt_wqueue_wakeup(rt_wqueue_t *queue, void *key)
 int rt_wqueue_wait(rt_wqueue_t *queue, int condition, int msec)
 {
     int tick;
-    rt_thread_t tid = rt_current_thread;
+    rt_thread_t tid = rt_thread_self();
     rt_timer_t  tmr = &(tid->thread_timer);
     struct rt_wqueue_node __wait;
     rt_base_t level;
@@ -81,6 +85,12 @@ int rt_wqueue_wait(rt_wqueue_t *queue, int condition, int msec)
     rt_list_init(&__wait.list);
 
     level = rt_hw_interrupt_disable();
+	if (queue->flag == RT_WQ_FLAG_WAKEUP)
+	{
+		/* already wakeup */
+		goto __exit_wakeup;
+	}
+
     rt_wqueue_add(queue, &__wait);
     rt_thread_suspend(tid);
 
@@ -97,6 +107,12 @@ int rt_wqueue_wait(rt_wqueue_t *queue, int condition, int msec)
 
     rt_schedule();
 
+    level = rt_hw_interrupt_disable();
+
+__exit_wakeup:
+	queue->flag = 0;
+    rt_hw_interrupt_enable(level);
+
     rt_wqueue_remove(&__wait);
 
     return 0;

include/rtdef.h

@@ -848,6 +848,9 @@ enum rt_device_class_type
 #define RT_DEVICE_CTRL_RTC_SET_ALARM    0x13            /**< set alarm */
 
 typedef struct rt_device *rt_device_t;
+/**
+ * operations set for device object
+ */
 struct rt_device_ops
 {
     /* common device interface */
@@ -859,6 +862,18 @@ struct rt_device_ops
     rt_err_t  (*control)(rt_device_t dev, int cmd, void *args);
 };
 
+#if defined(RT_USING_POSIX)
+/**
+ * WaitQueue structure
+ */
+struct rt_wqueue
+{
+	rt_uint32_t flag;
+	rt_list_t waiting_list;
+};
+typedef struct rt_wqueue rt_wqueue_t;
+#endif
+
 /**
  * Device structure
  */
@@ -891,7 +906,7 @@ struct rt_device
 
 #if defined(RT_USING_POSIX)
     const struct dfs_file_ops *fops;
-    rt_list_t wait_queue;
+    struct rt_wqueue wait_queue;
 #endif
 
     void                     *user_data;                /**< device private data */

src/device.c

@@ -29,6 +29,9 @@
  */
 
 #include <rtthread.h>
+#if defined(RT_USING_POSIX)
+#include <rtdevice.h> /* for wqueue_init */
+#endif
 
 #ifdef RT_USING_DEVICE
 
@@ -74,7 +77,7 @@ rt_err_t rt_device_register(rt_device_t dev,
 
 #if defined(RT_USING_POSIX)
     dev->fops = RT_NULL;
-    rt_list_init(&(dev->wait_queue));
+	rt_wqueue_init(&(dev->wait_queue));
 #endif
 
     return RT_EOK;