Add queue_send synchronous message sending. Right now only for SWCODEC. Actual usage to be added to playback and recording shortly in upcoming commits.

git-svn-id: svn://svn.rockbox.org/rockbox/trunk@11776 a1c6a512-1295-4272-9138-f99709370657

firmware/export/config.h

@@ -216,6 +216,10 @@
 #endif
 #endif
 
+#if CONFIG_CODEC == SWCODEC && !defined(BOOTLOADER)
+#define HAVE_EXTENDED_MESSAGING_AND_NAME
+#endif
+
 #if (CONFIG_CODEC == SWCODEC) && !defined(SIMULATOR) && !defined(BOOTLOADER)
 #define HAVE_PRIORITY_SCHEDULING
 #define HAVE_SCHEDULER_BOOSTCTRL

firmware/export/kernel.h

@@ -33,7 +33,7 @@
 #define QUEUE_LENGTH 16 /* MUST be a power of 2 */
 #define QUEUE_LENGTH_MASK (QUEUE_LENGTH - 1)
 
-/* System defined message ID's, occupying the top 5 bits of the event ID */
+/* System defined message ID's, occupying the top 8 bits of the event ID */
 #define SYS_EVENT                 (long)0x80000000 /* SYS events are negative */
 #define SYS_USB_CONNECTED         ((SYS_EVENT | ((long)1 << 27)))
 #define SYS_USB_CONNECTED_ACK     ((SYS_EVENT | ((long)2 << 27)))
@@ -55,12 +55,29 @@ struct event
     void *data;
 };
 
+#ifdef HAVE_EXTENDED_MESSAGING_AND_NAME
+struct queue_sender
+{
+    struct thread_entry *thread;
+    void                *retval;
+};
+
+struct queue_sender_list
+{
+    /* If non-NULL, there is a thread waiting for the corresponding event */
+    struct queue_sender *senders[QUEUE_LENGTH];
+    /* Send info for last message dequeued or NULL if replied or not sent */
+    struct queue_sender *curr_sender;
+};
+#endif /* HAVE_EXTENDED_MESSAGING_AND_NAME */
+
 struct event_queue
 {
     struct event events[QUEUE_LENGTH];
     struct thread_entry *thread;
     unsigned int read;
     unsigned int write;
+    struct queue_sender_list *send;
 };
 
 struct mutex
@@ -90,12 +107,18 @@ int tick_add_task(void (*f)(void));
 int tick_remove_task(void (*f)(void));
 
 extern void queue_init(struct event_queue *q, bool register_queue);
+extern void queue_enable_queue_send(struct event_queue *q, struct queue_sender_list *send);
 extern void queue_delete(struct event_queue *q);
 extern void queue_wait(struct event_queue *q, struct event *ev);
 extern void queue_wait_w_tmo(struct event_queue *q, struct event *ev, int ticks);
 extern void queue_post(struct event_queue *q, long id, void *data);
+#ifdef HAVE_EXTENDED_MESSAGING_AND_NAME
+extern void * queue_send(struct event_queue *q, long id, void *data);
+extern void queue_reply(struct event_queue *q, void *retval);
+extern bool queue_in_queue_send(struct event_queue *q);
+#endif /* HAVE_EXTENDED_MESSAGING_AND_NAME */
 extern bool queue_empty(const struct event_queue* q);
-void queue_clear(struct event_queue* q);
+extern void queue_clear(struct event_queue* q);
 extern void queue_remove_from_head(struct event_queue *q, long id);
 extern int queue_broadcast(long id, void *data);
 

firmware/export/thread.h

@@ -143,6 +143,13 @@ void switch_thread(bool save_context, struct thread_entry **blocked_list);
 void sleep_thread(int ticks);
 void block_thread(struct thread_entry **thread);
 void block_thread_w_tmo(struct thread_entry **thread, int timeout);
