#209 - Add generic socket setsockopt API

AGENTS.md

@@ -66,13 +66,15 @@ clang-format -style=file -i -fallback-style=none <files>
 
 ## Architecture
 
-**Single C++/CUDA shared library** (`libdaqiri.so`) exposing a C++ API through `#include <daqiri/daqiri.h>`. The public surface is intentionally flat free-function helpers (`get_rx_burst`, `get_packet_ptr`, `set_udp_header`, …) that all operate on opaque DAQIRI-owned buffers. Applications never touch engine types directly.
+**Single C++/CUDA shared library** (`libdaqiri.so`) exposing a C++ API through `#include <daqiri/daqiri.h>`. The public surface is intentionally flat free-function helpers (`get_rx_burst`, `get_packet_ptr`, `set_udp_header`, `socket_setsockopt`, …) that all operate on opaque DAQIRI-owned buffers or connection IDs. Applications never touch engine types directly.
 
 ### Engine abstraction
-`src/engine.h` defines `daqiri::Engine` — an (almost) ABC with ~50 virtual methods covering init, RX/TX burst dequeue/enqueue, header-fill helpers, buffer free, and RDMA connection setup. Engines live in `src/engines/<name>/` (`dpdk/`, `rdma/`, `socket/`, `ibverbs/`). `DAQIRI_ENGINE` selects the optional `dpdk` and `ibverbs` engines at CMake configure time; the `socket` engine is always built. The user-facing value `ibverbs` builds **two** internal engines that both use libibverbs: `rdma` (`src/engines/rdma/`, `DAQIRI_ENGINE_RDMA`, RoCE/InfiniBand for socket `roce://`) and `ibverbs` (`src/engines/ibverbs/`, `DAQIRI_ENGINE_IBVERBS`, the pure-DevX MPRQ raw-Ethernet engine). Each engine produces its own static library (`daqiri_dpdk`, `daqiri_rdma`, `daqiri_socket`, `daqiri_ibverbs`) linked into `daqiri_common`, and each adds a `DAQIRI_ENGINE_<NAME>=1` compile definition.
+`src/engine.h` defines `daqiri::Engine` — an (almost) ABC with ~50 virtual methods covering init, RX/TX burst dequeue/enqueue, header-fill helpers, buffer free, socket connection helpers, runtime TCP/UDP `setsockopt` passthrough, and RDMA connection setup. Engines live in `src/engines/<name>/` (`dpdk/`, `rdma/`, `socket/`, `ibverbs/`). `DAQIRI_ENGINE` selects the optional `dpdk` and `ibverbs` engines at CMake configure time; the `socket` engine is always built. The user-facing value `ibverbs` builds **two** internal engines that both use libibverbs: `rdma` (`src/engines/rdma/`, `DAQIRI_ENGINE_RDMA`, RoCE/InfiniBand for socket `roce://`) and `ibverbs` (`src/engines/ibverbs/`, `DAQIRI_ENGINE_IBVERBS`, the pure-DevX MPRQ raw-Ethernet engine). Each engine produces its own static library (`daqiri_dpdk`, `daqiri_rdma`, `daqiri_socket`, `daqiri_ibverbs`) linked into `daqiri_common`, and each adds a `DAQIRI_ENGINE_<NAME>=1` compile definition.
 
 `EngineType` (`include/daqiri/types.h`) is resolved from `(stream_type, engine)`: `raw` defaults to `EngineType::DPDK`; `raw` + `engine: "ibverbs"` selects `EngineType::IBVERBS` (the MPRQ engine); `socket` + a `roce://` endpoint (or `engine: "ibverbs"`) selects `EngineType::RDMA`. The stream-aware `config_engine_from_string(str, stream_type)` overload encodes the `ibverbs`→`{IBVERBS for raw, RDMA for socket}` split. `EngineFactory` (in `engine.h`) is a singleton that instantiates the active engine. `daqiri_init(...)` resolves which engine to use from the `NetworkConfig` and then delegates everything through the `Engine` vtable. There is only ever **one** active `Engine` per process.
 
