lifemaps/qt6-migration-pipeline
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
77c67c12ca4e723492266f595401e8fdbcc00d38
qt6-migration-pipeline/external/rti_connext_dds-7.7.0/include/ndds/reda/reda_concurrentQueue.h
T
torsten 77c67c12ca Execute the same instructions but now for tools in folders you are getti 
Entire-Metadata: .entire/metadata/fb87247f-2a75-407e-b336-cbc8c20ba5a6
Entire-Session: fb87247f-2a75-407e-b336-cbc8c20ba5a6
Entire-Strategy: manual-commit
2026-09-24 16:24:51 +00:00

1152 lines
43 KiB
C++
Raw Blame History

/*
* (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 <stdio.h>
#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 .---------.
<initial state> -----------> / attached \-----+
\___________/ | attach
| ^ | read
| | | write
| +--------+
destroy |
V
<final state>
\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.<P>
{\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, &currentInfo ) == 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, &currentInfo ) == 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 */
Reference in new issue View Git Blame Copy Permalink