+#ifdef HAVE_EXTENDED_MESSAGING_AND_NAME
+void set_irq_level_and_block_thread(struct thread_entry **thread, int level);
+#if 0
+void set_irq_level_and_block_thread_w_tmo(struct thread_entry **list,
+                                          int timeout, int level)
+#endif
+#endif
 void wakeup_thread(struct thread_entry **thread);
 #ifdef HAVE_PRIORITY_SCHEDULING
 int thread_set_priority(struct thread_entry *thread, int priority);

firmware/kernel.c

@@ -86,13 +86,74 @@ void yield(void)
 /****************************************************************************
  * Queue handling stuff
  ****************************************************************************/
+
+#ifdef HAVE_EXTENDED_MESSAGING_AND_NAME
+/* Moves waiting thread's descriptor to the current sender when a
+   message is dequeued */
+static void queue_fetch_sender(struct queue_sender_list *send,
+                               unsigned int i)
+{
+    int old_level = set_irq_level(HIGHEST_IRQ_LEVEL);
+    struct queue_sender **spp = &send->senders[i];
+
+    if(*spp)
+    {
+        send->curr_sender = *spp;
+        *spp = NULL;
+    }
+
+    set_irq_level(old_level);
+}
+
+/* Puts the specified return value in the waiting thread's return value
+   and wakes the thread  - a sender should be confirmed to exist first */
+static void queue_release_sender(struct queue_sender **sender, void *retval)
+{
+    (*sender)->retval = retval;
+    wakeup_thread(&(*sender)->thread);
+    *sender = NULL;
+}
+
+/* Releases any waiting threads that are queued with queue_send -
+   reply with NULL */
+static void queue_release_all_senders(struct event_queue *q)
+{
+    if(q->send)
+    {
+        unsigned int i;
+        for(i = q->read; i != q->write; i++)
+        {
+            struct queue_sender **spp =
+                &q->send->senders[i & QUEUE_LENGTH_MASK];
+            if(*spp)
+            {
+                queue_release_sender(spp, NULL);
+            }
+        }
+    }
+}
+
+/* Enables queue_send on the specified queue - caller allocates the extra
+   data structure */
+void queue_enable_queue_send(struct event_queue *q,
+                             struct queue_sender_list *send)
+{
+    q->send = send;
+    memset(send, 0, sizeof(*send));
+}
+#endif /* HAVE_EXTENDED_MESSAGING_AND_NAME */
+
+
 void queue_init(struct event_queue *q, bool register_queue)
 {
-    q->read = 0;
+    q->read   = 0;
-    q->write = 0;
+    q->write  = 0;
     q->thread = NULL;
-    
+#ifdef HAVE_EXTENDED_MESSAGING_AND_NAME
-    if (register_queue)
+    q->send   = NULL; /* No message sending by default */
+#endif
+
+    if(register_queue)
     {
         /* Add it to the all_queues array */
         all_queues[num_queues++] = q;
@@ -118,6 +179,12 @@ void queue_delete(struct event_queue *q)
 
     if(found)
     {
+#ifdef HAVE_EXTENDED_MESSAGING_AND_NAME
+        /* Release waiting threads and reply to any dequeued message
+           waiting for one. */
+        queue_release_all_senders(q);
+        queue_reply(q, NULL);
+#endif
         /* Move the following queues up in the list */
         for(;i < num_queues-1;i++)
         {
@@ -130,24 +197,44 @@ void queue_delete(struct event_queue *q)
 
 void queue_wait(struct event_queue *q, struct event *ev)
 {
-    if (q->read == q->write)
+    unsigned int rd;
+
+    if(q->read == q->write)
     {
         block_thread(&q->thread);
     }
 
-    *ev = q->events[(q->read++) & QUEUE_LENGTH_MASK];
+    rd = q->read++ & QUEUE_LENGTH_MASK;
+    *ev = q->events[rd];
+
+#ifdef HAVE_EXTENDED_MESSAGING_AND_NAME
+    if(q->send && q->send->senders[rd])
+    {
+        /* Get data for a waiting thread if one */
+        queue_fetch_sender(q->send, rd);
+    }
+#endif
 }
 
 void queue_wait_w_tmo(struct event_queue *q, struct event *ev, int ticks)
 {
-    if (q->read == q->write && ticks > 0)
+    if(q->read == q->write && ticks > 0)
     {
         block_thread_w_tmo(&q->thread, ticks);
     }
 
-    if (q->read != q->write)
+    if(q->read != q->write)
     {
-        *ev = q->events[(q->read++) & QUEUE_LENGTH_MASK];
+        unsigned int rd = q->read++ & QUEUE_LENGTH_MASK;
+        *ev = q->events[rd];
+
+#ifdef HAVE_EXTENDED_MESSAGING_AND_NAME
+        if(q->send && q->send->senders[rd])
+        {
+            /* Get data for a waiting thread if one */
+            queue_fetch_sender(q->send, rd);
+        }
+#endif
     }
     else
     {
@@ -157,20 +244,81 @@ void queue_wait_w_tmo(struct event_queue *q, struct event *ev, int ticks)
 
 void queue_post(struct event_queue *q, long id, void *data)
 {
-    int wr;
+    int oldlevel = set_irq_level(HIGHEST_IRQ_LEVEL);
-    int oldlevel;
+    unsigned int wr = q->write++ & QUEUE_LENGTH_MASK;
 
-    oldlevel = set_irq_level(HIGHEST_IRQ_LEVEL);
+    q->events[wr].id   = id;
-    wr = (q->write++) & QUEUE_LENGTH_MASK;
-
-    q->events[wr].id = id;
     q->events[wr].data = data;
-    
+
+#ifdef HAVE_EXTENDED_MESSAGING_AND_NAME
+    if(q->send)
+    {
+        struct queue_sender **spp = &q->send->senders[wr];
+
+        if(*spp)
+        {
+            /* overflow protect - unblock any thread waiting at this index */
+            queue_release_sender(spp, NULL);
+        }
+    }
+#endif
+
     wakeup_thread(&q->thread);
-    
     set_irq_level(oldlevel);
 }
 
+#ifdef HAVE_EXTENDED_MESSAGING_AND_NAME
+void * queue_send(struct event_queue *q, long id, void *data)
+{
+    int oldlevel = set_irq_level(HIGHEST_IRQ_LEVEL);
+    unsigned int wr = q->write++ & QUEUE_LENGTH_MASK;
+
+    q->events[wr].id   = id;
+    q->events[wr].data = data;
+
+    if(q->send)
+    {
+        struct queue_sender **spp = &q->send->senders[wr];
+        struct queue_sender sender;
+
+        if(*spp)
+        {
+            /* overflow protect - unblock any thread waiting at this index */
+            queue_release_sender(spp, NULL);
+        }
+
+        *spp = &sender;
+        sender.thread = NULL;
+
+        wakeup_thread(&q->thread);
+        set_irq_level_and_block_thread(&sender.thread, oldlevel);
+        return sender.retval;
+    }
+
+    /* Function as queue_post if sending is not enabled */
+    wakeup_thread(&q->thread);
+    set_irq_level(oldlevel);
+    return NULL;
+}
+
+#if 0 /* not used now but probably will be later */
+/* Query if the last message dequeued was added by queue_send or not */
+bool queue_in_queue_send(struct event_queue *q)
+{
+    return q->send && q->send->curr_sender;
+}
+#endif
+
+/* Replies with retval to any dequeued message sent with queue_send */
+void queue_reply(struct event_queue *q, void *retval)
+{
+    if(q->send && q->send->curr_sender)
+    {
+        queue_release_sender(&q->send->curr_sender, retval);
+    }
+}
+#endif /* HAVE_EXTENDED_MESSAGING_AND_NAME */
+
 bool queue_empty(const struct event_queue* q)
 {
     return ( q->read == q->write );
@@ -179,6 +327,13 @@ bool queue_empty(const struct event_queue* q)
 void queue_clear(struct event_queue* q)
 {
     int oldlevel = set_irq_level(HIGHEST_IRQ_LEVEL);
+
+#ifdef HAVE_EXTENDED_MESSAGING_AND_NAME
+    /* Release all thread waiting in the queue for a reply -
+       dequeued sent message will be handled by owning thread */
+    queue_release_all_senders(q);
+#endif
+
     q->read = 0;
     q->write = 0;
     set_irq_level(oldlevel);
@@ -188,9 +343,27 @@ void queue_remove_from_head(struct event_queue *q, long id)
 {
     int oldlevel = set_irq_level(HIGHEST_IRQ_LEVEL);
     
-    while (q->read != q->write && 
+    while(q->read != q->write)
-      q->events[(q->read) & QUEUE_LENGTH_MASK].id == id)
     {
+        unsigned int rd = q->read & QUEUE_LENGTH_MASK;
+
+        if(q->events[rd].id != id)
+        {
+            break;
+        }
+
+#ifdef HAVE_EXTENDED_MESSAGING_AND_NAME
+        if(q->send)
+        {
+            struct queue_sender **spp = &q->send->senders[rd];
+
+            if(*spp)
+            {
+                /* Release any thread waiting on this message */
+                queue_release_sender(spp, NULL);
+            }
+        }
+#endif
         q->read++;
     }
     

firmware/thread.c

@@ -39,6 +39,11 @@ static unsigned short highest_priority IBSS_ATTR;
 static int boosted_threads IBSS_ATTR;
 #endif
 
+#ifdef HAVE_EXTENDED_MESSAGING_AND_NAME
+#define STAY_IRQ_LEVEL -1
+static int switch_to_irq_level = STAY_IRQ_LEVEL;
+#endif
+
 /* Define to enable additional checks for blocking violations etc. */
 #define THREAD_EXTRA_CHECKS
 
@@ -388,6 +393,18 @@ void switch_thread(bool save_context, struct thread_entry **blocked_list)
     
         /* Rearrange thread lists as needed */
         change_thread_state(blocked_list);
+
+#ifdef HAVE_EXTENDED_MESSAGING_AND_NAME
+        /* This has to be done after the scheduler is finished with the
+           blocked_list pointer so that an IRQ can't kill us by attempting
+           a wake but before attempting any core sleep. */
+        if (switch_to_irq_level != STAY_IRQ_LEVEL)
+        {
+            int level = switch_to_irq_level;
+            switch_to_irq_level = STAY_IRQ_LEVEL;
+            set_irq_level(level);
+        }
+#endif
     }
     
     /* Go through the list of sleeping task to check if we need to wake up
@@ -471,6 +488,7 @@ void block_thread(struct thread_entry **list)
     /* Set the state to blocked and ask the scheduler to switch tasks,
      * this takes us off of the run queue until we are explicitly woken */
     SET_STATE(current->statearg, STATE_BLOCKED, 0);
+
     switch_thread(true, list);
 
 #ifdef HAVE_SCHEDULER_BOOSTCTRL
@@ -522,6 +540,23 @@ void block_thread_w_tmo(struct thread_entry **list, int timeout)
     *list = NULL;
 }
 
+#if defined(HAVE_EXTENDED_MESSAGING_AND_NAME) && !defined(SIMULATOR)
+void set_irq_level_and_block_thread(struct thread_entry **list, int level)
+{
+    switch_to_irq_level = level;
+    block_thread(list);
+}
+
+#if 0
+void set_irq_level_and_block_thread_w_tmo(struct thread_entry **list,
+                                          int timeout, int level)
+{
+    switch_to_irq_level = level;
+    block_thread_w_tmo(list, timeout);
+}
+#endif
+#endif /* HAVE_EXTENDED_MESSAGING_AND_NAME */
+
 void wakeup_thread(struct thread_entry **list)
 {
     struct thread_entry *thread;

uisimulator/sdl/kernel.c

@@ -18,6 +18,7 @@
  ****************************************************************************/
 
 #include <stdlib.h>
+#include "memory.h"
 #include "uisdl.h"
 #include "kernel.h"
 #include "thread-sdl.h"
@@ -32,13 +33,71 @@ int set_irq_level (int level)
     return (_lv = level);
 }
 
+#ifdef HAVE_EXTENDED_MESSAGING_AND_NAME
+/* Moves waiting thread's descriptor to the current sender when a
+   message is dequeued */
+static void queue_fetch_sender(struct queue_sender_list *send,
+                               unsigned int i)
+{
+    int old_level = set_irq_level(15<<4);
+    struct queue_sender **spp = &send->senders[i];
+
+    if(*spp)
+    {
+        send->curr_sender = *spp;
+        *spp = NULL;
+    }
+
+    set_irq_level(old_level);
+}
+
+/* Puts the specified return value in the waiting thread's return value
+   and wakes the thread  - a sender should be confirmed to exist first */
+static void queue_release_sender(struct queue_sender **sender, void *retval)
+{
+    (*sender)->retval = retval;
+    *sender = NULL;
+}
+
+/* Releases any waiting threads that are queued with queue_send -
+   reply with NULL */
+static void queue_release_all_senders(struct event_queue *q)
+{
+    if(q->send)
+    {
+        unsigned int i;
+        for(i = q->read; i != q->write; i++)
+        {
+            struct queue_sender **spp =
+                &q->send->senders[i & QUEUE_LENGTH_MASK];
+            if(*spp)
+            {
+                queue_release_sender(spp, NULL);
+            }
+        }
+    }
+}
+
+/* Enables queue_send on the specified queue - caller allocates the extra
+   data structure */
+void queue_enable_queue_send(struct event_queue *q,
+                             struct queue_sender_list *send)
+{
+    q->send = send;
+    memset(send, 0, sizeof(*send));
+}
+#endif /* HAVE_EXTENDED_MESSAGING_AND_NAME */
+
 void queue_init(struct event_queue *q, bool register_queue)
 {
     (void)register_queue;
     
-    q->read = 0;
+    q->read   = 0;
-    q->write = 0;
+    q->write  = 0;
     q->thread = NULL;
+#ifdef HAVE_EXTENDED_MESSAGING_AND_NAME
+    q->send   = NULL; /* No message sending by default */
+#endif
 }
 
 void queue_delete(struct event_queue *q)
@@ -48,12 +107,23 @@ void queue_delete(struct event_queue *q)
 
 void queue_wait(struct event_queue *q, struct event *ev)
 {
+    unsigned int rd;
+
     while(q->read == q->write)
     {
         switch_thread(true, NULL);
     }
 
-    *ev = q->events[(q->read++) & QUEUE_LENGTH_MASK];
+    rd = q->read++ & QUEUE_LENGTH_MASK;
+    *ev = q->events[rd];
+
+#ifdef HAVE_EXTENDED_MESSAGING_AND_NAME
+    if(q->send && q->send->senders[rd])
+    {
+        /* Get data for a waiting thread if one */
+        queue_fetch_sender(q->send, rd);
+    }
+#endif
 }
 
 void queue_wait_w_tmo(struct event_queue *q, struct event *ev, int ticks)
@@ -67,7 +137,16 @@ void queue_wait_w_tmo(struct event_queue *q, struct event *ev, int ticks)
 
     if(q->read != q->write)
     {
-        *ev = q->events[(q->read++) & QUEUE_LENGTH_MASK];
+        unsigned int rd = q->read++ & QUEUE_LENGTH_MASK;
+        *ev = q->events[rd];
+
+#ifdef HAVE_EXTENDED_MESSAGING_AND_NAME
+        if(q->send && q->send->senders[rd])
+        {
+            /* Get data for a waiting thread if one */
+            queue_fetch_sender(q->send, rd);
+        }
+#endif
     }
     else
     {
@@ -77,17 +156,82 @@ void queue_wait_w_tmo(struct event_queue *q, struct event *ev, int ticks)
 
 void queue_post(struct event_queue *q, long id, void *data)
 {
-    int wr;
+    int oldlevel = set_irq_level(15<<4);
-    int oldlevel;
+    unsigned int wr = q->write++ & QUEUE_LENGTH_MASK;
 
-    oldlevel = set_irq_level(15<<4);
+    q->events[wr].id   = id;
-    wr = (q->write++) & QUEUE_LENGTH_MASK;
-
-    q->events[wr].id = id;
     q->events[wr].data = data;
+
+#ifdef HAVE_EXTENDED_MESSAGING_AND_NAME
+    if(q->send)
+    {
+        struct queue_sender **spp = &q->send->senders[wr];
+
+        if(*spp)
+        {
+            /* overflow protect - unblock any thread waiting at this index */
+            queue_release_sender(spp, NULL);
+        }
+    }
+#endif
+
     set_irq_level(oldlevel);
 }
 
+#ifdef HAVE_EXTENDED_MESSAGING_AND_NAME
+void * queue_send(struct event_queue *q, long id, void *data)
+{
+    int oldlevel = set_irq_level(15<<4);
+    unsigned int wr = q->write++ & QUEUE_LENGTH_MASK;
+
+    q->events[wr].id   = id;
+    q->events[wr].data = data;
+
+    if(q->send)
+    {
+        struct queue_sender **spp = &q->send->senders[wr];
+        struct queue_sender sender;
+
+        if(*spp)
+        {
+            /* overflow protect - unblock any thread waiting at this index */
+            queue_release_sender(spp, NULL);
+        }
+
+        *spp = &sender;
+
+        set_irq_level(oldlevel);
+        while (*spp != NULL)
+        {
+            switch_thread(true, NULL);
+        }
+
+        return sender.retval;
+    }
+
+    /* Function as queue_post if sending is not enabled */
+    set_irq_level(oldlevel);
+    return NULL;
+}
+
+#if 0 /* not used now but probably will be later */
+/* Query if the last message dequeued was added by queue_send or not */
+bool queue_in_queue_send(struct event_queue *q)
+{
+    return q->send && q->send->curr_sender;
+}
+#endif
+
+/* Replies with retval to any dequeued message sent with queue_send */
+void queue_reply(struct event_queue *q, void *retval)
+{
+    if(q->send && q->send->curr_sender)
+    {
+        queue_release_sender(&q->send->curr_sender, retval);
+    }
+}
+#endif /* HAVE_EXTENDED_MESSAGING_AND_NAME */
+
 bool queue_empty(const struct event_queue* q)
 {
     return ( q->read == q->write );
@@ -96,6 +240,11 @@ bool queue_empty(const struct event_queue* q)
 void queue_clear(struct event_queue* q)
 {
     /* fixme: This is potentially unsafe in case we do interrupt-like processing */
+#ifdef HAVE_EXTENDED_MESSAGING_AND_NAME
+    /* Release all thread waiting in the queue for a reply -
+       dequeued sent message will be handled by owning thread */
+    queue_release_all_senders(q);
+#endif
     q->read = 0;
     q->write = 0;
 }
@@ -104,9 +253,27 @@ void queue_remove_from_head(struct event_queue *q, long id)
 {
     int oldlevel = set_irq_level(15<<4);
     
-    while (q->read != q->write && 
+    while(q->read != q->write)
-      q->events[(q->read) & QUEUE_LENGTH_MASK].id == id)
     {
+        unsigned int rd = q->read & QUEUE_LENGTH_MASK;
+
+        if(q->events[rd].id != id)
+        {
+            break;
+        }
+
+#ifdef HAVE_EXTENDED_MESSAGING_AND_NAME
+        if(q->send)
+        {
+            struct queue_sender **spp = &q->send->senders[rd];
+
+            if(*spp)
+            {
+                /* Release any thread waiting on this message */
+                queue_release_sender(spp, NULL);
+            }
+        }
+#endif
         q->read++;
     }