+The always-built socket engine implements Linux UDP/TCP streams directly. Applications that need kernel socket tuning call `socket_setsockopt(conn_id, level, optname, optval, optlen)` after resolving a TCP/UDP connection ID; DAQIRI passes the numeric Linux constants through without maintaining a symbolic option map. `socket_setsockopt` is not supported for `roce://` connections, which delegate to the RDMA/ibverbs path.
+
 The default `dpdk` raw engine (`src/engines/dpdk/`, `DpdkEngine`) programs RX steering, send-to-kernel fallbacks (`flow_isolation: true`), and `tx_eth_src` TX offloads via DPDK RTE Flow during `daqiri_init()`. Standard UDP/IP (group 3), flex-item (group 1) and eCPRI-over-Ethernet (group 2, EtherType `0xAEFE`, via `RTE_FLOW_ITEM_TYPE_ECPRI` matching message type and pc_id/rtc_id) RX flows use separate flow groups; `validate_config()` rejects mixing these flow classes per interface, duplicate `flex_item_id` values, and unknown `action.id` / `flex_item_id` values before NIC programming. The mlx5 PMD honors the native eCPRI flow item only under **firmware steering** (`dv_flow_en=1`); under HW steering (`dv_flow_en=2`, the default) the rule installs but silently never matches on ConnectX-class NICs. So `initialize()` auto-switches any interface carrying eCPRI RX flows to `dv_flow_en=1` (with a `WARN`), which means the async/template dynamic-RX-flow path is unavailable on that port. Flex-item parser handles are created per `(port, flex_item_id)` (scoped per interface). All programmed `rte_flow` rules and flex-item handles are tracked and destroyed in order on shutdown, init failure, and engine teardown (programmed flows → flex items → group-0 ETH jump rules). See `docs/benchmarks/raw_benchmarking.md` (Flow programming smoke test) for manual verification steps.
 
 The `ibverbs` raw engine (`src/engines/ibverbs/`, `IbverbsEngine`) drives a Mellanox/mlx5 Multi-Packet (striding) Receive Queue via **DevX** (`mlx5dv_devx_obj_create` against vendored PRM structs in `mlx5_prm_min.h`): a DevX CQ + striding RQ + TIR + `mlx5dv_dr` flow steering, with manual WQE/doorbell management and worker-driven cyclic refill. RX packets DMA strided into one pre-posted MR (host or GPU via `ibv_reg_dmabuf_mr`); a queue with >1 memory region instead uses a non-striding DevX *regular* RQ with multi-segment scatter WQEs for **physical** header-data split (header → CPU MR, payload → GPU MR). TX builds mlx5 send WQEs directly on a raw-packet QP's SQ (via `mlx5dv_init_obj`, bypassing `ibv_post_send`) from a slab of registered slots tracked by cyclic index counters, with NIC checksum offload and a `tx_eth_src` offload. It uses the libdpdk-free `daqiri::Ring`/`daqiri::ObjectPool` for the worker→app burst handoff (like the rdma engine — neither links DPDK) and drives the NIC through libibverbs/mlx5dv directly. Feature set: RX (MPRQ), TX, GPUDirect, physical/logical HDS, multi-queue 5-tuple flow steering with per-packet flow IDs, flex-item arbitrary-offset, IPv4-total-length and eCPRI-over-Ethernet (EtherType `0xAEFE`, message type + pc_id/rtc_id) flow matching (mlx5 flex parser / `misc_parameters_4`), per-packet RX hardware timestamps, accurate TX send scheduling (wait-on-time WAIT WQE), and GPU reorder/quantize. Because it uses the kernel netdev directly, `ensure_port_mtus` raises the netdev MTU at init to cover the configured frame size (jumbo frames silently drop on RX otherwise). Queues sharing a `cpu_core` are serviced round-robin by one poller thread.

README.md

@@ -43,6 +43,9 @@ DAQIRI provides direct NIC hardware access in userspace, bypassing the Linux ker
   raw ibverbs flows can also use hardware-only VLAN push/pop and VXLAN, GRE, or
   NVGRE encap/decap actions; socket/RDMA streams reject those tunnel actions.
 - **RDMA** — RDMA verbs (READ, WRITE, SEND) over RoCE on Ethernet NICs or InfiniBand.
+- **Linux socket control** — TCP/UDP socket streams expose connection IDs and
+  `socket_setsockopt()` for native Linux `setsockopt` tuning without YAML option
+  name mappings.
 - **Optional OpenTelemetry metrics** — Expose per-interface or per-queue packet,
   byte, and drop counters when built with `DAQIRI_ENABLE_OTEL_METRICS=ON`.
 

docs/api-reference/configuration.md

@@ -140,6 +140,11 @@ Endpoint addresses are URI strings. Supported schemes are `tcp://`, `udp://`, an
   fields accepted for older configs when a top-level engine override provides the
   transport.
 
