[rtems commit] score: Fine grained locking for message queues

Sebastian Huber sebh at rtems.org
Wed May 20 07:12:03 UTC 2015


Module:    rtems
Branch:    master
Commit:    cc18d7bec7b3c5515cb9e6cd9771d4b94309b3bd
Changeset: http://git.rtems.org/rtems/commit/?id=cc18d7bec7b3c5515cb9e6cd9771d4b94309b3bd

Author:    Sebastian Huber <sebastian.huber at embedded-brains.de>
Date:      Thu Apr 30 13:12:54 2015 +0200

score: Fine grained locking for message queues

Aggregate several critical sections into a bigger one.  Sending and
receiving messages is now protected by an ISR lock.  Thread dispatching
is only disabled in case a blocking operation is necessary.  The message
copy procedure is done inside the critical section (interrupts
disabled).  Thus this change may have a negative impact on the interrupt
latency in case very large messages are transferred.

Update #2273.

---

 cpukit/posix/include/rtems/posix/mqueueimpl.h  |  15 +++
 cpukit/posix/src/mqueuerecvsupp.c              |  15 ++-
 cpukit/posix/src/mqueuesendsupp.c              |  14 ++-
 cpukit/rtems/include/rtems/rtems/messageimpl.h |  15 +++
 cpukit/rtems/src/msgqbroadcast.c               |  12 ++-
 cpukit/rtems/src/msgqflush.c                   |  13 ++-
 cpukit/rtems/src/msgqreceive.c                 |  11 ++-
 cpukit/rtems/src/msgqsend.c                    |  12 ++-
 cpukit/rtems/src/msgqurgent.c                  |  11 ++-
 cpukit/score/include/rtems/score/coremsgimpl.h | 121 ++++++++++++++++++++++---
 cpukit/score/src/coremsgbroadcast.c            |  61 +++++--------
 cpukit/score/src/coremsgclose.c                |   8 +-
 cpukit/score/src/coremsgflush.c                |   9 +-
 cpukit/score/src/coremsginsert.c               |   6 --
 cpukit/score/src/coremsgseize.c                |  30 ++++--
 cpukit/score/src/coremsgsubmit.c               |  81 +++++++++--------
 doc/user/msg.t                                 |   5 +-
 17 files changed, 303 insertions(+), 136 deletions(-)

diff --git a/cpukit/posix/include/rtems/posix/mqueueimpl.h b/cpukit/posix/include/rtems/posix/mqueueimpl.h
index bfc850d..90269bf 100644
--- a/cpukit/posix/include/rtems/posix/mqueueimpl.h
+++ b/cpukit/posix/include/rtems/posix/mqueueimpl.h
@@ -250,6 +250,21 @@ RTEMS_INLINE_ROUTINE POSIX_Message_queue_Control_fd *_POSIX_Message_queue_Get_fd
     location
   );
 }
+
+RTEMS_INLINE_ROUTINE POSIX_Message_queue_Control_fd *
+_POSIX_Message_queue_Get_fd_interrupt_disable(
+  mqd_t              id,
+  Objects_Locations *location,
+  ISR_lock_Context  *lock_context
+)
+{
+  return (POSIX_Message_queue_Control_fd *) _Objects_Get_isr_disable(
+    &_POSIX_Message_queue_Information_fds,
+    (Objects_Id)id,
+    location,
+    lock_context
+  );
+}
  
 /**
  * @see _POSIX_Name_to_id().
diff --git a/cpukit/posix/src/mqueuerecvsupp.c b/cpukit/posix/src/mqueuerecvsupp.c
index bea19fe..2f9bb2d 100644
--- a/cpukit/posix/src/mqueuerecvsupp.c
+++ b/cpukit/posix/src/mqueuerecvsupp.c
@@ -54,20 +54,25 @@ ssize_t _POSIX_Message_queue_Receive_support(
   size_t                           length_out;
   bool                             do_wait;
   Thread_Control                  *executing;
+  ISR_lock_Context                 lock_context;
 
-  the_mq_fd = _POSIX_Message_queue_Get_fd( mqdes, &location );
+  the_mq_fd = _POSIX_Message_queue_Get_fd_interrupt_disable(
+    mqdes,
+    &location,
+    &lock_context
+  );
   switch ( location ) {
 
     case OBJECTS_LOCAL:
       if ( (the_mq_fd->oflag & O_ACCMODE) == O_WRONLY ) {
-        _Objects_Put( &the_mq_fd->Object );
+        _ISR_lock_ISR_enable( &lock_context );
         rtems_set_errno_and_return_minus_one( EBADF );
       }
 
       the_mq = the_mq_fd->Queue;
 
       if ( msg_len < the_mq->Message_queue.maximum_message_size ) {
-        _Objects_Put( &the_mq_fd->Object );
+        _ISR_lock_ISR_enable( &lock_context );
         rtems_set_errno_and_return_minus_one( EMSGSIZE );
       }
 
@@ -97,10 +102,10 @@ ssize_t _POSIX_Message_queue_Receive_support(
         msg_ptr,
         &length_out,
         do_wait,
-        timeout
+        timeout,
+        &lock_context
       );
 
-      _Objects_Put( &the_mq_fd->Object );
       if (msg_prio) {
         *msg_prio = _POSIX_Message_queue_Priority_from_core(
              executing->Wait.count
diff --git a/cpukit/posix/src/mqueuesendsupp.c b/cpukit/posix/src/mqueuesendsupp.c
index 2d6ddae..d73538a 100644
--- a/cpukit/posix/src/mqueuesendsupp.c
+++ b/cpukit/posix/src/mqueuesendsupp.c
@@ -64,6 +64,7 @@ int _POSIX_Message_queue_Send_support(
   CORE_message_queue_Status       msg_status;
   bool                            do_wait;
   Thread_Control                 *executing;
+  ISR_lock_Context                lock_context;
 
   /*
    * Validate the priority.
@@ -73,12 +74,16 @@ int _POSIX_Message_queue_Send_support(
   if ( msg_prio > MQ_PRIO_MAX )
     rtems_set_errno_and_return_minus_one( EINVAL );
 
-  the_mq_fd = _POSIX_Message_queue_Get_fd( mqdes, &location );
+  the_mq_fd = _POSIX_Message_queue_Get_fd_interrupt_disable(
+    mqdes,
+    &location,
+    &lock_context
+  );
   switch ( location ) {
 
     case OBJECTS_LOCAL:
       if ( (the_mq_fd->oflag & O_ACCMODE) == O_RDONLY ) {
-        _Objects_Put( &the_mq_fd->Object );
+        _ISR_lock_ISR_enable( &lock_context );
         rtems_set_errno_and_return_minus_one( EBADF );
       }
 
@@ -105,11 +110,10 @@ int _POSIX_Message_queue_Send_support(
         NULL,
         _POSIX_Message_queue_Priority_to_core( msg_prio ),
         do_wait,
-        timeout    /* no timeout */
+        timeout,   /* no timeout */
+        &lock_context
       );
 
