diff --git a/AGENTS.md b/AGENTS.md index ae942e67..cec042c5 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -19,6 +19,7 @@ CMake options (full table in `docs/getting-started.md`): - `DAQIRI_BUILD_PYTHON` — builds `pybind11` bindings from `python/`. - `DAQIRI_BUILD_EXAMPLES` — builds the benchmark executables (default `ON`). - `DAQIRI_ENABLE_OTEL_METRICS` — enables OpenTelemetry metrics instrumentation (default `OFF`). +- `DAQIRI_ENABLE_BURST_VALIDATION` — enables defensive burst bounds and backing-capacity validation in public packet accessors and setters (default `ON`; disable only for trusted-input performance testing). - `DAQIRI_REORDER_GPU_PROFILE` — enable CUDA event timing in the DPDK reorder kernels (off by default). - `DAQIRI_ENABLE_S3` — enable AWS SDK-backed asynchronous raw packet writes to S3 (off by default). - `DAQIRI_PREFER_SYSTEM_YAML_CPP` — prefer system `yaml-cpp` over the vendored `third_party/yaml-cpp` submodule (off by default; keep off when a conda/miniforge env is on `PATH`). @@ -124,7 +125,7 @@ The web docs live in `docs/` and are built with [MkDocs Material](https://squidf **Keeping docs in sync with code:** before committing changes, scan for the recurring drift hotspots: - **Stream-type list** (`src/engines/*/`) — README Engines table, `docs/getting-started.md`, `docs/concepts.md` (Stream Types section + Support and testing admonition), `docs/api-reference/configuration.md` -- **CMake options / `DAQIRI_ENGINE` default** (`src/CMakeLists.txt:137`) — README Quick Start, `docs/getting-started.md`, this file's Build & run section +- **CMake options / `DAQIRI_ENGINE` default** (`src/CMakeLists.txt`) — README Quick Start, `docs/getting-started.md`, this file's Build & run section - **Benchmark binary or YAML names** (`examples/`) — the benchmark table above, `docs/benchmarks/raw_benchmarking.md`, the "Choosing an example config" decision tree in `docs/tutorials/configuration-walkthrough.md` (every YAML must have a leaf; CI's `scripts/check_doc_refs.py` enforces coverage), and per-platform performance docs (`docs/benchmarks/performance-*.md`) - **Public API include** (`#include `; source files under `include/daqiri/`) — `docs/api-reference/index.md`, `docs/api-reference/cpp.md`, `docs/api-reference/python.md`; if the change adds or renames a user-facing concept, also `docs/concepts.md` - **Python bindings** (`python/daqiri_common_pybind.cpp`) — `docs/api-reference/python.md` (function reference tables, enums/classes tables, GIL Behavior section) diff --git a/CMakeLists.txt b/CMakeLists.txt index 67520967..d9ef8463 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -33,6 +33,8 @@ endif() option(DAQIRI_BUILD_PYTHON "Build Python bindings" OFF) option(DAQIRI_BUILD_EXAMPLES "Build standalone examples" ON) option(DAQIRI_ENABLE_OTEL_METRICS "Enable OpenTelemetry metrics instrumentation" OFF) +option(DAQIRI_ENABLE_BURST_VALIDATION + "Enable defensive burst bounds and backing capacity validation" ON) set(DAQIRI_ENGINE "dpdk ibverbs" CACHE STRING "Optional engine implementations to build: dpdk, ibverbs (Linux sockets are always built in)") add_subdirectory(src) diff --git a/docs/getting-started.md b/docs/getting-started.md index 1686708f..2c6284e0 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -195,6 +195,7 @@ Both methods use the same public C++ include: | `DAQIRI_ENGINE` | `"dpdk ibverbs"` | Space-separated list of optional engine implementations to compile in. Valid values: `dpdk` (Raw Ethernet) and `ibverbs`. `ibverbs` builds two libibverbs-based engines: RDMA/RoCE (for `stream_type: "socket"` with `roce://` endpoints) and a Mellanox/mlx5 Multi-Packet (striding) Receive Queue engine for `stream_type: "raw"` (opt in per stream with `engine: "ibverbs"`). Linux UDP/TCP sockets are always built in, so there is no `socket` value. | | `DAQIRI_BUILD_PYTHON` | `OFF` | Build pybind11 Python bindings. | | `DAQIRI_BUILD_EXAMPLES` | `ON` | Build benchmark executables. | +| `DAQIRI_ENABLE_BURST_VALIDATION` | `ON` | Enable defensive burst bounds and backing-capacity validation in public packet accessors and setters. Disable only for trusted-input performance testing after benchmarking. | | `DAQIRI_ENABLE_GDS` | `OFF` | Enable cuFile-backed burst file writes from CUDA device memory. Host-memory writes use POSIX APIs without GDS. | | `DAQIRI_ENABLE_OTEL_METRICS` | `OFF` | Enable OpenTelemetry C++ metrics instrumentation. When enabled, OpenTelemetry C++ API package metadata must be available to CMake. | | `DAQIRI_ENABLE_S3` | `OFF` | Enable AWS SDK-backed asynchronous raw packet writes to S3. | diff --git a/docs/tutorials/bare-metal-cmake-build.md b/docs/tutorials/bare-metal-cmake-build.md index 9fac5c0c..6a01ce8f 100644 --- a/docs/tutorials/bare-metal-cmake-build.md +++ b/docs/tutorials/bare-metal-cmake-build.md @@ -278,6 +278,7 @@ The override is honored because `src/CMakeLists.txt` only sets `CMAKE_CUDA_ARCHI - `-DBUILD_SHARED_LIBS=ON`: produces `libdaqiri.so` (recommended). With `OFF`, you get a static library. - `-DDAQIRI_BUILD_EXAMPLES=ON`: builds the `daqiri_bench_*` executables under `build/examples/`. Required for the smoke test in [Step 5.3](#53-smoke-test). On by default. +- `-DDAQIRI_ENABLE_BURST_VALIDATION=ON`: enables defensive burst bounds and backing-capacity validation in packet accessors and setters. On by default; turn off only for trusted-input performance testing after benchmarking. - `-DDAQIRI_REORDER_GPU_PROFILE=ON`: instruments the CUDA reorder kernels with CUDA event timing. Off by default; turn on only when profiling. ## Step 5: Build, install, and verify diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 21d3a086..e1048812 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -64,6 +64,8 @@ endif() option(DAQIRI_ENABLE_GDS "Enable cuFile-backed burst file writes from CUDA device memory" OFF) option(DAQIRI_ENABLE_OTEL_METRICS "Enable OpenTelemetry metrics instrumentation" OFF) option(DAQIRI_ENABLE_S3 "Enable AWS SDK-backed raw packet writes to S3" OFF) +option(DAQIRI_ENABLE_BURST_VALIDATION + "Enable defensive burst bounds and backing capacity validation" ON) set(DOCA_PATH /opt/mellanox/doca) if(DAQIRI_ENABLE_OTEL_METRICS) @@ -72,6 +74,11 @@ if(DAQIRI_ENABLE_OTEL_METRICS) else() add_compile_definitions(DAQIRI_ENABLE_OTEL_METRICS=0) endif() +if(DAQIRI_ENABLE_BURST_VALIDATION) + add_compile_definitions(DAQIRI_ENABLE_BURST_VALIDATION=1) +else() + add_compile_definitions(DAQIRI_ENABLE_BURST_VALIDATION=0) +endif() if(EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/../third_party/spdlog/include/spdlog/spdlog.h") message(STATUS "Using vendored spdlog headers from third_party/spdlog") diff --git a/src/burst_validation.h b/src/burst_validation.h new file mode 100644 index 00000000..30b5749e --- /dev/null +++ b/src/burst_validation.h @@ -0,0 +1,276 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include +#include +#include +#include +#include + +#include +#include + +#ifndef DAQIRI_ENABLE_BURST_VALIDATION +#define DAQIRI_ENABLE_BURST_VALIDATION 1 +#endif + +namespace daqiri::burst_validation { + +struct BurstLimits { + size_t num_pkts = 0; + int num_segs = 0; +}; + +static inline constexpr size_t kUnknownCapacity = std::numeric_limits::max(); + +inline constexpr bool strict_enabled() { + return DAQIRI_ENABLE_BURST_VALIDATION != 0; +} + +inline BurstLimits header_limits(const BurstParams* burst) { + if (burst == nullptr) { return {}; } + return {burst->hdr.hdr.num_pkts, burst->hdr.hdr.num_segs}; +} + +inline BurstLimits transport_limits(const BurstParams* burst) { + if (burst == nullptr) { return {}; } + return {burst->transport_hdr.num_pkts, burst->transport_hdr.num_segs}; +} + +inline Status validate_burst(const BurstParams* burst, const char* op_name) { + if (burst != nullptr) { return Status::SUCCESS; } + DAQIRI_LOG_ERROR("{}: burst is null", op_name); + return Status::NULL_PTR; +} + +inline Status validate_packet_index(const BurstParams* burst, + BurstLimits limits, + int idx, + const char* op_name) { + Status status = validate_burst(burst, op_name); + if (status != Status::SUCCESS) { return status; } + if (idx < 0 || (strict_enabled() && static_cast(idx) >= limits.num_pkts)) { + DAQIRI_LOG_ERROR("{}: packet index {} out of range [0, {})", op_name, idx, limits.num_pkts); + return Status::INVALID_PARAMETER; + } + return Status::SUCCESS; +} + +inline Status validate_packet_count(const BurstParams* burst, + BurstLimits limits, + size_t max_pkts, + const char* op_name) { + Status status = validate_burst(burst, op_name); + if (status != Status::SUCCESS) { return status; } + if (!strict_enabled()) { return Status::SUCCESS; } + if (limits.num_pkts == 0 || limits.num_pkts > max_pkts) { + DAQIRI_LOG_ERROR("{}: invalid packet count {} (max {})", + op_name, + limits.num_pkts, + max_pkts); + return Status::INVALID_PARAMETER; + } + return Status::SUCCESS; +} + +inline Status validate_segment_count(const BurstParams* burst, + BurstLimits limits, + const char* op_name) { + Status status = validate_burst(burst, op_name); + if (status != Status::SUCCESS) { return status; } + if (!strict_enabled()) { return Status::SUCCESS; } + if (limits.num_segs <= 0 || limits.num_segs > MAX_NUM_SEGS) { + DAQIRI_LOG_ERROR("{}: invalid segment count {}", op_name, limits.num_segs); + return Status::INVALID_PARAMETER; + } + return Status::SUCCESS; +} + +inline Status validate_segment_index(const BurstParams* burst, + BurstLimits limits, + int seg, + const char* op_name) { + Status status = validate_burst(burst, op_name); + if (status != Status::SUCCESS) { return status; } + if (!strict_enabled()) { + if (seg < 0 || seg >= MAX_NUM_SEGS) { + DAQIRI_LOG_ERROR("{}: segment index {} out of range [0, {})", op_name, seg, MAX_NUM_SEGS); + return Status::INVALID_PARAMETER; + } + return Status::SUCCESS; + } + status = validate_segment_count(burst, limits, op_name); + if (status != Status::SUCCESS) { return status; } + if (seg < 0 || seg >= limits.num_segs) { + DAQIRI_LOG_ERROR("{}: segment index {} out of range [0, {})", op_name, seg, limits.num_segs); + return Status::INVALID_PARAMETER; + } + return Status::SUCCESS; +} + +inline Status validate_segment_packet_storage(const BurstParams* burst, + BurstLimits limits, + int seg, + int idx, + bool require_packet_ptr, + bool require_length_ptr, + const char* op_name) { + Status status = validate_packet_index(burst, limits, idx, op_name); + if (status != Status::SUCCESS) { return status; } + status = validate_segment_index(burst, limits, seg, op_name); + if (status != Status::SUCCESS) { return status; } + if (!strict_enabled()) { return Status::SUCCESS; } + + if (require_packet_ptr) { + if (burst->pkts[seg] == nullptr) { + DAQIRI_LOG_ERROR("{}: packet pointer array is null for segment {}", op_name, seg); + return Status::NULL_PTR; + } + if (burst->pkts[seg][idx] == nullptr) { + DAQIRI_LOG_ERROR("{}: packet pointer is null for packet {} segment {}", op_name, idx, seg); + return Status::NULL_PTR; + } + } + + if (require_length_ptr && burst->pkt_lens[seg] == nullptr) { + DAQIRI_LOG_ERROR("{}: packet length array is null for segment {}", op_name, seg); + return Status::NULL_PTR; + } + + return Status::SUCCESS; +} + +inline Status validate_payload_write(const BurstParams* burst, + BurstLimits limits, + int idx, + const void* data, + int len, + size_t payload_capacity, + const char* op_name) { + Status status = validate_segment_packet_storage( + burst, limits, 0, idx, true, false, op_name); + if (status != Status::SUCCESS) { return status; } + if (!strict_enabled()) { return Status::SUCCESS; } + if (len < 0) { + DAQIRI_LOG_ERROR("{}: payload length {} is negative", op_name, len); + return Status::INVALID_PARAMETER; + } + if (len > 0 && data == nullptr) { + DAQIRI_LOG_ERROR("{}: payload source is null for non-empty payload", op_name); + return Status::NULL_PTR; + } + const size_t payload_len = static_cast(len); + if (payload_capacity != kUnknownCapacity && payload_len > payload_capacity) { + DAQIRI_LOG_ERROR("{}: payload length {} exceeds backing capacity {}", + op_name, + payload_len, + payload_capacity); + return Status::INVALID_PARAMETER; + } + return Status::SUCCESS; +} + +inline Status validate_packet_lengths(const BurstParams* burst, + BurstLimits limits, + int idx, + const std::initializer_list& lens, + const std::array& capacities, + bool require_packet_ptrs, + bool require_length_ptrs, + size_t max_segment_len, + const char* op_name, + uint32_t* total_len) { + if (total_len != nullptr) { *total_len = 0; } + + Status status = validate_packet_index(burst, limits, idx, op_name); + if (status != Status::SUCCESS) { return status; } + status = validate_segment_count(burst, limits, op_name); + if (status != Status::SUCCESS) { return status; } + + if (!strict_enabled()) { + uint64_t total = 0; + for (int len : lens) { + total += static_cast(len); + if (total > std::numeric_limits::max()) { + DAQIRI_LOG_ERROR("{}: total packet length {} exceeds uint32_t", op_name, total); + return Status::INVALID_PARAMETER; + } + } + if (total_len != nullptr) { *total_len = static_cast(total); } + return Status::SUCCESS; + } + + if (lens.size() != static_cast(limits.num_segs)) { + DAQIRI_LOG_ERROR("{}: got {} packet length(s), expected {} segment length(s)", + op_name, + lens.size(), + limits.num_segs); + return Status::INVALID_PARAMETER; + } + + uint64_t total = 0; + int seg = 0; + for (int len : lens) { + status = validate_segment_packet_storage( + burst, limits, seg, idx, require_packet_ptrs, require_length_ptrs, op_name); + if (status != Status::SUCCESS) { return status; } + if (len < 0) { + DAQIRI_LOG_ERROR("{}: packet length {} is negative for segment {}", + op_name, + len, + seg); + return Status::INVALID_PARAMETER; + } + + const size_t seg_len = static_cast(len); + if (seg_len > max_segment_len) { + DAQIRI_LOG_ERROR("{}: packet length {} exceeds maximum segment length {} for segment {}", + op_name, + seg_len, + max_segment_len, + seg); + return Status::INVALID_PARAMETER; + } + if (capacities[seg] != kUnknownCapacity && seg_len > capacities[seg]) { + DAQIRI_LOG_ERROR("{}: packet length {} exceeds backing capacity {} for segment {}", + op_name, + seg_len, + capacities[seg], + seg); + return Status::INVALID_PARAMETER; + } + total += seg_len; + if (total > std::numeric_limits::max()) { + DAQIRI_LOG_ERROR("{}: total packet length {} exceeds uint32_t", op_name, total); + return Status::INVALID_PARAMETER; + } + ++seg; + } + + if (total_len != nullptr) { *total_len = static_cast(total); } + return Status::SUCCESS; +} + +inline std::array unknown_capacities() { + std::array capacities{}; + capacities.fill(kUnknownCapacity); + return capacities; +} + +} // namespace daqiri::burst_validation diff --git a/src/engines/dpdk/daqiri_dpdk_engine.cpp b/src/engines/dpdk/daqiri_dpdk_engine.cpp index 5e2f93e6..38c310ff 100644 --- a/src/engines/dpdk/daqiri_dpdk_engine.cpp +++ b/src/engines/dpdk/daqiri_dpdk_engine.cpp @@ -39,6 +39,7 @@ #include #include "src/dpdk_log.h" +#include "src/burst_validation.h" #include "daqiri_dpdk_engine.h" #include "src/kernels.h" #include @@ -468,6 +469,79 @@ struct UDPPkt { uint8_t payload[]; } __attribute__((packed)); +static size_t mbuf_data_capacity(const rte_mbuf* mbuf) { + if (mbuf == nullptr || mbuf->buf_len < mbuf->data_off) { return 0; } + return static_cast(mbuf->buf_len - mbuf->data_off); +} + +static Status validate_dpdk_first_segment_write(BurstParams* burst, + int idx, + size_t bytes_required, + const char* op_name) { + const auto limits = burst_validation::header_limits(burst); + Status status = burst_validation::validate_segment_packet_storage( + burst, limits, 0, idx, true, false, op_name); + if (status != Status::SUCCESS) { return status; } + if (!burst_validation::strict_enabled()) { return Status::SUCCESS; } + + const auto* mbuf = reinterpret_cast(burst->pkts[0][idx]); + const size_t capacity = mbuf_data_capacity(mbuf); + if (bytes_required > capacity) { + DAQIRI_LOG_ERROR("{}: write size {} exceeds first-segment backing capacity {}", + op_name, + bytes_required, + capacity); + return Status::INVALID_PARAMETER; + } + return Status::SUCCESS; +} + +static Status dpdk_packet_length_capacities(BurstParams* burst, + int idx, + std::array* capacities, + const char* op_name) { + if (capacities == nullptr) { return Status::NULL_PTR; } + *capacities = burst_validation::unknown_capacities(); + if (!burst_validation::strict_enabled()) { return Status::SUCCESS; } + + const auto limits = burst_validation::header_limits(burst); + Status status = burst_validation::validate_packet_index(burst, limits, idx, op_name); + if (status != Status::SUCCESS) { return status; } + status = burst_validation::validate_segment_count(burst, limits, op_name); + if (status != Status::SUCCESS) { return status; } + + for (int seg = 0; seg < limits.num_segs; ++seg) { + status = burst_validation::validate_segment_packet_storage( + burst, limits, seg, idx, true, false, op_name); + if (status != Status::SUCCESS) { return status; } + (*capacities)[static_cast(seg)] = + mbuf_data_capacity(reinterpret_cast(burst->pkts[seg][idx])); + } + return Status::SUCCESS; +} + +static Status validate_dpdk_packet_lengths(BurstParams* burst, + int idx, + const std::initializer_list& lens, + uint32_t* total_len, + const char* op_name) { + std::array capacities{}; + Status status = dpdk_packet_length_capacities(burst, idx, &capacities, op_name); + if (status != Status::SUCCESS) { return status; } + + return burst_validation::validate_packet_lengths( + burst, + burst_validation::header_limits(burst), + idx, + lens, + capacities, + true, + false, + std::numeric_limits::max(), + op_name, + total_len); +} + static inline uint64_t convert_rx_timestamp_ticks_to_ns(uint64_t timestamp, const RxTimestampConversion& conversion) { if (!conversion.valid || conversion.ticks_per_second == 0) { return 0; } @@ -6199,45 +6273,74 @@ int DpdkEngine::tx_lb_worker(void* arg) { /* daqiri interface implementations */ void* DpdkEngine::get_segment_packet_ptr(BurstParams* burst, int seg, int idx) { - if (burst == nullptr || idx < 0 || seg < 0 || seg >= MAX_NUM_SEGS) { return nullptr; } + const auto limits = burst_validation::header_limits(burst); + if ((burst == nullptr) || + burst_validation::validate_segment_packet_storage( + burst, limits, seg, idx, true, false, "DpdkEngine::get_segment_packet_ptr") + != Status::SUCCESS) { + return nullptr; + } if ((burst->hdr.hdr.burst_flags & kBurstFlagDpdkReordered) != 0U) { - if (seg != 0 || burst->pkts[0] == nullptr) { return nullptr; } + if (seg != 0) { return nullptr; } return burst->pkts[0][idx]; } return rte_pktmbuf_mtod(reinterpret_cast(burst->pkts[seg][idx]), void*); } void* DpdkEngine::get_packet_ptr(BurstParams* burst, int idx) { - if (burst == nullptr || idx < 0) { return nullptr; } - if ((burst->hdr.hdr.burst_flags & kBurstFlagDpdkReordered) != 0U) { - if (burst->pkts[0] == nullptr) { return nullptr; } - return burst->pkts[0][idx]; - } - return rte_pktmbuf_mtod(reinterpret_cast(burst->pkts[0][idx]), void*); + return get_segment_packet_ptr(burst, 0, idx); } uint32_t DpdkEngine::get_segment_packet_length(BurstParams* burst, int seg, int idx) { - if (burst == nullptr || idx < 0 || seg < 0 || seg >= MAX_NUM_SEGS) { return 0; } + if (burst == nullptr) { return 0; } + const auto limits = burst_validation::header_limits(burst); if ((burst->hdr.hdr.burst_flags & kBurstFlagDpdkReordered) != 0U) { - if (seg != 0 || burst->pkt_lens[0] == nullptr) { return 0; } + if (seg != 0 || + burst_validation::validate_segment_packet_storage( + burst, limits, seg, idx, false, true, "DpdkEngine::get_segment_packet_length") + != Status::SUCCESS) { + return 0; + } return burst->pkt_lens[0][idx]; } + if (burst_validation::validate_segment_packet_storage( + burst, limits, seg, idx, true, false, "DpdkEngine::get_segment_packet_length") + != Status::SUCCESS) { + return 0; + } return reinterpret_cast(burst->pkts[seg][idx])->data_len; } uint32_t DpdkEngine::get_packet_length(BurstParams* burst, int idx) { - if (burst == nullptr || idx < 0) { return 0; } + if (burst == nullptr) { return 0; } + const auto limits = burst_validation::header_limits(burst); if ((burst->hdr.hdr.burst_flags & kBurstFlagDpdkReordered) != 0U) { - if (burst->pkt_lens[0] == nullptr) { return 0; } + if (burst_validation::validate_segment_packet_storage( + burst, limits, 0, idx, false, true, "DpdkEngine::get_packet_length") + != Status::SUCCESS) { + return 0; + } return burst->pkt_lens[0][idx]; } + if (burst_validation::validate_segment_packet_storage( + burst, limits, 0, idx, true, false, "DpdkEngine::get_packet_length") + != Status::SUCCESS) { + return 0; + } return reinterpret_cast(burst->pkts[0][idx])->pkt_len; } FlowId DpdkEngine::get_packet_flow_id(BurstParams* burst, int idx) { - if (burst == nullptr || idx < 0) { return 0; } + if (burst == nullptr) { return 0; } if ((burst->hdr.hdr.burst_flags & kBurstFlagDpdkReordered) != 0U) { return 0; } - if (idx >= static_cast(burst->hdr.hdr.num_pkts) || burst->pkts[0] == nullptr) { + if (burst_validation::validate_segment_packet_storage( + burst, + burst_validation::header_limits(burst), + 0, + idx, + true, + false, + "DpdkEngine::get_packet_flow_id") != Status::SUCCESS) { return 0; } const auto* mbuf = reinterpret_cast(burst->pkts[0][idx]); @@ -6248,11 +6351,11 @@ FlowId DpdkEngine::get_packet_flow_id(BurstParams* burst, int idx) { Status DpdkEngine::get_packet_rx_timestamp(BurstParams* burst, int idx, uint64_t* timestamp_ns) { if (burst == nullptr || timestamp_ns == nullptr) { return Status::NULL_PTR; } *timestamp_ns = 0; - if (idx < 0 || idx >= static_cast(burst->hdr.hdr.num_pkts)) { - return Status::INVALID_PARAMETER; - } if ((burst->hdr.hdr.burst_flags & kBurstFlagDpdkReordered) != 0U) { + Status status = burst_validation::validate_packet_index( + burst, burst_validation::header_limits(burst), idx, "DpdkEngine::get_packet_rx_timestamp"); + if (status != Status::SUCCESS) { return status; } if (idx != 0 || burst->custom_pkt_data == nullptr) { return Status::INVALID_PARAMETER; } auto ctx = std::static_pointer_cast(burst->custom_pkt_data); if (!ctx || !ctx->rx_timestamp_ns_valid) { return Status::NOT_SUPPORTED; } @@ -6260,7 +6363,15 @@ Status DpdkEngine::get_packet_rx_timestamp(BurstParams* burst, int idx, uint64_t return Status::SUCCESS; } - if (burst->pkts[0] == nullptr) { return Status::INVALID_PARAMETER; } + Status status = burst_validation::validate_segment_packet_storage( + burst, + burst_validation::header_limits(burst), + 0, + idx, + true, + false, + "DpdkEngine::get_packet_rx_timestamp"); + if (status != Status::SUCCESS) { return status; } auto* mbuf = reinterpret_cast(burst->pkts[0][idx]); if (mbuf == nullptr || burst->hdr.hdr.port_id >= rx_timestamp_conversions_.size()) { return Status::INVALID_PARAMETER; @@ -6278,9 +6389,15 @@ Status DpdkEngine::get_packet_rx_timestamp(BurstParams* burst, int idx, uint64_t Status DpdkEngine::set_packet_tx_time(BurstParams* burst, int idx, uint64_t timestamp) { if (burst == nullptr) { return Status::NULL_PTR; } - if (idx < 0 || idx >= static_cast(burst->hdr.hdr.num_pkts)) { - return Status::INVALID_PARAMETER; - } + Status status = burst_validation::validate_segment_packet_storage( + burst, + burst_validation::header_limits(burst), + 0, + idx, + true, + false, + "DpdkEngine::set_packet_tx_time"); + if (status != Status::SUCCESS) { return status; } if (timestamp_dynfield_offset_ < 0 || tx_timestamp_dynflag_mask_ == 0) { return Status::NOT_SUPPORTED; } @@ -6306,6 +6423,18 @@ void* DpdkEngine::get_packet_extra_info(BurstParams* burst, int idx) { } Status DpdkEngine::get_tx_packet_burst(BurstParams* burst) { + if (burst == nullptr) { return Status::NULL_PTR; } + const auto limits = burst_validation::header_limits(burst); + Status status = burst_validation::validate_segment_count( + burst, limits, "DpdkEngine::get_tx_packet_burst"); + if (status != Status::SUCCESS) { return status; } + status = burst_validation::validate_packet_count( + burst, + limits, + static_cast(std::numeric_limits::max()), + "DpdkEngine::get_tx_packet_burst"); + if (status != Status::SUCCESS) { return status; } + const uint32_t key = generate_queue_key(burst->hdr.hdr.port_id, burst->hdr.hdr.q_id); const auto q_it = tx_dpdk_q_map_.find(key); if (q_it == tx_dpdk_q_map_.end() || q_it->second == nullptr) { @@ -6316,6 +6445,12 @@ Status DpdkEngine::get_tx_packet_burst(BurstParams* burst) { return Status::INVALID_PARAMETER; } const auto& q = q_it->second; + if (q->pools.size() < static_cast(limits.num_segs)) { + DAQIRI_LOG_ERROR("TX queue has {} pool(s), burst requires {} segment(s)", + q->pools.size(), + limits.num_segs); + return Status::INVALID_PARAMETER; + } const auto burst_pool = tx_burst_buffers.find(key); if (burst_pool == tx_burst_buffers.end()) { @@ -6342,6 +6477,11 @@ Status DpdkEngine::get_tx_packet_burst(BurstParams* burst) { } Status DpdkEngine::set_eth_header(BurstParams* burst, int idx, char* dst_addr) { + if (dst_addr == nullptr) { return Status::NULL_PTR; } + Status status = validate_dpdk_first_segment_write( + burst, idx, sizeof(UDPPkt), "DpdkEngine::set_eth_header"); + if (status != Status::SUCCESS) { return status; } + auto mbuf = reinterpret_cast(burst->pkts[0][idx]); auto mbuf_data = rte_pktmbuf_mtod(mbuf, UDPPkt*); memcpy(reinterpret_cast(&mbuf_data->eth.dst_addr), @@ -6354,6 +6494,15 @@ Status DpdkEngine::set_eth_header(BurstParams* burst, int idx, char* dst_addr) { Status DpdkEngine::set_ipv4_header(BurstParams* burst, int idx, int ip_len, uint8_t proto, unsigned int src_host, unsigned int dst_host) { + if (ip_len < 0 || + static_cast(ip_len) > + std::numeric_limits::max() - sizeof(rte_ipv4_hdr)) { + return Status::INVALID_PARAMETER; + } + Status status = validate_dpdk_first_segment_write( + burst, idx, sizeof(UDPPkt), "DpdkEngine::set_ipv4_header"); + if (status != Status::SUCCESS) { return status; } + auto mbuf = reinterpret_cast(burst->pkts[0][idx]); auto mbuf_data = rte_pktmbuf_mtod(mbuf, UDPPkt*); mbuf_data->ip.next_proto_id = proto; @@ -6367,6 +6516,15 @@ Status DpdkEngine::set_ipv4_header(BurstParams* burst, int idx, int ip_len, uint Status DpdkEngine::set_udp_header(BurstParams* burst, int idx, int udp_len, uint16_t src_port, uint16_t dst_port) { + if (udp_len < 0 || + static_cast(udp_len) > + std::numeric_limits::max() - sizeof(rte_udp_hdr)) { + return Status::INVALID_PARAMETER; + } + Status status = validate_dpdk_first_segment_write( + burst, idx, sizeof(UDPPkt), "DpdkEngine::set_udp_header"); + if (status != Status::SUCCESS) { return status; } + auto mbuf = reinterpret_cast(burst->pkts[0][idx]); auto mbuf_data = rte_pktmbuf_mtod(mbuf, UDPPkt*); @@ -6378,17 +6536,59 @@ Status DpdkEngine::set_udp_header(BurstParams* burst, int idx, int udp_len, uint } Status DpdkEngine::set_udp_payload(BurstParams* burst, int idx, void* data, int len) { + size_t payload_capacity = burst_validation::kUnknownCapacity; + bool header_fits = true; + if (burst_validation::strict_enabled() && burst != nullptr && + burst_validation::validate_segment_packet_storage( + burst, + burst_validation::header_limits(burst), + 0, + idx, + true, + false, + "DpdkEngine::set_udp_payload") + == Status::SUCCESS) { + const auto* mbuf = reinterpret_cast(burst->pkts[0][idx]); + const size_t capacity = mbuf_data_capacity(mbuf); + header_fits = capacity >= sizeof(UDPPkt); + payload_capacity = capacity >= sizeof(UDPPkt) ? capacity - sizeof(UDPPkt) : 0; + } + Status status = burst_validation::validate_payload_write( + burst, + burst_validation::header_limits(burst), + idx, + data, + len, + payload_capacity, + "DpdkEngine::set_udp_payload"); + if (status != Status::SUCCESS) { return status; } + if (!header_fits) { return Status::INVALID_PARAMETER; } + auto mbuf = reinterpret_cast(burst->pkts[0][idx]); auto mbuf_data = rte_pktmbuf_mtod(mbuf, UDPPkt*); - rte_memcpy(mbuf_data->payload, data, len); + if (len > 0) { rte_memcpy(mbuf_data->payload, data, len); } return Status::SUCCESS; } bool DpdkEngine::is_tx_burst_available(BurstParams* burst) { + if (burst == nullptr) { return false; } + const auto limits = burst_validation::header_limits(burst); + if (burst_validation::validate_segment_count( + burst, limits, "DpdkEngine::is_tx_burst_available") != Status::SUCCESS) { + return false; + } + if (burst_validation::validate_packet_count( + burst, + limits, + static_cast(std::numeric_limits::max()), + "DpdkEngine::is_tx_burst_available") != Status::SUCCESS) { + return false; + } const uint32_t key = generate_queue_key(burst->hdr.hdr.port_id, burst->hdr.hdr.q_id); const auto item = tx_dpdk_q_map_.find(key); - if (item == tx_dpdk_q_map_.end()) { + if (item == tx_dpdk_q_map_.end() || + item->second->pools.size() < static_cast(limits.num_segs)) { return false; } @@ -6403,9 +6603,14 @@ bool DpdkEngine::is_tx_burst_available(BurstParams* burst) { Status DpdkEngine::set_packet_lengths(BurstParams* burst, int idx, const std::initializer_list& lens) { uint32_t ttl_len = 0; + Status status = validate_dpdk_packet_lengths( + burst, idx, lens, &ttl_len, "DpdkEngine::set_packet_lengths"); + if (status != Status::SUCCESS) { return status; } + + int seg_idx = 0; for (int seg = 0; seg < burst->hdr.hdr.num_segs; seg++) { - reinterpret_cast(burst->pkts[seg])[idx]->data_len = *(lens.begin() + seg); - ttl_len += *(lens.begin() + seg); + reinterpret_cast(burst->pkts[seg])[idx]->data_len = *(lens.begin() + seg_idx); + ++seg_idx; } reinterpret_cast(burst->pkts[0])[idx]->pkt_len = ttl_len; @@ -6418,15 +6623,34 @@ Status DpdkEngine::set_packet_lengths(BurstParams* burst, int idx, Status DpdkEngine::set_all_packet_lengths(BurstParams* burst, const std::initializer_list& lens) { if (burst == nullptr) { return Status::NULL_PTR; } + if (burst_validation::validate_segment_count( + burst, + burst_validation::header_limits(burst), + "DpdkEngine::set_all_packet_lengths") != Status::SUCCESS) { + return Status::INVALID_PARAMETER; + } + if (burst->hdr.hdr.num_pkts == 0 || + burst->hdr.hdr.num_pkts > static_cast(std::numeric_limits::max())) { + return Status::INVALID_PARAMETER; + } std::array seg_lens{}; uint32_t ttl_len = 0; + const auto num_pkts = static_cast(burst->hdr.hdr.num_pkts); + for (int idx = 0; idx < num_pkts; ++idx) { + uint32_t packet_len = 0; + Status status = validate_dpdk_packet_lengths( + burst, idx, lens, &packet_len, "DpdkEngine::set_all_packet_lengths"); + if (status != Status::SUCCESS) { return status; } + ttl_len = packet_len; + } + + int lens_idx = 0; for (int seg = 0; seg < burst->hdr.hdr.num_segs; ++seg) { - seg_lens[seg] = static_cast(*(lens.begin() + seg)); - ttl_len += seg_lens[seg]; + seg_lens[seg] = static_cast(*(lens.begin() + lens_idx)); + ++lens_idx; } - const auto num_pkts = static_cast(burst->hdr.hdr.num_pkts); for (int seg = 0; seg < burst->hdr.hdr.num_segs; ++seg) { auto** mbufs = reinterpret_cast(burst->pkts[seg]); for (int idx = 0; idx < num_pkts; ++idx) { mbufs[idx]->data_len = seg_lens[seg]; } @@ -6444,25 +6668,57 @@ void DpdkEngine::free_packet_segment(BurstParams* burst, int seg, int pkt) { if (burst == nullptr) { return; } if ((burst->hdr.hdr.burst_flags & kBurstFlagDpdkReordered) != 0U) { release_reorder_output_context(burst); - if (seg == 0 && burst->pkt_lens[0] != nullptr && pkt >= 0 - && pkt < static_cast(burst->hdr.hdr.num_pkts)) { + if (seg == 0 && burst->pkt_lens[0] != nullptr && + burst_validation::validate_packet_index( + burst, + burst_validation::header_limits(burst), + pkt, + "DpdkEngine::free_packet_segment") == Status::SUCCESS) { burst->pkt_lens[0][pkt] = 0; } return; } + if (burst_validation::validate_segment_packet_storage( + burst, + burst_validation::header_limits(burst), + seg, + pkt, + true, + false, + "DpdkEngine::free_packet_segment") != Status::SUCCESS) { + return; + } rte_pktmbuf_free_seg(reinterpret_cast(burst->pkts[seg])[pkt]); } void DpdkEngine::free_all_segment_packets(BurstParams* burst, int seg) { if (burst == nullptr) { return; } + const auto limits = burst_validation::header_limits(burst); if ((burst->hdr.hdr.burst_flags & kBurstFlagDpdkReordered) != 0U) { release_reorder_output_context(burst); - if (seg == 0 && burst->pkt_lens[0] != nullptr) { - for (int p = 0; p < burst->hdr.hdr.num_pkts; p++) { burst->pkt_lens[0][p] = 0; } + if (seg == 0 && burst->pkt_lens[0] != nullptr && + burst_validation::validate_packet_count( + burst, + limits, + static_cast(std::numeric_limits::max()), + "DpdkEngine::free_all_segment_packets") == Status::SUCCESS) { + for (int p = 0; p < static_cast(burst->hdr.hdr.num_pkts); p++) { + burst->pkt_lens[0][p] = 0; + } } return; } - for (int p = 0; p < burst->hdr.hdr.num_pkts; p++) { + if (burst_validation::validate_segment_index( + burst, limits, seg, "DpdkEngine::free_all_segment_packets") != Status::SUCCESS || + burst_validation::validate_packet_count( + burst, + limits, + static_cast(std::numeric_limits::max()), + "DpdkEngine::free_all_segment_packets") != Status::SUCCESS || + burst->pkts[seg] == nullptr) { + return; + } + for (int p = 0; p < static_cast(burst->hdr.hdr.num_pkts); p++) { rte_pktmbuf_free_seg(reinterpret_cast(burst->pkts[seg])[p]); } } @@ -6471,24 +6727,53 @@ void DpdkEngine::free_packet(BurstParams* burst, int pkt) { if (burst == nullptr) { return; } if ((burst->hdr.hdr.burst_flags & kBurstFlagDpdkReordered) != 0U) { release_reorder_output_context(burst); - if (burst->pkt_lens[0] != nullptr && pkt >= 0 && pkt < static_cast(burst->hdr.hdr.num_pkts)) { + if (burst->pkt_lens[0] != nullptr && + burst_validation::validate_packet_index( + burst, burst_validation::header_limits(burst), pkt, "DpdkEngine::free_packet") == + Status::SUCCESS) { burst->pkt_lens[0][pkt] = 0; } return; } + if (burst_validation::validate_segment_packet_storage( + burst, + burst_validation::header_limits(burst), + 0, + pkt, + true, + false, + "DpdkEngine::free_packet") != Status::SUCCESS) { + return; + } rte_pktmbuf_free(reinterpret_cast(burst->pkts[0])[pkt]); } void DpdkEngine::free_all_packets(BurstParams* burst) { if (burst == nullptr) { return; } + const auto limits = burst_validation::header_limits(burst); if ((burst->hdr.hdr.burst_flags & kBurstFlagDpdkReordered) != 0U) { release_reorder_output_context(burst); - if (burst->pkt_lens[0] != nullptr) { - for (int p = 0; p < burst->hdr.hdr.num_pkts; p++) { burst->pkt_lens[0][p] = 0; } + if (burst->pkt_lens[0] != nullptr && + burst_validation::validate_packet_count( + burst, + limits, + static_cast(std::numeric_limits::max()), + "DpdkEngine::free_all_packets") == Status::SUCCESS) { + for (int p = 0; p < static_cast(burst->hdr.hdr.num_pkts); p++) { + burst->pkt_lens[0][p] = 0; + } } return; } - for (int p = 0; p < burst->hdr.hdr.num_pkts; p++) { + if (burst_validation::validate_packet_count( + burst, + limits, + static_cast(std::numeric_limits::max()), + "DpdkEngine::free_all_packets") != Status::SUCCESS || + burst->pkts[0] == nullptr) { + return; + } + for (int p = 0; p < static_cast(burst->hdr.hdr.num_pkts); p++) { rte_pktmbuf_free(reinterpret_cast(burst->pkts[0])[p]); } } @@ -6507,8 +6792,14 @@ void DpdkEngine::free_rx_burst(BurstParams* burst) { return; } - for (int seg = 0; seg < burst->hdr.hdr.num_segs; seg++) { - rte_mempool_put(rx_burst_buffer, (void*)burst->pkts[seg]); + const auto limits = burst_validation::header_limits(burst); + if (burst_validation::validate_segment_count( + burst, limits, "DpdkEngine::free_rx_burst") == Status::SUCCESS) { + for (int seg = 0; seg < burst->hdr.hdr.num_segs; seg++) { + if (burst->pkts[seg] != nullptr) { + rte_mempool_put(rx_burst_buffer, (void*)burst->pkts[seg]); + } + } } burst->hdr.hdr.num_pkts = 0; @@ -6517,11 +6808,19 @@ void DpdkEngine::free_rx_burst(BurstParams* burst) { } void DpdkEngine::free_tx_burst(BurstParams* burst) { + if (burst == nullptr) { return; } const uint32_t key = generate_queue_key(burst->hdr.hdr.port_id, burst->hdr.hdr.q_id); const auto burst_pool = tx_burst_buffers.find(key); - for (int seg = 0; seg < burst->hdr.hdr.num_segs; seg++) { - rte_mempool_put(burst_pool->second, (void*)burst->pkts[seg]); + const auto limits = burst_validation::header_limits(burst); + if (burst_pool != tx_burst_buffers.end() && + burst_validation::validate_segment_count( + burst, limits, "DpdkEngine::free_tx_burst") == Status::SUCCESS) { + for (int seg = 0; seg < burst->hdr.hdr.num_segs; seg++) { + if (burst->pkts[seg] != nullptr) { + rte_mempool_put(burst_pool->second, (void*)burst->pkts[seg]); + } + } } burst->hdr.hdr.num_pkts = 0; @@ -6574,6 +6873,18 @@ Status DpdkEngine::get_tx_metadata_buffer(BurstParams** burst) { } Status DpdkEngine::send_tx_burst(BurstParams* burst) { + if (burst == nullptr) { return Status::NULL_PTR; } + const auto limits = burst_validation::header_limits(burst); + Status status = burst_validation::validate_segment_count( + burst, limits, "DpdkEngine::send_tx_burst"); + if (status != Status::SUCCESS) { return status; } + status = burst_validation::validate_packet_count( + burst, + limits, + static_cast(std::numeric_limits::max()), + "DpdkEngine::send_tx_burst"); + if (status != Status::SUCCESS) { return status; } + uint32_t key = generate_queue_key(burst->hdr.hdr.port_id, burst->hdr.hdr.q_id); const auto ring = tx_rings.find(key); @@ -6642,6 +6953,14 @@ void DpdkEngine::print_stats() { uint64_t DpdkEngine::get_burst_tot_byte(BurstParams* burst) { if (burst == nullptr) { return 0; } + const auto limits = burst_validation::header_limits(burst); + if (burst_validation::validate_packet_count( + burst, + limits, + static_cast(std::numeric_limits::max()), + "DpdkEngine::get_burst_tot_byte") != Status::SUCCESS) { + return 0; + } uint64_t total = 0; if ((burst->hdr.hdr.burst_flags & kBurstFlagDpdkReordered) != 0U) { if (burst->pkt_lens[0] != nullptr) { diff --git a/src/engines/ibverbs/daqiri_ibverbs_engine.cpp b/src/engines/ibverbs/daqiri_ibverbs_engine.cpp index 95627f9c..ce946453 100644 --- a/src/engines/ibverbs/daqiri_ibverbs_engine.cpp +++ b/src/engines/ibverbs/daqiri_ibverbs_engine.cpp @@ -17,6 +17,7 @@ #include "src/engines/ibverbs/daqiri_ibverbs_engine.h" #include "src/engines/ibverbs/mlx5_prm_min.h" +#include "src/burst_validation.h" #include "src/kernels.h" #include @@ -32,6 +33,7 @@ #include #include #include +#include #include #include #include @@ -2904,6 +2906,16 @@ void IbverbsEngine::free_packet(BurstParams* burst, int pkt) { if (burst == nullptr) { return; } + if (burst->hdr.hdr.burst_flags & DAQIRI_BURST_FLAG_REORDERED) { + return; + } + if (burst_validation::validate_packet_index( + burst, + burst_validation::header_limits(burst), + pkt, + "IbverbsEngine::free_packet") != Status::SUCCESS) { + return; + } IbvRxQueue* q = find_rx_queue(burst->hdr.hdr.port_id, burst->hdr.hdr.q_id); if (q == nullptr) { return; @@ -2921,6 +2933,13 @@ void IbverbsEngine::free_all_packets(BurstParams* burst) { return; } if (burst->hdr.hdr.burst_flags & IBV_TX_BURST_FLAG) { + if (burst_validation::validate_packet_count( + burst, + burst_validation::header_limits(burst), + std::numeric_limits::max(), + "IbverbsEngine::free_all_packets") != Status::SUCCESS) { + return; + } // TX burst the app allocated but never sent (a local fill failure): roll // back the allocation. The bench only does this for the just-allocated // burst, so these are the most-recent (unposted) cyclic slots. @@ -2935,6 +2954,13 @@ void IbverbsEngine::free_all_packets(BurstParams* burst) { if (q == nullptr) { return; } + if (burst_validation::validate_packet_count( + burst, + burst_validation::header_limits(burst), + std::numeric_limits::max(), + "IbverbsEngine::free_all_packets") != Status::SUCCESS) { + return; + } uint16_t* wqe_arr = burst_wqe_arr(burst); uint16_t* strd_arr = burst_strd_arr(burst); const int num = static_cast(burst->hdr.hdr.num_pkts); @@ -2983,22 +3009,74 @@ void IbverbsEngine::free_rx_metadata(BurstParams* burst) { // Accessors // --------------------------------------------------------------------------- void* IbverbsEngine::get_packet_ptr(BurstParams* burst, int idx) { - return burst->pkts[0][idx]; + return get_segment_packet_ptr(burst, 0, idx); } uint32_t IbverbsEngine::get_packet_length(BurstParams* burst, int idx) { - return burst->pkt_lens[0][idx]; + uint64_t total = 0; + const auto limits = burst_validation::header_limits(burst); + if (burst_validation::validate_packet_index( + burst, limits, idx, "IbverbsEngine::get_packet_length") != Status::SUCCESS || + burst_validation::validate_segment_count( + burst, limits, "IbverbsEngine::get_packet_length") != Status::SUCCESS) { + return 0; + } + for (int seg = 0; seg < limits.num_segs; ++seg) { + if (burst_validation::validate_segment_packet_storage( + burst, + limits, + seg, + idx, + false, + true, + "IbverbsEngine::get_packet_length") != Status::SUCCESS) { + return 0; + } + total += burst->pkt_lens[seg][idx]; + if (total > std::numeric_limits::max()) { return 0; } + } + return static_cast(total); } void* IbverbsEngine::get_segment_packet_ptr(BurstParams* burst, int seg, int idx) { + if (burst_validation::validate_segment_packet_storage( + burst, + burst_validation::header_limits(burst), + seg, + idx, + true, + false, + "IbverbsEngine::get_segment_packet_ptr") != Status::SUCCESS) { + return nullptr; + } return burst->pkts[seg][idx]; } uint32_t IbverbsEngine::get_segment_packet_length(BurstParams* burst, int seg, int idx) { + if (burst_validation::validate_segment_packet_storage( + burst, + burst_validation::header_limits(burst), + seg, + idx, + false, + true, + "IbverbsEngine::get_segment_packet_length") != Status::SUCCESS) { + return 0; + } return burst->pkt_lens[seg][idx]; } FlowId IbverbsEngine::get_packet_flow_id(BurstParams* burst, int idx) { + if (burst == nullptr || (burst->hdr.hdr.burst_flags & DAQIRI_BURST_FLAG_REORDERED) != 0U) { + return 0; + } + if (burst_validation::validate_packet_index( + burst, + burst_validation::header_limits(burst), + idx, + "IbverbsEngine::get_packet_flow_id") != Status::SUCCESS) { + return 0; + } // Per-packet MARK tag captured from the CQE (set by the flow's tag action). // Distinguishes flows that share a queue. Falls back to the per-queue flow_id // for untagged packets (tag 0), e.g. catch-all traffic or single-flow configs. @@ -3032,9 +3110,12 @@ Status IbverbsEngine::get_packet_rx_timestamp(BurstParams* burst, int idx, uint6 return Status::NULL_PTR; } *timestamp_ns = 0; - if (idx < 0 || idx >= static_cast(burst->hdr.hdr.num_pkts)) { - return Status::INVALID_PARAMETER; - } + Status status = burst_validation::validate_packet_index( + burst, + burst_validation::header_limits(burst), + idx, + "IbverbsEngine::get_packet_rx_timestamp"); + if (status != Status::SUCCESS) { return status; } // Reordered output bursts are GPU result buffers, not raw packets -- no // per-packet HW timestamp is carried. if (burst->hdr.hdr.burst_flags & DAQIRI_BURST_FLAG_REORDERED) { @@ -3055,9 +3136,17 @@ void* IbverbsEngine::get_packet_extra_info(BurstParams* burst, int idx) { } uint64_t IbverbsEngine::get_burst_tot_byte(BurstParams* burst) { + if (burst == nullptr || + burst_validation::validate_packet_count( + burst, + burst_validation::header_limits(burst), + std::numeric_limits::max(), + "IbverbsEngine::get_burst_tot_byte") != Status::SUCCESS) { + return 0; + } uint64_t tot = 0; for (size_t i = 0; i < burst->hdr.hdr.num_pkts; i++) { - tot += burst->pkt_lens[0][i]; + tot += get_packet_length(burst, static_cast(i)); } return tot; } @@ -3548,6 +3637,31 @@ IbvTxQueue* IbverbsEngine::find_tx_queue(int port, int q) { return nullptr; } +size_t IbverbsEngine::tx_segment_capacity(const BurstParams* burst, int seg) { + if (burst == nullptr || seg < 0) { return 0; } + auto* q = find_tx_queue(burst->hdr.hdr.port_id, burst->hdr.hdr.q_id); + if (q == nullptr || seg >= static_cast(q->regions.size())) { return 0; } + return q->regions[seg].slot_size; +} + +size_t IbverbsEngine::tx_payload_capacity(const BurstParams* burst) { + const size_t capacity = tx_segment_capacity(burst, 0); + return capacity >= sizeof(UDPIPV4Pkt) ? capacity - sizeof(UDPIPV4Pkt) : 0; +} + +std::array IbverbsEngine::tx_segment_capacities(const BurstParams* burst) { + std::array capacities = burst_validation::unknown_capacities(); + if (!burst_validation::strict_enabled()) { return capacities; } + if (burst == nullptr) { return capacities; } + auto* q = find_tx_queue(burst->hdr.hdr.port_id, burst->hdr.hdr.q_id); + if (q == nullptr) { return capacities; } + const int num_segs = std::min(static_cast(q->regions.size()), MAX_NUM_SEGS); + for (int seg = 0; seg < num_segs; ++seg) { + capacities[seg] = q->regions[seg].slot_size; + } + return capacities; +} + Status IbverbsEngine::create_tx_raw_qp(IbvTxQueue& q) { q.cq = ibv_create_cq(q.ctx, static_cast(q.num_slots) + 1, nullptr, nullptr, 0); if (q.cq == nullptr) { @@ -3795,19 +3909,40 @@ Status IbverbsEngine::get_tx_metadata_buffer(BurstParams** burst) { } bool IbverbsEngine::is_tx_burst_available(BurstParams* burst) { + if (burst == nullptr) { return false; } IbvTxQueue* q = find_tx_queue(burst->hdr.hdr.port_id, burst->hdr.hdr.q_id); if (q == nullptr) { return false; } + const auto limits = burst_validation::header_limits(burst); + if (burst_validation::validate_packet_count( + burst, + limits, + std::numeric_limits::max(), + "IbverbsEngine::is_tx_burst_available") != Status::SUCCESS) { + return false; + } const uint64_t in_flight = q->alloc_head - q->completed_tail.load(std::memory_order_acquire); return (q->num_slots - in_flight) >= static_cast(burst->hdr.hdr.num_pkts); } Status IbverbsEngine::get_tx_packet_burst(BurstParams* burst) { + if (burst == nullptr) { return Status::NULL_PTR; } IbvTxQueue* q = find_tx_queue(burst->hdr.hdr.port_id, burst->hdr.hdr.q_id); if (q == nullptr) { return Status::INVALID_PARAMETER; } + const auto limits = burst_validation::header_limits(burst); + Status status = burst_validation::validate_packet_count( + burst, + limits, + std::numeric_limits::max(), + "IbverbsEngine::get_tx_packet_burst"); + if (status != Status::SUCCESS) { return status; } + if (q->num_segs <= 0 || q->num_segs > MAX_NUM_SEGS || + q->regions.size() < static_cast(q->num_segs)) { + return Status::INVALID_PARAMETER; + } const unsigned n = static_cast(burst->hdr.hdr.num_pkts); burst->hdr.hdr.num_segs = q->num_segs; // Hand out the next n cyclic slots if they fit under num_slots in flight. No @@ -3836,11 +3971,21 @@ Status IbverbsEngine::get_tx_packet_burst(BurstParams* burst) { Status IbverbsEngine::set_packet_lengths(BurstParams* burst, int idx, const std::initializer_list& lens) { + const Status status = burst_validation::validate_packet_lengths( + burst, + burst_validation::header_limits(burst), + idx, + lens, + tx_segment_capacities(burst), + true, + true, + std::numeric_limits::max(), + "IbverbsEngine::set_packet_lengths", + nullptr); + if (status != Status::SUCCESS) { return status; } + int seg = 0; for (int v : lens) { - if (seg >= MAX_NUM_SEGS) { - break; - } burst->pkt_lens[seg][idx] = static_cast(v); ++seg; } @@ -3848,6 +3993,21 @@ Status IbverbsEngine::set_packet_lengths(BurstParams* burst, int idx, } Status IbverbsEngine::set_eth_header(BurstParams* burst, int idx, char* dst_addr) { + if (dst_addr == nullptr) { return Status::NULL_PTR; } + Status status = burst_validation::validate_segment_packet_storage( + burst, + burst_validation::header_limits(burst), + 0, + idx, + true, + false, + "IbverbsEngine::set_eth_header"); + if (status != Status::SUCCESS) { return status; } + if (burst_validation::strict_enabled() && + sizeof(UDPIPV4Pkt) > tx_segment_capacity(burst, 0)) { + return Status::INVALID_PARAMETER; + } + auto* pkt = static_cast(burst->pkts[0][idx]); memcpy(pkt->eth.h_dest, dst_addr, 6); pkt->eth.h_proto = htons(0x0800); // IPv4 @@ -3861,6 +4021,23 @@ Status IbverbsEngine::set_eth_header(BurstParams* burst, int idx, char* dst_addr Status IbverbsEngine::set_ipv4_header(BurstParams* burst, int idx, int ip_len, uint8_t proto, unsigned int src_host, unsigned int dst_host) { + if (ip_len < 0 || static_cast(ip_len) > std::numeric_limits::max()) { + return Status::INVALID_PARAMETER; + } + Status status = burst_validation::validate_segment_packet_storage( + burst, + burst_validation::header_limits(burst), + 0, + idx, + true, + false, + "IbverbsEngine::set_ipv4_header"); + if (status != Status::SUCCESS) { return status; } + if (burst_validation::strict_enabled() && + sizeof(UDPIPV4Pkt) > tx_segment_capacity(burst, 0)) { + return Status::INVALID_PARAMETER; + } + auto* pkt = static_cast(burst->pkts[0][idx]); pkt->ip.version = 4; pkt->ip.ihl = 5; @@ -3874,6 +4051,23 @@ Status IbverbsEngine::set_ipv4_header(BurstParams* burst, int idx, int ip_len, u Status IbverbsEngine::set_udp_header(BurstParams* burst, int idx, int udp_len, uint16_t src_port, uint16_t dst_port) { + if (udp_len < 0 || static_cast(udp_len) > std::numeric_limits::max()) { + return Status::INVALID_PARAMETER; + } + Status status = burst_validation::validate_segment_packet_storage( + burst, + burst_validation::header_limits(burst), + 0, + idx, + true, + false, + "IbverbsEngine::set_udp_header"); + if (status != Status::SUCCESS) { return status; } + if (burst_validation::strict_enabled() && + sizeof(UDPIPV4Pkt) > tx_segment_capacity(burst, 0)) { + return Status::INVALID_PARAMETER; + } + auto* pkt = static_cast(burst->pkts[0][idx]); pkt->udp.source = htons(src_port); pkt->udp.dest = htons(dst_port); @@ -3883,8 +4077,24 @@ Status IbverbsEngine::set_udp_header(BurstParams* burst, int idx, int udp_len, u } Status IbverbsEngine::set_udp_payload(BurstParams* burst, int idx, void* data, int len) { + if (burst == nullptr) { return Status::NULL_PTR; } + if (burst_validation::strict_enabled() && + tx_segment_capacity(burst, 0) < sizeof(UDPIPV4Pkt)) { + return Status::INVALID_PARAMETER; + } + const Status status = burst_validation::validate_payload_write( + burst, + burst_validation::header_limits(burst), + idx, + data, + len, + burst_validation::strict_enabled() ? tx_payload_capacity(burst) + : burst_validation::kUnknownCapacity, + "IbverbsEngine::set_udp_payload"); + if (status != Status::SUCCESS) { return status; } + auto* p = static_cast(burst->pkts[0][idx]) + sizeof(UDPIPV4Pkt); - memcpy(p, data, static_cast(len)); + if (len > 0) { memcpy(p, data, static_cast(len)); } return Status::SUCCESS; } @@ -3896,6 +4106,16 @@ Status IbverbsEngine::set_udp_payload(BurstParams* burst, int idx, void* data, i // far in the past wraps). Returns NOT_SUPPORTED when the device lacks wait-on-time // or a real-time clock. Status IbverbsEngine::set_packet_tx_time(BurstParams* burst, int idx, uint64_t time) { + Status status = burst_validation::validate_segment_packet_storage( + burst, + burst_validation::header_limits(burst), + 0, + idx, + true, + false, + "IbverbsEngine::set_packet_tx_time"); + if (status != Status::SUCCESS) { return status; } + IbvTxQueue* q = find_tx_queue(burst->hdr.hdr.port_id, burst->hdr.hdr.q_id); if (q == nullptr) { return Status::INVALID_PARAMETER; @@ -4020,6 +4240,18 @@ void IbverbsEngine::post_tx_burst(IbvTxQueue& q, BurstParams* burst) { // overlaps the next fill batch with the post of the previous one and beat the // single-thread inline model in measurement. Status IbverbsEngine::send_tx_burst(BurstParams* burst) { + if (burst == nullptr) { return Status::NULL_PTR; } + const auto limits = burst_validation::header_limits(burst); + Status status = burst_validation::validate_segment_count( + burst, limits, "IbverbsEngine::send_tx_burst"); + if (status != Status::SUCCESS) { return status; } + status = burst_validation::validate_packet_count( + burst, + limits, + std::numeric_limits::max(), + "IbverbsEngine::send_tx_burst"); + if (status != Status::SUCCESS) { return status; } + IbvTxQueue* q = find_tx_queue(burst->hdr.hdr.port_id, burst->hdr.hdr.q_id); if (q == nullptr) { return Status::INVALID_PARAMETER; diff --git a/src/engines/ibverbs/daqiri_ibverbs_engine.h b/src/engines/ibverbs/daqiri_ibverbs_engine.h index 92f2e243..6e127511 100644 --- a/src/engines/ibverbs/daqiri_ibverbs_engine.h +++ b/src/engines/ibverbs/daqiri_ibverbs_engine.h @@ -24,6 +24,7 @@ #include #include +#include #include #include #include @@ -456,6 +457,9 @@ class IbverbsEngine : public Engine { // drains each send_ring (post) + reclaims completions. void tx_worker(std::vector group); IbvTxQueue* find_tx_queue(int port, int q); + size_t tx_segment_capacity(const BurstParams* burst, int seg); + size_t tx_payload_capacity(const BurstParams* burst); + std::array tx_segment_capacities(const BurstParams* burst); // ---- helpers ---- static int mr_access_to_ibv(uint32_t access); diff --git a/src/engines/rdma/daqiri_rdma_engine.cpp b/src/engines/rdma/daqiri_rdma_engine.cpp index e9d76743..c59a7f15 100644 --- a/src/engines/rdma/daqiri_rdma_engine.cpp +++ b/src/engines/rdma/daqiri_rdma_engine.cpp @@ -21,12 +21,13 @@ #include #include #include -#include #include #include +#include #include #include "src/daqiri_ring.h" #include "src/daqiri_pool.h" +#include "src/burst_validation.h" #include "src/metrics.h" #include "daqiri_rdma_engine.h" @@ -98,28 +99,68 @@ bool RdmaEngine::set_config_and_initialize(const NetworkConfig& cfg) { return initialized_; } +size_t RdmaEngine::local_mr_capacity(const BurstParams* burst) const { + if (burst == nullptr) { return 0; } + const auto mr_it = mrs_.find(std::string(burst->transport_hdr.local_mr_name)); + if (mr_it == mrs_.end()) { return 0; } + return mr_it->second.params_.adj_size_; +} + // Common DAQIRI functions Status RdmaEngine::set_packet_lengths(BurstParams* burst, int idx, const std::initializer_list& lens) { - assert(lens.size() == 1); // Split not supported yet + std::array capacities = burst_validation::unknown_capacities(); + if (burst_validation::strict_enabled()) { capacities[0] = local_mr_capacity(burst); } + const Status status = burst_validation::validate_packet_lengths( + burst, + burst_validation::transport_limits(burst), + idx, + lens, + capacities, + true, + true, + std::numeric_limits::max(), + "RdmaEngine::set_packet_lengths", + nullptr); + if (status != Status::SUCCESS) { return status; } burst->pkt_lens[0][idx] = lens.begin()[0]; return Status::SUCCESS; } void* RdmaEngine::get_segment_packet_ptr(BurstParams* burst, int seg, int idx) { + if (burst_validation::validate_segment_packet_storage( + burst, + burst_validation::transport_limits(burst), + seg, + idx, + true, + false, + "RdmaEngine::get_segment_packet_ptr") != Status::SUCCESS) { + return nullptr; + } return burst->pkts[seg][idx]; } void* RdmaEngine::get_packet_ptr(BurstParams* burst, int idx) { - return burst->pkts[0][idx]; + return get_segment_packet_ptr(burst, 0, idx); } uint32_t RdmaEngine::get_segment_packet_length(BurstParams* burst, int seg, int idx) { + if (burst_validation::validate_segment_packet_storage( + burst, + burst_validation::transport_limits(burst), + seg, + idx, + false, + true, + "RdmaEngine::get_segment_packet_length") != Status::SUCCESS) { + return 0; + } return burst->pkt_lens[seg][idx]; } uint32_t RdmaEngine::get_packet_length(BurstParams* burst, int idx) { - return burst->pkt_lens[0][idx]; + return get_segment_packet_length(burst, 0, idx); } Status RdmaEngine::set_eth_header(BurstParams* burst, int idx, char* dst_addr) { @@ -140,7 +181,17 @@ Status RdmaEngine::set_udp_header(BurstParams* burst, int idx, int udp_len, uint } Status RdmaEngine::set_udp_payload(BurstParams* burst, int idx, void* data, int len) { - memcpy(burst->pkts[0][idx], data, len); + const Status status = burst_validation::validate_payload_write( + burst, + burst_validation::transport_limits(burst), + idx, + data, + len, + burst_validation::strict_enabled() ? local_mr_capacity(burst) + : burst_validation::kUnknownCapacity, + "RdmaEngine::set_udp_payload"); + if (status != Status::SUCCESS) { return status; } + if (len > 0) { memcpy(burst->pkts[0][idx], data, static_cast(len)); } return Status::SUCCESS; } @@ -431,7 +482,17 @@ inline int RdmaEngine::set_affinity(int cpu_core) { } Status RdmaEngine::send_tx_burst(BurstParams* burst) { - if (burst == nullptr) { return Status::INVALID_PARAMETER; } + if (burst == nullptr) { return Status::NULL_PTR; } + const auto limits = burst_validation::transport_limits(burst); + Status status = burst_validation::validate_segment_count( + burst, limits, "RdmaEngine::send_tx_burst"); + if (status != Status::SUCCESS) { return status; } + status = burst_validation::validate_packet_count( + burst, + limits, + MAX_RDMA_BATCH, + "RdmaEngine::send_tx_burst"); + if (status != Status::SUCCESS) { return status; } daqiri::Ring* ring; @@ -906,6 +967,8 @@ RDMAOpCode RdmaEngine::rdma_get_opcode(BurstParams* burst) { Status RdmaEngine::rdma_set_header(BurstParams* burst, RDMAOpCode op_code, uintptr_t conn_id, bool is_server, int num_pkts, uint64_t wr_id, const std::string& local_mr_name) { + if (burst == nullptr) { return Status::NULL_PTR; } + if (num_pkts <= 0 || num_pkts > MAX_RDMA_BATCH) { return Status::INVALID_PARAMETER; } burst->transport_hdr.opcode = op_code; set_connection_id(burst, conn_id); burst->transport_hdr.server = is_server; @@ -1195,8 +1258,16 @@ Status RdmaEngine::get_tx_packet_burst(BurstParams* burst) { if (burst == nullptr) { return Status::INVALID_PARAMETER; } // RDMA isn't allowing split segments yet - assert(burst->transport_hdr.num_segs == 1); - assert(burst->transport_hdr.num_pkts <= MAX_RDMA_BATCH); + const auto limits = burst_validation::transport_limits(burst); + Status status = burst_validation::validate_segment_count( + burst, limits, "RdmaEngine::get_tx_packet_burst"); + if (status != Status::SUCCESS) { return status; } + if (limits.num_segs != 1 || limits.num_pkts == 0 || limits.num_pkts > MAX_RDMA_BATCH) { + DAQIRI_LOG_ERROR("Invalid RDMA TX burst shape: num_pkts={} num_segs={}", + limits.num_pkts, + limits.num_segs); + return Status::INVALID_PARAMETER; + } auto burst_pool = mem_pools_.find(burst->transport_hdr.local_mr_name); if (burst_pool == mem_pools_.end()) { DAQIRI_LOG_ERROR("Failed to look up burst pool name for MR {}", @@ -1232,6 +1303,13 @@ Status RdmaEngine::get_tx_packet_burst(BurstParams* burst) { } bool RdmaEngine::is_tx_burst_available(BurstParams* burst) { + if (burst == nullptr) { return false; } + const auto limits = burst_validation::transport_limits(burst); + if (burst_validation::validate_segment_count( + burst, limits, "RdmaEngine::is_tx_burst_available") != Status::SUCCESS || + limits.num_segs != 1 || limits.num_pkts == 0 || limits.num_pkts > MAX_RDMA_BATCH) { + return false; + } auto burst_pool = mem_pools_.find(burst->transport_hdr.local_mr_name); if (burst_pool == mem_pools_.end()) { DAQIRI_LOG_ERROR("Failed to look up burst pool name for MR {}", @@ -1624,10 +1702,18 @@ void RdmaEngine::free_tx_burst(BurstParams* burst) { if (burst == nullptr) { return; } auto burst_pool = mem_pools_.find(burst->transport_hdr.local_mr_name); - if (burst->transport_hdr.num_pkts > 0 && burst->pkts[0] != nullptr && + const auto limits = burst_validation::transport_limits(burst); + const bool valid_packet_count = + burst_validation::validate_packet_count( + burst, + limits, + MAX_RDMA_BATCH, + "RdmaEngine::free_tx_burst") == Status::SUCCESS; + if (valid_packet_count && burst->pkts[0] != nullptr && burst_pool != mem_pools_.end()) { + const auto num_pkts = static_cast(burst->transport_hdr.num_pkts); unsigned ret = burst_pool->second->enqueue_bulk(reinterpret_cast(burst->pkts[0]), - burst->transport_hdr.num_pkts); + num_pkts); if (ret != burst->transport_hdr.num_pkts) { DAQIRI_LOG_CRITICAL( "Asked to free {} packets, only enqueued {}", burst->transport_hdr.num_pkts, ret); diff --git a/src/engines/rdma/daqiri_rdma_engine.h b/src/engines/rdma/daqiri_rdma_engine.h index aa614409..fe0d9fd1 100644 --- a/src/engines/rdma/daqiri_rdma_engine.h +++ b/src/engines/rdma/daqiri_rdma_engine.h @@ -17,6 +17,7 @@ #pragma once +#include #include #include #include @@ -228,6 +229,7 @@ class RdmaEngine : public Engine { int setup_thread_params(rdma_thread_params* params, bool is_server); int destroy_thread_params(rdma_thread_params* params); RDMAOpCode ibv_opcode_to_daqiri_opcode(ibv_wc_opcode opcode); + size_t local_mr_capacity(const BurstParams* burst) const; }; }; // namespace daqiri diff --git a/src/engines/socket/daqiri_socket_engine.cpp b/src/engines/socket/daqiri_socket_engine.cpp index b700243a..21dcb01a 100644 --- a/src/engines/socket/daqiri_socket_engine.cpp +++ b/src/engines/socket/daqiri_socket_engine.cpp @@ -21,6 +21,8 @@ #include "daqiri_socket_engine.h" +#include "src/burst_validation.h" + #include #include #include @@ -34,6 +36,7 @@ #include #include +#include #include #include @@ -49,6 +52,14 @@ namespace { constexpr size_t kMaxUdpPayloadBytes = 65507; +burst_validation::BurstLimits cleanup_header_limits(const BurstParams* burst) { + if (burst == nullptr) { return {}; } + return { + burst->hdr.hdr.num_pkts, + std::clamp(burst->hdr.hdr.num_segs, 0, MAX_NUM_SEGS), + }; +} + bool parse_ipv4_addr(const std::string& ip, uint16_t port, sockaddr_in* addr) { if (addr == nullptr) { return false; } @@ -200,16 +211,28 @@ uint32_t SocketEngine::get_packet_length(BurstParams* burst, int idx) { } void* SocketEngine::get_segment_packet_ptr(BurstParams* burst, int seg, int idx) { - if (burst == nullptr || seg < 0 || seg >= MAX_NUM_SEGS || burst->pkts[seg] == nullptr || idx < 0 || - idx >= static_cast(burst->hdr.hdr.num_pkts)) { + if (burst_validation::validate_segment_packet_storage( + burst, + burst_validation::header_limits(burst), + seg, + idx, + true, + false, + "SocketEngine::get_segment_packet_ptr") != Status::SUCCESS) { return nullptr; } return burst->pkts[seg][idx]; } uint32_t SocketEngine::get_segment_packet_length(BurstParams* burst, int seg, int idx) { - if (burst == nullptr || seg < 0 || seg >= MAX_NUM_SEGS || burst->pkt_lens[seg] == nullptr || idx < 0 || - idx >= static_cast(burst->hdr.hdr.num_pkts)) { + if (burst_validation::validate_segment_packet_storage( + burst, + burst_validation::header_limits(burst), + seg, + idx, + false, + true, + "SocketEngine::get_segment_packet_length") != Status::SUCCESS) { return 0; } return burst->pkt_lens[seg][idx]; @@ -239,14 +262,14 @@ Status SocketEngine::get_tx_packet_burst(BurstParams* burst) { #endif } - if (burst == nullptr) { return Status::INVALID_PARAMETER; } + if (burst == nullptr) { return Status::NULL_PTR; } const auto* ep = endpoint_for_port(burst->hdr.hdr.port_id); const auto num_pkts = burst->hdr.hdr.num_pkts > 0 ? burst->hdr.hdr.num_pkts : (ep ? ep->tx_batch_size : 1); const auto pkt_size = ep ? ep->max_packet_size : static_cast(65536); - if (num_pkts == 0) { - DAQIRI_LOG_ERROR("TX burst requested with zero packets"); + if (num_pkts == 0 || num_pkts > static_cast(std::numeric_limits::max())) { + DAQIRI_LOG_ERROR("TX burst requested with invalid packet count {}", num_pkts); return Status::INVALID_PARAMETER; } @@ -318,12 +341,18 @@ Status SocketEngine::set_udp_header(BurstParams* burst, int idx, int udp_len, ui } Status SocketEngine::set_udp_payload(BurstParams* burst, int idx, void* data, int len) { - if (burst == nullptr || burst->pkts[0] == nullptr || idx < 0 || - idx >= static_cast(burst->hdr.hdr.num_pkts) || data == nullptr || len < 0) { - return Status::INVALID_PARAMETER; - } - - std::memcpy(burst->pkts[0][idx], data, static_cast(len)); + const Status status = burst_validation::validate_payload_write( + burst, + burst_validation::header_limits(burst), + idx, + data, + len, + burst_validation::strict_enabled() ? packet_capacity(burst) + : burst_validation::kUnknownCapacity, + "SocketEngine::set_udp_payload"); + if (status != Status::SUCCESS) { return status; } + + if (len > 0) { std::memcpy(burst->pkts[0][idx], data, static_cast(len)); } return Status::SUCCESS; } @@ -350,17 +379,39 @@ Status SocketEngine::set_packet_lengths(BurstParams* burst, int idx, #endif } - if (burst == nullptr || burst->pkt_lens[0] == nullptr || idx < 0 || - idx >= static_cast(burst->hdr.hdr.num_pkts) || lens.size() == 0) { - return Status::INVALID_PARAMETER; - } + std::array capacities = burst_validation::unknown_capacities(); + if (burst_validation::strict_enabled()) { capacities[0] = packet_capacity(burst); } + const Status status = burst_validation::validate_packet_lengths( + burst, + burst_validation::header_limits(burst), + idx, + lens, + capacities, + true, + true, + std::numeric_limits::max(), + "SocketEngine::set_packet_lengths", + nullptr); + if (status != Status::SUCCESS) { return status; } burst->pkt_lens[0][idx] = static_cast(*(lens.begin())); return Status::SUCCESS; } void SocketEngine::free_all_segment_packets(BurstParams* burst, int seg) { - if (burst == nullptr || seg < 0 || seg >= MAX_NUM_SEGS || burst->pkts[seg] == nullptr) { return; } + if (burst_validation::validate_segment_index( + burst, + cleanup_header_limits(burst), + seg, + "SocketEngine::free_all_segment_packets") != Status::SUCCESS || + burst_validation::validate_packet_count( + burst, + burst_validation::header_limits(burst), + static_cast(std::numeric_limits::max()), + "SocketEngine::free_all_segment_packets") != Status::SUCCESS || + burst->pkts[seg] == nullptr) { + return; + } for (size_t i = 0; i < burst->hdr.hdr.num_pkts; ++i) { auto* pkt = reinterpret_cast(burst->pkts[seg][i]); delete[] pkt; @@ -369,16 +420,26 @@ void SocketEngine::free_all_segment_packets(BurstParams* burst, int seg) { } void SocketEngine::free_all_packets(BurstParams* burst) { - if (burst == nullptr) { return; } - const int num_segs = std::clamp(burst->hdr.hdr.num_segs, 0, MAX_NUM_SEGS); - for (int seg = 0; seg < num_segs; ++seg) { + if (burst_validation::validate_segment_count( + burst, + cleanup_header_limits(burst), + "SocketEngine::free_all_packets") != Status::SUCCESS) { + return; + } + for (int seg = 0; seg < cleanup_header_limits(burst).num_segs; ++seg) { free_all_segment_packets(burst, seg); } } void SocketEngine::free_packet_segment(BurstParams* burst, int seg, int pkt) { - if (burst == nullptr || seg < 0 || seg >= MAX_NUM_SEGS || burst->pkts[seg] == nullptr || pkt < 0 || - pkt >= static_cast(burst->hdr.hdr.num_pkts)) { + if (burst_validation::validate_segment_packet_storage( + burst, + cleanup_header_limits(burst), + seg, + pkt, + true, + false, + "SocketEngine::free_packet_segment") != Status::SUCCESS) { return; } auto* data = reinterpret_cast(burst->pkts[seg][pkt]); @@ -547,7 +608,14 @@ void SocketEngine::print_stats() { } uint64_t SocketEngine::get_burst_tot_byte(BurstParams* burst) { - if (burst == nullptr || burst->pkt_lens[0] == nullptr) { return 0; } + if (burst == nullptr || burst->pkt_lens[0] == nullptr || + burst_validation::validate_packet_count( + burst, + burst_validation::header_limits(burst), + static_cast(std::numeric_limits::max()), + "SocketEngine::get_burst_tot_byte") != Status::SUCCESS) { + return 0; + } uint64_t total = 0; for (size_t i = 0; i < burst->hdr.hdr.num_pkts; ++i) { total += burst->pkt_lens[0][i]; @@ -837,7 +905,19 @@ Status SocketEngine::send_tx_burst(BurstParams* burst) { #endif } - if (burst == nullptr || burst->pkts[0] == nullptr || burst->pkt_lens[0] == nullptr) { + if (burst == nullptr) { return Status::NULL_PTR; } + const auto limits = burst_validation::header_limits(burst); + Status status = burst_validation::validate_segment_count( + burst, limits, "SocketEngine::send_tx_burst"); + if (status != Status::SUCCESS) { return status; } + status = burst_validation::validate_packet_count( + burst, + limits, + static_cast(std::numeric_limits::max()), + "SocketEngine::send_tx_burst"); + if (status != Status::SUCCESS) { return status; } + + if (burst->pkts[0] == nullptr || burst->pkt_lens[0] == nullptr) { return Status::INVALID_PARAMETER; } @@ -868,7 +948,7 @@ Status SocketEngine::send_tx_burst(BurstParams* burst) { } } - Status status = Status::SUCCESS; + status = Status::SUCCESS; size_t sent_pkts = 0; uint64_t sent_bytes = 0; const auto requested_pkts = static_cast(burst->hdr.hdr.num_pkts); @@ -987,6 +1067,12 @@ uint32_t SocketEngine::select_batch_size(const std::vector& queue return static_cast(std::max(1, queues.front().common_.batch_size_)); } +size_t SocketEngine::packet_capacity(const BurstParams* burst) const { + if (burst == nullptr) { return 0; } + const auto* ep = endpoint_for_port(burst->hdr.hdr.port_id); + return ep != nullptr ? ep->max_packet_size : static_cast(65536); +} + std::shared_ptr SocketEngine::register_connection( int fd, uint16_t port, uint16_t queue, int if_index, bool server_side, bool is_udp, const std::shared_ptr& rx_queue, bool start_rx_thread) { diff --git a/src/engines/socket/daqiri_socket_engine.h b/src/engines/socket/daqiri_socket_engine.h index dcb8ff17..66689805 100644 --- a/src/engines/socket/daqiri_socket_engine.h +++ b/src/engines/socket/daqiri_socket_engine.h @@ -18,6 +18,7 @@ #pragma once #include +#include #include #include #include @@ -181,6 +182,7 @@ class SocketEngine : public Engine { uint16_t select_queue_id(const std::vector& queues) const; uint16_t select_queue_id(const std::vector& queues) const; uint32_t select_batch_size(const std::vector& queues) const; + size_t packet_capacity(const BurstParams* burst) const; Status pop_rx_burst(const std::shared_ptr& qstate, BurstParams** burst); void push_rx_burst(const std::shared_ptr& qstate, BurstParams* burst);