+Linux TCP/UDP socket options are intentionally not configured in YAML. Apply them
+after connection setup with `socket_setsockopt(conn_id, level, optname, optval,
+optlen)`, using the numeric constants from the target system headers. The API is
+not supported for `roce://` endpoints.
+
 When using RoCE, set `stream_type: "socket"` and use `roce://` endpoint addresses
 plus a `roce_config` block for transport settings. A RoCE URI may include
 `?engine=ibverbs`; when omitted, `ibverbs` is the default and only supported RoCE

docs/api-reference/cpp.md

@@ -705,6 +705,7 @@ workflow sections above show the common call order and ownership rules.
 | `socket_connect_to_server(server_addr, server_port, src_addr, &conn_id)` | Connect a socket client using an explicit source address. |
 | `socket_get_port_queue(conn_id, &port, &queue)` | Resolve a socket connection to port/queue routing. |
 | `socket_get_server_conn_id(server_addr, server_port, &conn_id)` | Look up a server-side socket connection ID. |
+| `socket_setsockopt(conn_id, level, optname, optval, optlen)` | Apply a Linux `setsockopt` option to an existing TCP/UDP socket connection. |
 | `rdma_connect_to_server(server_addr, server_port, &conn_id)` | Connect an RDMA client to a server. |
 | `rdma_connect_to_server(server_addr, server_port, src_addr, &conn_id)` | Connect an RDMA client using an explicit source address. |
 | `rdma_get_port_queue(conn_id, &port, &queue)` | Resolve an RDMA connection to port/queue routing. |

docs/api-reference/index.md

@@ -33,7 +33,9 @@ configuration is the source of truth for queue IDs, memory placement,
 stream-type / engine / endpoint selection, flow routing, and static TX
 tunnel/VLAN transforms. Dynamic RX flow APIs can add and delete runtime
 queue-steering rules and, on raw DPDK or raw ibverbs, runtime RX decap/pop
-rules using the same ordered action model.
+rules using the same ordered action model. TCP/UDP socket options are also
+runtime state: after resolving a connection ID, applications can call
+`socket_setsockopt()` with native Linux `level` and option constants.
 
 The configuration schema lives in the
 [Configuration YAML Reference](configuration.md). For an annotated

docs/api-reference/python.md

@@ -382,10 +382,15 @@ Socket helpers return connection IDs and queue routing information. The
 returned port and queue can then be used with the normal burst APIs:
 
 ```python
+import socket
+
 status, conn_id = daqiri.socket_connect_to_server("192.0.2.10", 5000)
 if status == daqiri.Status.SUCCESS:
     status, port, queue = daqiri.socket_get_port_queue(conn_id)
 
+    # Socket options use OS constants directly; DAQIRI does not map option names.
+    daqiri.socket_setsockopt(conn_id, socket.SOL_SOCKET, socket.SO_RCVBUF, 8 * 1024 * 1024)
+
     status, rx_burst = daqiri.get_rx_burst(port, queue)
     if status == daqiri.Status.SUCCESS and rx_burst is not None:
         try:
@@ -406,6 +411,11 @@ if status == daqiri.Status.SUCCESS:
         daqiri.free_tx_metadata(tx_burst)
 ```
 
+`socket_setsockopt(conn_id, level, optname, value)` applies Linux `setsockopt`
+to an existing TCP/UDP socket connection. `value` may be `bool`, `int`, `str`,
+`bytes`, `bytearray`, or another bytes-like buffer. For structured options such
+as timeouts, pack the native struct yourself and pass it as bytes.
+
 RDMA follows the same tuple-returning style:
 
 ```python
@@ -587,6 +597,7 @@ The workflow sections above show the common call order and ownership rules.
 | `socket_connect_to_server(server_addr, server_port[, src_addr])` | Return `(Status, conn_id)`. |
 | `socket_get_port_queue(conn_id)` | Return `(Status, port, queue)`. |
 | `socket_get_server_conn_id(server_addr, server_port)` | Return `(Status, conn_id)`. |
+| `socket_setsockopt(conn_id, level, optname, value)` | Apply a Linux socket option to an existing TCP/UDP socket connection. |
 | `rdma_connect_to_server(server_addr, server_port[, src_addr])` | Return `(Status, conn_id)`. |
 | `rdma_get_port_queue(conn_id)` | Return `(Status, port, queue)`. |
 | `rdma_get_server_conn_id(server_addr, server_port)` | Return `(Status, conn_id)`. |

docs/benchmarks/socket_benchmarking.md

@@ -158,6 +158,11 @@ The shipped configs run both endpoints on `127.0.0.1` and are useful for a smoke
 
 For an on-wire namespace test, use separate server and client YAML files. The important fields are the endpoint URI scheme, namespace IPs, server port, `max_payload_size`, memory-region `buf_size`, and benchmark `message_size`.
 
+Applications can tune the underlying TCP/UDP socket after resolving a connection ID
+with `socket_connect_to_server()` or `socket_get_server_conn_id()`. Use
+`socket_setsockopt(conn_id, level, optname, optval, optlen)` with the integer
+constants from your system headers.
+
 Server-side UDP template:
 
 ```yaml

