diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 6900e30f..50e5b8d8 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -60,6 +60,7 @@ add_subdirectory(lib) add_subdirectory(tools) if(ENABLE_TESTS) + enable_testing() add_subdirectory(tests) add_subdirectory(utests) endif(ENABLE_TESTS) diff --git a/src/soapysdr/CMakeLists.txt b/src/soapysdr/CMakeLists.txt index fdeb801c..48ecb442 100644 --- a/src/soapysdr/CMakeLists.txt +++ b/src/soapysdr/CMakeLists.txt @@ -17,7 +17,7 @@ SOAPY_SDR_MODULE_UTIL( DESTINATION ${CMAKE_INSTALL_LIBDIR}/SoapySDR/modules${SOAPY_SDR_ABI_VERSION}/ SOURCES - usdr_soapy.cpp usdr_soapy_reg.cpp + usdr_soapy.cpp usdr_soapy_reg.cpp rx_packet_buffer.cpp LIBRARIES usdr ) @@ -25,6 +25,16 @@ SOAPY_SDR_MODULE_UTIL( if(ENABLE_TESTS) add_executable(test_usdr_soapy tests/test_usdr_soapy.c) target_link_libraries(test_usdr_soapy ${SoapySDR_LIBRARIES}) + set_target_properties(test_usdr_soapy PROPERTIES + RUNTIME_OUTPUT_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/tests + ) + + add_executable(test_rx_packet_buffer tests/test_rx_packet_buffer.cpp rx_packet_buffer.cpp) + target_link_libraries(test_rx_packet_buffer usdr) + set_target_properties(test_rx_packet_buffer PROPERTIES + RUNTIME_OUTPUT_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/tests + ) + add_test(NAME rx_packet_buffer COMMAND $) endif(ENABLE_TESTS) option(ENABLE_SOAPY_HIL_TESTS "Enable SoapySDR hardware-in-the-loop tests" OFF) diff --git a/src/soapysdr/README.md b/src/soapysdr/README.md new file mode 100644 index 00000000..858411b8 --- /dev/null +++ b/src/soapysdr/README.md @@ -0,0 +1,72 @@ +# SoapySDR USDR support + +This directory contains the `usdr` SoapySDR module. + +## Device discovery + +Use the `driver=usdr` filter when listing USDR devices. Without the driver key, +SoapySDR asks every installed module to enumerate, and unrelated modules such +as audio devices may appear in the output. + +USB devices can be selected by their topology path, for example +`bus=usb@3/1/6`: `3` is the USB bus, `1` is the downstream port, and `6` is the +device address. This is useful when several USDR boards are connected to stable +USB ports. It is not a serial number and may change if the board is moved to +another port or hub, and the device/address part may also change after +unplugging and plugging the board back into the same port. + +```sh +SoapySDRUtil --find="driver=usdr" +SoapySDRUtil --find="driver=usdr,bus=usb@3/1/6" +SoapySDRUtil --find="driver=usdr,bus=pci,device=usdr0" +``` + +## Soapy Parameters + +Device arguments are passed when enumerating or opening the device, for example +`SoapySDRUtil --find="driver=usdr,bus=usb@3/1/6"` or +`SoapySDR.Device({"driver": "usdr", "bus": "usb@3/1/6"})`. + +| Argument | Scope | Description | +| --- | --- | --- | +| `driver=usdr` | discovery/open | Selects the USDR Soapy module. Use this to avoid unrelated devices from other Soapy modules. | +| `bus=` | discovery/open | Selects a USB or PCI bus path, for example `usb@3/1/6` or `pci`. | +| `device=` | discovery/open | Selects a lower-level device name, for example `usdr0` with `bus=pci`. | +| `dev=` | open | Packed lower-level device string. Explicit `bus`, `device`, `fe`, `extclk`, and `extref` arguments override values from `dev`. | +| `fe=` | discovery/open | Selects a frontend when supported by the lower level. | +| `extclk=` | discovery/open | Passes external clock selection to the lower level. | +| `extref=` | discovery/open | Passes external reference selection to the lower level. | +| `rxGapFill=none` | open | Default RX timestamp gap fill mode for streams created by this device. | +| `rxGapFill=zero` | open | Default RX timestamp gap fill mode that fills timestamp gaps with zero samples. | + +Stream arguments are passed to `setupStream()`. + +| Argument | Direction | Default | Description | +| --- | --- | --- | --- | +| `bufferLength=` | RX/TX | automatic | Hardware packet size over the link. RX values must be either `0`/automatic or in the supported range checked by the driver. | +| `linkFormat=CS16` | RX/TX | `CS16` | Complex int16 link format. TX currently supports `CS16` only. | +| `linkFormat=CS12` | RX | `CS16` | Complex int12 link format for RX when the user stream format is `CF32`. | +| `floatScale=1.0` | RX/TX | `1.0` | Stream float scaling. Values other than `1.0` are currently rejected. | +| `rxGapFill=none` | RX | device default | Keeps only real samples and exposes packet loss as a timestamp jump. | +| `rxGapFill=zero` | RX | device default | Keeps only real samples internally, but fills timestamp gaps with zero samples when data is returned from `readStream()`. | + +Advanced and debug open arguments are intended for development and diagnostics: + +| Argument | Description | +| --- | --- | +| `loglevel=` | Overrides USDR log level. The `SOAPY_USDR_LOGLEVEL` environment variable can also set the default on Linux. | +| `calls=1` | Enables verbose Soapy call logging. | +| `desired_rx_pkt=` | Overrides the default RX packet size used when `bufferLength` is not specified. | +| `rx12bit=1` | Forces RX wire format to 12-bit mode. | +| `rxdump=` | Dumps received samples to a file for debugging. | +| `txcorr=` | Applies the existing TX correction/debug path. | +| `refclk=` | Recognized for reference clock selection; currently logs the request. | +| `rx_bw=` / `tx_bw=` | Applies a bandwidth value at open time. Prefer standard `setBandwidth()` for normal applications. | + +`SOAPY_USDR_ARGS` may be used to override open arguments from the environment +with a packed Soapy kwargs string. + +## Testing + +See [tests/README.md](tests/README.md) for Python and C hardware-in-the-loop +smoke tests, stream buffer checks, and CTest integration. diff --git a/src/soapysdr/rx_packet_buffer.cpp b/src/soapysdr/rx_packet_buffer.cpp new file mode 100644 index 00000000..fc36518c --- /dev/null +++ b/src/soapysdr/rx_packet_buffer.cpp @@ -0,0 +1,257 @@ +// Copyright (c) 2026 Wavelet Lab +// SPDX-License-Identifier: MIT + +#include "rx_packet_buffer.h" + +#include +#include +#include + +RxPacketBuffer::RxPacketBuffer(pusdr_dms_t stream, + unsigned channels, + size_t samples_per_packet, + size_t bytes_per_packet, + GapFill gap_fill, + RecvFunction recv_function) + : _stream(stream) + , _recv(recv_function ? recv_function : &usdr_dms_recv) + , _channels(channels) + , _samples_per_packet(samples_per_packet) + , _bytes_per_packet(bytes_per_packet) + , _bytes_per_sample(bytes_per_packet / samples_per_packet) + , _gap_fill(gap_fill) + , _capacity(0) + , _read_pos(0) + , _write_pos(0) + , _available(0) + , _next_output_time(0) + , _time_valid(false) +{ + if (_stream == nullptr || _channels == 0 || _samples_per_packet == 0 || + _bytes_per_packet == 0 || _bytes_per_sample == 0 || + (_bytes_per_packet % _samples_per_packet) != 0) { + throw std::invalid_argument("RxPacketBuffer: invalid stream packet configuration"); + } + + _packet_buffers.resize(_channels); + _packet_ptrs.resize(_channels); + for (unsigned i = 0; i < _channels; i++) { + _packet_buffers[i].resize(_bytes_per_packet); + _packet_ptrs[i] = _packet_buffers[i].data(); + } + + ensureCapacity(_bytes_per_packet * 16); +} + +size_t RxPacketBuffer::bytesPerElems(size_t elems) const +{ + return elems * _bytes_per_sample; +} + +size_t RxPacketBuffer::elemsPerBytes(size_t bytes) const +{ + return bytes / _bytes_per_sample; +} + +void RxPacketBuffer::ensureCapacity(size_t requested_bytes) +{ + if (requested_bytes <= _capacity) { + return; + } + + size_t new_capacity = std::max(_bytes_per_packet * 16, _capacity); + if (new_capacity == 0) { + new_capacity = _bytes_per_packet * 16; + } + while (new_capacity < requested_bytes) { + new_capacity *= 2; + } + + std::vector> new_buffers(_channels); + for (unsigned ch = 0; ch < _channels; ch++) { + new_buffers[ch].resize(new_capacity); + if (_available == 0) { + continue; + } + + const size_t first = std::min(_available, _capacity - _read_pos); + std::memcpy(new_buffers[ch].data(), _buffers[ch].data() + _read_pos, first); + if (first < _available) { + std::memcpy(new_buffers[ch].data() + first, _buffers[ch].data(), _available - first); + } + } + + _buffers.swap(new_buffers); + _capacity = new_capacity; + _read_pos = 0; + _write_pos = _available; +} + +void RxPacketBuffer::appendPacket(const usdr_dms_recv_nfo_t &nfo) +{ + const size_t reported_samples = (nfo.totsyms != 0) ? nfo.totsyms : _samples_per_packet; + const size_t packet_samples = std::min(reported_samples, _samples_per_packet); + const size_t packet_bytes = bytesPerElems(packet_samples); + + if (!_time_valid) { + _next_output_time = nfo.fsymtime; + _time_valid = true; + } + + if (!_segments.empty()) { + const Segment &last = _segments.back(); + const dm_time_t expected_time = last.start_time + (dm_time_t)last.samples; + if (expected_time == nfo.fsymtime) { + _segments.back().samples += packet_samples; + } else if (_gap_fill == GAP_FILL_ZERO && nfo.fsymtime > expected_time) { + _segments.push_back({nfo.fsymtime, packet_samples}); + } else { + dropBufferedData(); + _next_output_time = nfo.fsymtime; + _time_valid = true; + _segments.push_back({nfo.fsymtime, packet_samples}); + } + } else if (nfo.fsymtime < _next_output_time || + (_gap_fill == GAP_FILL_NONE && nfo.fsymtime != _next_output_time)) { + dropBufferedData(); + _next_output_time = nfo.fsymtime; + _time_valid = true; + _segments.push_back({nfo.fsymtime, packet_samples}); + } else { + _segments.push_back({nfo.fsymtime, packet_samples}); + } + + ensureCapacity(_available + packet_bytes); + for (unsigned ch = 0; ch < _channels; ch++) { + const size_t first = std::min(packet_bytes, _capacity - _write_pos); + std::memcpy(_buffers[ch].data() + _write_pos, _packet_buffers[ch].data(), first); + if (first < packet_bytes) { + std::memcpy(_buffers[ch].data(), _packet_buffers[ch].data() + first, packet_bytes - first); + } + } + + _write_pos = (_write_pos + packet_bytes) % _capacity; + _available += packet_bytes; +} + +void RxPacketBuffer::readRealBytes(void * const *buffs, size_t dst_offset_bytes, size_t bytes) +{ + for (unsigned ch = 0; ch < _channels; ch++) { + unsigned char *dst = static_cast(buffs[ch]) + dst_offset_bytes; + const size_t first = std::min(bytes, _capacity - _read_pos); + std::memcpy(dst, _buffers[ch].data() + _read_pos, first); + if (first < bytes) { + std::memcpy(dst + first, _buffers[ch].data(), bytes - first); + } + } + + _read_pos = (_read_pos + bytes) % _capacity; + _available -= bytes; + size_t samples = elemsPerBytes(bytes); + _next_output_time += (dm_time_t)samples; + while (samples != 0 && !_segments.empty()) { + Segment &segment = _segments.front(); + if (samples < segment.samples) { + segment.start_time += (dm_time_t)samples; + segment.samples -= samples; + break; + } + samples -= segment.samples; + _segments.pop_front(); + } +} + +void RxPacketBuffer::writeZeros(void * const *buffs, size_t dst_offset_bytes, size_t bytes) +{ + for (unsigned ch = 0; ch < _channels; ch++) { + unsigned char *dst = static_cast(buffs[ch]) + dst_offset_bytes; + std::memset(dst, 0, bytes); + } + _next_output_time += (dm_time_t)elemsPerBytes(bytes); +} + +size_t RxPacketBuffer::outputAvailableSamples() const +{ + if (!_time_valid || _segments.empty()) { + return 0; + } + + dm_time_t cursor = _next_output_time; + size_t samples = 0; + for (const Segment &segment: _segments) { + if (segment.start_time > cursor) { + if (_gap_fill != GAP_FILL_ZERO) { + break; + } + samples += (size_t)(segment.start_time - cursor); + cursor = segment.start_time; + } else if (segment.start_time < cursor) { + return 0; + } + + samples += segment.samples; + cursor += (dm_time_t)segment.samples; + } + return samples; +} + +void RxPacketBuffer::dropBufferedData() +{ + _read_pos = 0; + _write_pos = 0; + _available = 0; + _segments.clear(); +} + +int RxPacketBuffer::read(void * const *buffs, + size_t elems, + long timeout_us, + dm_time_t &sample_time, + usdr_dms_recv_nfo_t &nfo) +{ + const size_t requested_bytes = bytesPerElems(elems); + ensureCapacity(requested_bytes + _bytes_per_packet); + + while (outputAvailableSamples() < elems) { + const int res = _recv(_stream, _packet_ptrs.data(), timeout_us / 1000, &nfo); + if (res != 0) { + return res; + } + appendPacket(nfo); + } + + sample_time = _next_output_time; + size_t copied = 0; + while (copied < requested_bytes) { + if (_segments.empty()) { + break; + } + + const Segment &first = _segments.front(); + if (first.start_time > _next_output_time) { + const size_t zero_samples = std::min((size_t)(first.start_time - _next_output_time), + elemsPerBytes(requested_bytes - copied)); + const size_t zero_bytes = bytesPerElems(zero_samples); + writeZeros(buffs, copied, zero_bytes); + copied += zero_bytes; + continue; + } + + const size_t real_bytes = std::min(bytesPerElems(first.samples), requested_bytes - copied); + readRealBytes(buffs, copied, real_bytes); + copied += real_bytes; + } + return 0; +} + +void RxPacketBuffer::reset() +{ + dropBufferedData(); + _next_output_time = 0; + _time_valid = false; +} + +bool RxPacketBuffer::empty() const +{ + return _segments.empty() && _available == 0; +} diff --git a/src/soapysdr/rx_packet_buffer.h b/src/soapysdr/rx_packet_buffer.h new file mode 100644 index 00000000..3b968496 --- /dev/null +++ b/src/soapysdr/rx_packet_buffer.h @@ -0,0 +1,79 @@ +// Copyright (c) 2026 Wavelet Lab +// SPDX-License-Identifier: MIT + +#ifndef RX_PACKET_BUFFER_H +#define RX_PACKET_BUFFER_H + +#include +#include +#include +#include + +#include "../lib/models/dm_stream.h" + +class RxPacketBuffer +{ +public: + enum GapFill { + GAP_FILL_NONE, + GAP_FILL_ZERO + }; + + typedef int (*RecvFunction)(pusdr_dms_t stream, + void **buffs, + unsigned timeout_ms, + usdr_dms_recv_nfo_t *nfo); + + RxPacketBuffer(pusdr_dms_t stream, + unsigned channels, + size_t samples_per_packet, + size_t bytes_per_packet, + GapFill gap_fill = GAP_FILL_NONE, + RecvFunction recv_function = nullptr); + + int read(void * const *buffs, + size_t elems, + long timeout_us, + dm_time_t &sample_time, + usdr_dms_recv_nfo_t &nfo); + + void reset(); + bool empty() const; + +private: + struct Segment { + dm_time_t start_time; + size_t samples; + }; + + size_t bytesPerElems(size_t elems) const; + size_t elemsPerBytes(size_t bytes) const; + void ensureCapacity(size_t requested_bytes); + void appendPacket(const usdr_dms_recv_nfo_t &nfo); + void readRealBytes(void * const *buffs, size_t dst_offset_bytes, size_t bytes); + void writeZeros(void * const *buffs, size_t dst_offset_bytes, size_t bytes); + size_t outputAvailableSamples() const; + void dropBufferedData(); + + pusdr_dms_t _stream; + RecvFunction _recv; + unsigned _channels; + size_t _samples_per_packet; + size_t _bytes_per_packet; + size_t _bytes_per_sample; + GapFill _gap_fill; + + std::vector> _buffers; + std::vector> _packet_buffers; + std::vector _packet_ptrs; + std::deque _segments; + + size_t _capacity; + size_t _read_pos; + size_t _write_pos; + size_t _available; + dm_time_t _next_output_time; + bool _time_valid; +}; + +#endif diff --git a/src/soapysdr/tests/README.md b/src/soapysdr/tests/README.md index 7f7c632a..779713ea 100644 --- a/src/soapysdr/tests/README.md +++ b/src/soapysdr/tests/README.md @@ -1,9 +1,10 @@ -# SoapySDR USDR hardware tests +# SoapySDR USDR tests -This directory contains hardware-in-the-loop tests for the `usdr` SoapySDR -module. They are intentionally Python scripts, not unit tests: the goal is to -exercise the installed Soapy API against a real board and print a readable -capability report. +This directory contains SoapySDR tests for the `usdr` module. The Python and C +smoke tests are hardware-in-the-loop tests: they exercise the installed Soapy +API against a real board and print a readable capability report. The +`test_rx_packet_buffer` target is a local unit test for packet buffering logic +and does not require hardware. ## Quick control-plane smoke test @@ -11,8 +12,8 @@ capability report. python3 src/soapysdr/tests/soapy_usdr_hil.py --device "driver=usdr" ``` -The script enumerates/open the device, detects available RX/TX software and -hardware channels, checks common Get/List functions, validates ranges, and +The script enumerates and opens the device, detects available RX/TX software +and hardware channels, checks common Get/List functions, validates ranges, and round-trips safe control values for sample rate, frequency, bandwidth, and gain where supported. @@ -27,7 +28,32 @@ python3 src/soapysdr/tests/soapy_usdr_hil.py \ ``` The stream test needs Python `numpy`, because SoapySDR Python bindings expect -array-like sample buffers for `readStream()`. +array-like sample buffers for `readStream()`. When RX streaming is enabled, the +test also reads several sizes different from `bufferLength` to exercise the +packet buffering path and timestamp continuity. + +By default, timestamp gaps are not filled and appear as timestamp jumps. Use +`--rx-gap-fill zero` to request zero-filled gaps: + +```sh +python3 src/soapysdr/tests/soapy_usdr_hil.py \ + --device "driver=usdr" \ + --rx-stream \ + --rx-gap-fill zero +``` + +## TX streaming smoke test + +TX streaming is opt-in because it transmits samples: + +```sh +python3 src/soapysdr/tests/soapy_usdr_hil.py \ + --device "driver=usdr" \ + --tx-stream +``` + +The TX smoke test writes more samples than the stream MTU in one `writeStream()` +call, so it exercises the Soapy-side chunking path. ## C API smoke test @@ -37,15 +63,30 @@ from `test_usdr_soapy.c`: ```sh cmake -S src -B build -DENABLE_TESTS=ON cmake --build build --target test_usdr_soapy -build/soapysdr/test_usdr_soapy -Q +build/soapysdr/tests/test_usdr_soapy -Q ``` Use `-Q` for query/control-plane checks only. Omit it to include RX streaming: ```sh -build/soapysdr/test_usdr_soapy -c 2 -i 4096 -n 4 +build/soapysdr/tests/test_usdr_soapy -c 2 -i 4096 -n 4 ``` +The C RX stream smoke test also performs variable-size `readStream()` calls +over the configured hardware packet size. Add `-Z` to enable zero-filled RX +timestamp gaps, or `-T` to include the TX `writeStream()` chunking smoke test. + +## Packet buffer unit test + +```sh +cmake -S src -B build -DENABLE_TESTS=ON +cmake --build build --target test_rx_packet_buffer +build/soapysdr/tests/test_rx_packet_buffer +``` + +This test validates variable RX read sizes, timestamp-gap no-fill mode, and +zero-fill mode including very large virtual gaps. + ## CTest integration Hardware tests are opt-in so CI without an SDR device remains green: @@ -55,3 +96,9 @@ cmake -S src -B build -DENABLE_TESTS=ON -DENABLE_SOAPY_HIL_TESTS=ON cmake --build build ctest --test-dir build -R soapy_usdr_hil --output-on-failure ``` + +The packet-buffer unit test can be run without hardware: + +```sh +ctest --test-dir build -R rx_packet_buffer --output-on-failure +``` diff --git a/src/soapysdr/tests/soapy_usdr_hil.py b/src/soapysdr/tests/soapy_usdr_hil.py index 790fb663..1fb3f7d2 100755 --- a/src/soapysdr/tests/soapy_usdr_hil.py +++ b/src/soapysdr/tests/soapy_usdr_hil.py @@ -533,15 +533,43 @@ def check_rx_stream(runner: Runner, dev: Any, args: argparse.Namespace) -> None: frequency = pick_in_range(freq_ranges, args.frequency) bandwidth = pick_in_range(bw_ranges, args.bandwidth) if bw_ranges else 0.0 + def parse_sample_sizes(value: str) -> List[int]: + sizes: List[int] = [] + for item in value.split(","): + item = item.strip() + if not item: + continue + size = int(item) + if size <= 0: + raise ValueError(f"stream read size should be positive: {size}") + sizes.append(size) + return sizes + + def check_next_timestamp(prev: Dict[str, Any], item: Dict[str, Any]) -> None: + if prev["timeNs"] == 0 or item["timeNs"] == 0: + return + delta_samples = round((item["timeNs"] - prev["timeNs"]) * sample_rate / 1e9) + if abs(delta_samples - prev["ret"]) > 1: + raise AssertionError( + f"timestamp step mismatch: expected {prev['ret']} samples, got {delta_samples}" + ) + def run_stream() -> Dict[str, Any]: dev.setSampleRate(SOAPY_SDR_RX, channel, sample_rate) dev.setFrequency(SOAPY_SDR_RX, channel, frequency) if bandwidth > 0: dev.setBandwidth(SOAPY_SDR_RX, channel, bandwidth) - stream_args = {"bufferLength": str(args.rx_samples), "linkFormat": args.link_format} + stream_args = { + "bufferLength": str(args.rx_samples), + "linkFormat": args.link_format, + "rxGapFill": args.rx_gap_fill, + } stream = dev.setupStream(SOAPY_SDR_RX, "CF32", [channel], stream_args) reads: List[Dict[str, Any]] = [] + variable_reads: List[Dict[str, Any]] = [] + variable_sizes = parse_sample_sizes(args.rx_variable_sizes) + max_read_size = max([args.rx_samples] + variable_sizes) try: mtu = int(dev.getStreamMTU(stream)) inactive = dev.readStream(stream, [np.empty(args.rx_samples, np.complex64)], args.rx_samples, timeoutUs=10000) @@ -560,14 +588,80 @@ def run_stream() -> Dict[str, Any]: "mean_abs": float(np.mean(np.abs(buff[:ret]))), } ) + prev_read: Optional[Dict[str, Any]] = reads[-1] if reads else None + for requested in variable_sizes: + buff = np.empty(max_read_size, np.complex64) + result = dev.readStream(stream, [buff], requested, timeoutUs=args.timeout_ms * 1000) + ret = int(getattr(result, "ret", result)) + if ret != requested: + raise AssertionError(f"readStream requested {requested}, returned {ret}") + item = { + "requested": requested, + "ret": ret, + "flags": int(getattr(result, "flags", 0)), + "timeNs": int(getattr(result, "timeNs", 0)), + "mean_abs": float(np.mean(np.abs(buff[:ret]))), + } + if prev_read is not None: + check_next_timestamp(prev_read, item) + variable_reads.append(item) + prev_read = item dev.deactivateStream(stream) - return {"channel": channel, "mtu": mtu, "inactive_read_ret": int(getattr(inactive, "ret", inactive)), "reads": reads} + return { + "channel": channel, + "mtu": mtu, + "inactive_read_ret": int(getattr(inactive, "ret", inactive)), + "reads": reads, + "variable_reads": variable_reads, + } finally: dev.closeStream(stream) runner.report["rx_stream"] = runner.check("RX setup/activate/read/deactivate/close stream", run_stream) +def check_tx_stream(runner: Runner, dev: Any, args: argparse.Namespace) -> None: + tx_channels = int(dev.getNumChannels(SOAPY_SDR_TX)) + if tx_channels <= 0: + runner.skip("TX stream", "device has no TX channels") + return + + try: + import numpy as np + except Exception as exc: + runner.skip("TX stream", f"numpy is not available: {exc}") + return + + channel = min(args.tx_channel, tx_channels - 1) + sample_rate = pick_in_range(dev.getSampleRateRange(SOAPY_SDR_TX, channel), args.sample_rate) + frequency = pick_in_range(dev.getFrequencyRange(SOAPY_SDR_TX, channel), args.frequency) + + def run_stream() -> Dict[str, Any]: + dev.setSampleRate(SOAPY_SDR_TX, channel, sample_rate) + dev.setFrequency(SOAPY_SDR_TX, channel, frequency) + stream = dev.setupStream(SOAPY_SDR_TX, "CF32", [channel], {"bufferLength": str(args.tx_packet_samples)}) + active = False + try: + mtu = int(dev.getStreamMTU(stream)) + write_elems = max(args.tx_samples, mtu * 2) + buff = np.zeros(write_elems, np.complex64) + dev.activateStream(stream) + active = True + result = dev.writeStream(stream, [buff], write_elems, timeoutUs=args.timeout_ms * 1000) + ret = int(getattr(result, "ret", result)) + dev.deactivateStream(stream) + active = False + if ret != write_elems: + raise AssertionError(f"writeStream requested {write_elems}, returned {ret}") + return {"channel": channel, "mtu": mtu, "requested": write_elems, "ret": ret} + finally: + if active: + dev.deactivateStream(stream) + dev.closeStream(stream) + + runner.report["tx_stream"] = runner.check("TX large writeStream chunking", run_stream) + + def parse_args(argv: Optional[Sequence[str]] = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--device", default="driver=usdr", help="SoapySDR device string") @@ -579,6 +673,16 @@ def parse_args(argv: Optional[Sequence[str]] = None) -> argparse.Namespace: parser.add_argument("--rx-channel", type=int, default=0, help="RX channel to stream") parser.add_argument("--rx-samples", type=int, default=4096, help="Samples per RX read") parser.add_argument("--rx-reads", type=int, default=4, help="Number of RX reads") + parser.add_argument( + "--rx-variable-sizes", + default="17,1024,4095,4096,4097,8193", + help="Comma-separated RX read sizes used to test packet buffering", + ) + parser.add_argument("--rx-gap-fill", default="none", choices=("none", "zero"), help="RX timestamp gap fill mode") + parser.add_argument("--tx-stream", action="store_true", help="Run TX writeStream chunking smoke test") + parser.add_argument("--tx-channel", type=int, default=0, help="TX channel to stream") + parser.add_argument("--tx-packet-samples", type=int, default=4096, help="Hardware packet size for TX stream setup") + parser.add_argument("--tx-samples", type=int, default=8192, help="Minimum TX samples to write in one large call") parser.add_argument("--link-format", default="CS16", choices=("CS16", "CS12"), help="RX link format") parser.add_argument("--timeout-ms", type=int, default=1000, help="Stream timeout") parser.add_argument("--json", dest="json_path", default="", help="Optional JSON report path") @@ -614,6 +718,8 @@ def main(argv: Optional[Sequence[str]] = None) -> int: check_control_plane(runner, dev, args) if args.rx_stream: check_rx_stream(runner, dev, args) + if args.tx_stream: + check_tx_stream(runner, dev, args) finally: if hasattr(dev, "close"): dev.close() diff --git a/src/soapysdr/tests/test_rx_packet_buffer.cpp b/src/soapysdr/tests/test_rx_packet_buffer.cpp new file mode 100644 index 00000000..f5eab4b8 --- /dev/null +++ b/src/soapysdr/tests/test_rx_packet_buffer.cpp @@ -0,0 +1,221 @@ +// Copyright (c) 2026 Wavelet Lab +// SPDX-License-Identifier: MIT + +#include "../rx_packet_buffer.h" + +#include +#include +#include +#include +#include + +struct FakeStream +{ + unsigned channels = 2; + unsigned samples_per_packet = 4; + dm_time_t next_time = 100; + dm_time_t gap_time = 108; + unsigned recv_count = 0; + bool jump_after_first_packet = false; +}; + +static int fake_recv(pusdr_dms_t stream, void **buffs, unsigned /*timeout_ms*/, usdr_dms_recv_nfo_t *nfo) +{ + FakeStream *fake = reinterpret_cast(stream); + + for (unsigned ch = 0; ch < fake->channels; ch++) { + int16_t *out = static_cast(buffs[ch]); + for (unsigned i = 0; i < fake->samples_per_packet; i++) { + out[i] = (int16_t)(ch * 1000 + fake->recv_count * fake->samples_per_packet + i); + } + } + + nfo->fsymtime = fake->next_time; + nfo->totsyms = fake->samples_per_packet; + nfo->totlost = 0; + nfo->max_parts = 0; + nfo->extra = 0; + + fake->recv_count++; + if (fake->jump_after_first_packet && fake->recv_count == 1) { + fake->next_time = fake->gap_time; + } else { + fake->next_time += fake->samples_per_packet; + } + + return 0; +} + +static void assert_samples(const std::vector &samples, int first_value) +{ + for (size_t i = 0; i < samples.size(); i++) { + assert(samples[i] == first_value + (int)i); + } +} + +static void test_variable_read_sizes() +{ + FakeStream fake; + RxPacketBuffer buffer(reinterpret_cast(&fake), + fake.channels, + fake.samples_per_packet, + fake.samples_per_packet * sizeof(int16_t), + RxPacketBuffer::GAP_FILL_NONE, + fake_recv); + + usdr_dms_recv_nfo_t nfo = {}; + dm_time_t sample_time = 0; + + std::vector ch0(5); + std::vector ch1(5); + void *buffs[] = { ch0.data(), ch1.data() }; + + int res = buffer.read(buffs, 3, 100000, sample_time, nfo); + assert(res == 0); + assert(sample_time == 100); + assert(fake.recv_count == 1); + assert_samples(std::vector(ch0.begin(), ch0.begin() + 3), 0); + assert_samples(std::vector(ch1.begin(), ch1.begin() + 3), 1000); + + std::memset(ch0.data(), 0, ch0.size() * sizeof(ch0[0])); + std::memset(ch1.data(), 0, ch1.size() * sizeof(ch1[0])); + res = buffer.read(buffs, 5, 100000, sample_time, nfo); + assert(res == 0); + assert(sample_time == 103); + assert(fake.recv_count == 2); + assert_samples(ch0, 3); + assert_samples(ch1, 1003); + + std::memset(ch0.data(), 0, ch0.size() * sizeof(ch0[0])); + std::memset(ch1.data(), 0, ch1.size() * sizeof(ch1[0])); + res = buffer.read(buffs, 2, 100000, sample_time, nfo); + assert(res == 0); + assert(sample_time == 108); + assert(fake.recv_count == 3); + assert_samples(std::vector(ch0.begin(), ch0.begin() + 2), 8); + assert_samples(std::vector(ch1.begin(), ch1.begin() + 2), 1008); +} + +static void test_timestamp_gap_drops_buffered_tail() +{ + FakeStream fake; + fake.jump_after_first_packet = true; + RxPacketBuffer buffer(reinterpret_cast(&fake), + fake.channels, + fake.samples_per_packet, + fake.samples_per_packet * sizeof(int16_t), + RxPacketBuffer::GAP_FILL_NONE, + fake_recv); + + usdr_dms_recv_nfo_t nfo = {}; + dm_time_t sample_time = 0; + + std::vector ch0(4); + std::vector ch1(4); + void *buffs[] = { ch0.data(), ch1.data() }; + + int res = buffer.read(buffs, 2, 100000, sample_time, nfo); + assert(res == 0); + assert(sample_time == 100); + assert(fake.recv_count == 1); + assert_samples(std::vector(ch0.begin(), ch0.begin() + 2), 0); + + res = buffer.read(buffs, 4, 100000, sample_time, nfo); + assert(res == 0); + assert(sample_time == 108); + assert(fake.recv_count == 2); + assert_samples(ch0, 4); + assert_samples(ch1, 1004); +} + +static void test_timestamp_gap_zero_fill() +{ + FakeStream fake; + fake.jump_after_first_packet = true; + fake.gap_time = 108; + RxPacketBuffer buffer(reinterpret_cast(&fake), + fake.channels, + fake.samples_per_packet, + fake.samples_per_packet * sizeof(int16_t), + RxPacketBuffer::GAP_FILL_ZERO, + fake_recv); + + usdr_dms_recv_nfo_t nfo = {}; + dm_time_t sample_time = 0; + + std::vector ch0(8, -1); + std::vector ch1(8, -1); + void *buffs[] = { ch0.data(), ch1.data() }; + + int res = buffer.read(buffs, 2, 100000, sample_time, nfo); + assert(res == 0); + assert(sample_time == 100); + assert(fake.recv_count == 1); + assert_samples(std::vector(ch0.begin(), ch0.begin() + 2), 0); + + res = buffer.read(buffs, 8, 100000, sample_time, nfo); + assert(res == 0); + assert(sample_time == 102); + assert(fake.recv_count == 2); + assert(ch0[0] == 2); + assert(ch0[1] == 3); + assert(ch1[0] == 1002); + assert(ch1[1] == 1003); + for (size_t i = 2; i < 6; i++) { + assert(ch0[i] == 0); + assert(ch1[i] == 0); + } + assert(ch0[6] == 4); + assert(ch0[7] == 5); + assert(ch1[6] == 1004); + assert(ch1[7] == 1005); +} + +static void test_large_timestamp_gap_zero_fill_is_virtual() +{ + FakeStream fake; + fake.jump_after_first_packet = true; + fake.gap_time = 1000000000ULL; + RxPacketBuffer buffer(reinterpret_cast(&fake), + fake.channels, + fake.samples_per_packet, + fake.samples_per_packet * sizeof(int16_t), + RxPacketBuffer::GAP_FILL_ZERO, + fake_recv); + + usdr_dms_recv_nfo_t nfo = {}; + dm_time_t sample_time = 0; + + std::vector ch0(6, -1); + std::vector ch1(6, -1); + void *buffs[] = { ch0.data(), ch1.data() }; + + int res = buffer.read(buffs, 2, 100000, sample_time, nfo); + assert(res == 0); + assert(sample_time == 100); + assert(fake.recv_count == 1); + + res = buffer.read(buffs, 6, 100000, sample_time, nfo); + assert(res == 0); + assert(sample_time == 102); + assert(fake.recv_count == 2); + assert(ch0[0] == 2); + assert(ch0[1] == 3); + assert(ch1[0] == 1002); + assert(ch1[1] == 1003); + for (size_t i = 2; i < ch0.size(); i++) { + assert(ch0[i] == 0); + assert(ch1[i] == 0); + } +} + +int main() +{ + test_variable_read_sizes(); + test_timestamp_gap_drops_buffered_tail(); + test_timestamp_gap_zero_fill(); + test_large_timestamp_gap_zero_fill_is_virtual(); + + std::cout << "RxPacketBuffer tests passed" << std::endl; + return 0; +} diff --git a/src/soapysdr/tests/test_usdr_soapy.c b/src/soapysdr/tests/test_usdr_soapy.c index 2dfc59ef..01a87cd1 100644 --- a/src/soapysdr/tests/test_usdr_soapy.c +++ b/src/soapysdr/tests/test_usdr_soapy.c @@ -27,8 +27,8 @@ static int fail(const char *what) static void usage(const char *argv0) { - printf("Usage: %s [-d bus] [-c channels] [-i packet_size] [-n reads] [-r rate] [-f freq] [-b bw] [-g gain] [-Q]\n", argv0); - printf(" -d bus USDR bus string, e.g. usb@3/3/6 or pci,device=/dev/usdr0\n"); + printf("Usage: %s [-d bus] [-c channels] [-i packet_size] [-n reads] [-r rate] [-f freq] [-b bw] [-g gain] [-Q] [-T] [-Z]\n", argv0); + printf(" -d bus USDR bus string, e.g. usb@3/1/6 or pci,device=/dev/usdr0\n"); printf(" -c channels RX channels to stream, default 1\n"); printf(" -i samples RX samples per read, default 4096\n"); printf(" -n reads RX reads, default 4\n"); @@ -37,6 +37,8 @@ static void usage(const char *argv0) printf(" -b bw Preferred bandwidth, default 1e6\n"); printf(" -g gain Preferred gain, default 15\n"); printf(" -Q Query/control-plane only, skip RX stream\n"); + printf(" -T Include TX writeStream chunking smoke test\n"); + printf(" -Z Fill RX timestamp gaps with zero samples\n"); } static void print_kwargs(const SoapySDRKwargs *kwargs) @@ -345,7 +347,21 @@ static int check_channel(SoapySDRDevice *sdr, int direction, const char *label, return EXIT_SUCCESS; } -static int run_rx_stream(SoapySDRDevice *sdr, unsigned channels, unsigned packet_size, unsigned reads) +static int check_timestamp_step(long long prev_time_ns, int prev_ret, long long time_ns, double sample_rate) +{ + if (prev_time_ns == 0 || time_ns == 0 || prev_ret <= 0 || sample_rate <= 0.0) { + return EXIT_SUCCESS; + } + double delta_samples = (double)(time_ns - prev_time_ns) * sample_rate / 1e9; + if (delta_samples < (double)prev_ret - 1.0 || delta_samples > (double)prev_ret + 1.0) { + printf("Timestamp step mismatch: expected %d samples, got %.3f\n", prev_ret, delta_samples); + return EXIT_FAILURE; + } + return EXIT_SUCCESS; +} + +static int run_rx_stream(SoapySDRDevice *sdr, unsigned channels, unsigned packet_size, unsigned reads, + double sample_rate, bool zero_fill_gaps) { int status = EXIT_FAILURE; SoapySDRStream *rx_stream = NULL; @@ -367,7 +383,8 @@ static int run_rx_stream(SoapySDRDevice *sdr, unsigned channels, unsigned packet } for (unsigned i = 0; i < channels; i++) { act_channels[i] = i; - const size_t buffer_size = 2u * packet_size * sizeof(float); + const size_t max_read_size = packet_size * 2u + 1u; + const size_t buffer_size = 2u * max_read_size * sizeof(float); if (usdr_alignalloc(&buffs[i], STREAM_ALIGN, buffer_size) != 0) { goto cleanup; } @@ -379,8 +396,10 @@ static int run_rx_stream(SoapySDRDevice *sdr, unsigned channels, unsigned packet SoapySDRKwargs stream_args = {}; SoapySDRKwargs_set(&stream_args, "bufferLength", packet_size_str); SoapySDRKwargs_set(&stream_args, "linkFormat", SOAPY_SDR_CS16); + SoapySDRKwargs_set(&stream_args, "rxGapFill", zero_fill_gaps ? "zero" : "none"); - printf("\nRX stream: channels=%u packet_size=%u reads=%u\n", channels, packet_size, reads); + printf("\nRX stream: channels=%u packet_size=%u reads=%u gap_mode=%s\n", + channels, packet_size, reads, zero_fill_gaps ? "zero" : "none"); #if (SOAPY_SDR_API_VERSION < 0x00080000) if (SoapySDRDevice_setupStream(sdr, &rx_stream, SOAPY_SDR_RX, SOAPY_SDR_CF32, act_channels, channels, &stream_args) != 0) { SoapySDRKwargs_clear(&stream_args); @@ -409,6 +428,8 @@ static int run_rx_stream(SoapySDRDevice *sdr, unsigned channels, unsigned packet fail("activateStream"); goto cleanup; } + long long prev_time_ns = 0; + int prev_ret = 0; for (unsigned i = 0; i < reads; i++) { flags = 0; time_ns = 0; @@ -418,7 +439,43 @@ static int run_rx_stream(SoapySDRDevice *sdr, unsigned channels, unsigned packet SoapySDRDevice_deactivateStream(sdr, rx_stream, 0, 0); goto cleanup; } + if (check_timestamp_step(prev_time_ns, prev_ret, time_ns, sample_rate) != EXIT_SUCCESS) { + SoapySDRDevice_deactivateStream(sdr, rx_stream, 0, 0); + goto cleanup; + } + prev_time_ns = time_ns; + prev_ret = ret; + } + + const unsigned variable_sizes[] = { + 17u, + packet_size / 2u, + packet_size - 1u, + packet_size, + packet_size + 1u, + packet_size * 2u + 1u + }; + printf("RX variable read sizes:"); + for (size_t i = 0; i < sizeof(variable_sizes) / sizeof(variable_sizes[0]); i++) { + const unsigned requested = variable_sizes[i]; + flags = 0; + time_ns = 0; + ret = SoapySDRDevice_readStream(sdr, rx_stream, buffs, requested, &flags, &time_ns, 100000); + printf(" %u=>%d", requested, ret); + if (ret != (int)requested) { + printf("\n"); + SoapySDRDevice_deactivateStream(sdr, rx_stream, 0, 0); + goto cleanup; + } + if (check_timestamp_step(prev_time_ns, prev_ret, time_ns, sample_rate) != EXIT_SUCCESS) { + printf("\n"); + SoapySDRDevice_deactivateStream(sdr, rx_stream, 0, 0); + goto cleanup; + } + prev_time_ns = time_ns; + prev_ret = ret; } + printf("\n"); if (SoapySDRDevice_deactivateStream(sdr, rx_stream, 0, 0) != 0) { fail("deactivateStream"); goto cleanup; @@ -441,6 +498,81 @@ static int run_rx_stream(SoapySDRDevice *sdr, unsigned channels, unsigned packet return status; } +static int run_tx_stream(SoapySDRDevice *sdr, unsigned packet_size) +{ + int status = EXIT_FAILURE; + SoapySDRStream *tx_stream = NULL; + size_t tx_channels = SoapySDRDevice_getNumChannels(sdr, SOAPY_SDR_TX); + if (tx_channels == 0) { + printf("\nSKIP TX stream: device has no TX channels\n"); + return EXIT_SUCCESS; + } + + size_t channel = 0; + void *buff = NULL; + const size_t write_size = packet_size * 2u; + const size_t buffer_size = 2u * write_size * sizeof(float); + if (usdr_alignalloc(&buff, STREAM_ALIGN, buffer_size) != 0) { + return EXIT_FAILURE; + } + memset(buff, 0, buffer_size); + const void *buffs[1] = {buff}; + + char packet_size_str[32]; + snprintf(packet_size_str, sizeof(packet_size_str), "%u", packet_size); + SoapySDRKwargs stream_args = {}; + SoapySDRKwargs_set(&stream_args, "bufferLength", packet_size_str); + size_t channels[1] = {channel}; + + printf("\nTX stream chunking: channel=%zu packet_size=%u write_size=%zu\n", channel, packet_size, write_size); +#if (SOAPY_SDR_API_VERSION < 0x00080000) + if (SoapySDRDevice_setupStream(sdr, &tx_stream, SOAPY_SDR_TX, SOAPY_SDR_CF32, channels, 1, &stream_args) != 0) { + SoapySDRKwargs_clear(&stream_args); + fail("setupStream(TX)"); + goto cleanup; + } +#else + tx_stream = SoapySDRDevice_setupStream(sdr, SOAPY_SDR_TX, SOAPY_SDR_CF32, channels, 1, &stream_args); + if (tx_stream == NULL) { + SoapySDRKwargs_clear(&stream_args); + fail("setupStream(TX)"); + goto cleanup; + } +#endif + SoapySDRKwargs_clear(&stream_args); + + size_t mtu = SoapySDRDevice_getStreamMTU(sdr, tx_stream); + printf("TX stream MTU: %zu\n", mtu); + if (SoapySDRDevice_activateStream(sdr, tx_stream, 0, 0, 0) != 0) { + fail("activateStream(TX)"); + goto cleanup; + } + int flags = 0; + int ret = SoapySDRDevice_writeStream(sdr, tx_stream, buffs, write_size, &flags, 0, 100000); + printf("writeStream ret=%d flags=%d\n", ret, flags); + if (SoapySDRDevice_deactivateStream(sdr, tx_stream, 0, 0) != 0) { + fail("deactivateStream(TX)"); + goto cleanup; + } + if (ret != (int)write_size) { + goto cleanup; + } + if (SoapySDRDevice_closeStream(sdr, tx_stream) != 0) { + tx_stream = NULL; + fail("closeStream(TX)"); + goto cleanup; + } + tx_stream = NULL; + status = EXIT_SUCCESS; + +cleanup: + if (tx_stream != NULL) { + SoapySDRDevice_closeStream(sdr, tx_stream); + } + usdr_alignfree(buff); + return status; +} + int main(int argc, char **argv) { const char *device = ""; @@ -452,9 +584,11 @@ int main(int argc, char **argv) double bandwidth = 1e6; double gain = 15.0; bool query_only = false; + bool tx_stream = false; + bool zero_fill_gaps = false; int opt; - while ((opt = getopt(argc, argv, "hd:c:i:n:r:f:b:g:Q")) != -1) { + while ((opt = getopt(argc, argv, "hd:c:i:n:r:f:b:g:QTZ")) != -1) { switch (opt) { case 'd': device = optarg; break; case 'c': channels = (unsigned)atoi(optarg); break; @@ -465,6 +599,8 @@ int main(int argc, char **argv) case 'b': bandwidth = atof(optarg); break; case 'g': gain = atof(optarg); break; case 'Q': query_only = true; break; + case 'T': tx_stream = true; break; + case 'Z': zero_fill_gaps = true; break; case 'h': default: usage(argv[0]); @@ -520,7 +656,10 @@ int main(int argc, char **argv) status = check_channel(sdr, SOAPY_SDR_TX, "TX", i, sample_rate, rx_freq, bandwidth, gain); } if (status == EXIT_SUCCESS && !query_only) { - status = run_rx_stream(sdr, channels, packet_size, reads); + status = run_rx_stream(sdr, channels, packet_size, reads, sample_rate, zero_fill_gaps); + } + if (status == EXIT_SUCCESS && tx_stream) { + status = run_tx_stream(sdr, packet_size); } int unmake_status = SoapySDRDevice_unmake(sdr); diff --git a/src/soapysdr/usdr_soapy.cpp b/src/soapysdr/usdr_soapy.cpp index 259c8043..a23e02bb 100644 --- a/src/soapysdr/usdr_soapy.cpp +++ b/src/soapysdr/usdr_soapy.cpp @@ -57,6 +57,17 @@ bool usdrSoapyIsDeviceArg(const std::string &key) return key == "dev" || USDR_SOAPY_DEVICE_ARGS.count(key) != 0; } +static RxPacketBuffer::GapFill parse_rx_gap_fill(const std::string &value, const char *context) +{ + if (value == "none") { + return RxPacketBuffer::GAP_FILL_NONE; + } + if (value == "zero") { + return RxPacketBuffer::GAP_FILL_ZERO; + } + throw std::runtime_error(std::string(context) + "([rxGapFill=" + value + "]) unsupported mode"); +} + std::shared_ptr usdr_handle::get(const std::string& name) { auto idx = s_created.find(name); @@ -431,6 +442,9 @@ SoapyUSDR::SoapyUSDR(const SoapySDR::Kwargs &args_orig) if (args.count("calls")) { _dump_calls = atoi(args.at("calls").c_str()) ? true : false; } + if (args.count("rxGapFill")) { + _rx_gap_fill = parse_rx_gap_fill(args.at("rxGapFill"), "SoapyUSDR::SoapyUSDR"); + } usdrlog_setlevel(NULL, loglevel); @@ -1671,6 +1685,19 @@ SoapySDR::ArgInfoList SoapyUSDR::getStreamArgsInfo(const int direction, const si argInfos.push_back(info); } + if (direction == SOAPY_SDR_RX) { + SoapySDR::ArgInfo info; + info.key = "rxGapFill"; + info.name = "RX Gap Fill"; + info.description = "How RX stream buffering fills timestamp gaps."; + info.type = SoapySDR::ArgInfo::STRING; + info.options.push_back("none"); + info.optionNames.push_back("Expose timestamp jumps"); + info.options.push_back("zero"); + info.optionNames.push_back("Fill missing samples with zeroes"); + info.value = "none"; + argInfos.push_back(info); + } return argInfos; } @@ -1709,6 +1736,7 @@ SoapySDR::Stream *SoapyUSDR::setupStream( } unsigned pktSamples = 0; + RxPacketBuffer::GapFill rx_gap_fill = _rx_gap_fill; if (args.count("linkFormat")) { const std::string& link_fmt = args.at("linkFormat"); @@ -1740,6 +1768,10 @@ SoapySDR::Stream *SoapyUSDR::setupStream( } } + if (args.count("rxGapFill")) { + rx_gap_fill = parse_rx_gap_fill(args.at("rxGapFill"), "SoapyUSDR::setupStream"); + } + if (direction == SOAPY_SDR_RX && _force_rx_wire12bit) { wire12bit = true; } @@ -1813,6 +1845,12 @@ SoapySDR::Stream *SoapyUSDR::setupStream( if (direction == SOAPY_SDR_RX) { _rx_log_chans = num_channels; + ustr->rx_gap_fill = rx_gap_fill; + ustr->rx_direct_buffs.resize(num_channels); + ustr->rxbuf.reset(new RxPacketBuffer(ustr->strm, num_channels, + ustr->nfo.pktsyms, + ustr->nfo.pktbszie, + rx_gap_fill)); } else { _tx_log_chans = num_channels; } @@ -1833,12 +1871,7 @@ void SoapyUSDR::closeStream(SoapySDR::Stream *stream) ustr->strm = NULL; } - if (ustr->rxcbuf.size() > 0) { - for (unsigned i = 0; i < ustr->rxcbuf.size(); i++) { - ring_circbuf_destroy(ustr->rxcbuf[i]); - } - ustr->rxcbuf.resize(0); - } + ustr->rxbuf.reset(); ustr->setup = false; } @@ -1913,89 +1946,53 @@ int SoapyUSDR::readStream( numElems = std::min(numElems, (size_t)ustr->nfo.pktsyms); } - if (ustr->rxcbuf.size() > 0) { - size_t req_bytes = numElems * ustr->nfo.pktbszie / ustr->nfo.pktsyms; - do { - // fprintf(stderr, "rxcb wpos=%lld rpos=%lld req_bytes=%lld\n", - // (long long)ustr->rxcbuf->wpos, - // (long long)ustr->rxcbuf->rpos, - // (long long)req_bytes); - - bool have_all_channels = true; - for (unsigned i = 0; i < ustr->rxcbuf.size(); i++) { - if (ring_circbuf_rspace(ustr->rxcbuf[i]) < req_bytes) { - have_all_channels = false; - break; - } - } - - if (have_all_channels) { - for (unsigned i = 0; i < ustr->rxcbuf.size(); i++) { - // TODO: decide how to handle alignment requirements for user-provided stream buffers. - ring_circbuf_read(ustr->rxcbuf[i], buffs[i], req_bytes); - } - - flags &= ~SOAPY_SDR_HAS_TIME; - timeNs = 0; - return numElems; - } + if (!ustr->rxbuf) { + return SOAPY_SDR_STREAM_ERROR; + } - // We don't have enough data here - std::vector chans(ustr->rxcbuf.size()); - for (unsigned i = 0; i < ustr->rxcbuf.size(); i++) { - chans[i] = ring_circbuf_wptr(ustr->rxcbuf[i]); - } + dm_time_t sample_time = 0; + size_t returned_elems = numElems; + const bool direct_packet = + (ustr->rx_gap_fill == RxPacketBuffer::GAP_FILL_NONE) && + ustr->rxbuf->empty() && + (numElems == (size_t)ustr->nfo.pktsyms); - res = usdr_dms_recv(ustr->strm, chans.data(), timeoutUs / 1000, &nfo); - if (res == 0) { - for (unsigned i = 0; i < ustr->rxcbuf.size(); i++) { - ustr->rxcbuf[i]->wpos += ustr->nfo.pktbszie; - } - last_recv_pkt_time = nfo.fsymtime; - } - } while (res == 0); + if (direct_packet) { + if (ustr->rx_direct_buffs.size() < _rx_log_chans) { + ustr->rx_direct_buffs.resize(_rx_log_chans); + } + for (unsigned i = 0; i < _rx_log_chans; i++) { + ustr->rx_direct_buffs[i] = buffs[i]; + } - return SOAPY_SDR_TIMEOUT; + res = usdr_dms_recv(ustr->strm, ustr->rx_direct_buffs.data(), timeoutUs / 1000, &nfo); + if (res) { + return SOAPY_SDR_TIMEOUT; + } + sample_time = nfo.fsymtime; + returned_elems = (nfo.totsyms != 0) ? std::min((size_t)nfo.totsyms, numElems) : numElems; } else { - if (numElems != ustr->nfo.pktsyms) { - size_t blksz; - blksz = ustr->nfo.pktsyms * 16; - while (blksz < numElems * 2) - blksz <<= 1; - - size_t blksz_bytes = blksz * ustr->nfo.pktbszie / ustr->nfo.pktsyms; - - SoapySDR::logf(SOAPY_SDR_ERROR, "SoapyUSDR::readStream(%s) requested %d but block is configured for %d, injecting jitter buffer of %d bytes. Performance will be degraded", - ustr->stream, numElems, ustr->nfo.pktsyms, blksz_bytes); - - ustr->rxcbuf.resize(_rx_log_chans); - for (unsigned i = 0; i < _rx_log_chans; i++) { - ustr->rxcbuf[i] = ring_circbuf_create(blksz_bytes); - } - - // Reenter - return readStream(stream, buffs, numElems, flags, timeNs, timeoutUs); + res = ustr->rxbuf->read(buffs, numElems, timeoutUs, sample_time, nfo); + if (res) { + return SOAPY_SDR_TIMEOUT; } + } - // TODO: decide how to handle alignment requirements for user-provided stream buffers. - res = usdr_dms_recv(ustr->strm, (void**)buffs, timeoutUs / 1000, &nfo); - - if (rd && res == 0) { - const size_t bytes_per_channel = nfo.totsyms * ustr->nfo.pktbszie / ustr->nfo.pktsyms; - for (unsigned i = 0; i < _rx_log_chans; i++) { - fwrite(buffs[i], bytes_per_channel, 1, rd); - } - - const float marker[2] = { -2.0f, 2.0f }; - fwrite(marker, sizeof(marker), 1, rd); + if (rd) { + const size_t bytes_per_channel = returned_elems * ustr->nfo.pktbszie / ustr->nfo.pktsyms; + for (unsigned i = 0; i < _rx_log_chans; i++) { + fwrite(buffs[i], bytes_per_channel, 1, rd); } - flags |= SOAPY_SDR_HAS_TIME; - timeNs = SoapySDR::ticksToTimeNs(nfo.fsymtime, _actual_rx_rate); - - last_recv_pkt_time = nfo.fsymtime; - return (res) ? SOAPY_SDR_TIMEOUT : nfo.totsyms; + const float marker[2] = { -2.0f, 2.0f }; + fwrite(marker, sizeof(marker), 1, rd); } + + flags |= SOAPY_SDR_HAS_TIME; + timeNs = SoapySDR::ticksToTimeNs((long long)sample_time, _actual_rx_rate); + + last_recv_pkt_time = sample_time + returned_elems; + return returned_elems; } int SoapyUSDR::writeStream(SoapySDR::Stream *stream, @@ -2022,13 +2019,39 @@ int SoapyUSDR::writeStream(SoapySDR::Stream *stream, avg_gap = (1 - alpha) * avg_gap + alpha * lag; } - SoapySDR::logf(SOAPY_SDR_DEBUG, "writeStream::writeStream(%s) @ %lld num %d should be %d\n", ustr->stream, ts, numElems, ustr->nfo.pktsyms); + SoapySDR::logf(SOAPY_SDR_DEBUG, "writeStream::writeStream(%s) @ %lld num %d mtu %d\n", + ustr->stream, ts, (unsigned)numElems, ustr->nfo.pktsyms); + + const size_t mtu = ustr->nfo.pktsyms; + if (mtu == 0 || ustr->nfo.pktbszie == 0 || _tx_log_chans == 0 || + (ustr->nfo.pktbszie % ustr->nfo.pktsyms) != 0) { + return SOAPY_SDR_STREAM_ERROR; + } + + const size_t bytes_per_sample = ustr->nfo.pktbszie / ustr->nfo.pktsyms; + std::vector chunk_buffs(_tx_log_chans); + size_t sent = 0; + int res = 0; + while (sent < numElems) { + const size_t chunk_elems = std::min(mtu, numElems - sent); + const size_t offset = sent * bytes_per_sample; + for (unsigned i = 0; i < _tx_log_chans; i++) { + chunk_buffs[i] = static_cast(buffs[i]) + offset; + } + + const dm_time_t chunk_ts = (ts >= 0) ? (dm_time_t)(ts + sent) : (dm_time_t)-1; + // TODO: decide how to handle alignment requirements for user-provided stream buffers. + res = usdr_dms_send(ustr->strm, chunk_buffs.data(), (unsigned)chunk_elems, + chunk_ts, timeoutUs / 1000); + if (res) { + break; + } + + sent += chunk_elems; + } - unsigned toSend = numElems; - // TODO: decide how to handle alignment requirements for user-provided stream buffers. - int res = usdr_dms_send(ustr->strm, (const void **) buffs, numElems, ts, timeoutUs / 1000); if (this->calc_ts >= 0) - this->calc_ts += numElems; + this->calc_ts += sent; if (tx_pkts % 1000 == 0) { SoapySDR::logf(_dump_calls ? SOAPY_SDR_ERROR : SOAPY_SDR_TRACE, @@ -2036,7 +2059,7 @@ int SoapyUSDR::writeStream(SoapySDR::Stream *stream, } tx_pkts++; - return (res) ? SOAPY_SDR_TIMEOUT : toSend; + return (res && sent == 0) ? SOAPY_SDR_TIMEOUT : (int)sent; } int SoapyUSDR::readStreamStatus( diff --git a/src/soapysdr/usdr_soapy.h b/src/soapysdr/usdr_soapy.h index 0811430b..04fc9eeb 100644 --- a/src/soapysdr/usdr_soapy.h +++ b/src/soapysdr/usdr_soapy.h @@ -10,11 +10,10 @@ #include #include #include +#include +#include "rx_packet_buffer.h" #include "../lib/models/dm_all.h" -extern "C" { -#include "../common/ring_circbuf.h" -} SoapySDR::Kwargs usdrSoapyDeviceArgs(const SoapySDR::Kwargs &args); std::string usdrSoapyDeviceString(const SoapySDR::Kwargs &args); @@ -440,7 +439,9 @@ class SoapyUSDR : public SoapySDR::Device bool setup = false; std::atomic active; - std::vector rxcbuf; + RxPacketBuffer::GapFill rx_gap_fill = RxPacketBuffer::GAP_FILL_NONE; + std::vector rx_direct_buffs; + std::unique_ptr rxbuf; }; uint64_t max_sw_chans(const int direction) const; @@ -463,6 +464,7 @@ class SoapyUSDR : public SoapySDR::Device unsigned _desired_rx_pkt; + RxPacketBuffer::GapFill _rx_gap_fill = RxPacketBuffer::GAP_FILL_NONE; bool _force_rx_wire12bit = false; bool _dump_calls = false; diff --git a/src/tools/python/example_fft_rx_soapy.py b/src/tools/python/example_fft_rx_soapy.py index 0396725a..c77f8486 100755 --- a/src/tools/python/example_fft_rx_soapy.py +++ b/src/tools/python/example_fft_rx_soapy.py @@ -16,7 +16,7 @@ Parameter `device` can be any SoapySDR device string, e.g. "driver=usdr" or "driver=usdr,bus=" device_bus can be: - bus=pci,device=, where is the PCI device Path (e.g. /dev/usdr0) - - bus=usb@, where is the USB address (e.g. 3/3/6 for bus 3, device 3, function 6) + - bus=usb@, where is the USB address (e.g. 3/1/6 for bus 3, port 1, device 6) """ from __future__ import annotations diff --git a/src/tools/python/example_fft_tx_soapy.py b/src/tools/python/example_fft_tx_soapy.py index 6d012da3..b6223fac 100755 --- a/src/tools/python/example_fft_tx_soapy.py +++ b/src/tools/python/example_fft_tx_soapy.py @@ -17,7 +17,7 @@ Parameter `device` can be any SoapySDR device string, e.g. "driver=usdr" or "driver=usdr,bus=" device_bus can be: - bus=pci,device=, where is the PCI device Path (e.g. /dev/usdr0) - - bus=usb@, where is the USB address (e.g. 3/3/6 for bus 3, device 3, function 6) + - bus=usb@, where is the USB address (e.g. 3/1/6 for bus 3, port 1, device 6) """ from __future__ import annotations