-      _Objects_Put( &the_mq_fd->Object );
-
       /*
        *  If we had to block, then this is where the task returns
        *  after it wakes up.  The returned status is correct for
diff --git a/cpukit/rtems/include/rtems/rtems/messageimpl.h b/cpukit/rtems/include/rtems/rtems/messageimpl.h
index fa9e573..2399d65 100644
--- a/cpukit/rtems/include/rtems/rtems/messageimpl.h
+++ b/cpukit/rtems/include/rtems/rtems/messageimpl.h
@@ -139,6 +139,21 @@ RTEMS_INLINE_ROUTINE Message_queue_Control *_Message_queue_Get (
      _Objects_Get( &_Message_queue_Information, id, location );
 }
 
+RTEMS_INLINE_ROUTINE Message_queue_Control *
+_Message_queue_Get_interrupt_disable(
+  Objects_Id         id,
+  Objects_Locations *location,
+  ISR_lock_Context  *lock_context
+)
+{
+  return (Message_queue_Control *) _Objects_Get_isr_disable(
+    &_Message_queue_Information,
+    id,
+    location,
+    lock_context
+  );
+}
+
 RTEMS_INLINE_ROUTINE Message_queue_Control *_Message_queue_Allocate( void )
 {
   return (Message_queue_Control *)
diff --git a/cpukit/rtems/src/msgqbroadcast.c b/cpukit/rtems/src/msgqbroadcast.c
index 64ea80e..aabbf3f 100644
--- a/cpukit/rtems/src/msgqbroadcast.c
+++ b/cpukit/rtems/src/msgqbroadcast.c
@@ -40,6 +40,7 @@ rtems_status_code rtems_message_queue_broadcast(
   Message_queue_Control          *the_message_queue;
   Objects_Locations               location;
   CORE_message_queue_Status       core_status;
+  ISR_lock_Context                lock_context;
 
   if ( !buffer )
     return RTEMS_INVALID_ADDRESS;
@@ -47,7 +48,11 @@ rtems_status_code rtems_message_queue_broadcast(
   if ( !count )
     return RTEMS_INVALID_ADDRESS;
 
-  the_message_queue = _Message_queue_Get( id, &location );
+  the_message_queue = _Message_queue_Get_interrupt_disable(
+    id,
+    &location,
+    &lock_context
+  );
   switch ( location ) {
 
     case OBJECTS_LOCAL:
@@ -61,10 +66,9 @@ rtems_status_code rtems_message_queue_broadcast(
                       #else
                         NULL,
                       #endif
-                      count
+                      count,
+                      &lock_context
                     );
-
-      _Objects_Put( &the_message_queue->Object );
       return
         _Message_queue_Translate_core_message_queue_return_code( core_status );
 
diff --git a/cpukit/rtems/src/msgqflush.c b/cpukit/rtems/src/msgqflush.c
index 7ae7ef4..809c243 100644
--- a/cpukit/rtems/src/msgqflush.c
+++ b/cpukit/rtems/src/msgqflush.c
@@ -54,16 +54,23 @@ rtems_status_code rtems_message_queue_flush(
 {
   Message_queue_Control          *the_message_queue;
   Objects_Locations               location;
+  ISR_lock_Context                lock_context;
 
   if ( !count )
     return RTEMS_INVALID_ADDRESS;
 
-  the_message_queue = _Message_queue_Get( id, &location );
+  the_message_queue = _Message_queue_Get_interrupt_disable(
+    id,
+    &location,
+    &lock_context
+  );
   switch ( location ) {
 
     case OBJECTS_LOCAL:
-      *count = _CORE_message_queue_Flush( &the_message_queue->message_queue );
-      _Objects_Put( &the_message_queue->Object );
+      *count = _CORE_message_queue_Flush(
+        &the_message_queue->message_queue,
+        &lock_context
+      );
       return RTEMS_SUCCESSFUL;
 
 #if defined(RTEMS_MULTIPROCESSING)
diff --git a/cpukit/rtems/src/msgqreceive.c b/cpukit/rtems/src/msgqreceive.c
index db09cfe..2b9a4e7 100644
--- a/cpukit/rtems/src/msgqreceive.c
+++ b/cpukit/rtems/src/msgqreceive.c
@@ -42,6 +42,7 @@ rtems_status_code rtems_message_queue_receive(
   Objects_Locations               location;
   bool                            wait;
   Thread_Control                 *executing;
+  ISR_lock_Context                lock_context;
 
   if ( !buffer )
     return RTEMS_INVALID_ADDRESS;
@@ -49,7 +50,11 @@ rtems_status_code rtems_message_queue_receive(
   if ( !size )
     return RTEMS_INVALID_ADDRESS;
 
-  the_message_queue = _Message_queue_Get( id, &location );
+  the_message_queue = _Message_queue_Get_interrupt_disable(
+    id,
+    &location,
+    &lock_context
+  );
   switch ( location ) {
 
     case OBJECTS_LOCAL:
@@ -66,9 +71,9 @@ rtems_status_code rtems_message_queue_receive(
         buffer,
         size,
         wait,
-        timeout
+        timeout,
+        &lock_context
       );
-      _Objects_Put( &the_message_queue->Object );
       return _Message_queue_Translate_core_message_queue_return_code(
         executing->Wait.return_code
       );
diff --git a/cpukit/rtems/src/msgqsend.c b/cpukit/rtems/src/msgqsend.c
index 34b7c29..fb3979e 100644
--- a/cpukit/rtems/src/msgqsend.c
+++ b/cpukit/rtems/src/msgqsend.c
@@ -62,11 +62,16 @@ rtems_status_code rtems_message_queue_send(
   Message_queue_Control           *the_message_queue;
   Objects_Locations                location;
   CORE_message_queue_Status        status;
+  ISR_lock_Context                 lock_context;
 
   if ( !buffer )
     return RTEMS_INVALID_ADDRESS;
 
-  the_message_queue = _Message_queue_Get( id, &location );
+  the_message_queue = _Message_queue_Get_interrupt_disable(
+    id,
+    &location,
+    &lock_context
+  );
   switch ( location ) {
 
     case OBJECTS_LOCAL:
@@ -77,11 +82,10 @@ rtems_status_code rtems_message_queue_send(
         id,
         MESSAGE_QUEUE_MP_HANDLER,
         false,   /* sender does not block */
