/* * Copyright (c) 2024 Real-Time Innovations, Inc. 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 netio_receiver_h #define netio_receiver_h #include "netio/netio_dll.h" #include "netio/netio_common.h" #ifndef netio_cap_include_manager_h #include "netio_cap/netio_cap_manager.h" #endif #ifdef __cplusplus extern "C" { #endif struct RTIClock; struct REDABuffer; struct REDAExclusiveArea; struct REDAWorker; struct REDAWeakReference; struct REDADatabase; struct RTIEventActiveObjectListener; struct RTINetioConfigurator; /* ================================================================= */ /* Component: Receiver */ /* ================================================================= */ /*e \file * \ingroup RTINetioReceiverComponent * \brief Receiver component interfaces and definitions. */ /* ================================================================= */ /* Receiver Property */ /* ================================================================= */ /*e \defgroup RTINetioReceiverPropertyClass RTINetioReceiverProperty * \ingroup RTINetioReceiverComponent * \brief Transport independent receiver property. */ /*e \ingroup RTINetioReceiverPropertyClass * \brief Transport independent property of the receiver owning the * ReceiveResources created by the transports. */ struct RTINetioReceiverProperty { /*e * bitmap specifies which property the user is specifying actively. * Those not indicated should be left as default. * Boolean property such as whether blocking receive is to be used can * be specified through just bitmap. * * @see RTI_NETIO_RECEIVER_PROPERTY_BIT_POLLED * RTI_NETIO_RECEIVER_PROPERTY_BIT_TIMESTAMP * RTI_NETIO_RECEIVER_PROPERTY_BIT_BOUND */ int bitmap; /*e Control resource growth. Helps avoid memory allocation. */ struct REDAFastBufferPoolGrowthProperty resourceGrowth; /*e Control entryport growth. Helps avoid memory allocation. */ struct REDAFastBufferPoolGrowthProperty entryportGrowth; /*e * Print warning message when processing a received message operation time * exceeds processMessageTimeBasedLoggingThreshold. * See also processMessageTimeBasedLoggingExplicitPorts. * Default: INFINITE */ RTINtpTime processMessageTimeBasedLoggingThreshold; /*e * Print warning message when processing a received message operation time * exceeds processMessageTimeBasedLoggingThreshold using a port in * processMessageTimeBasedLoggingExplicitPorts. * * Filter out the ports on which we do time-based logging. * * The ports will be separated by comma ','. They can be described using * regular expresion. For example "76*,1234". * Apply to all the ports if processMessageTimeBasedLoggingExplicitPorts is NULL. * Default: (NULL) all the ports. * * This string is: * - Allocated in: RTINetioReceiverProperty_copy(). * - Released in RTINetioReceiver_delete(). */ char *processMessageTimeBasedLoggingExplicitPorts; }; /*e \ingroup RTINetioReceiverPropertyClass * \brief If this bit of property bitmap is set, * the receiver is a polled receiver, and a blocking receiver otherwise. * * @see RTINetioReceiverProperty#bitmap */ #define RTI_NETIO_RECEIVER_PROPERTY_BIT_POLLED (0x1) /*e \ingroup RTINetioReceiverPropertyClass * \brief If this bit of property bitmap is unset, the receiver will not * timestamp messages. * * @see RTINetioReceiverProperty#bitmap */ #define RTI_NETIO_RECEIVER_PROPERTY_BIT_TIMESTAMP (0x2) /*e \ingroup RTINetioReceiverPropertyClass * \brief If this bit of property bitmap is set, the receiver will bound each * receive thread to a specific receive resource (meaning that each receive * thread will only serve one specific resource). * * @see RTINetioReceiverProperty#bitmap */ #define RTI_NETIO_RECEIVER_PROPERTY_BIT_BOUND (0x4) /*e \ingroup RTINetioReceiverPropertyClass * \brief Use to initialize the property on the stack before modifying the * relevant property. */ #define RTI_NETIO_RECEIVER_PROPERTY_DEFAULT \ { \ RTI_NETIO_RECEIVER_PROPERTY_BIT_TIMESTAMP \ | RTI_NETIO_RECEIVER_PROPERTY_BIT_BOUND, /* bitmap */ \ {16, REDA_FAST_BUFFER_POOL_UNLIMITED, REDA_FAST_BUFFER_POOL_UNLIMITED}, /* resourceGrowth */ \ {8, REDA_FAST_BUFFER_POOL_UNLIMITED, REDA_FAST_BUFFER_POOL_UNLIMITED}, /* entryportGrowth */ \ {RTI_NTP_TIME_SEC_MAX, RTI_NTP_TIME_FRAC_AUTO}, /* processMessageTimeBasedLoggingThreshold */ \ NULL /* processMessageTimeBasedLoggingExplicitPorts */ \ } #define RTINetioReceiverProperty_areThreadsBoundToRR(property__) \ (((property__)->bitmap & RTI_NETIO_RECEIVER_PROPERTY_BIT_BOUND) != 0) /* ================================================================= */ /* Receiver */ /* ================================================================= */ /*e \defgroup RTINetioReceiverClass RTINetioReceiver * \ingroup RTINetioReceiverComponent * \brief The Receiver API. * * See RTINetio_Receiver_public UML diagram. */ /*e \ingroup RTINetioReceiverClass * \brief Abstracts the transport resources that receive messages. * * It plays the role similar to that of a socket used to receive * messages in the BSD-sockets API. The RTINetioReceiver has knowledge * of (is configured with) all possible entryports that it may * use. This allows the underlying tranport plugin to reserve any resources * that will be needed to optimize preformance--some modern * high-performance transport API's such as VI-Arch require this to * establish a "virtual connection". It may also be required to enforce * CoS properties such as limited flow rates. * * 'ReceiveResources' are transport-specific because they * abstract a local address (such as a port) whose format and * interpretation depends on the transport (e.g. IPv4, IPv6 or shared * memory). Although the RTINetioReceiver logically keeps a list of * 'ReceiveResources', the receiveResources themselves "belong" * to the transport (as it the transport is the only one that knows how * to interpret them). When the receiver is asked to add an entryport, * the Receiver contacts the appropriate transport plugin(s) to * "create the ReceiveResource" and then keeps track of it. * ReceiveResources in turn remember the transport plugin to * which they belong. * * @invariant A receiver is either polled (RTINetioReceiver_isPolled) * or blocking (RTINetioReceiver_isBlocking), and ReceiveResources and * transport plugins used by the Receiver are ALL blocking or * polled, i.e. the same as the receiver. If blocking receiver, * number of threads servicing the receiver >= number of ReceiveResources * to be serviced (determined by RTINetioReceiver_getRequiredThreadCount). * * @mtsafety See RTINetio_Receiver_public UML diagram. * * @see RTINetioConfigurator */ struct RTINetioReceiver; /*e \ingroup RTINetioReceiverClass * \brief Add another way(s) for the receiver to receive message. * * A ReceiveResource is a queue of incoming messages on ONE network * interface, which may be belong to a multicast group. A ReceiveResource * is identified by the port, group, and a CoS hint. * * An Entryport is a logical group of all ReceiveResources with * the same port and group, but using different transport plugin * instances to listen on different interface(s). * * @pre Receiver enabled. * * @post receiver caches the entryport and ReceiveResources * that can serve it. The cached ReceiveResource will be dispatched * in future call (not necessarily the very next) of * RTINetioReceiver_receiveFast. If there is no transport plugin instance * agreeing to service the entryport, the receiver still * remembers the entryport in hope that a new transport * plugin instance may be added in the future that can service it. * * @param entryportWRPtrOut. \b Out. Storage for the weak reference to the * new Entryport Table record to be created, so that it can be * removed safely later. * @param receiveResourceWRPtrOut. \b Out. If not null, populated with the WR to * the RR Table record that is being created. * @param entryportIn \b In. The entryport (description). * * The \b entryportIn.locator.transport specifies the class of transport * plugin instances to which this entryport applies. * * The \b entryportIn.locator.address is an optional \b group multicast * address. It must be be INVALID if not listening to a multicast group. * If using multicast, like IPv4 multicast, should be the multicast * address to join, like 225.0.0.1. * * The \b entryportIn.locator.port may NOT be invalid. * The port must be unique in a transport plugin instance, but may * (NOT must) be duplicate across another transport plugin instance. * For example, both the IPv4 unicast and IPv4 multicast transport * plugin instances cannot bind to the same port at the same * time unless REUSEPORT setsockopt() is performed on the receiving socket. * * The \b entryportIn.cos is a hint to the transport, when creating the * underlying resource. For example, to create a special socket with * IP_TOS option, look at this parameter to decide whether to * perform setsockopt(IP_TOS). * * The \b entryportIn.aliasList is a list of transport plugin instance * aliases that should serve this request. Together with transport class * and serviceRange of a transport plugin instance, this restricts which * transport plugin instances will attempt to serve an entryport. * * Examples are: * * * @return RTI_TRUE on success, RTI_FALSE otherwise. * * @see RTINetioReceiver_receiveFast RTINetioReceiver_removeEntryport */ extern RTINetioDllExport int RTINetioReceiver_addEntryport( struct RTINetioReceiver *me, struct REDAWeakReference *entryportWRPtrOut, struct REDAWeakReference *receiveResourceWRPtrOut, const struct RTINetioLocatorInfo *entryportIn, struct REDAWorker *worker); /*e \ingroup RTINetioReceiverClass * \brief Remove one way for the receiver to receive message. * * @post The receiver will not receive on the entryport any more. * Any currently blocking receive operation is cancelled. Resources * used for this entryport is eventually reclaimed by the reaper thread. * * @param entryportWRPtr \b In. Obtained from RTINetioReceiver_addEntryport * * @return REDACursorFailReasonClass * * @see RTINetioReceiver_addEntryport */ extern RTINetioDllExport int RTINetioReceiver_removeEntryport( struct RTINetioReceiver *me, const struct REDAWeakReference *entryportWRPtrIn, struct REDAWorker *worker); /*e \ingroup RTINetioReceiverClass * \brief Check there are entry ports in use (i.e., if there is receive * resources). * * @param hasEntryPortsInUse \b Out. Cannot be NULL. If any active entry port * is found, this will be set to RTI_TRUE. * * @param transportPluginWRPtrIn \b In. If different to NULL, it will only check * entry ports for the matching transport plugin. * * @return RTI_TRUE on success, RTI_FALSE otherwise. * * @see RTINetioReceiver_addEntryport */ extern RTINetioDllExport int RTINetioReceiver_checkEntryports( struct RTINetioReceiver *me, RTIBool *hasEntryPortsInUse, const struct REDAWeakReference *transportPluginWRPtrIn, struct REDAWorker *worker); /*e * \ingroup RTINetioReceiverPropertyClass * \brief Indicates that data was read. */ #define RTI_NETIO_RECEIVER_RECEIVE_RETCODE_OK 1 /*e * \ingroup RTINetioReceiverPropertyClass * \brief Indicates that no data was read. */ #define RTI_NETIO_RECEIVER_RECEIVE_RETCODE_NO_DATA 0 /*e * \ingroup RTINetioReceiverPropertyClass * \brief Indicates that a generic error happened during the receive call. */ #define RTI_NETIO_RECEIVER_RECEIVE_RETCODE_UNBLOCK -1 /*e * \ingroup RTINetioReceiverPropertyClass * \brief Indicates that the bound receive resource was not found or that the * access to the resource failed. In this case, if there is a thread bound to * the receive resource, the RTINetioReceiver_receiveFast() caller needs to * destroy the thread. */ #define RTI_NETIO_RECEIVER_RECEIVE_RETCODE_NOT_FOUND -2 /*e * \ingroup RTINetioReceiverPropertyClass * \brief Indicates that the RTINetioReceiver_receiveFast() caller should call * to RTINetioReceiver_returnLoanAndDisownResource. */ #define RTI_NETIO_RECEIVER_RECEIVE_RETCODE_NEEDS_DISOWN -3 /*e * \ingroup RTINetioReceiverClass * \brief Receive message on previously added ReceiveResources. * * First, a receiveResource is dispatched (according to "fair" algorithm) * for the calling thread to attempt a read. * * * @pre A valid receiver with at least one ReceiveResource * previously added. * * @post A just serviced ReceiveResource will not be serviced right away * until all other ReceiveResource have been serviced equally. For * blocking receiver, the ReceiveResource will remain blocked until * explictly disowned by the caller or the next time receiveFast() * is called again with the ReceiveResource still set valid in * message received (only as a convenience). * * @param messageOut \b Out. Pre-allocated structure to contain * information about (potentially) received message. The message is * "on loan" from the transport until RTINetioMessage_finalize * is called. * * @param resourceWRInOut \b InOut. Returns a reference to the * ReceiveResource on which this message was received. * * @param bufferIn \b In. Scratch buffer to copy * (if zero-copy receive not used) received data into. * * @return RTI_NETIO_RECEIVER_RECEIVE_RETCODE_OK if data was read or a retcode * indicating what went wrong otherwise. * * @see RTINetioReceiver_addEntryport RTINetio_destroyReceiver */ extern RTINetioDllExport int RTINetioReceiver_receiveFast( struct RTINetioReceiver *me, struct RTINetioMessage *messageOut, struct REDAWeakReference *resourceWRInOut, const struct REDABuffer *bufferIn, struct REDAWorker *worker); /*e \ingroup RTINetioReceiverClass * \brief For a blocking receiver, how many threads are required to service * all the ReceiveResources simultaneously? * * @pre blocking receiver. * * @return ReceiveResource count of 0 or greater on success, -1 on failure. */ extern RTINetioDllExport int RTINetioReceiver_getRequiredThreadCount( struct RTINetioReceiver *me, struct REDAWorker *worker); /*e \ingroup RTINetioReceiverClass * * @brief Create a receiver that can handle multiple transports. * * @pre Valid args. * * @post A new receiver is ready to accept entryports for transport plugins. * * @param clockIn \b In. Optional if receiver does not timestamp. * * @param dbIn \b In. In-memory database. * * @param listenerIn \b In. Because of complex object interactions, the * receiver may NOT be deleted at any time if there is an object * that use this receiver. In that case, that object can implement * the RTIEventActiveObjectListener#onStopped method to be called back * when the receiver is ok to be destroyed. If there is no such * interaction, this argument may be NULL. * * @param configuratorIn \b In. Netio configurator object. * * @param propertyIn \b In. NULL if default property is to be used. * * @param woker \b In. Worker. * * @return NULL on failure. * * @see RTINetioReceiver_addEntryport RTINetioReceiver_requestDeletion */ extern RTINetioDllExport struct RTINetioReceiver *RTINetioReceiver_new( struct RTIClock *clockIn, struct REDADatabase *dbIn, struct RTIEventActiveObjectListener *listenerIn, struct RTINetioConfigurator *configuratorIn, const struct RTINetioReceiverProperty *propertyIn, struct RTINetioCapManager *netioCapManagerIn, struct REDAWorker *workerIn); /*e \ingroup RTINetioReceiverClass * \brief Get the receiver ready for shutdown. * This function will wake up all receive thread, to be ready for shutdown. * NOTE that this does NOT guarantee that these threads actually get to * run and get out of the receiver. * * @post Any workers blocked on receiver are signalled for wake up. * Eventually, all the ReceiveResources will be reclaimed, and the receive * threads will be stopped, at which point it is safe to call * RTINetioReceiver_shutdown. */ extern RTINetioDllExport RTIBool RTINetioReceiver_preShutdownWakeup( struct RTINetioReceiver *me, struct REDAWorker *worker); /*e \ingroup RTINetioReceiverClass * \brief Get the receiver ready for destruction. * * @pre preshutdownWakeup has been called succesfully, and it is already * in the STOPPED state. */ extern RTINetioDllExport RTIBool RTINetioReceiver_shutdown( struct RTINetioReceiver *me, struct REDAWorker *worker); /*e \ingroup RTINetioReceiverClass * \brief Destructor. * * @pre receiver stopped */ extern RTINetioDllExport void RTINetioReceiver_delete( struct RTINetioReceiver *me, struct REDAWorker *worker); /*e * \ingroup RTINetioReceiverClass * \brief Binds a thread to a receive resource. * * @param me \b In. Netio Receiver. * @param resourceRecordWR \b In. WeakReference to the receive resource * to which we are returning the message loan. * @param receiveThreadWorker \b In. Worker associated with the thread to bound * to the receive resource. * @param woker \b In. Worker. */ extern RTINetioDllExport RTIBool RTINetioReceiver_bindResource( struct RTINetioReceiver *me, struct REDAWeakReference *resourceRecordWR, struct REDAWorker *receiveThreadWorker, struct REDAWorker *worker); /*e * \ingroup RTINetioReceiverClass * \brief Done with this message. Only supported under resource bound mode. * * @pre Valid message previously passed as an out param into a successful call * of RTINetioReceiver_receiveFast. Receiver is configured in resource bound * mode. * @behavior idempotent. May be macroed for performance. * @post Any OS resource used for this message is reclaimed, and the message * should not be read. The message object itself can now be used for * another RTINetioReceiver_receiveFast. * * @param self \b In. Netio Receiver. * @param resourceRecordWR \b In. WeakReference to the receive resource * to which we are returning the message loan. * @param loanedMessage \b In. Message to return. * @param woker \b In. Worker. * * @mtsafety Called from the receive thread the receive resource is bound to. * This function does not take any EA, it is not safe to call it for the same * resource from different threads. */ extern RTINetioDllExport void RTINetioReceiver_returnLoan( struct RTINetioReceiver *self, struct REDAWeakReference *resourceRecordWR, NDDS_Transport_Message_t *loanedMessage, struct REDAWorker *worker); /*e * \ingroup RTINetioReceiverClass * \brief Notify we are done with this message and potentially disown and * destroy a ReceiveResource. * * Behavior is different depending on the mode we are operating: * 1) If we are under receive resource BOUND mode, this function will check if * the receive resource has any users (entry ports) left, and it will disown * & destroy the receive resource if that is the case (otherwise it will be * equivalent to call to RTINetioReceiver_returnLoan, with the additional * cost of taking both resource table and record EAs). * * 2) If we are NOT under receive resource BOUND mode, this function will disown * the receive resource so other workers can service it, and additionally it * will destroy the receive resource if the resource has no users (entry * ports) left. * * @pre Valid message previously passed as an out param into a successful call * of RTINetioReceiver_receiveFast. * @behavior idempotent. May be macroed for performance. * @post Any OS resource used for this message is reclaimed, and the message * should not be read. The message object itself can now be used for * another RTINetioReceiver_receiveFast. * * @param self \b In. Netio Receiver. * @param resourceRecordWR \b In. WeakReference to the receive resource * we are disowning (and to which we are returning the message loan). * @param loanedMessage \b In. Message to return. * @param woker \b In. Worker. * * @mtsafety See UML model. This function takes both resource table and record * EAs. Called from (any) receive thread. */ extern RTINetioDllExport void RTINetioReceiver_returnLoanAndDisownResource( struct RTINetioReceiver *self, struct REDAWeakReference *resourceRecordWR, NDDS_Transport_Message_t *loanedMessage, struct REDAWorker *worker); #ifdef __cplusplus } /* extern "C" */ #endif #endif /* netio_receiver_h */