WEBRC congestion control (RFC 3738), unverified: no conformance claimed

CMakeLists.txt

@@ -50,7 +50,7 @@ endif()
 
 target_sources(flute
   PRIVATE
-  src/Receiver.cpp src/Transmitter.cpp src/AlcPacket.cpp src/File.cpp src/EncodingSymbol.cpp src/FileDeliveryTable.cpp src/IpSec.cpp
+  src/Receiver.cpp src/Transmitter.cpp src/AlcPacket.cpp src/File.cpp src/EncodingSymbol.cpp src/FileDeliveryTable.cpp src/IpSec.cpp src/Webrc.cpp
     utils/base64.cpp
   PUBLIC
     include/Receiver.h include/Transmitter.h include/File.h

include/AlcPacket.h

@@ -16,11 +16,31 @@
 #pragma once
 #include <stddef.h>
 #include <stdint.h>
+#include <optional>
 #include <vector>
 #include "flute_types.h"
 #include "EncodingSymbol.h"
 
 namespace LibFlute {
+
+ /**
+  *  Short-format Congestion Control Information, RFC 3738 clause 5.1.
+  *
+  *  Thirty-two bits, which is the width the LCT header's C=0 selects, so carrying it costs nothing
+  *  the header did not already spend. Absent, the field is sent as zeros, which is what a session
+  *  with no congestion control building block sends and what TS 26.346 clause 7.2.7 requires of a
+  *  3GPP session: "The length of the CCI (Congestion Control Identifier) field shall be 32 bits and
+  *  it is assigned a value of zero (C=0)."
+  */
+  struct CongestionControlInfo {
+    /** CTSI. "CTSI indicates the index of the current time slot." */
+    uint8_t current_time_slot_index = 0;
+    /** CN. "CN for the base channel is T, and the CNs for the wave channels are 0 through T-1." */
+    uint8_t channel_number = 0;
+    /** PSN. "The PSN of each packet is scoped by its CN value." */
+    uint16_t packet_sequence_number = 0;
+  };
+
   /**
    *  A class for parsing and creating ALC packets
    */
@@ -47,7 +67,8 @@ namespace LibFlute {
       *  @param close_object_flag Set the LCT Close Object flag (RFC 3451 clause 5.1, 'B' bit) on this packet
       */
       AlcPacket(uint64_t tsi, uint16_t toi, FecOti fec_oti, const std::vector<EncodingSymbol>& symbols, size_t max_size, uint32_t fdt_instance_id,
-                bool close_session_flag = false, bool close_object_flag = false);
+                bool close_session_flag = false, bool close_object_flag = false,
+                const std::optional<CongestionControlInfo>& cci = std::nullopt);
 
      /**
       *  Default destructor.
@@ -74,6 +95,16 @@ namespace LibFlute {
       */
       size_t header_length() const  { return _lct_header.lct_header_len * 4; };
 
+     /**
+      *  Congestion Control Information carried by a received packet, if the field held one.
+      *
+      *  A session running no building block sends the field as zeros, which reads back as a
+      *  well-formed value naming channel 0 with sequence number 0. Only a receiver that knows the
+      *  session runs WEBRC should interpret it, which is why this is a plain accessor rather than
+      *  something the parser acts on. Absent for a packet this object built for sending.
+      */
+      std::optional<CongestionControlInfo> congestion_control_info() const { return _cci; };
+
      /**
       *  Get the FDT instance ID
       */
@@ -136,6 +167,7 @@ namespace LibFlute {
       ContentEncoding _content_encoding = ContentEncoding::NONE;
       FecOti _fec_oti = {};
       bool _has_fti = false;
+      std::optional<CongestionControlInfo> _cci;
 
       char* _buffer = nullptr;
       size_t _len;

include/Receiver.h

@@ -18,11 +18,13 @@
 #include <boost/bind/bind.hpp>
 #include <atomic>
 #include <memory>
+#include <set>
 #include <string>
 #include <map>
 #include <mutex>
 #include "File.h"
 #include "FileDeliveryTable.h"
+#include "Webrc.h"
 
 namespace LibFlute {
   /**
@@ -62,7 +64,9 @@ namespace LibFlute {
       Receiver( const std::string& iface, const std::string& address,
           short port, uint64_t tsi,
           boost::asio::io_context& io_context,
-          const std::string& source_address = "");
+          const std::string& source_address = "",
+          Profile profile = Profile::Ts26517,
+          const std::optional<Webrc::SessionChannels>& webrc = std::nullopt);
 
      /**
       *  Destructor. Marks the receiver as no longer alive so that any async_receive_from
@@ -115,6 +119,28 @@ namespace LibFlute {
       void register_close_notification_callback(close_notification_callback_t cb) { _close_cb = cb; };
 
       void stop() { _running = false; }
+
+     /**
+      *  Join an additional multicast channel of this session, or leave one.
+      *
+      *  A multiple rate congestion control building block adjusts a receiver's rate by changing
+      *  which of a session's channels it is joined to. RFC 5775 clause 2.1: "An ALC session
+      *  comprises multiple channels originating at a single sender". Nothing in the library calls
+      *  these yet: they are the capability such a building block needs, and are usable on their own
+      *  for a session announced on more than one address.
+      *
+      *  The interface, and the source where the session is source-specific, are those the Receiver
+      *  was constructed with. Joining a group already joined, or leaving one not joined, does
+      *  nothing and reports false.
+      *
+      *  @param group Multicast group to join or leave, of the same family as the session.
+      *  @return True if the membership actually changed.
+      */
+      bool join_channel(const std::string& group);
+      bool leave_channel(const std::string& group);
+
+     /** Groups currently joined, including the one given at construction. */
+      std::set<std::string> joined_channels() const { return _joined_groups; };
     private:
 
       void handle_receive_from(const boost::system::error_code& error,
@@ -148,7 +174,28 @@ namespace LibFlute {
        *  the per-packet check costs no parsing. Empty for an any-source session. */
       std::optional<boost::asio::ip::address> _expected_source;
 
+      Profile _profile = Profile::Ts26517;
+      /** The building block, present only where the session runs one. */
+      std::unique_ptr<Webrc::ReceiverController> _webrc;
+      Webrc::SessionChannels _webrc_channels;
+      Webrc::Derived _webrc_derived{};
+      /** Last packet sequence number seen per channel number, for detecting loss. */
+      std::vector<std::optional<uint16_t>> _webrc_last_psn;
+      std::unique_ptr<boost::asio::steady_timer> _webrc_epoch_timer;
+      void start_webrc_epoch_timer();
+      void on_webrc_epoch();
+      void note_webrc_packet(const AlcPacket& alc);
+
+      /** Join or leave one multicast group, source-specific or not, on the session's interface. */
+      bool set_group_membership(const boost::asio::ip::address& group, bool join);
+
       std::string _mcast_address;
+      /** Retained so a channel joined after construction uses the same interface and, where the
+       *  session is source-specific, the same source. */
+      std::string _iface;
+      std::string _ssm_source;
+      /** Groups currently joined, the constructor's own among them. */
+      std::set<std::string> _joined_groups;
 
       completion_callback_t _completion_cb = nullptr;
       close_notification_callback_t _close_cb = nullptr;

include/Transmitter.h

@@ -20,13 +20,17 @@
 #include <atomic>
 #include <chrono>
 #include <queue>
+#include <utility>
+#include <vector>
+#include <memory>
 #include <string>
 #include <map>
 #include <set>
 #include <mutex>
 #include <optional>
 //#include "File.h"
 #include "AlcPacket.h"
+#include "Webrc.h"
 #include "FileDeliveryTable.h"
 
 namespace LibFlute {
@@ -539,6 +543,63 @@ namespace LibFlute {
       */
       const boost::asio::ip::udp::endpoint &endpoint() const { return _endpoint; };
 
+     /**
+      *  Add a further channel to this session, or remove one.
+      *
+      *  A multiple rate congestion control building block sends to several channels at different
+      *  rates and lets each receiver choose how many it is joined to. RFC 5775 clause 2.1: "An ALC
+      *  session comprises multiple channels originating at a single sender". Nothing in the library
+      *  drives these yet; they are the capability such a building block needs.
+      *
+      *  Channel 0 is always the one given at construction, WEBRC's base channel. An added channel
+      *  takes the same socket options and source-address binding.
+      *
+      *  @param address Destination address for the new channel.
+      *  @param port Destination port.
+      *  @return Index of the new channel, or 0 on failure; 0 is never an added channel.
+      */
+      size_t add_channel(const std::string& address, unsigned short port);
+
+     /**
+      *  Run the WEBRC congestion control building block over this session's channels.
+      *
+      *  RFC 5775 clause 2.2: "At a minimum, implementations of ALC MUST support [RFC3738]." This is
+      *  the sender half. Packets are distributed over a base channel and T wave channels according
+      *  to the schedule of RFC 3738 clause 3.1, and each carries the Congestion Control Information
+      *  of clause 5.1 so a receiver can tell the channels apart.
+      *
+      *  Refused under a 3GPP profile: TS 26.346 clause 7.2.4 excludes congestion control for MBMS
+      *  download and clause 7.2.7 fixes the CCI at a 32-bit zero.
+      *
+      *  @param params WEBRC inputs; the derived cycle length T decides how many addresses are needed.
+      *  @param wave_channel_addresses One address per wave channel, T of them, each distinct from
+      *         the session's own address, which serves as the base channel.
+      *  @throws std::runtime_error under a 3GPP profile, or if the wrong number of addresses is
+      *          given, so a misconfigured session fails at setup rather than on the wire.
+      */
+      void enable_webrc(const Webrc::Parameters& params,
+                        const std::vector<std::pair<std::string, unsigned short>>& wave_channel_addresses);
+
+     /** Whether the WEBRC building block is running on this session. */
+      bool webrc_enabled() const { return _webrc.has_value(); };
+
+     /** The Congestion Control Information the next packet on a channel would carry, for tests. */
+      std::optional<CongestionControlInfo> webrc_cci_for(size_t channel_index) const;
+
+     /**
+      *  Remove a channel added by add_channel(). Channel 0 cannot be removed. Indices above the one
+      *  removed shift down, so remove from the highest index first when removing several.
+      */
+      bool remove_channel(size_t index);
+
+     /** Number of channels in the session, always at least 1. */
+      size_t channel_count() const { return 1 + _extra_channels.size(); };
+
+     /** Destination of a channel by index; 0 is the one given at construction. */
+      const boost::asio::ip::udp::endpoint &channel_endpoint(size_t index) const {
+        return index == 0 ? _endpoint : _extra_channels.at(index - 1).endpoint;
+      };
+
      /**
       * Set UDP Address for FLUTE session
       *
@@ -719,6 +780,34 @@ namespace LibFlute {
        *  what may be signalled, so it cannot change once a session is running. */
       Profile _profile;
       boost::asio::ip::udp::socket _socket;
+
+      /** One further channel of the session, beyond the one given at construction. */
+      struct Channel {
+        boost::asio::ip::udp::endpoint endpoint;
+        std::unique_ptr<boost::asio::ip::udp::socket> socket;
+      };
+      std::vector<Channel> _extra_channels;
+
+      /** WEBRC state, present only while the building block is running. */
+      struct WebrcState {
+        Webrc::Parameters params;
+        Webrc::Derived derived;
+        std::chrono::steady_clock::time_point slot_started;
+        uint32_t ctsi = 0;
+        /** Packet sequence number per channel, indexed by channel number; the base channel is T. */
+        std::vector<uint16_t> psn;
+        /** Send credit per channel, indexed as psn is. A channel accrues credit at its own rate
+         *  and spends one on each packet, so packets land on the channels in proportion to the
+         *  rates the schedule dictates. */
+        std::vector<double> credit;
+      };
+      std::optional<WebrcState> _webrc;
+      size_t _webrc_next_channel = 0;
+      /** Socket a channel sends on; channel 0 is the session's own. */
+      boost::asio::ip::udp::socket& channel_socket(size_t index) {
+        return index == 0 ? _socket : *_extra_channels.at(index - 1).socket;
+      };
+      void advance_webrc_slot_if_due();
       boost::asio::io_context& _io_context;
       boost::asio::steady_timer _send_timer;
       boost::asio::steady_timer _fdt_timer;