-        0        /* no timeout */
+        0,       /* no timeout */
+        &lock_context
       );
 
-      _Objects_Put( &the_message_queue->Object );
-
       /*
        *  Since this API does not allow for blocking sends, we can directly
        *  return the returned status.
diff --git a/cpukit/rtems/src/msgqurgent.c b/cpukit/rtems/src/msgqurgent.c
index 85a9d4f..e6ae5ef 100644
--- a/cpukit/rtems/src/msgqurgent.c
+++ b/cpukit/rtems/src/msgqurgent.c
@@ -45,11 +45,16 @@ rtems_status_code rtems_message_queue_urgent(
   Message_queue_Control           *the_message_queue;
   Objects_Locations                location;
   CORE_message_queue_Status        status;
+  ISR_lock_Context                 lock_context;
 
   if ( !buffer )
     return RTEMS_INVALID_ADDRESS;
 
-  the_message_queue = _Message_queue_Get( id, &location );
+  the_message_queue = _Message_queue_Get_interrupt_disable(
+    id,
+    &location,
+    &lock_context
+  );
   switch ( location ) {
 
     case OBJECTS_LOCAL:
@@ -60,9 +65,9 @@ rtems_status_code rtems_message_queue_urgent(
         id,
         MESSAGE_QUEUE_MP_HANDLER,
         false,   /* sender does not block */
-        0        /* no timeout */
+        0,       /* no timeout */
+        &lock_context
       );
-      _Objects_Put( &the_message_queue->Object );
 
       /*
        *  Since this API does not allow for blocking sends, we can directly
diff --git a/cpukit/score/include/rtems/score/coremsgimpl.h b/cpukit/score/include/rtems/score/coremsgimpl.h
index 1f67969..51b5f37 100644
--- a/cpukit/score/include/rtems/score/coremsgimpl.h
+++ b/cpukit/score/include/rtems/score/coremsgimpl.h
@@ -21,6 +21,7 @@
 
 #include <rtems/score/coremsg.h>
 #include <rtems/score/chainimpl.h>
+#include <rtems/score/threaddispatch.h>
 #include <rtems/score/threadqimpl.h>
 
 #include <limits.h>
@@ -172,11 +173,13 @@ void _CORE_message_queue_Close(
  *  number of messages flushed from the queue is returned.
  *
  *  @param[in] the_message_queue points to the message queue to flush
+ *  @param[in] lock_context The lock context of the interrupt disable.
  *
  *  @retval This method returns the number of message pending messages flushed.
  */
 uint32_t   _CORE_message_queue_Flush(
-  CORE_message_queue_Control *the_message_queue
+  CORE_message_queue_Control *the_message_queue,
+  ISR_lock_Context           *lock_context
 );
 
 #if defined(FUNCTIONALITY_NOT_CURRENTLY_USED_BY_ANY_API)