docs/getting-started.md

@@ -201,6 +201,11 @@ Both methods use the same public C++ include:
 | `DAQIRI_PREFER_SYSTEM_YAML_CPP` | `OFF` | Prefer a system-installed `yaml-cpp` over the vendored `third_party/yaml-cpp` submodule. Keep `OFF` if a conda/miniforge env is on `PATH`. |
 | `BUILD_SHARED_LIBS` | — | Build as shared library. |
 
+Linux UDP/TCP sockets are always available. Applications that need kernel socket
+tuning can call `socket_setsockopt()` after resolving a TCP/UDP connection ID,
+passing the numeric `level` and option constants from system headers; DAQIRI does
+not maintain symbolic socket-option mappings in YAML.
+
 For Raw Ethernet (`stream_type: "raw"`), `daqiri_init()` validates that each `rx.flows`
 entry's legacy `action.id` or final ordered `actions:` queue action matches an
 `rx.queues` ID on the same interface, then programs flow rules into the NIC.

docs/tutorials/configuration-walkthrough.md

@@ -12,7 +12,7 @@ hide:
 DAQIRI exposes a single API on top of multiple packet I/O stacks, selected at runtime with `stream_type` and endpoint URI schemes such as `udp://`, `tcp://`, and `roce://`. Pick the row that matches your hardware and the role of the other endpoint:
 
 - **Raw Ethernet** — `stream_type: "raw"`. Kernel-bypass with GPUDirect zero-copy. Highest performance. Requires an [NVIDIA ConnectX-class NIC](https://www.nvidia.com/en-us/networking/ethernet-adapters/); `tx_port` and `rx_port` can share one physical NIC for a single-host closed-loop bench, or be split across two hosts.
-- **Socket — UDP / TCP** — `stream_type: "socket"` with `udp://` or `tcp://` endpoints. Plain Linux kernel sockets. No NIC, no privileges, no special CMake flags. Useful as a comparison baseline and as a path to first results on a system without an NVIDIA NIC.
+- **Socket — UDP / TCP** — `stream_type: "socket"` with `udp://` or `tcp://` endpoints. Plain Linux kernel sockets. No NIC, no privileges, no special CMake flags. Useful as a comparison baseline and as a path to first results on a system without an NVIDIA NIC. Socket options are runtime API calls: resolve a connection ID, then pass native Linux constants to `socket_setsockopt()` rather than adding option names to YAML.
 - **Socket — RoCE (RDMA)** — `stream_type: "socket"` and `roce://` endpoints. RDMA verbs over Ethernet, with a server/client connection model and a NIC-level reliable transport. Primarily intended for setups where **one** endpoint is a third-party RoCE implementation (FPGA, instrument, customer black box). When both peers run DAQIRI, prefer an upper-layer library such as MPI / NCCL / UCX instead.
 
 If you don't have any NIC at all, the `*_sw_loopback*` variants of the Raw Ethernet configs need no hardware — useful for first-time build verification.

include/daqiri/common.h

@@ -18,6 +18,7 @@
 #pragma once
 #include <daqiri/logging.hpp>
 #include <daqiri/types.h>
+#include <cstddef>
 #include <memory>
 #include <optional>
 #include <stdint.h>
@@ -902,6 +903,8 @@ Status socket_get_port_queue(uintptr_t conn_id, uint16_t *port,
                              uint16_t *queue);
 Status socket_get_server_conn_id(const std::string &server_addr,
                                  uint16_t server_port, uintptr_t *conn_id);
+Status socket_setsockopt(uintptr_t conn_id, int level, int optname,
+                         const void *optval, size_t optlen);
 
 // RDMA functions
 Status rdma_connect_to_server(const std::string &server_addr,

python/daqiri_common_pybind.cpp

@@ -442,6 +442,47 @@ struct PyS3Writer {
   }
 };
 
+Status socket_setsockopt_from_python(uintptr_t conn_id,
+                                     int level,
+                                     int optname,
+                                     py::object value) {
+  if (PyBool_Check(value.ptr())) {
+    const int opt_value = value.cast<bool>() ? 1 : 0;
+    py::gil_scoped_release release;
+    return socket_setsockopt(conn_id, level, optname, &opt_value, sizeof(opt_value));
+  }
+
+  if (py::isinstance<py::int_>(value)) {
+    const int opt_value = value.cast<int>();
+    py::gil_scoped_release release;
+    return socket_setsockopt(conn_id, level, optname, &opt_value, sizeof(opt_value));
+  }
+
+  if (py::isinstance<py::str>(value)) {
+    const std::string opt_value = value.cast<std::string>();
+    py::gil_scoped_release release;
+    return socket_setsockopt(
+        conn_id, level, optname, opt_value.data(), opt_value.size());
+  }
+
+  if (py::isinstance<py::bytes>(value)) {
+    const std::string opt_value = value.cast<std::string>();
+    py::gil_scoped_release release;
+    return socket_setsockopt(
+        conn_id, level, optname, opt_value.data(), opt_value.size());
+  }
+
+  if (PyObject_CheckBuffer(value.ptr())) {
+    py::buffer buffer = value.cast<py::buffer>();
+    py::buffer_info info = buffer.request();
+    const size_t nbytes = buffer_nbytes(info);
+    py::gil_scoped_release release;
+    return socket_setsockopt(conn_id, level, optname, info.ptr, nbytes);
+  }
+
+  throw py::type_error("setsockopt value must be bool, int, str, bytes, or a buffer");
+}
+
 void bind_enums(py::module_ &m) {
   py::enum_<Status>(m, "Status")
       .value("SUCCESS", Status::SUCCESS)
@@ -1287,6 +1328,8 @@ PYBIND11_MODULE(_daqiri, m) {
         return py::make_tuple(status, conn_id);
       },
       "server_addr"_a, "server_port"_a);
+  m.def("socket_setsockopt", &socket_setsockopt_from_python, "conn_id"_a,
+        "level"_a, "optname"_a, "value"_a);
 
   m.def(
       "rdma_connect_to_server",

src/common.cpp

@@ -789,6 +789,12 @@ Status socket_get_server_conn_id(const std::string& server_addr, uint16_t server
   return g_daqiri_engine->socket_get_server_conn_id(server_addr, server_port, conn_id);
 }
 
+Status socket_setsockopt(uintptr_t conn_id, int level, int optname, const void* optval,
+                         size_t optlen) {
+  ASSERT_DAQIRI_ENGINE_INITIALIZED();
+  return g_daqiri_engine->socket_setsockopt(conn_id, level, optname, optval, optlen);
+}
+
 // RDMA Functions
 Status rdma_connect_to_server(const std::string& server_addr, uint16_t server_port,
                               uintptr_t* conn_id) {

src/engine.cpp

@@ -757,6 +757,12 @@ Status Engine::socket_get_server_conn_id(const std::string& server_addr, uint16_
   return Status::NOT_SUPPORTED;
 }
 
+Status Engine::socket_setsockopt(uintptr_t conn_id, int level, int optname, const void* optval,
+                                  size_t optlen) {
+  DAQIRI_LOG_CRITICAL("Socket setsockopt not implemented");
+  return Status::NOT_SUPPORTED;
+}
+
 Status Engine::rdma_connect_to_server(const std::string& dst_addr, uint16_t dst_port,
                                        uintptr_t* conn_id) {
   DAQIRI_LOG_CRITICAL("RDMA connect to server not implemented");

src/engine.h

@@ -115,6 +115,8 @@ class Engine {
   virtual Status socket_get_port_queue(uintptr_t conn_id, uint16_t* port, uint16_t* queue);
   virtual Status socket_get_server_conn_id(const std::string& server_addr, uint16_t server_port,
                                            uintptr_t* conn_id);
+  virtual Status socket_setsockopt(uintptr_t conn_id, int level, int optname, const void* optval,
+                                   size_t optlen);
 
   virtual Status rdma_connect_to_server(const std::string& dst_addr, uint16_t dst_port,
                                         uintptr_t* conn_id);