/* * (c) Copyright, Real-Time Innovations, 2002-2016. * All rights reserved. * No duplications, whole or partial, manual or electronic, may be made * without express written permission. Any such copies, or * revisions thereof, must display this notice unaltered. * This code contains trade secrets of Real-Time Innovations, Inc. */ #ifndef reda_concurrentQueue_h #define reda_concurrentQueue_h #include #include "osapi/osapi_type.h" #include "reda/reda_dll.h" #ifdef __cplusplus extern "C" { #endif /* Doxygen file & module documentation ==================================*/ /*e \file \brief Defines the REDAConcurrentQueue */ /*e \defgroup REDAConcurrentQueueClass REDAConcurrentQueue \ingroup REDAModule \brief An implementation of a circular buffer. Appropriate for bufferered, message-oriented exchange between multiple readers and writers. The management infrastructure keeps track of the number and sizes of the messages in the buffer, the location where each individual message resides, and the size and location of available space to write new messages. In terms of the implementation, all the data and supporting structures are placed in a contigious chunk of memory. This memory can be provided as part of the initialization of the buffer so that it can be easily placed in shared memory by the application. The REDAConcurrentQueue has a management part and a data storage part. The management part enables concurrent access to the data storage. The management part includes an integer array that stores the size of each individual message in the buffer. The data storage is a contiguous memory buffer that can hold several individual messages. \verbatim Int Array +---+---+---+---+---+-- |SZ1|SZ2|SZ3| | |..... +---+---+---+---+---+-- Buffer |<--- SZ1 -->| SZ2|<---- SZ3 ----->| +------------+----+----------------+-------- | MSG1 |MSG2| MSG3 | ..... +------------+----+----------------+-------- \endverbatim The integer array stores the sizes SZ1, SZ2 and so on for messages MSG1, MSG2, respectively. The REDAConcurrentQueue management uses pointers for different locations in the buffer and indices for the integer array. The buffer pointers correspond to empty slots, slots with unread data, and slots ready for reuse. The indices correspond to the sizes of the above slots. A size of -1 is used for slot n, i.e. SZn = -1, to indicate that the corresponding buffer slot is reserved for writing data MSGn. A valid size SZn = size is used to indicate that a slot has unread data MSGn. A size SZn = -size is used to indicate that the slot can be released for reuse. The lifecycle of the queue is illustrated in the following state machine which applies to each user of the REDAConcurrentQueue \verbatim create .---------. -----------> / attached \-----+ \___________/ | attach | ^ | read | | | write | +--------+ destroy | V \endverbatim The REDAConcurrentQueue can be used by meny threads/processes to concurrently read and write. Each "user" of the queue should attach to the REDACOncurrentQueue using its own private handle. The concurent use of the queue also requires the use of two external "Exclusive Areas" (mutexes) that ensure protection of the internal critical sections. These two exclusive areas are called the ReadEA and the WriteEA. The folloing pseudo-code illustrates how a somebody writers to the queue: \code // Assume MY_MEMORY points to an allocated memory region // Assume writer attaches to an existing queue: struct REDAConcurrentQueue q; char *whereToWrite; int finishedHandle; RTIBool returnStatus; REDAConcurrentQueue_attach(q, MY_MEMORY); // Start the write. Must be done inside the WriteExclusiveArea EnterExclusiveArea(WRITE_EA); returnStatus = REDAConcurrentQueue_startWriteEA( q, &finishedHandle, &whereToWrite, NUM_BYTES_TO_WRITE, WRITER_COOKIE); LeaveExclusiveArea(WRITE_EA); // Copy the data into the handle returned in whereToWrite // This can be done outside the exclusive area memcpy(whereToWrite, MY_DATA, NUM_BYTES_TO_WRITE); // Tell the queue we are done: REDAConcurrentQueue_finishWrite(q, &finishedHandle, NUM_BYTES_TO_WRITE); \endcode The folloing pseudo-code illustrates how a somebody reads from the queue: \code // Assume MY_MEMORY points to an allocated memory region // Assume reader creates the queue: struct REDAConcurrentQueue q; int finishHandle = -1; char *readLocation = NULL; int numBytes = 0; REDAConcurrentQueue_create(q, messageCountMax, messageSizeMax, MY_MEMORY, NUM_BYTES_IN_MY_MEMORY); // Start the read. Must be done inside the WriteExclusiveArea EnterExclusiveArea(READ_EA); numBytes = REDAConcurrentQueue_startReadEA(rmb, &finishHandle, &readLocation); LeaveExclusiveArea(READ_EA); if (numBytes <= 0) { // failed. Do not read anything. // Do NOT call REDAConcurrentQueue_finishReadEA() either return ...; } // Access the data // Use readLocation[0..numBytes] // Indicate the read has completed EnterExclusiveArea(READ_EA); REDAConcurrentQueue_finishReadEA(rmb, finishHandle); LeaveExclusiveArea(READ_EA); \endcode @mtsafety This module assumes two exclusive areas: WRITE_EA and READ_EA. Most functions do not need to be invoked inside either exclusive area. The few functions that do indicate so in the documentation and have the suffix EA or EA in their name to further hint their proper use. Note that the module itself does not implement the exclusive area, rather, it assumes that the caller will implement its own mutex facility wrapping the calls that need to be inside the exclusive area. This allows the module to be kept simple and makes it easier to integrate with other modules that may have their own critical sections.