@@ -215,6 +218,7 @@ uint32_t   _CORE_message_queue_Flush(
  *         a thread that is unblocked is actually a remote thread.
  *  @param[out] count points to the variable that will contain the
  *         number of tasks that are sent this message
+ *  @param[in] lock_context The lock context of the interrupt disable.
  *  @retval @a *count will contain the number of messages sent
  *  @retval indication of the successful completion or reason for failure
  */
@@ -224,7 +228,8 @@ CORE_message_queue_Status _CORE_message_queue_Broadcast(
   size_t                                     size,
   Objects_Id                                 id,
   CORE_message_queue_API_mp_support_callout  api_message_queue_mp_support,
-  uint32_t                                  *count
+  uint32_t                                  *count,
+  ISR_lock_Context                          *lock_context
 );
 
 /**
@@ -250,6 +255,7 @@ CORE_message_queue_Status _CORE_message_queue_Broadcast(
  *         if the message queue is full.
  *  @param[in] timeout is the maximum number of clock ticks that the calling
  *         thread is willing to block if the message queue is full.
+ *  @param[in] lock_context The lock context of the interrupt disable.
  *  @retval indication of the successful completion or reason for failure
  */
 CORE_message_queue_Status _CORE_message_queue_Submit(
@@ -261,7 +267,8 @@ CORE_message_queue_Status _CORE_message_queue_Submit(
   CORE_message_queue_API_mp_support_callout  api_message_queue_mp_support,
   CORE_message_queue_Submit_types            submit_type,
   bool                                       wait,
-  Watchdog_Interval                          timeout
+  Watchdog_Interval                          timeout,
+  ISR_lock_Context                          *lock_context
 );
 
 /**
@@ -287,6 +294,7 @@ CORE_message_queue_Status _CORE_message_queue_Submit(
  *         if the message queue is empty.
  *  @param[in] timeout is the maximum number of clock ticks that the calling
  *         thread is willing to block if the message queue is empty.
+ *  @param[in] lock_context The lock context of the interrupt disable.
  *
  *  @retval indication of the successful completion or reason for failure.
  *          On success, the location pointed to @a size_p will contain the
@@ -305,7 +313,8 @@ void _CORE_message_queue_Seize(
   void                            *buffer,
   size_t                          *size_p,
   bool                             wait,
-  Watchdog_Interval                timeout
+  Watchdog_Interval                timeout,
+  ISR_lock_Context                *lock_context
 );
 
 /**
@@ -338,8 +347,9 @@ RTEMS_INLINE_ROUTINE CORE_message_queue_Status _CORE_message_queue_Send(
   size_t                                     size,
   Objects_Id                                 id,
   CORE_message_queue_API_mp_support_callout  api_message_queue_mp_support,
-  bool                                    wait,
-  Watchdog_Interval                          timeout
+  bool                                       wait,
+  Watchdog_Interval                          timeout,
+  ISR_lock_Context                          *lock_context
 )
 {
   return _CORE_message_queue_Submit(
@@ -351,7 +361,8 @@ RTEMS_INLINE_ROUTINE CORE_message_queue_Status _CORE_message_queue_Send(
     api_message_queue_mp_support,
     CORE_MESSAGE_QUEUE_SEND_REQUEST,
     wait,     /* sender may block */
-    timeout   /* timeout interval */
+    timeout,  /* timeout interval */
+    lock_context
   );
 }
 
