diff options
Diffstat (limited to 'cpukit/rtems/src/msg.c')
-rw-r--r-- | cpukit/rtems/src/msg.c | 524 |
1 files changed, 199 insertions, 325 deletions
diff --git a/cpukit/rtems/src/msg.c b/cpukit/rtems/src/msg.c index a69ae777c6..36c6bd7a79 100644 --- a/cpukit/rtems/src/msg.c +++ b/cpukit/rtems/src/msg.c @@ -14,19 +14,20 @@ */ #include <rtems/system.h> -#include <rtems/rtems/status.h> -#include <rtems/rtems/attr.h> +#include <rtems/sysstate.h> #include <rtems/core/chain.h> #include <rtems/core/isr.h> -#include <rtems/rtems/message.h> +#include <rtems/core/coremsg.h> #include <rtems/core/object.h> -#include <rtems/rtems/options.h> #include <rtems/core/states.h> -#include <rtems/rtems/support.h> #include <rtems/core/thread.h> #include <rtems/core/wkspace.h> #include <rtems/core/mpci.h> -#include <rtems/sysstate.h> +#include <rtems/rtems/status.h> +#include <rtems/rtems/attr.h> +#include <rtems/rtems/message.h> +#include <rtems/rtems/options.h> +#include <rtems/rtems/support.h> /*PAGE * @@ -72,56 +73,24 @@ void _Message_queue_Manager_initialization( * _Message_queue_Allocate * * Allocate a message queue and the space for its messages + * + * Input parameters: + * the_message_queue - the message queue to allocate message buffers + * count - maximum message and reserved buffer count + * max_message_size - maximum size of each message + * + * Output parameters: + * the_message_queue - set if successful, NULL otherwise */ Message_queue_Control *_Message_queue_Allocate ( - unsigned32 count, - unsigned32 max_message_size + unsigned32 count, + unsigned32 max_message_size ) { - Message_queue_Control *mq = 0; - unsigned32 message_buffering_required; - unsigned32 allocated_message_size; - - mq = - (Message_queue_Control *)_Objects_Allocate(&_Message_queue_Information); - - if (mq == 0) - goto failed; - - mq->maximum_message_size = max_message_size; - - /* - * round size up to multiple of a ptr for chain init - */ - - allocated_message_size = max_message_size; - if (allocated_message_size & (sizeof(unsigned32) - 1)) { - allocated_message_size += sizeof(unsigned32); - allocated_message_size &= ~(sizeof(unsigned32) - 1); - } - - message_buffering_required = - count * (allocated_message_size + sizeof(Message_queue_Buffer_control)); - - mq->message_buffers = - (Message_queue_Buffer *) _Workspace_Allocate(message_buffering_required); - - if (mq->message_buffers == 0) - goto failed; - - _Chain_Initialize - (&mq->Inactive_messages, - mq->message_buffers, - count, - allocated_message_size + sizeof(Message_queue_Buffer_control) - ); - return mq; + return + (Message_queue_Control *)_Objects_Allocate(&_Message_queue_Information); -failed: - if (mq) - _Message_queue_Free(mq); - return (Message_queue_Control *) 0; } /*PAGE @@ -132,11 +101,11 @@ failed: * a message queue data structure. * * Input parameters: - * name - user defined queue name - * count - maximum message and reserved buffer count + * name - user defined queue name + * count - maximum message and reserved buffer count * max_message_size - maximum size of each message - * attribute_set - process method - * id - pointer to queue + * attribute_set - process method + * id - pointer to queue * * Output parameters: * id - queue id @@ -153,11 +122,13 @@ rtems_status_code rtems_message_queue_create( ) { register Message_queue_Control *the_message_queue; + CORE_message_queue_Attributes the_message_queue_attributes; + boolean is_global; if ( !rtems_is_name_valid( name ) ) return RTEMS_INVALID_NAME; - if ( _Attributes_Is_global( attribute_set ) && + if ( (is_global = _Attributes_Is_global( attribute_set ) ) && !_System_state_Is_multiprocessing ) return RTEMS_MP_NOT_CONFIGURED; @@ -174,46 +145,52 @@ rtems_status_code rtems_message_queue_create( * and then just send smaller msgs from remote (or all) nodes. */ - if ( _Attributes_Is_global( attribute_set ) && - (_MPCI_table->maximum_packet_size < max_message_size)) - { - return RTEMS_INVALID_SIZE; - } + if ( is_global && (_MPCI_table->maximum_packet_size < max_message_size) ) + return RTEMS_INVALID_SIZE; + #endif _Thread_Disable_dispatch(); /* protects object pointer */ - the_message_queue = _Message_queue_Allocate(count, max_message_size); + the_message_queue = _Message_queue_Allocate( count, max_message_size ); if ( !the_message_queue ) { _Thread_Enable_dispatch(); return RTEMS_TOO_MANY; } - if ( _Attributes_Is_global( attribute_set ) && - !( _Objects_MP_Allocate_and_open( &_Message_queue_Information, name, - the_message_queue->Object.id, FALSE ) ) ) { + if ( is_global && + !( _Objects_MP_Allocate_and_open( &_Message_queue_Information, + name, the_message_queue->Object.id, FALSE ) ) ) { _Message_queue_Free( the_message_queue ); _Thread_Enable_dispatch(); return RTEMS_TOO_MANY; } - the_message_queue->maximum_pending_messages = count; - the_message_queue->attribute_set = attribute_set; - the_message_queue->number_of_pending_messages = 0; - _Chain_Initialize_empty( &the_message_queue->Pending_messages ); + if (_Attributes_Is_priority( attribute_set ) ) + the_message_queue_attributes.discipline = + CORE_MESSAGE_QUEUE_DISCIPLINES_PRIORITY; + else + the_message_queue_attributes.discipline = + CORE_MESSAGE_QUEUE_DISCIPLINES_FIFO; + + if ( ! _CORE_message_queue_Initialize( + &the_message_queue->message_queue, + OBJECTS_RTEMS_MESSAGE_QUEUES, + &the_message_queue_attributes, + count, + max_message_size, + _Message_queue_MP_Send_extract_proxy ) ) { + if ( is_global ) + _Objects_MP_Close( + &_Message_queue_Information, the_message_queue->Object.id); - _Thread_queue_Initialize( - &the_message_queue->Wait_queue, - OBJECTS_RTEMS_MESSAGE_QUEUES, - _Attributes_Is_priority( attribute_set ) ? - THREAD_QUEUE_DISCIPLINE_PRIORITY : THREAD_QUEUE_DISCIPLINE_FIFO, - STATES_WAITING_FOR_MESSAGE, - _Message_queue_MP_Send_extract_proxy, - RTEMS_TIMEOUT - ); + _Message_queue_Free( the_message_queue ); + _Thread_Enable_dispatch(); + return RTEMS_TOO_MANY; + } _Objects_Open( &_Message_queue_Information, @@ -223,7 +200,7 @@ rtems_status_code rtems_message_queue_create( *id = the_message_queue->Object.id; - if ( _Attributes_Is_global( attribute_set ) ) + if ( is_global ) _Message_queue_MP_Send_process_packet( MESSAGE_QUEUE_MP_ANNOUNCE_CREATE, the_message_queue->Object.id, @@ -304,14 +281,11 @@ rtems_status_code rtems_message_queue_delete( _Objects_Close( &_Message_queue_Information, &the_message_queue->Object ); - if ( the_message_queue->number_of_pending_messages != 0 ) - (void) _Message_queue_Flush_support( the_message_queue ); - else - _Thread_queue_Flush( - &the_message_queue->Wait_queue, - _Message_queue_MP_Send_object_was_deleted, - RTEMS_OBJECT_WAS_DELETED - ); + _CORE_message_queue_Close( + &the_message_queue->message_queue, + _Message_queue_MP_Send_object_was_deleted, + CORE_MESSAGE_QUEUE_STATUS_WAS_DELETED + ); _Message_queue_Free( the_message_queue ); @@ -346,6 +320,7 @@ rtems_status_code rtems_message_queue_delete( * Input parameters: * id - pointer to message queue * buffer - pointer to message buffer + * size - size of message to sent urgently * * Output parameters: * RTEMS_SUCCESSFUL - if successful @@ -358,7 +333,7 @@ rtems_status_code rtems_message_queue_send( unsigned32 size ) { - return _Message_queue_Submit(id, buffer, size, MESSAGE_QUEUE_SEND_REQUEST); + return( _Message_queue_Submit(id, buffer, size, MESSAGE_QUEUE_SEND_REQUEST) ); } /*PAGE @@ -371,10 +346,11 @@ rtems_status_code rtems_message_queue_send( * Input parameters: * id - pointer to message queue * buffer - pointer to message buffer + * size - size of message to sent urgently * * Output parameters: * RTEMS_SUCCESSFUL - if successful - * error code - if unsuccessful + * error code - if unsuccessful */ rtems_status_code rtems_message_queue_urgent( @@ -383,7 +359,7 @@ rtems_status_code rtems_message_queue_urgent( unsigned32 size ) { - return _Message_queue_Submit(id, buffer, size, MESSAGE_QUEUE_URGENT_REQUEST); + return(_Message_queue_Submit(id, buffer, size, MESSAGE_QUEUE_URGENT_REQUEST)); } /*PAGE @@ -396,11 +372,12 @@ rtems_status_code rtems_message_queue_urgent( * Input parameters: * id - pointer to message queue * buffer - pointer to message buffer + * size - size of message to broadcast * count - pointer to area to store number of threads made ready * * Output parameters: * count - number of threads made ready - * RTEMS_SUCCESSFUL - if successful + * RTEMS_SUCCESSFUL - if successful * error code - if unsuccessful */ @@ -413,8 +390,7 @@ rtems_status_code rtems_message_queue_broadcast( { register Message_queue_Control *the_message_queue; Objects_Locations location; - Thread_Control *the_thread; - unsigned32 number_broadcasted; + CORE_message_queue_Status core_status; the_message_queue = _Message_queue_Get( id, &location ); switch ( location ) { @@ -434,44 +410,21 @@ rtems_status_code rtems_message_queue_broadcast( ); case OBJECTS_LOCAL: - { - Thread_Wait_information *waitp; - unsigned32 constrained_size; - - number_broadcasted = 0; - while ( (the_thread = - _Thread_queue_Dequeue(&the_message_queue->Wait_queue)) ) { - waitp = &the_thread->Wait; - number_broadcasted += 1; - - constrained_size = size; - if (size > the_message_queue->maximum_message_size) - constrained_size = the_message_queue->maximum_message_size; - - _Message_queue_Copy_buffer(buffer, - waitp->return_argument, - constrained_size); - - *(rtems_unsigned32 *)the_thread->Wait.return_argument_1 = size; - - if ( !_Objects_Is_local_id( the_thread->Object.id ) ) { - the_thread->receive_packet->return_code = RTEMS_SUCCESSFUL; - - _Message_queue_MP_Send_response_packet( - MESSAGE_QUEUE_MP_RECEIVE_RESPONSE, - id, - the_thread - ); - } - } + core_status = _CORE_message_queue_Broadcast( + &the_message_queue->message_queue, + buffer, + size, + id, + _Message_queue_Core_message_queue_mp_support, + count + ); + _Thread_Enable_dispatch(); - *count = number_broadcasted; - return RTEMS_SUCCESSFUL; - } + return + _Message_queue_Translate_core_message_queue_return_code( core_status ); - default: - return RTEMS_INTERNAL_ERROR; } + return RTEMS_INTERNAL_ERROR; /* unreached - only to remove warnings */ } /*PAGE @@ -484,24 +437,26 @@ rtems_status_code rtems_message_queue_broadcast( * Input parameters: * id - queue id * buffer - pointer to message buffer + * size - size of message receive * option_set - options on receive * timeout - number of ticks to wait * * Output parameters: * RTEMS_SUCCESSFUL - if successful - * error code - if unsuccessful + * error code - if unsuccessful */ rtems_status_code rtems_message_queue_receive( Objects_Id id, void *buffer, - unsigned32 *size_p, + unsigned32 *size, unsigned32 option_set, rtems_interval timeout ) { register Message_queue_Control *the_message_queue; Objects_Locations location; + boolean wait; the_message_queue = _Message_queue_Get( id, &location ); switch ( location ) { @@ -514,21 +469,29 @@ rtems_status_code rtems_message_queue_receive( MESSAGE_QUEUE_MP_RECEIVE_REQUEST, id, buffer, - size_p, + size, option_set, timeout ); case OBJECTS_LOCAL: - if ( ! _Message_queue_Seize(the_message_queue, - option_set, - buffer, - size_p)) - { - _Thread_queue_Enqueue( &the_message_queue->Wait_queue, timeout ); - } + if ( _Options_Is_no_wait( option_set ) ) + wait = FALSE; + else + wait = TRUE; + + _CORE_message_queue_Seize( + &the_message_queue->message_queue, + the_message_queue->Object.id, + buffer, + size, + wait, + timeout + ); _Thread_Enable_dispatch(); - return _Thread_Executing->Wait.return_code; + return( _Message_queue_Translate_core_message_queue_return_code( + _Thread_Executing->Wait.return_code ) ); + } return RTEMS_INTERNAL_ERROR; /* unreached - only to remove warnings */ @@ -572,16 +535,13 @@ rtems_status_code rtems_message_queue_flush( MESSAGE_QUEUE_MP_FLUSH_REQUEST, id, 0, /* buffer not used */ - 0, /* size_p */ + 0, /* size */ 0, /* option_set not used */ MPCI_DEFAULT_TIMEOUT ); case OBJECTS_LOCAL: - if ( the_message_queue->number_of_pending_messages != 0 ) - *count = _Message_queue_Flush_support( the_message_queue ); - else - *count = 0; + *count = _CORE_message_queue_Flush( &the_message_queue->message_queue ); _Thread_Enable_dispatch(); return RTEMS_SUCCESSFUL; } @@ -591,125 +551,14 @@ rtems_status_code rtems_message_queue_flush( /*PAGE * - * _Message_queue_Seize - * - * This kernel routine dequeues a message, copies the message buffer to - * a given destination buffer, and frees the message buffer to the - * inactive message pool. - * - * Input parameters: - * the_message_queue - pointer to message queue - * option_set - options on receive - * the_buffer - pointer to message buffer to be filled - * - * Output parameters: - * TRUE - if message received or RTEMS_NO_WAIT and no message - * FALSE - if thread is to block - * - * NOTE: Dependent on BUFFER_LENGTH - * - * INTERRUPT LATENCY: - * available - * wait - */ - -boolean _Message_queue_Seize( - Message_queue_Control *the_message_queue, - rtems_option option_set, - void *buffer, - unsigned32 *size_p -) -{ - ISR_Level level; - Message_queue_Buffer_control *the_message; - Thread_Control *executing; - - executing = _Thread_Executing; - executing->Wait.return_code = RTEMS_SUCCESSFUL; - _ISR_Disable( level ); - if ( the_message_queue->number_of_pending_messages != 0 ) { - the_message_queue->number_of_pending_messages -= 1; - - the_message = _Message_queue_Get_pending_message( the_message_queue ); - _ISR_Enable( level ); - *size_p = the_message->Contents.size; - _Message_queue_Copy_buffer( the_message->Contents.buffer, buffer, *size_p ); - _Message_queue_Free_message_buffer(the_message_queue, the_message ); - return( TRUE ); - } - - if ( _Options_Is_no_wait( option_set ) ) { - _ISR_Enable( level ); - executing->Wait.return_code = RTEMS_UNSATISFIED; - return( TRUE ); - } - - the_message_queue->Wait_queue.sync = TRUE; - executing->Wait.queue = &the_message_queue->Wait_queue; - executing->Wait.id = the_message_queue->Object.id; - executing->Wait.option = option_set; - executing->Wait.return_argument = (void *)buffer; - executing->Wait.return_argument_1 = (void *)size_p; - _ISR_Enable( level ); - return FALSE; -} - -/*PAGE - * - * _Message_queue_Flush_support - * - * This message manager routine removes all messages from a message queue - * and returns them to the inactive message pool. - * - * Input parameters: - * the_message_queue - pointer to message queue - * - * Output parameters: - * returns - number of messages placed on inactive chain - * - * INTERRUPT LATENCY: - * only case - */ - -unsigned32 _Message_queue_Flush_support( - Message_queue_Control *the_message_queue -) -{ - ISR_Level level; - Chain_Node *inactive_first; - Chain_Node *message_queue_first; - Chain_Node *message_queue_last; - unsigned32 count; - - _ISR_Disable( level ); - inactive_first = the_message_queue->Inactive_messages.first; - message_queue_first = the_message_queue->Pending_messages.first; - message_queue_last = the_message_queue->Pending_messages.last; - - the_message_queue->Inactive_messages.first = message_queue_first; - message_queue_last->next = inactive_first; - inactive_first->previous = message_queue_last; - message_queue_first->previous = - _Chain_Head( &the_message_queue->Inactive_messages ); - - _Chain_Initialize_empty( &the_message_queue->Pending_messages ); - - count = the_message_queue->number_of_pending_messages; - the_message_queue->number_of_pending_messages = 0; - _ISR_Enable( level ); - return count; -} - -/*PAGE - * * _Message_queue_Submit * - * This routine implements the directives q_send and q_urgent. It - * processes a message that is to be submitted to the designated - * message queue. The message will either be processed as a send - * send message which it will be inserted at the rear of the queue - * or it will be processed as an urgent message which will be inserted - * at the front of the queue. + * This routine implements the directives rtems_message_queue_send + * and rtems_message_queue_urgent. It processes a message that is + * to be submitted to the designated message queue. The message will + * either be processed as a send send message which it will be inserted + * at the rear of the queue or it will be processed as an urgent message + * which will be inserted at the front of the queue. * * Input parameters: * id - pointer to message queue @@ -729,10 +578,9 @@ rtems_status_code _Message_queue_Submit( Message_queue_Submit_types submit_type ) { - register Message_queue_Control *the_message_queue; - Objects_Locations location; - Thread_Control *the_thread; - Message_queue_Buffer_control *the_message; + register Message_queue_Control *the_message_queue; + Objects_Locations location; + CORE_message_queue_Status core_status; the_message_queue = _Message_queue_Get( id, &location ); switch ( location ) @@ -766,75 +614,101 @@ rtems_status_code _Message_queue_Submit( } case OBJECTS_LOCAL: - if (size > the_message_queue->maximum_message_size) - { - _Thread_Enable_dispatch(); - return RTEMS_INVALID_SIZE; - } - - /* - * Is there a thread currently waiting on this message queue? - */ - - the_thread = _Thread_queue_Dequeue( &the_message_queue->Wait_queue ); - if ( the_thread ) - { - _Message_queue_Copy_buffer( - buffer, - the_thread->Wait.return_argument, - size - ); - *(rtems_unsigned32 *)the_thread->Wait.return_argument_1 = size; - - if ( !_Objects_Is_local_id( the_thread->Object.id ) ) { - the_thread->receive_packet->return_code = RTEMS_SUCCESSFUL; - - _Message_queue_MP_Send_response_packet( - MESSAGE_QUEUE_MP_RECEIVE_RESPONSE, - id, - the_thread - ); - - } - _Thread_Enable_dispatch(); - return RTEMS_SUCCESSFUL; - } - - /* - * No one waiting on this one currently. - * Allocate a message buffer and store it away - */ - - if ( the_message_queue->number_of_pending_messages == - the_message_queue->maximum_pending_messages ) { - _Thread_Enable_dispatch(); - return RTEMS_TOO_MANY; - } - - the_message = _Message_queue_Allocate_message_buffer(the_message_queue); - if ( the_message == 0) { - _Thread_Enable_dispatch(); - return RTEMS_UNSATISFIED; - } - - _Message_queue_Copy_buffer( buffer, the_message->Contents.buffer, size ); - the_message->Contents.size = size; - - the_message_queue->number_of_pending_messages += 1; - switch ( submit_type ) { case MESSAGE_QUEUE_SEND_REQUEST: - _Message_queue_Append( the_message_queue, the_message ); + core_status = _CORE_message_queue_Send( + &the_message_queue->message_queue, + buffer, + size, + id, + _Message_queue_Core_message_queue_mp_support + ); break; case MESSAGE_QUEUE_URGENT_REQUEST: - _Message_queue_Prepend( the_message_queue, the_message ); + core_status = _CORE_message_queue_Urgent( + &the_message_queue->message_queue, + buffer, + size, + id, + _Message_queue_Core_message_queue_mp_support + ); break; + default: + core_status = CORE_MESSAGE_QUEUE_STATUS_SUCCESSFUL; + return RTEMS_INTERNAL_ERROR; /* should never get here */ } _Thread_Enable_dispatch(); - return RTEMS_SUCCESSFUL; + return _Message_queue_Translate_core_message_queue_return_code( + core_status ); - default: - return RTEMS_INTERNAL_ERROR; /* And they were such nice boys, too! */ } + return RTEMS_INTERNAL_ERROR; /* unreached - only to remove warnings */ +} + +/*PAGE + * + * _Message_queue_Translate_core_message_queue_return_code + * + * Input parameters: + * the_message_queue_status - message_queue status code to translate + * + * Output parameters: + * rtems status code - translated RTEMS status code + * + */ + +rtems_status_code _Message_queue_Translate_core_message_queue_return_code ( + unsigned32 the_message_queue_status +) +{ + switch ( the_message_queue_status ) { + case CORE_MESSAGE_QUEUE_STATUS_SUCCESSFUL: + return RTEMS_SUCCESSFUL; + case CORE_MESSAGE_QUEUE_STATUS_INVALID_SIZE: + return RTEMS_INVALID_SIZE; + case CORE_MESSAGE_QUEUE_STATUS_TOO_MANY: + return RTEMS_TOO_MANY; + case CORE_MESSAGE_QUEUE_STATUS_UNSATISFIED: + return RTEMS_UNSATISFIED; + case CORE_MESSAGE_QUEUE_STATUS_UNSATISFIED_NOWAIT: + return RTEMS_UNSATISFIED; + case CORE_MESSAGE_QUEUE_STATUS_WAS_DELETED: + return RTEMS_OBJECT_WAS_DELETED; + case CORE_MESSAGE_QUEUE_STATUS_TIMEOUT: + return RTEMS_TIMEOUT; + case THREAD_STATUS_PROXY_BLOCKING: + return THREAD_STATUS_PROXY_BLOCKING; + } + _Internal_error_Occurred( /* XXX */ + INTERNAL_ERROR_RTEMS_API, + TRUE, + the_message_queue_status + ); + return RTEMS_INTERNAL_ERROR; /* unreached - only to remove warnings */ +} + +/*PAGE + * + * _Message_queue_Core_message_queue_mp_support + * + * Input parameters: + * the_thread - the remote thread the message was submitted to + * id - id of the message queue + * + * Output parameters: NONE + */ + +void _Message_queue_Core_message_queue_mp_support ( + Thread_Control *the_thread, + Objects_Id id +) +{ + the_thread->receive_packet->return_code = RTEMS_SUCCESSFUL; + + _Message_queue_MP_Send_response_packet( + MESSAGE_QUEUE_MP_RECEIVE_RESPONSE, + id, + the_thread + ); } |