{\b Assumption}: that a write to an integer is an atomic operation and so is the reading of an integer so that if one thread writes and another reads an integer from the same location, the reader will never see an integer that wasn't written (e.g. half of the one write and the other half from teh next write...). @todo This module needs to have an additional function REDAConcurrentQueue_open(char *address) (or attach()) that allows an application to attach to a concurrent queue that is placed at the given address. We also need methods: getMessageCountMax(), getMessageSizeMax(), getMemAddressNumBytes(). This will allow an application to attach to a queue created by another aplication, get/verify the properties of the queue, and start using it. Otherwise this work must be done by the user of the message queue. @see REDAConcurrentQueue_create REDAConcurrentQueue_attach REDAConcurrentQueue_startWriteEA REDAConcurrentQueue_finishWrite REDAConcurrentQueue_startReadEA REDAConcurrentQueue_finishReadEA */ /*e \ingroup REDAConcurrentQueueClass \brief Number of bytes to align the beginning of each message. */ #define REDA_CONCURRENT_QUEUE_MSG_ALIGN_BYTES (8) /*e \ingroup REDAConcurrentQueueClass \brief Special cookie to indicate that there is no writer The value of this cookie is defined to be ZERO. Applications can assume this will be the case. There are two reasons for selecting ZERO: (1) Using zero saves the additiona logic to correct for endianess when storing/retrieving from the shared area. (2) This value is less likely to collide with most uses of the cookie (e.g. storing a PID or a timestamp) */ #define REDA_CONCURRENT_QUEUE_COOKIE_INVALID (0) /*e \ingroup REDAConcurrentQueueClass Opaque handle initialized by the _init method */ struct REDAConcurrentQueueHandle; /*e \ingroup REDAConcurrentQueueClass \brief Data structure describing the size parameters of the REDAConcurrentQueue This structure can be used to retrieve the sizing parameters that were used to create the REDAConcurrentQueue, that is the paramaters passed to REDAConcurrentQueue_create \code REDAConcurrentQueueDesc qDesc; REDAConcurrentQueue_getQueueDesc(myQueue, &qDesc); // Now you can use the contents of qDesc: if (qDesc._maxDataBytes < NEEDED_BYTES ) { ... } \endcode */ struct REDAConcurrentQueueDesc { /*e Maximum number of bytes of all messages in the buffer */ int _maxDataBytes; /*e Maximum number of bytes on a single write/read */ int _messageSizeMax; /*e Maximum number of individual messages the buffer can hold */ int _messageCountMax; /*e * The memory address passed to the create or attach call. We return * (void *) because this member gets cast back and forth between (char *) * and (struct REDAConcurrentQueueHeader *), and casting from one type to * another violates EXP39-C: * https://wiki.sei.cmu.edu/confluence/display/c/EXP39-C.+Do+not+access+a+variable+through+a+pointer+of+an+incompatible+type * . But casting to and from a (void *) is fine. */ const void *_memAddress; }; /*e \ingroup REDAConcurrentQueueClass \brief Struture used to get information in the current state of the queue This structure can be used to retrieve the state of the Queue. \code REDAConcurrentQueueStateInfo qInfo; REDAConcurrentQueue_getQueueStateInfo(myQueue, &qInfo); // Now you can use the contents of qInfo: if ( qInfo._nextToBeReadMessageIsReady ) { REDAConcurrentQueue_startReadEA(myQueue, ...) } \endcode Possible states and interpretation of the structure follows: \verbatim ( _nextToBeReadMsgIsReadyToRead == FALSE && _nextToBeReadMsgIsBeingWritten == FALSE) \endverbatim Indicates that there is no information on future messages to be read as there is currently no message either ready or in the process of being read. In this situation _nextToBeReadMsgSize, _nextToBeReadMsgWriteFinishHandle, and _nextToBeReadMsgWriteFinishHandle are filled sentinel values: _nextToBeReadMsgSize==-1, _nextToBeReadMsgWriteFinishHandle==0, _nextToBeReadMsgWriteFinishHandle==0. \verbatim ( _nextToBeReadMsgIsReadyToRead == FALSE && _nextToBeReadMsgIsBeingWritten == TRUE) \endverbatim Indicates that there is no message ready to be read (i.e. a call to REDAConcurrentQueue_startReadEA would fail) but there is one being written. In this case _nextToBeReadMsgSize, _nextToBeReadMsgWriteFinishHandle, and _nextToBeReadMsgWriteFinishHandle are filled with the proper information for that future-to-be-ready message. \verbatim ( _nextToBeReadMsgIsReadyToRead == TRUE && _nextToBeReadMsgIsBeingWritten == FALSE) \endverbatim Indicates that there is a message ready to be read (i.e. a call to REDAConcurrentQueue_startReadEA would succeed) . In this case _nextToBeReadMsgSize, _nextToBeReadMsgWriteFinishHandle, and _nextToBeReadMsgWriteFinishHandle are filled with the proper information for that ready message. \verbatim ( _haveMsgBeingRead == FALSE ) \endverbatim Indicates that there are no messages currently being read. I.e. there is no reader that has called REDAConcurrentQueue_startReadEA and has not yet called the corresponding REDAConcurrentQueue_finishReadEA In this situation _nextBeingReadMsgSize, _nextBeingReadMsgReadFinishHandle, and _nextBeingReadMsgReaderCookie are filled sentinel values: _nextBeingReadMsgSize==-1, _nextBeingReadMsgReadFinishHandle==0, _nextBeingReadMsgReaderCookie==0. In all cases _bytesWrittenCounter and _bytesFullyReadCounter are filled with the proper values. */ struct REDAConcurrentQueueStateInfo { /*e Total number of bytes ever written to the queue since it was created */ unsigned int _bytesWrittenCounter; /*e Total number of bytes ever read from the queue since it was created */ unsigned int _bytesFullyReadCounter; /*e Boolean flag indicating whether the next message in the queue is ready to be read. In other words, whether the next call to REDAConcurrentQueue_startReadEA will return a message */ RTIBool _nextToBeReadMsgIsReadyToRead; /*e Boolean flag indicating whether the next message in the queue is still being written. In other words, whether a writer has called REDAConcurrentQueue_startWriteEA but has not yet called REDAConcurrentQueue_finishWriteEA */ RTIBool _nextToBeReadMsgIsBeingWritten; /*e Size of the message that will be returned by the next call to REDAConcurrentQueue_startReadEA, assuming it is known. */ int _nextToBeReadMsgSize; /*e In the event that _nextToBeeadMsgIsBeingWritten==RTI_TRUE, this field contains the value of the finishedHandle that the writer writing that message should be used as a parameter to REDAConcurrentQueue_finishWriteEA */ int _nextToBeReadMsgWriteFinishHandle; /*e In the event that _nextToBeeadMsgIsBeingWritten==RTI_TRUE, this field contains the value of the cookie that the writer writing that message used in the call to REDAConcurrentQueue_startWriteEA */ unsigned int _nextToBeReadMsgWriterCookie; /*e Boolean flag indicating whether there is one (or more) messages in the process of being read. */ RTIBool _haveMsgBeingRead; /*e In the event that _haveMsgBeingRead==RTI_TRUE this field contains the size of the 'first' message that is still being read. "First" detemined by the order in which REDAConcurrentQueue_startReadEA was called. */ int _nextBeingReadMsgSize; /*e In the event that _haveMsgBeingRead==RTI_TRUE this field contains the finishHandle of the 'first' message that is still being read. That is the finishHandle that should be passed to the REDAConcurrentQueue_finishReadEA to indicate the message has been completely read */ int _nextBeingReadMsgReadFinishHandle; /*e In the event that _haveMsgBeingRead==RTI_TRUE this field contains the value of the cookie that the reader reading that message used in the call to REDAConcurrentQueue_startReadEA */ unsigned int _nextBeingReadMsgReaderCookie; }; struct REDAConcurrentQueueVersion { char major; char minor; }; extern REDADllVariable const struct REDAConcurrentQueueVersion REDA_CONCURRENT_QUEUE_VERSION_BEFORE_BUG_14240_FIX; extern REDADllVariable const struct REDAConcurrentQueueVersion REDA_CONCURRENT_QUEUE_VERSION_AFTER_BUG_14240_FIX; extern REDADllVariable const struct REDAConcurrentQueueVersion REDA_CONCURRENT_QUEUE_VERSION_AFTER_WRITER_COOKIE_SUPPORT; extern REDADllVariable const struct REDAConcurrentQueueVersion REDA_CONCURRENT_QUEUE_VERSION_DEFAULT; /* * Default version values for the concurrent queue. This must always be the * latest queue version. * See REDA_CONCURRENT_QUEUE_VERSION_DEFAULT. */ #define REDA_CONCURRENT_QUEUE_VERSION_DEFAULT_MAJOR (4) #define REDA_CONCURRENT_QUEUE_VERSION_DEFAULT_MINOR (0) #define REDAConcurrentQueueProperty_INITIALIZER \ { \ { \ REDA_CONCURRENT_QUEUE_VERSION_DEFAULT_MAJOR, \ REDA_CONCURRENT_QUEUE_VERSION_DEFAULT_MINOR \ } \ } struct REDAConcurrentQueueProperty { struct REDAConcurrentQueueVersion version; }; extern REDADllExport RTIBool REDAConcurrentQueue_createWithProperty( struct REDAConcurrentQueueHandle *q, int messageCountMax, int messageSizeMax, char *memAddress, int memAddressNumBytes, const struct REDAConcurrentQueueProperty *property); /*e \ingroup REDAConcurrentQueueClass \brief Creates a ConcurrentQueue and 'formats' the memory region (memAddress) to hold the contents of the queue. @pre Pre-allocated pointer to struct REDAConcurrentQueue, pointer to a an allocated memory region that is reserved for the exclusive use of the ConcurrentQueue. This memory region should be "unformatted", that is, there should not have been a previous call to REDAConcurrentQueueClass_create that used that same memory region. @post The memory pointed by memAddress has been initialized to hold the queue and "formatted" to match the requested messageCountMax and messageSizeMax. The internal pointers of q are initalized to be able to use the queue. This function should be the first REDAConcurrentQueue function called on a memory region and it should be called exactly once on each memory region. Subsequent users of the queue should call REDAConcurrentQueue_attach Either the writer or reader (or a third party) can create the queue and there can be any number of writers and readers. It is also possible for the same entity to both read and write from the queue using the same handle. The queue logic is robust to the use of any kind of memory region as long as it can be read/written using normal assignment operations. In particular it can be local memory, shared memory, a bus-based piece of memory, or even a memory-mapped region that uses some underneath to implement distributed-shared memory underneath. To account for these multi-processor situations REDACircualarQueue performs endianness checks and corrections in all its accesses to the underlying queueMemory. @param q \b InOut. "this". Must point to valid memory big enough to hold struct REDAConcurrentQueue @param messageCountMax \b In. The maximum number of messages the queue needs to buffer (i.e. messages that have been written and not yet fully read). The minimum value for messageCountMax is "1", but higher values will allow better performance as writers and readers can then be more decoupled from each other. @param messageSizeMax \b In. maximum size of an individual message that the will be written on any call to startWriteEA() @param memAddress \b In. pointer to pre-allocated memory of size memAddressNumBytes. Must be aligned to REDA_CONCURRENT_QUEUE_MSG_ALIGN_BYTES. @param memAddressNumBytes \b In. total size of the memory buffer available. This buffer will contain all the buffered messages, as well as the managemnt infra-structure. The bigger the buffer the more messages that can be accomodated. The exact value of the size is not important as REDAConcurrentQueue will transparently enforce the limits imposed by messageCountMax and memAddressNumBytes. The only impact would be performence. A useful function to properly size this buffer is REDAConcurrentQueue_getSizeRequired(messageSizeMax, messageCountMax, maxBytesBuffered). @return On success RTI_TRUE, RTI_FALSE otherwise. @exception RTI_LOG_PRECONDITION_FAILURE @see REDAConcurrentQueue_attach REDAConcurrentQueue_getSizeRequired REDAConcurrentQueue_startWriteEA */ extern REDADllExport RTIBool REDAConcurrentQueue_create( struct REDAConcurrentQueueHandle *q, int messageCountMax, int messageSizeMax, char *memAddress, int memAddressNumBytes); /*e \ingroup REDAConcurrentQueueClass \brief Makes a previously-created queue innacessible to new applications wanting to attach to it. Calling this operation will make the queue innacessible to applications trying to attach new handles to the queue. The queue, however, is not removed and existing handles attached to the queue can still be used to access the queue. This operation does not 'detach' the handle form the queue. The handle can still be used to read and write from the queue. @pre A REDAConcurrentQueueHandle attached (via create or attach) to a queue. @post Marks the queue as 'destroyed' subsequent attempts to attach to the queue will fail. Existing handles already attached to the queue can remain operational. @param q \b InOut. "this". Must point to valid memory big enough to hold struct REDAConcurrentQueue @see REDAConcurrentQueue_create REDAConcurrentQueue_attach */ extern REDADllExport RTIBool REDAConcurrentQueue_unlink(struct REDAConcurrentQueueHandle *q); /*e \ingroup REDAConcurrentQueueClass \brief Initializes the structure of the circular buffer from and existing (previously-created) queue @pre Pre-allocated pointer to struct REDAConcurrentQueue. A pointer to a memoryRegion that has been "formatted" by a prior call to REDAConcurentQueue_create @post The internal pointers of q are initialized to point to the storage indicated by memAddress and the queue parameters (num messages max message size, etc.) are initialized from the ones in the attached queue. This function should be called by all users of the REDAConcurrentQueue with the exception of the one that creates the Queue. The function will attach the passed REDAConcurrentQueue "q" parameter to the Queue stored in the memory pointed by "memAddress". This function will check that the specified memory has been formatted to contain a REDAConcurentQueue by a previous call to REDAConcurrentQueue_create and fail if that is not the case. @param q \b InOut. "this". Must point to valid memory big enough to hold struct REDAConcurrentQueue @param memAddress \b In. pointer to pre-allocated memory of size memAddressNumBytes. Must be aligned to REDA_CONCURRENT_QUEUE_MSG_ALIGN_BYTES. @return On success RTI_TRUE, RTI_FALSE otherwise. @exception RTI_LOG_PRECONDITION_FAILURE @see REDAConcurrentQueue_create REDAConcurrentQueue_getSizeRequired REDAConcurrentQueue_startWriteEA */ extern REDADllExport RTIBool REDAConcurrentQueue_attach( struct REDAConcurrentQueueHandle *q, char *memAddress); /*e \ingroup REDAConcurrentQueueClass \brief Detaches a previously-attached handle from a queue. Calling this operation will invalidate the handle and render it unusable for accessing the queue. It is the responsibility of the caller to not have any outstanding loans (read or write) on the queue. In other words, any calls to REDAConcurrentQueue_startReadEA must have been matched with corresponding subsequent calls to REDAConcurrentQueue_finishReadEA and similarly any calls to REDAConcurrentQueue_startWriteEA must have been matched with sucequent calls to REDAConcurrentQueue_finishWriteEA. This operation does not verify that the handle has no outstanding loans on the queue. If it does, those loans will remain outstanding until somebody forcefully releases them using REDAConcurrentQueue_finishReadEA or REDAConcurrentQueue_finishWriteEA with the proper finishedHandle. @pre A REDAConcurrentQueueHandle attached (via create or attach) to a queue. @post Upon successful completion the handle will be detached from the queue and it will not be usable. The application should not call any queue operations using the detached handle. The queue is otherwise not affected as it does not keep track of the number of attched handles. @param q \b InOut. "this". Must point to valid memory big enough to hold struct REDAConcurrentQueue @see REDAConcurrentQueue_create REDAConcurrentQueue_attach */ extern REDADllExport RTIBool REDAConcurrentQueue_detach(struct REDAConcurrentQueueHandle *q); /*e \ingroup REDAConcurrentQueueClass \brief Calculate the size of buffer to pass as the memAddress argument of init(). @param messageCountMax \b In. The maximum number of messages the queue needs to buffer (i.e. messages that have been written and not yet fully read). The minimum value for messageCountMax is "1", but higher values will allow better performance as writers and readers can then be more decoupled from each other. @param messageSizeMax \b In. maximum size of an individual message that the will be written on any call to startWriteEA() @param maxBytesBuffered \b In. maximum bytes the queue will hold. @param queueVersion \b In. Version of the concurrent queue. @return Bytes required to hold maxBytesBuffered of message content. The size returned includes all the overhead needed for the management of the message buffers. @exception none. @mtsafety safe @see REDAConcurrentQueue_create */ extern REDADllExport int REDAConcurrentQueue_getSizeRequired( int messageSizeMax, int messageCountMax, int maxBytesBuffered, const struct REDAConcurrentQueueVersion *queueVersion); /*e \ingroup REDAConcurrentQueueClass \brief Reserves the resources needed for the write and gives it to the caller to write to the appropriate location. The function must be used in conjunction with REDAConcurrentQueue_finishWrite(). \code int finishHandle = -1; char *writeLocation = NULL; RTIBool ok; // get resources needed to write NUM_BYTES (assume even) EnterExclusiveArea(WRITE_EA_for_rmb); ok = REDAConcurrentQueue_startWriteEA(rmb, &finishHandle, &writeLocation, NUM_BYTES); LeaveExclusiveArea(WRITE_EA_for_rmb); if (!ok) { // failed. Do not write anything. // Should not call REDAConcurrentQueue_finishWrite() either return ...; } // do the write memcpy(writeLocation, MY_DATA_PTR1, NUM_BYTES/2); memcpy(writeLocation+NUM_BYTES/2, MY_DATA_PTR2, NUM_BYTES/2); // Indicate the write has completed REDAConcurrentQueue_finishWrite(rmb, finishHandle, NUM_BYTES); \endcode @pre This function must be invoked inside the WRITE_EA exclusive area. REDAConcurrentQueue must have space to accomodate the desired size bytes. whereTo must be non-NULL. @post the bytes are written @param q \b In. Handle to the REDAConcurrentQueue @param finishedHandle \b out. A place where to save the handle to use when calling REDAConcurrentQueue_finishWrite() @param whereTo \b out. A place where to save the pointer to the memory that the caller should use to write the data. @param msgSize \b in. The total number of bytes the caller intends to write as part of the message. @param cookie \b in. A cookie to associate with the message which can be retrieved using REDAConcurrentQueues_getQueueStateInfo_ReadEA The cookie is treated as an opaque value by REDAConcurrentQueue it is only meaningful to the caller. The special value '0' is reserved to indicate an invalid cookie. @return RTI_TRUE if the function succeds in obtaining all the resources or RTI_FALSE if the function failed. The function may fail if msgSize exceeds the messageSizeMax, or if the existing data plus msgSize would exceed total space available for data (as supposed to overhead) in the queue, or if the number of messages messageCountMax are exceeded. On failure the queue is unmodified. @exception none. @mtsafety NOT safe. Caller must call this within the WRITE_EA @see REDAConcurrentQueue_finishWrite */ extern REDADllExport RTIBool REDAConcurrentQueue_startWriteEA( struct REDAConcurrentQueueHandle *q, int *finishedHandle, char **whereTo, int msgSize, unsigned int cookie); /*e \ingroup REDAConcurrentQueueClass \brief Indicates a corresponding REDAConcurrentQueue_startWriteEA() has finished and return the buffer loaned where the data has been written. @post Updates the size for that messageIndex to indicate the true size as well as the fact that the data is now valid. @param q \b In. Handle to the REDAConcurrentQueue @param finishedHandle \b In. The handle returned to the REDAConcurrentQueue_startWrite() must be of type int. It represents the message index for that message. @param msgSize \b In. The size of the message written. Must match the size passed to the corresponding call to REDAConcurrentQueue_startWriteEA() @exception none. @mtsafety safe @see REDAConcurrentQueue_startWriteEA */ extern REDADllExport void REDAConcurrentQueue_finishWrite( struct REDAConcurrentQueueHandle *q, int finishedHandle, int msgSize); /*e \ingroup REDAConcurrentQueueClass \brief Does a "zero-copy" read returning a pointer to an internal buffer containing the data in the "first" (FIFO) buffer. When the reader is done it must call REDAConcurrentQueue_finishReadEA(). This call obtains a loan of the 'data' pointer that the caller can use to access the data read. The loan is returned when finishReadEA() is called, after which the 'data' pointer may no longer be used to access the data. \code int finishHandle = -1; char *readLocation = NULL; int numBytes = 0; // get resources needed to write NUM_BYTES EnterExclusiveArea(READ_EA_for_rmb); numBytes = REDAConcurrentQueue_startReadEA(rmb, &finishHandle, &readLocation); LeaveExclusiveArea(READ_EA_for_rmb); if (numBytes <= 0) { // failed. Do not read anything. // Do NOT call REDAConcurrentQueue_finishReadEA() either return ...; } // Access the data // Use readLocation[0..numBytes] // Indicate the read has completed EnterExclusiveArea(READ_EA_for_rmb); REDAConcurrentQueue_finishReadEA(rmb, finishHandle); LeaveExclusiveArea(READ_EA_for_rmb); \endcode @pre This function must be invoked inside the READ_EA exclusive area. @param q \b in. Handle to the REDAConcurrentQueue @param finishedHandle \b out. A place where to save the handle to use when calling REDAConcurrentQueue_finishReadEA() @param whereFrom \b out. A place where to save the pointer to the memory that the caller should use to read the data. @param readerCookie \b in. A cookie to associate with the message being read which can be retrieved by REDAConcurrentQueue_getQueueStateInfo_ReadEA. The cookie is treated as an opaque value by REDAConcurrentQueue it is only meaningful to the caller. The special value '0' is reserved to indicate an invalid cookie. @return Number of bytes to read. @exception none. @mtsafety Not safe. Caller must call this function inside READ_EA exclusive area. @see REDAConcurrentQueue_finishReadEA */ extern REDADllExport int REDAConcurrentQueue_startReadEA( struct REDAConcurrentQueueHandle *q, int *finishedHandle, char **whereFrom, unsigned int readerCookie); /*e \ingroup REDAConcurrentQueueClass \brief Indicates that the prior call to REDAConcurrentQueue_startReadEA() has completed and buffer loaned can now be released. @pre This function must be invoked inside the READ_EA exclusive area. Prior successful invocation to readEA() without intervening coresponding call to finishReadEA. @pre finishedHandle is the handle returned in the corresponding call to REDAConcurrentQueue_readMsg() which corresponded to the index of the message being read. @param q \b in. Handle to the REDAConcurrentQueue @param finishedHandle \b In. The handle gotten from readEA(). @exception none. @mtsafety Not safe. Caller must call this function inside READ_EA exclusive area. @see REDAConcurrentQueue_startReadEA */ extern REDADllExport void REDAConcurrentQueue_finishReadEA( struct REDAConcurrentQueueHandle *q, int finishedHandle); /*e \ingroup REDAConcurrentQueueClass \brief Retrieves information on the sizing paramaters of the queue. @pre valid queue pointer The information retrieved matches what was used to create the queue as the size of a queue is inmmutable. @param q \b In. Handle to the REDAConcurrentQueue @param descOut \b Out. Upon return this parameter will be filled with information describing the size of the queue. @see REDAConcurrentQueue_create */ extern REDADllExport void REDAConcurrentQueue_getQueueDesc( struct REDAConcurrentQueueHandle *q, struct REDAConcurrentQueueDesc *descOut); /*e * \ingroup REDAConcurrentQueueClass * * \brief Retrieves information on the state of the queue. * * @pre valid queue pointer * * The information retrieved can be used for a variety of purposes including * detection of stuck writers and readers and flushing those stuck * writers and readers. This use-case is described in the documentation of * REDAConcurrentQueue_isWriterPotentiallyStuck * REDAConcurrentQueue_isReaderPotentiallyStuck * * @param q \b In. Handle to the REDAConcurrentQueue * @param infoOut \b Out. A REDAConcurrentQueueStateInfo object that will be * filled with the information describing the state of the queue. * * @attention If a writer crashes while calling startWriteEA, and another * writer restores the state from the backup at the same time that the reader * is calling this function, the value of * REDAConcurrentQueueStateInfo::_bytesWrittenCounter cannot be trusted, since * it may be more than the actual number of bytes written. However, once the * backup is restored, any subsequent call to this function will return the * correct value. * * @see REDAConcurrentQueue_isWriterPotentiallyStuck * REDAConcurrentQueue_isReaderPotentiallyStuck */ extern REDADllExport void REDAConcurrentQueue_getQueueStateInfoReadEA( struct REDAConcurrentQueueHandle *q, struct REDAConcurrentQueueStateInfo *infoOut); /*e \ingroup REDAConcurrentQueueClass \brief Checks whether the writer writing the 'next' message to be read has made progress since the last time the function was call. @pre valid queue pointer @post none This operation should be called periodically or whenever the application using the queue has reason to suspect that there may be a writer 'stuck' writing to the queue. The operation is useful to detect teh situation where a writer calls REDAConcurrentQueue_startWriteEA but never calls REDAConcurrentQueue_finishWrite The operation tells the caller whether there is a writer currently writing the 'next message' that would be retreived by REDAConcurrentQueue_startReadEA and whether that writer was also writing that same message the last time REDAConcurrentQueue_checkForStuckWriter was called. If this is the case, the function returns RTI_TRUE, otherwise it returns RTI_FALSE. The fact that the function returns RTI_TRUE is just an indication that the writer whose cookie is returned is potentially stuck. The ultimate responsibility from determining whether it is truly stuck rests with the caller which can either use a time-based approach (no progress beyond a certain time threshold) or a more direct approach (store the PID in the cookie and check with the Operating System that the process is still alive). If the caller determines that the writer is indeed stuck, it can then use the information stored in currentInfo to skip over the message that the stuck writer was writing. The following pseudo-code illustrates the use case: \code struct REDAConcurrentQueueStateInfo previousInfo; struct REDAConcurrentQueueStateInfo currentInfo; int readFinishedHandle; char *whereFrom; RTIBool firstTime = RTI_TRUE; while (1) { previousInfo = currentInfo; REDAConcurrentQueue_getQueueStateInfoReadEA(q, currentInfo); // The first time previousInfo is not set correctly if ( firstTime) { continue; } firstTime = RTI_FALSE; if (REDAConcurrentQueue_isWriterPotentiallyStuck( q, &previousInfo, ¤tInfo ) == RTI_TRUE ) { // Writer is potentially stuck // Make determination, for example if cookie is a PID // check that the process with that PID is still alive if ( ! STUCK ) { continue; } // We have determined it is stuck. Un-stick it and // skip the partially-written message REDAConcurrentQueue_finishWrite( q, queueStateInfo._nextToBeReadMsgWriteFinishHandle, queueStateInfo._nextToBeReadMsgSize); REDAConcurrentQueue_startReadEA(q, &readFinishedHandle, &whereFrom, MY_READER_COOKIE); REDAConcurrentQueue_finishReadEA(q, readFinishedHandle); } // wait for a while before checking again sleep(...); } \endcode @param q \b in. Handle to the REDAConcurrentQueue @param previousInfo \b in. Should be filled with the result of a previous (not the most recent) call to REDAConcurrentQueue_getQueueStateInfo_ReadEA @param currentInfo \b in. Should be filled with the result of the most recent call to REDAConcurrentQueue_getQueueStateInfo_ReadEA @return RTI_FALSE if there is no reason to think there is a stuck writer. I.e. either there is nobody writing the 'next' message to be read, or it is not the same writer or message than the previous time REDAConcurrentQueue_checkForStuckWriter_ReadEA was called. In other words, there was progress since the last call. The two cases can be distingushed by the value of the cookie which will be set to the cookie of the writer currently writing the 'next' message that would be read, or to ZERO if there is no such writer. @return RTI_TRUE if there is a writer currently writing the next message and it is the same writer and the same message as the last time REDAConcurrentQueue_checkForStuckWriter_ReadEA was called. In other words there was no progress and therefore the writer whose cookie is returned in the output parameter 'writerCookie' is potentially stuck. @mt-safety multi-thread safe. */ extern REDADllExport RTIBool REDAConcurrentQueue_isWriterPotentiallyStuck( struct REDAConcurrentQueueHandle *q, struct REDAConcurrentQueueStateInfo *previousInfo, struct REDAConcurrentQueueStateInfo *currentInfo); /*e \ingroup REDAConcurrentQueueClass \brief Checks whether there is a reader in the procees of reading a message that is potentially stuck. Ie. has called REDAConcurrentQueue_startReadEA but not REDAConcurrentQueue_finishReadEA. @pre valid queue pointer @post none This operation should be called periodically or whenever the application using the queue has reason to suspect that there may be a reader 'stuck' reading from the queue. The operation is useful to detect teh situation where a reader calls REDAConcurrentQueue_startReadEA but never calls REDAConcurrentQueue_finishReadEA. The operation tells the caller whether there is a writer currently reading a message and whether that reader was also reading that same message the last time REDAConcurrentQueue_checkForStuckReader was called. If this is the case, the function returns RTI_TRUE, otherwise it returns RTI_FALSE. The fact that the function returns RTI_TRUE is just an indication that the reader whose cookie is returned is potentially stuck. The ultimate responsibility from determining whether it is truly stuck rests with the caller which can either use a time-based approach (no progress beyond a certain time threshold) or a more direct approach (store the PID in the cookie and check with the Operating System that the process is still alive). If the caller determines that the reader is indeed stuck, it can then use the information stored in currentInfo to flush the message that the stuck reader was reading. The following pseudo-code illustrates the use case: \code struct REDAConcurrentQueueStateInfo previousInfo; struct REDAConcurrentQueueStateInfo currentInfo; RTIBool firstTime = RTI_TRUE; while (1) { previousInfo = currentInfo; REDAConcurrentQueue_getQueueStateInfoReadEA(q, currentInfo); // The first time previousInfo is not set correctly if ( firstTime) { continue; } firstTime = RTI_FALSE; if (REDAConcurrentQueue_isReaderPotentiallyStuck( q, &previousInfo, ¤tInfo ) == RTI_TRUE ) { // Reader is potentially stuck // Make determination, for example if cookie is a PID // check that the process with that PID is still alive if ( ! STUCK ) { continue; } // We have determined it is stuck. Un-stick it REDAConcurrentQueue_finishReadEA( q, queueStateInfo._nextBeinReadMsgReadFinishHandle); } // wait for a while before checking again sleep(...); } \endcode @param q \b in. Handle to the REDAConcurrentQueue @param previousInfo \b in. Should be filled with the result of a previous (not the most recent) call to REDAConcurrentQueue_getQueueStateInfo_ReadEA @param currentInfo \b in. Should be filled with the result of the most recent call to REDAConcurrentQueue_getQueueStateInfo_ReadEA */ extern REDADllExport RTIBool REDAConcurrentQueue_isReaderPotentiallyStuck( struct REDAConcurrentQueueHandle *q, struct REDAConcurrentQueueStateInfo *previousInfo, struct REDAConcurrentQueueStateInfo *currentInfo); /*e \ingroup REDAConcurrentQueueClass @brief Skips all partially read or 'ready to read' messages. @pre The caller must be sure that there are no other applications that are currently reading messages, that is, that have called readEA but have not yet called finishReadEA. This function must be invoked inside the READ_EA exclusive area. @post After successful return from this call we can have one of two possible cases. Either the queue will be empty, or else there are some messages being written and the next message to be read is not still ready (i.e. it is still being written). This operation is useful if an application wants to skip all the messages that are currently either being read, or ready to read. The typical use case is when a reader application attaches to an existing queue that was being used by readers that are no longer active. The REDAConcurentQueue has no means of determining whether any current readers are still active. This is the responsibility of the appliction calling this function. To make this determination, the caller may use the helper operations and REDAConcurrentQueue_isReaderPotentiallyStuck and REDAConcurrentQueue_getQueueStateInfoReadEA and wrap some additional logic around them. @param q \b in. Handle to the REDAConcurrentQueue @param readerCookie \b in. A cookie to associate with the messages being read which can be retrieved by REDAConcurrentQueue_getQueueStateInfo_ReadEA. The cookie is treated as an opaque value by REDAConcurrentQueue it is only meaningful to the caller. The special value '0' is reserved to indicate an invalid cookie. @exception none. @mtsafety Not safe. Caller must call this function inside READ_EA exclusive area. Moreover the caller must know that there are no other active readers that have called 'readEA' but not yet called 'finishReadEA' */ extern REDADllExport void REDAConcurrentQueue_flushReadEA( struct REDAConcurrentQueueHandle *q, int readCookie); /*e \ingroup REDAConcurrentQueueClass * * @brief Get the cookie with the (truncated) PID value of the writer. * * @param queueHandle \b in. Handle to the REDAConcurrentQueue. * * @param bufferHandle \b in. Handle to the loaned buffer. * * @return Writer cookie used to include information about the source PID * in Network Capture. */ extern REDADllExport const RTI_UINT32 *REDAConcurrentQueue_getCookieFromHandle( struct REDAConcurrentQueueHandle *queueHandle, int *bufferHandle); #ifdef __cplusplus } /* extern "C" */ #endif /* pick up hidden performance boosting macros and optimizations */ #include "reda/reda_concurrentQueue_impl.h" #endif /* reda_concurrentQueue_h */