@@ -364,8 +375,9 @@ RTEMS_INLINE_ROUTINE CORE_message_queue_Status _CORE_message_queue_Urgent(
   size_t                                     size,
   Objects_Id                                 id,
   CORE_message_queue_API_mp_support_callout  api_message_queue_mp_support,
-  bool                                    wait,
-  Watchdog_Interval                          timeout
+  bool                                       wait,
+  Watchdog_Interval                          timeout,
+  ISR_lock_Context                          *lock_context
 )
 {
   return _CORE_message_queue_Submit(
@@ -377,10 +389,46 @@ RTEMS_INLINE_ROUTINE CORE_message_queue_Status _CORE_message_queue_Urgent(
     api_message_queue_mp_support,
     CORE_MESSAGE_QUEUE_URGENT_REQUEST,
     wait,     /* sender may block */
-    timeout   /* timeout interval */
+    timeout,  /* timeout interval */
+    lock_context
  );
 }
 
+RTEMS_INLINE_ROUTINE void _CORE_message_queue_Acquire(
+  CORE_message_queue_Control *the_message_queue,
+  ISR_lock_Context           *lock_context
+)
+{
+  _Thread_queue_Acquire( &the_message_queue->Wait_queue, lock_context );
+}
+
+RTEMS_INLINE_ROUTINE void _CORE_message_queue_Acquire_critical(
+  CORE_message_queue_Control *the_message_queue,
+  ISR_lock_Context           *lock_context
+)
+{
+  _Thread_queue_Acquire_critical( &the_message_queue->Wait_queue, lock_context );
+
+  #if defined(RTEMS_MULTIPROCESSING)
+    /*
+     * In case RTEMS_MULTIPROCESSING is enabled, then we have to prevent
+     * deletion of the executing thread after the thread queue operations.
+     */
+    _Thread_Dispatch_disable();
+  #endif
+}
+
+RTEMS_INLINE_ROUTINE void _CORE_message_queue_Release(
+  CORE_message_queue_Control *the_message_queue,
+  ISR_lock_Context           *lock_context
+)
+{
+  _Thread_queue_Release( &the_message_queue->Wait_queue, lock_context );
+  #if defined(RTEMS_MULTIPROCESSING)
+    _Thread_Dispatch_enable( _Per_CPU_Get() );
+  #endif
+}
+
 /**
  * This routine copies the contents of the source message buffer
  * to the destination message buffer.
@@ -404,7 +452,7 @@ _CORE_message_queue_Allocate_message_buffer (
 )
 {
    return (CORE_message_queue_Buffer_control *)
-     _Chain_Get( &the_message_queue->Inactive_messages );
+     _Chain_Get_unprotected( &the_message_queue->Inactive_messages );
 }
 
 /**
@@ -416,7 +464,7 @@ RTEMS_INLINE_ROUTINE void _CORE_message_queue_Free_message_buffer (
   CORE_message_queue_Buffer_control *the_message
 )
 {
-  _Chain_Append( &the_message_queue->Inactive_messages, &the_message->Node );
+  _Chain_Append_unprotected( &the_message_queue->Inactive_messages, &the_message->Node );
 }
 
 /**
@@ -510,6 +558,55 @@ RTEMS_INLINE_ROUTINE bool _CORE_message_queue_Is_priority(
            the_message_queue, the_handler, the_argument )
 #endif
 
+RTEMS_INLINE_ROUTINE Thread_Control *_CORE_message_queue_Dequeue_receiver(
+  CORE_message_queue_Control      *the_message_queue,
+  const void                      *buffer,
+  size_t                           size,
+  CORE_message_queue_Submit_types  submit_type,
+  ISR_lock_Context                *lock_context
+)
+{
+  Thread_Control *the_thread;
+
+  /*
+   *  If there are pending messages, then there can't be threads
+   *  waiting for us to send them a message.
+   *
+   *  NOTE: This check is critical because threads can block on
+   *        send and receive and this ensures that we are broadcasting
+   *        the message to threads waiting to receive -- not to send.
+   */
+  if ( the_message_queue->number_of_pending_messages != 0 ) {
+    return NULL;
+  }
+
+  /*
+   *  There must be no pending messages if there is a thread waiting to
+   *  receive a message.
+   */
+  the_thread = _Thread_queue_First_locked( &the_message_queue->Wait_queue );
+  if ( the_thread == NULL ) {
+    return NULL;
+  }
+
+   *(size_t *) the_thread->Wait.return_argument = size;
+   the_thread->Wait.count = (uint32_t) submit_type;
+
+  _CORE_message_queue_Copy_buffer(
+    buffer,
+    the_thread->Wait.return_argument_second.mutable_object,
+    size
+  );
+
+  _Thread_queue_Extract_critical(
+    &the_message_queue->Wait_queue,
+    the_thread,
+    lock_context
+  );
+
+  return the_thread;
+}
+
 /** @} */
 
 #ifdef __cplusplus
diff --git a/cpukit/score/src/coremsgbroadcast.c b/cpukit/score/src/coremsgbroadcast.c
index ff9f3ec..3ee587c 100644
--- a/cpukit/score/src/coremsgbroadcast.c
+++ b/cpukit/score/src/coremsgbroadcast.c
@@ -20,7 +20,6 @@
 
 #include <rtems/score/coremsgimpl.h>
 #include <rtems/score/objectimpl.h>
