rx basics

Author: pavel-kirienko

libudpard/udpard.c

@@ -36,15 +36,17 @@
 
 typedef unsigned char byte_t; ///< For compatibility with platforms where byte size is not 8 bits.
 
-#define RX_SLOT_COUNT (UDPARD_PRIORITY_MAX + 1U)
-#define BIG_BANG      INT64_MIN
+#define BIG_BANG INT64_MIN
 
 #define KILO 1000L
 #define MEGA 1000000LL
 
 /// Sessions will be garbage-collected after being idle for this long, along with unfinished transfers, if any.
 #define SESSION_LIFETIME (60 * MEGA)
 
+/// The maximum number of incoming transfers that can be in the state of incomplete reassembly simultaneously.
+#define RX_SLOT_COUNT (UDPARD_PRIORITY_MAX + 1U)
+
 #define UDP_PORT               9382U
 #define IPv4_MCAST_PREFIX      0xEF000000UL
 #define IPv4_MCAST_SUFFIX_MASK 0x007FFFFFUL
@@ -574,3 +576,85 @@ void udpard_tx_free(const udpard_tx_mem_resources_t memory, udpard_tx_item_t* co
 }
 
 // ---------------------------------------------  RX PIPELINE  ---------------------------------------------
+
+/// All but the transfer metadata: fields that change from frame to frame within the same transfer.
+typedef struct
+{
+    uint32_t           index;
+    uint32_t           offset; ///< Offset of this fragment's payload within the full transfer payload.
+    bool               end_of_transfer;
+    udpard_bytes_t     payload; ///< Also contains the transfer CRC (but not the header CRC).
+    udpard_bytes_mut_t origin;  ///< The entirety of the free-able buffer passed from the application.
+} rx_frame_base_t;
+
+/// Full frame state.
+typedef struct
+{
+    rx_frame_base_t base;
+    meta_t          meta;
+} rx_frame_t;
+
+static bool rx_validate_memory_resources(const udpard_rx_memory_resources_t memory)
+{
+    return (memory.session.alloc != NULL) && (memory.session.free != NULL) && //
+           (memory.fragment.alloc != NULL) && (memory.fragment.free != NULL);
+}
+
+/// This is designed to be convertible to/from UdpardFragment, so that the application could be
+/// given a linked list of these objects represented as a list of UdpardFragment.
+typedef struct rx_fragment_t
+{
+    udpard_fragment_t base;
+    udpard_tree_t     tree;
+} rx_fragment_t;
+
+/// Internally, the RX pipeline is arranged as follows:
+///
+/// - There is one port per subscription or an RPC-service listener. Within the port, there are N sessions,
+///   one session per remote node emitting transfers on this port (i.e., on this subject, or sending
+///   request/response of this service). Sessions are constructed dynamically in memory provided by
+///   UdpardMemoryResource.
+///
+/// - Per session, there are UDPARD_NETWORK_INTERFACE_COUNT_MAX interface states to support interface redundancy.
+///
+/// - Per interface, there are RX_SLOT_COUNT slots; a slot keeps the state of a transfer in the process of being
+///   reassembled which includes its payload fragments.
+///
+/// Port -> Session -> Interface -> Slot -> Fragments.
+///
+/// Consider the following examples, where A,B,C denote distinct multi-frame transfers:
+///
+///     A0 A1 A2 B0 B1 B2    -- two transfers without OOO frames; both accepted
+///     A2 A0 A1 B0 B2 B1    -- two transfers with OOO frames; both accepted
+///     A0 A1 B0 A2 B1 B2    -- two transfers with interleaved frames; both accepted (this is why we need 2 slots)
+///     B1 A2 A0 C0 B0 A1 C1 -- if we have only 2 slots: B evicted by C; A and C accepted, B dropped
+///     B0 A0 A1 C0 B1 A2 C1 -- ditto
+///     A0 A1 C0 B0 A2 C1 B1 -- if we have only 2 slots: A evicted by B; B and C accepted, A dropped
+///
+/// TODO: Early truncation is really easy to add since every frame carries the payload offset from the origin.
+typedef struct
+{
+    udpard_microsecond_t ts;          ///< Timestamp of the earliest frame; TIMESTAMP_UNSET upon restart.
+    uint64_t             transfer_id; ///< When first constructed, this shall be set to UINT64_MAX (unreachable value).
+    uint32_t             max_index;   ///< Maximum observed frame index in this transfer (so far); zero upon restart.
+    uint32_t             eot_index;   ///< Frame index where the EOT flag was observed; FRAME_INDEX_UNSET upon restart.
+    uint32_t             accepted_frames; ///< Number of frames accepted so far.
+    size_t               payload_size;
+    udpard_tree_t*       fragments;
+} rx_slot_t;
+
+typedef struct
+{
+    udpard_microsecond_t ts_usec; ///< The timestamp of the last valid transfer to arrive on this interface.
+    rx_slot_t            slots[RX_SLOT_COUNT];
+} rx_iface_t;
+
+/// Keep in mind that we have a dedicated session object per remote node per port; this means that the states
+/// kept here are specific per remote node, as it should be.
+typedef struct
+{
+    udpard_tree_t index_remote_uid;
+    uint64_t      remote_uid;
+    // TODO: a structure for transfer-ID storage and completeness marking.
+    rx_iface_t ifaces[UDPARD_NETWORK_INTERFACE_COUNT_MAX];
+} rx_session_t;

libudpard/udpard.h

@@ -236,6 +236,10 @@ typedef struct udpard_fragment_t
     /// The application can use this pointer to free the outer buffer after the payload has been consumed.
     udpard_bytes_mut_t origin;
 
+    /// Zero-based index and byte offset of this fragment view from the beginning of the transfer payload.
+    size_t   offset;
+    uint32_t index;
+
     /// When the fragment is no longer needed, this deleter shall be used to free the origin buffer.
     /// We provide a dedicated deleter per fragment to allow NIC drivers to manage the memory directly,
     /// which allows DMA access to the fragment data without copying.
@@ -561,17 +565,23 @@ typedef struct udpard_rx_transfer_t
     /// The total size of the payload available to the application, in bytes, is provided for convenience;
     /// it is the sum of the sizes of all its fragments. For example, if the sender emitted a transfer of 2000
     /// bytes split into two frames, 1408 bytes in the first frame and 592 bytes in the second frame,
-    /// then the payload_size will be 2000 and the payload buffer will contain two fragments of 1408 and 592 bytes.
-    /// The transfer CRC is not included here. If the received payload exceeds the configured extent,
-    /// the excess payload will be discarded and the payload_size will be set to the extent.
+    /// then the payload_size_stored will be 2000 and the payload buffer will contain two fragments of 1408 and
+    /// 592 bytes. The transfer CRC is not included here. If the received payload exceeds the configured extent,
+    /// the excess payload will be discarded and the payload_size_stored will be set to the extent.
     ///
     /// The application is given ownership of the payload buffer, so it is required to free it after use;
     /// this requires freeing both the handles and the payload buffers they point to.
     /// Beware that different memory resources may have been used to allocate the handles and the payload buffers;
     /// the application is responsible for freeing them using the correct memory resource.
     ///
     /// If the payload is empty, the corresponding buffer pointers may be NULL.
-    size_t            payload_size;
+    size_t payload_size_stored;
+
+    /// The original size of the transfer payload before extent-based truncation, in bytes.
+    /// This value is provided for informational purposes only; the application should not attempt to access
+    /// the excess payload as it has already been discarded. Cannot be less than payload_size_stored.
+    size_t payload_size_wire;
+
     udpard_fragment_t payload;
 } udpard_rx_transfer_t;