-#include <rtems/score/thread.h>
 
 CORE_message_queue_Status _CORE_message_queue_Broadcast(
   CORE_message_queue_Control                *the_message_queue,
@@ -33,55 +32,45 @@ CORE_message_queue_Status _CORE_message_queue_Broadcast(
     Objects_Id                                 id __attribute__((unused)),
     CORE_message_queue_API_mp_support_callout  api_message_queue_mp_support __attribute__((unused)),
   #endif
-  uint32_t                                  *count
+  uint32_t                                  *count,
+  ISR_lock_Context                          *lock_context
 )
 {
-  Thread_Control          *the_thread;
-  uint32_t                 number_broadcasted;
-  Thread_Wait_information *waitp;
+  Thread_Control             *the_thread;
+  uint32_t                    number_broadcasted;
 
   if ( size > the_message_queue->maximum_message_size ) {
+    _ISR_lock_ISR_enable( lock_context );
     return CORE_MESSAGE_QUEUE_STATUS_INVALID_SIZE;
   }
 
-  /*
-   *  If there are pending messages, then there can't be threads
-   *  waiting for us to send them a message.
-   *
-   *  NOTE: This check is critical because threads can block on
-   *        send and receive and this ensures that we are broadcasting
-   *        the message to threads waiting to receive -- not to send.
-   */
+  number_broadcasted = 0;
 
-  if ( the_message_queue->number_of_pending_messages != 0 ) {
-    *count = 0;
-    return CORE_MESSAGE_QUEUE_STATUS_SUCCESSFUL;
-  }
+  _CORE_message_queue_Acquire_critical( the_message_queue, lock_context );
 
-  /*
-   *  There must be no pending messages if there is a thread waiting to
-   *  receive a message.
-   */
-  number_broadcasted = 0;
-  while ((the_thread =
-          _Thread_queue_Dequeue(&the_message_queue->Wait_queue))) {
-    waitp = &the_thread->Wait;
+  while (
+    ( the_thread =
+      _CORE_message_queue_Dequeue_receiver(
+        the_message_queue,
+        buffer,
+        size,
+        0,
+        lock_context
+      )
+    )
+  ) {
     number_broadcasted += 1;
 
-    _CORE_message_queue_Copy_buffer(
-      buffer,
-      waitp->return_argument_second.mutable_object,
-      size
-    );
+#if defined(RTEMS_MULTIPROCESSING)
+    if ( !_Objects_Is_local_id( the_thread->Object.id ) )
+      (*api_message_queue_mp_support) ( the_thread, id );
+#endif
 
-    *(size_t *) the_thread->Wait.return_argument = size;
+    _CORE_message_queue_Acquire( the_message_queue, lock_context );
+  }
 
-    #if defined(RTEMS_MULTIPROCESSING)
-      if ( !_Objects_Is_local_id( the_thread->Object.id ) )
-        (*api_message_queue_mp_support) ( the_thread, id );
-    #endif
+  _CORE_message_queue_Release( the_message_queue, lock_context );
 
-  }
   *count = number_broadcasted;
   return CORE_MESSAGE_QUEUE_STATUS_SUCCESSFUL;
 }
diff --git a/cpukit/score/src/coremsgclose.c b/cpukit/score/src/coremsgclose.c
index d808a4e..e0d370d 100644
--- a/cpukit/score/src/coremsgclose.c
+++ b/cpukit/score/src/coremsgclose.c
@@ -18,11 +18,7 @@
 #include "config.h"
 #endif
 
-#include <rtems/system.h>
-#include <rtems/score/chain.h>
-#include <rtems/score/isr.h>
 #include <rtems/score/coremsgimpl.h>
-#include <rtems/score/thread.h>
 #include <rtems/score/wkspace.h>
 
 void _CORE_message_queue_Close(
@@ -31,6 +27,7 @@ void _CORE_message_queue_Close(
   uint32_t                    status
 )
 {
+  ISR_lock_Context lock_context;
 
   /*
    *  This will flush blocked threads whether they were blocked on
@@ -49,7 +46,8 @@ void _CORE_message_queue_Close(
    *  the flush satisfying any blocked senders as a side-effect.
    */
 
-  (void) _CORE_message_queue_Flush( the_message_queue );
+  _ISR_lock_ISR_disable( &lock_context );
+  (void) _CORE_message_queue_Flush( the_message_queue, &lock_context );
 
   (void) _Workspace_Free( the_message_queue->message_buffers );
 
diff --git a/cpukit/score/src/coremsgflush.c b/cpukit/score/src/coremsgflush.c
index 05683f0..f67dcf2 100644
--- a/cpukit/score/src/coremsgflush.c
+++ b/cpukit/score/src/coremsgflush.c
@@ -20,13 +20,12 @@
 #endif
 
 #include <rtems/score/coremsgimpl.h>
-#include <rtems/score/isr.h>
 
 uint32_t   _CORE_message_queue_Flush(
-  CORE_message_queue_Control *the_message_queue
+  CORE_message_queue_Control *the_message_queue,
+  ISR_lock_Context           *lock_context
 )
 {
-  ISR_Level   level;
   Chain_Node *inactive_head;
   Chain_Node *inactive_first;
   Chain_Node *message_queue_first;
@@ -60,7 +59,7 @@ uint32_t   _CORE_message_queue_Flush(
    *  fixed execution time that only deals with pending messages.
    */
 
-  _ISR_Disable( level );
+  _CORE_message_queue_Acquire_critical( the_message_queue, lock_context );
 
   count = the_message_queue->number_of_pending_messages;
   if ( count != 0 ) {
@@ -79,6 +78,6 @@ uint32_t   _CORE_message_queue_Flush(
     _Chain_Initialize_empty( &the_message_queue->Pending_messages );
   }
 
-  _ISR_Enable( level );
+  _CORE_message_queue_Release( the_message_queue, lock_context );
   return count;
 }
diff --git a/cpukit/score/src/coremsginsert.c b/cpukit/score/src/coremsginsert.c
index 28407ba..0a73af8 100644
--- a/cpukit/score/src/coremsginsert.c
+++ b/cpukit/score/src/coremsginsert.c
@@ -19,7 +19,6 @@
 #endif
 
 #include <rtems/score/coremsgimpl.h>
-#include <rtems/score/isrlevel.h>
 
 #if defined(RTEMS_SCORE_COREMSG_ENABLE_MESSAGE_PRIORITY)
 static bool _CORE_message_queue_Order(
@@ -45,7 +44,6 @@ void _CORE_message_queue_Insert_message(
 )
 {
   Chain_Control *pending_messages;
-  ISR_Level      level;
 #if defined(RTEMS_SCORE_COREMSG_ENABLE_NOTIFICATION)
   bool           notify;
 #endif
@@ -53,8 +51,6 @@ void _CORE_message_queue_Insert_message(
   _CORE_message_queue_Set_message_priority( the_message, submit_type );
   pending_messages = &the_message_queue->Pending_messages;
 
-  _ISR_Disable( level );
-
 #if defined(RTEMS_SCORE_COREMSG_ENABLE_NOTIFICATION)
   notify = ( the_message_queue->number_of_pending_messages == 0 );
 #endif
@@ -74,8 +70,6 @@ void _CORE_message_queue_Insert_message(
     _Chain_Prepend_unprotected( pending_messages, &the_message->Node );
   }
 
-  _ISR_Enable( level );
-
   #if defined(RTEMS_SCORE_COREMSG_ENABLE_NOTIFICATION)
     /*
      *  According to POSIX, does this happen before or after the message
diff --git a/cpukit/score/src/coremsgseize.c b/cpukit/score/src/coremsgseize.c
index ec6cf8c..0d1c36f 100644
--- a/cpukit/score/src/coremsgseize.c
+++ b/cpukit/score/src/coremsgseize.c
@@ -33,18 +33,17 @@ void _CORE_message_queue_Seize(
   void                            *buffer,
   size_t                          *size_p,
   bool                             wait,
-  Watchdog_Interval                timeout
+  Watchdog_Interval                timeout,
+  ISR_lock_Context                *lock_context
 )
 {
-  ISR_lock_Context                   lock_context;
   CORE_message_queue_Buffer_control *the_message;
 
   executing->Wait.return_code = CORE_MESSAGE_QUEUE_STATUS_SUCCESSFUL;
-  _Thread_queue_Acquire( &the_message_queue->Wait_queue, &lock_context );
+  _CORE_message_queue_Acquire_critical( the_message_queue, lock_context );
   the_message = _CORE_message_queue_Get_pending_message( the_message_queue );
   if ( the_message != NULL ) {
     the_message_queue->number_of_pending_messages -= 1;
-    _Thread_queue_Release( &the_message_queue->Wait_queue, &lock_context );
 
     *size_p = the_message->Contents.size;
     executing->Wait.count =
@@ -61,6 +60,7 @@ void _CORE_message_queue_Seize(
        *  So return immediately.
        */
       _CORE_message_queue_Free_message_buffer(the_message_queue, the_message);
+      _CORE_message_queue_Release( the_message_queue, lock_context );
       return;
     #else
     {
@@ -73,12 +73,15 @@ void _CORE_message_queue_Seize(
        *  NOTE: If we note that the queue was not full before this receive,
        *  then we can avoid this dequeue.
        */
-      the_thread = _Thread_queue_Dequeue( &the_message_queue->Wait_queue );
-      if ( !the_thread ) {
+      the_thread = _Thread_queue_First_locked(
+        &the_message_queue->Wait_queue
+      );
+      if ( the_thread == NULL ) {
         _CORE_message_queue_Free_message_buffer(
           the_message_queue,
           the_message
         );
+        _CORE_message_queue_Release( the_message_queue, lock_context );
         return;
       }
 
@@ -103,13 +106,21 @@ void _CORE_message_queue_Seize(
          the_message,
          _CORE_message_queue_Get_message_priority( the_message )
       );
+      _Thread_queue_Extract_critical(
+        &the_message_queue->Wait_queue,
+        the_thread,
+        lock_context
+      );
+      #if defined(RTEMS_MULTIPROCESSING)
+        _Thread_Dispatch_enable( _Per_CPU_Get() );
+      #endif
       return;
     }
     #endif
   }
 
   if ( !wait ) {
-    _Thread_queue_Release( &the_message_queue->Wait_queue, &lock_context );
+    _CORE_message_queue_Release( the_message_queue, lock_context );
     executing->Wait.return_code = CORE_MESSAGE_QUEUE_STATUS_UNSATISFIED_NOWAIT;
     return;
   }
@@ -125,6 +136,9 @@ void _CORE_message_queue_Seize(
     STATES_WAITING_FOR_MESSAGE,
     timeout,
     CORE_MESSAGE_QUEUE_STATUS_TIMEOUT,
-    &lock_context
+    lock_context
   );
+  #if defined(RTEMS_MULTIPROCESSING)
+    _Thread_Dispatch_enable( _Per_CPU_Get() );
+  #endif
 }
diff --git a/cpukit/score/src/coremsgsubmit.c b/cpukit/score/src/coremsgsubmit.c
index 0d0965f..37f8579 100644
--- a/cpukit/score/src/coremsgsubmit.c
+++ b/cpukit/score/src/coremsgsubmit.c
@@ -38,36 +38,39 @@ CORE_message_queue_Status _CORE_message_queue_Submit(
   #endif
   CORE_message_queue_Submit_types            submit_type,
   bool                                       wait,
-  Watchdog_Interval                          timeout
+  Watchdog_Interval                          timeout,
+  ISR_lock_Context                          *lock_context
 )
 {
-  CORE_message_queue_Buffer_control   *the_message;
-  Thread_Control                      *the_thread;
+  CORE_message_queue_Buffer_control *the_message;
+  Thread_Control                    *the_thread;
 
   if ( size > the_message_queue->maximum_message_size ) {
+    _ISR_lock_ISR_enable( lock_context );
     return CORE_MESSAGE_QUEUE_STATUS_INVALID_SIZE;
   }
 
+  _CORE_message_queue_Acquire_critical( the_message_queue, lock_context );
+
   /*
    *  Is there a thread currently waiting on this message queue?
    */
-  if ( the_message_queue->number_of_pending_messages == 0 ) {
-    the_thread = _Thread_queue_Dequeue( &the_message_queue->Wait_queue );
-    if ( the_thread ) {
-      _CORE_message_queue_Copy_buffer(
-        buffer,
-        the_thread->Wait.return_argument_second.mutable_object,
-        size
-      );
-      *(size_t *) the_thread->Wait.return_argument = size;
-      the_thread->Wait.count = (uint32_t) submit_type;
-
-      #if defined(RTEMS_MULTIPROCESSING)
-        if ( !_Objects_Is_local_id( the_thread->Object.id ) )
-          (*api_message_queue_mp_support) ( the_thread, id );
-      #endif
-      return CORE_MESSAGE_QUEUE_STATUS_SUCCESSFUL;
-    }
+
+  the_thread = _CORE_message_queue_Dequeue_receiver(
+    the_message_queue,
+    buffer,
+    size,
+    submit_type,
+    lock_context
+  );
+  if ( the_thread != NULL ) {
+    #if defined(RTEMS_MULTIPROCESSING)
+      if ( !_Objects_Is_local_id( the_thread->Object.id ) )
+        (*api_message_queue_mp_support) ( the_thread, id );
+
+      _Thread_Dispatch_enable( _Per_CPU_Get() );
+    #endif
+    return CORE_MESSAGE_QUEUE_STATUS_SUCCESSFUL;
   }
 
   /*
@@ -77,23 +80,25 @@ CORE_message_queue_Status _CORE_message_queue_Submit(
   the_message =
       _CORE_message_queue_Allocate_message_buffer( the_message_queue );
   if ( the_message ) {
+    the_message->Contents.size = size;
+    _CORE_message_queue_Set_message_priority( the_message, submit_type );
     _CORE_message_queue_Copy_buffer(
       buffer,
       the_message->Contents.buffer,
       size
     );
-    the_message->Contents.size = size;
-    _CORE_message_queue_Set_message_priority( the_message, submit_type );
 
     _CORE_message_queue_Insert_message(
        the_message_queue,
        the_message,
        submit_type
     );
+    _CORE_message_queue_Release( the_message_queue, lock_context );
     return CORE_MESSAGE_QUEUE_STATUS_SUCCESSFUL;
   }
 
   #if !defined(RTEMS_SCORE_COREMSG_ENABLE_BLOCKING_SEND)
+    _CORE_message_queue_Release( the_message_queue, lock_context );
     return CORE_MESSAGE_QUEUE_STATUS_TOO_MANY;
   #else
     /*
@@ -102,6 +107,7 @@ CORE_message_queue_Status _CORE_message_queue_Submit(
      *  on the queue.
      */
     if ( !wait ) {
+      _CORE_message_queue_Release( the_message_queue, lock_context );
       return CORE_MESSAGE_QUEUE_STATUS_TOO_MANY;
     }
 
@@ -110,6 +116,7 @@ CORE_message_queue_Status _CORE_message_queue_Submit(
      *  deadly to block in an ISR.
      */
     if ( _ISR_Is_in_progress() ) {
+      _CORE_message_queue_Release( the_message_queue, lock_context );
       return CORE_MESSAGE_QUEUE_STATUS_UNSATISFIED;
     }
 
@@ -119,20 +126,22 @@ CORE_message_queue_Status _CORE_message_queue_Submit(
      *  it as a variable.  Doing this emphasizes how dangerous it
      *  would be to use this variable prior to here.
      */
-    {
-      executing->Wait.id = id;
-      executing->Wait.return_argument_second.immutable_object = buffer;
-      executing->Wait.option = (uint32_t) size;
-      executing->Wait.count = submit_type;
-
-      _Thread_queue_Enqueue(
-        &the_message_queue->Wait_queue,
-        executing,
-        STATES_WAITING_FOR_MESSAGE,
-        timeout,
-        CORE_MESSAGE_QUEUE_STATUS_TIMEOUT
-      );
-    }
+    executing->Wait.id = id;
+    executing->Wait.return_argument_second.immutable_object = buffer;
+    executing->Wait.option = (uint32_t) size;
+    executing->Wait.count = submit_type;
+
+    _Thread_queue_Enqueue_critical(
+      &the_message_queue->Wait_queue,
+      executing,
+      STATES_WAITING_FOR_MESSAGE,
+      timeout,
+      CORE_MESSAGE_QUEUE_STATUS_TIMEOUT,
+      lock_context
+    );
+    #if defined(RTEMS_MULTIPROCESSING)
+      _Thread_Dispatch_enable( _Per_CPU_Get() );
+    #endif
 
     return CORE_MESSAGE_QUEUE_STATUS_UNSATISFIED_WAIT;
   #endif
diff --git a/doc/user/msg.t b/doc/user/msg.t
index f58d677..eb3cb35 100644
--- a/doc/user/msg.t
+++ b/doc/user/msg.t
@@ -50,7 +50,10 @@ wait for a message to arrive at a queue.  Also, a task may poll
 a queue for the arrival of a message.
 
 The maximum length message which can be sent is set
-on a per message queue basis.
+on a per message queue basis.  The message content must be copied in general
+to/from an internal buffer of the message queue or directly to a peer in
+certain cases.  This copy operation is performed with interrupts disabled.  So
+it is advisable to keep the messages as short as possible.
 
 @subsection Building a Message Queue Attribute Set
 




More information about the vc mailing list