diff --git a/NOTICE b/NOTICE index ff1b993a0b..a388a3a9b5 100644 --- a/NOTICE +++ b/NOTICE @@ -84,3 +84,11 @@ Nuand: https://github.com/Nuand/bladeRF/tree/master?tab=License-1-ov-file Credits for https://github.com/pothosware/SoapySDR Pothosware: BSL 1.0 License + +Credits for https://github.com/zeromq/libzmq +ZeroMQ authors: Mozilla Public License Version 2.0 + +Credits for source code: + - radio/zmq/zmq_imported.cpp + - radio/zmq/zmq_imported.h +Software Radio Systems Limited: BSD-3-Clause-Open-MPI diff --git a/ci-scripts/docker/Dockerfile.unittest.ubuntu b/ci-scripts/docker/Dockerfile.unittest.ubuntu index 3c95c678e8..722da5f0db 100644 --- a/ci-scripts/docker/Dockerfile.unittest.ubuntu +++ b/ci-scripts/docker/Dockerfile.unittest.ubuntu @@ -13,11 +13,12 @@ RUN apt-get update && \ DEBIAN_FRONTEND=noninteractive apt-get upgrade --yes && \ DEBIAN_FRONTEND=noninteractive apt-get install --yes \ libgtest-dev \ - libyaml-cpp-dev + libyaml-cpp-dev \ + libzmq3-dev RUN rm -Rf /oai-ran WORKDIR /oai-ran COPY . . WORKDIR /oai-ran/build -RUN cmake -GNinja -DENABLE_TESTS=ON -DCMAKE_BUILD_TYPE=Debug -DSANITIZE_ADDRESS=True .. && ninja tests +RUN cmake -GNinja -DENABLE_TESTS=ON -DOAI_ZMQ=ON -DCMAKE_BUILD_TYPE=Debug -DSANITIZE_ADDRESS=True .. && ninja tests diff --git a/common/config/config_load_configmodule.c b/common/config/config_load_configmodule.c index 087c8f622d..a47e1958ae 100644 --- a/common/config/config_load_configmodule.c +++ b/common/config/config_load_configmodule.c @@ -282,12 +282,9 @@ configmodule_interface_t *load_configmodule(int argc, cfgmode = strdup(CONFIG_LIBCONFIGFILE); } } - static configmodule_interface_t *cfgptr = NULL; - if (cfgptr) - fprintf(stderr, "ERROR: Call load_configmodule more than one time\n"); // The macros are not thread safe print_params and similar - cfgptr = calloc(sizeof(configmodule_interface_t), 1); + configmodule_interface_t *cfgptr = calloc(sizeof(configmodule_interface_t), 1); if (!cfgptr) { fprintf(stderr, "ERROR: cannot allocate a memory for configuration\n"); return NULL; diff --git a/common/config/config_load_configmodule.h b/common/config/config_load_configmodule.h index 5e535f67f7..a47b6b1801 100644 --- a/common/config/config_load_configmodule.h +++ b/common/config/config_load_configmodule.h @@ -10,6 +10,9 @@ #ifndef INCLUDE_CONFIG_LOADCONFIGMODULE_H #define INCLUDE_CONFIG_LOADCONFIGMODULE_H +#ifdef __cplusplus +extern "C" { +#endif #include #include @@ -117,5 +120,8 @@ void write_parsedcfg(configmodule_interface_t *cfg); extern void free_configmodule(void); #define CONFIG_PRINTF_ERROR(f, x... ) if (isLogInitDone ()) { LOG_E(ENB_APP,f,x);} else {printf(f,x);}; if ( !CONFIG_ISFLAGSET(CONFIG_NOABORTONCHKF) ) exit_fun("exit because configuration failed\n"); +#ifdef __cplusplus +} +#endif #endif /* INCLUDE_CONFIG_LOADCONFIGMODULE_H */ diff --git a/common/config/config_paramdesc.h b/common/config/config_paramdesc.h index efb5c956d2..ccca1dbd51 100644 --- a/common/config/config_paramdesc.h +++ b/common/config/config_paramdesc.h @@ -181,6 +181,11 @@ typedef struct paramdef { { \ OPTNAME(name), HELPSTR(help), PARAMFLAG(flags), .u64ptr = ptr, .defuintval = defval, PARAMTYPE(TYPE_UINT64), .numelt = 0 \ } +#define STRINGLISTPARAM(name, help, flags, ptr, defval) \ + { \ + OPTNAME(name), HELPSTR(help), PARAMFLAG(flags), .strptr = ptr, .defstrlistval = defval, PARAMTYPE(TYPE_STRINGLIST), \ + .numelt = 0 \ + } typedef struct paramlist_def { char listname[MAX_OPTNAME_SIZE]; diff --git a/doc/README.md b/doc/README.md index d1dfc2dc4d..02be593a7e 100644 --- a/doc/README.md +++ b/doc/README.md @@ -108,6 +108,7 @@ Some directories under `radio` contain READMEs: - [fhi_72](../radio/fhi_72/README.md) - [vrtsim](../radio/vrtsim/README.md) - [rf_emulator](../radio/emulator/README.md) +- [zmq](../radio/zmq/README.md) The other SDRs (AW2S, LimeSDR, ...) have no READMEs. diff --git a/radio/CMakeLists.txt b/radio/CMakeLists.txt index 5a8028253d..285224cd32 100644 --- a/radio/CMakeLists.txt +++ b/radio/CMakeLists.txt @@ -51,3 +51,8 @@ add_boolean_option(OAI_RF_EMULATOR ON "Activate OAI's RF emulator" OFF) if(OAI_RF_EMULATOR) add_subdirectory(emulator) endif() + +add_boolean_option(OAI_ZMQ OFF "Activate OAI's ZMQ radio" OFF) +if(OAI_ZMQ) + add_subdirectory(zmq) +endif() diff --git a/radio/zmq/CMakeLists.txt b/radio/zmq/CMakeLists.txt new file mode 100644 index 0000000000..ce23cb5c9d --- /dev/null +++ b/radio/zmq/CMakeLists.txt @@ -0,0 +1,21 @@ +# SPDX-License-Identifier: LicenseRef-CSSL-1.0 + +pkg_check_modules(zmq REQUIRED libzmq) +add_library(ring_buffer ring_buffer.cpp) +add_library(zmq_imported zmq_imported.cpp) +target_include_directories(zmq_imported PUBLIC ./) +target_link_libraries(zmq_imported PUBLIC ${zmq_LIBRARIES} log_headers) + +target_include_directories(ring_buffer PUBLIC ./) +add_library(oai_zmqdevif SHARED zmq_radio.cpp) +target_link_libraries(oai_zmqdevif PUBLIC ${zmq_LIBRARIES} ring_buffer log_headers zmq_imported) +target_include_directories(oai_zmqdevif PUBLIC ${zmq_INCLUDE_DIRS}) +set_target_properties(oai_zmqdevif PROPERTIES LIBRARY_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}) + +add_library(zmq_radio_static STATIC zmq_radio.cpp) +target_link_libraries(zmq_radio_static PUBLIC ${zmq_LIBRARIES} ring_buffer log_headers zmq_imported) +target_include_directories(zmq_radio_static PUBLIC ${zmq_INCLUDE_DIRS}) + +if (ENABLE_TESTS) + add_subdirectory(tests) +endif() diff --git a/radio/zmq/README.md b/radio/zmq/README.md new file mode 100644 index 0000000000..48c66d6d82 --- /dev/null +++ b/radio/zmq/README.md @@ -0,0 +1,40 @@ + + +# Overview + +This library implements ZMQ-based radio driver + +# Architecture + +The radio simulates RX/TX pairs between different processes over network or on the same machine +as ZMQ REQ/REP socket pairs. All the antennas have to be configured and connected before the +simulation can start. + +# Limitations + +You cannot reconnect a device once the simulation has started. The simulation has to be restarted +from scratch in order to add devices or restart a process. + +# Requirements + +Depends on zmq library being installed in the system. On ubuntu: `libzmq3-dev`, on RHEL/Fedora: `zeromq-devel` + +# Usage + +## Simple 1-to-1 antenna mapping + +Add `--device.name oai_zmqdevif` to load the library in UE / gNB process. Add `--zmq.[0].tx_channels ` and +`--zmq.[0].rx_channels ` to define ZMQ REQ/REP pairs. + +On the opposite side, load the library like specified above but invert the `rx` and `tx` channels. This way all +antennas of UE will be directly mapped to all antennas of the gNB. + +### Example: + +``` +sudo ./nr-uesoftmodem -r 106 --numerology 1 --band 78 -C 3619200000 --device.name oai_zmqdevif --zmq.[0].tx_channels tcp://127.0.0.1:4557 --zmq.[0].rx_channels tcp://127.0.0.1:4556 --ssb 516 +``` + +``` +sudo ./nr-softmodem -O ../targets/PROJECTS/GENERIC-NR-5GC/CONF/gnb.sa.band78.fr1.106PRB.usrpb210.conf --gNBs.[0].min_rxtxtime 6 --device.name oai_zmqdevif --zmq.[0].tx_channels tcp://127.0.0.1:4556 --zmq.[0].rx_channels tcp://127.0.0.1:4557 +``` diff --git a/radio/zmq/ring_buffer.cpp b/radio/zmq/ring_buffer.cpp new file mode 100644 index 0000000000..c4fdd94b29 --- /dev/null +++ b/radio/zmq/ring_buffer.cpp @@ -0,0 +1,167 @@ +/* + * SPDX-License-Identifier: LicenseRef-CSSL-1.0 + */ + +#include "ring_buffer.h" +#include +#include +#include + +ring_buffer::ring_buffer(size_t max_size) : max_size_(max_size) +{ + buffer_ = std::make_unique(max_size); +} + +size_t ring_buffer::push_samples(const cf_t *samples, const size_t nsamps) +{ + size_t overflow = 0; + // if nsamps > max_size skip nsamps - max_size samples + size_t nsamps_left = nsamps; + if (nsamps > max_size_) { + samples += nsamps - max_size_; + nsamps_left = max_size_; + overflow += nsamps - max_size_; + } + + // Detect overflow + if (size_ + nsamps_left > max_size_) { + size_t newtail__pos = (head_ + nsamps_left) % max_size_; + overflow += (size_ + nsamps_left) - max_size_; + tail_ = newtail__pos; + } + + size_t first_chunk = std::min(nsamps_left, max_size_ - head_); + memcpy(&buffer_[head_], samples, first_chunk * sizeof(cf_t)); + head_ = (head_ + first_chunk) % max_size_; + samples += first_chunk; + nsamps_left -= first_chunk; + if (nsamps_left > 0) { + memcpy(&buffer_[0], samples, nsamps_left * sizeof(cf_t)); + head_ = nsamps_left; + } + + size_ = std::min(size_ + nsamps, max_size_); + + return overflow; +} + +size_t ring_buffer::push_zeros(const size_t num_zeros) +{ + size_t overflow = 0; + // if nsamps > max_size skip nsamps - max_size samples + size_t nsamps_left = num_zeros; + if (num_zeros > max_size_) { + nsamps_left = max_size_; + overflow += num_zeros - max_size_; + } + + // Detect overflow + if (size_ + nsamps_left > max_size_) { + size_t new_tail_pos = (head_ + nsamps_left) % max_size_; + overflow += (size_ + nsamps_left) - max_size_; + tail_ = new_tail_pos; + } + + size_t first_chunk = std::min(nsamps_left, max_size_ - head_); + memset(&buffer_[head_], 0, first_chunk * sizeof(cf_t)); + head_ = (head_ + first_chunk) % max_size_; + nsamps_left -= first_chunk; + if (nsamps_left > 0) { + memset(&buffer_[0], 0, nsamps_left * sizeof(cf_t)); + head_ = nsamps_left; + } + + size_ = std::min(size_ + num_zeros, max_size_); + + return overflow; +} + +size_t ring_buffer::pop_samples(cf_t *samples, size_t num_samples) +{ + size_t samples_to_pop = std::min(size_, num_samples); + if (samples_to_pop > 0) { + if (tail_ + samples_to_pop > max_size_) { + size_t first_chunk = max_size_ - tail_; + memcpy(samples, &buffer_[tail_], first_chunk * sizeof(cf_t)); + memcpy(samples + first_chunk, &buffer_[0], (samples_to_pop - first_chunk) * sizeof(cf_t)); + } else { + memcpy(samples, &buffer_[tail_], samples_to_pop * sizeof(cf_t)); + } + tail_ = (tail_ + samples_to_pop) % max_size_; + size_ -= samples_to_pop; + return samples_to_pop; + } + return 0; +} + +void ring_buffer::clear_samples() +{ + head_ = 0; + tail_ = 0; + size_ = 0; +} + +void ring_buffer::reset() +{ + clear_samples(); +} + +size_t ring_buffer::size() const +{ + return size_; +} + +size_t overflow_buffer::push_samples(const cf_t *samples, size_t nsamps) +{ + std::lock_guard lock(mutex_); + size_t overflow = buffer_.push_samples(samples, nsamps); + zeros_to_send_ += overflow; + return overflow; +} + +size_t overflow_buffer::push_zeros(size_t num_zeros) +{ + std::lock_guard lock(mutex_); + size_t overflow = buffer_.push_zeros(num_zeros); + zeros_to_send_ += overflow; + return overflow; +} + +size_t overflow_buffer::pop_samples(cf_t *samples, size_t num_samples) +{ + std::lock_guard lock(mutex_); + size_t samples_popped = 0; + if (zeros_to_send_ > 0) { + size_t num_zeros = std::min(zeros_to_send_, num_samples); + memset(samples, 0, num_zeros * sizeof(cf_t)); + zeros_to_send_ -= num_zeros; + samples += num_zeros; + num_samples -= num_zeros; + samples_popped += num_zeros; + } + + if (num_samples > 0) { + samples_popped += buffer_.pop_samples(samples, num_samples); + } + return samples_popped; +} + +void overflow_buffer::reset() +{ + std::lock_guard lock(mutex_); + buffer_.reset(); + zeros_to_send_ = buffer_.size() / 2; +} + +void overflow_buffer::clear_samples() +{ + std::lock_guard lock(mutex_); + buffer_.clear_samples(); + zeros_to_send_ = 0; +} + +size_t overflow_buffer::size() +{ + std::lock_guard lock(mutex_); + return buffer_.size() + zeros_to_send_; +} diff --git a/radio/zmq/ring_buffer.h b/radio/zmq/ring_buffer.h new file mode 100644 index 0000000000..40a877427e --- /dev/null +++ b/radio/zmq/ring_buffer.h @@ -0,0 +1,46 @@ +/* + * SPDX-License-Identifier: LicenseRef-CSSL-1.0 + */ + +#pragma once +#include +#include +#include +#include "common/platform_types.h" + +// A basic cirular sample buffer class +class ring_buffer { + private: + std::unique_ptr buffer_; + size_t head_ = 0; + size_t tail_ = 0; + size_t size_ = 0; + size_t max_size_; + + public: + ring_buffer(size_t max_size = 614400); + size_t push_samples(const cf_t *samples, size_t nsamps); + size_t push_zeros(size_t num_zeros); + size_t pop_samples(cf_t *samples, size_t num_samples); + void reset(); + void clear_samples(); + size_t size() const; +}; + +// A thread-safe wrapper around ring_buffer that counts overflows +class overflow_buffer { + ring_buffer buffer_; + size_t zeros_to_send_; + std::mutex mutex_; + + public: + overflow_buffer(size_t max_size = 614400) : buffer_(max_size), zeros_to_send_(0) + { + } + size_t push_samples(const cf_t *samples, size_t nsamps); + size_t push_zeros(size_t num_zeros); + size_t pop_samples(cf_t *samples, size_t num_samples); + void reset(); + void clear_samples(); + size_t size(); +}; diff --git a/radio/zmq/tests/CMakeLists.txt b/radio/zmq/tests/CMakeLists.txt new file mode 100644 index 0000000000..aa46be194b --- /dev/null +++ b/radio/zmq/tests/CMakeLists.txt @@ -0,0 +1,13 @@ +# SPDX-License-Identifier: LicenseRef-CSSL-1.0 + +add_executable(test_ring_buffer test_ring_buffer.cpp) +target_link_libraries(test_ring_buffer PRIVATE ring_buffer GTest::gtest) +add_dependencies(tests test_ring_buffer) +add_test(NAME test_ring_buffer + COMMAND ./test_ring_buffer) + +add_executable(test_zmq_radio test_zmq_radio.cpp) +target_link_libraries(test_zmq_radio PRIVATE zmq_radio_static UTIL CONFIG_LIB GTest::gtest zmq) +add_dependencies(tests test_zmq_radio) +add_test(NAME test_zmq_radio + COMMAND ./test_zmq_radio) diff --git a/radio/zmq/tests/test_ring_buffer.cpp b/radio/zmq/tests/test_ring_buffer.cpp new file mode 100644 index 0000000000..8687139b13 --- /dev/null +++ b/radio/zmq/tests/test_ring_buffer.cpp @@ -0,0 +1,221 @@ +/* + * SPDX-License-Identifier: LicenseRef-CSSL-1.0 + */ + +#include +#include "ring_buffer.h" + +TEST(CircularBuffer, simplePushPop) +{ + ring_buffer cb(10); + cf_t data[10]; + for (int i = 0; i < 10; i++) { + data[i] = { (float)i, (float)(i + 1)}; + } + ASSERT_EQ(cb.push_samples(data, 10), 0); + + cf_t read_data[10]; + ASSERT_EQ(cb.pop_samples(read_data, 10), 10); + for (int i = 0; i < 10; i++) { + ASSERT_EQ(read_data[i].r, data[i].r); + ASSERT_EQ(read_data[i].i, data[i].i); + } + + ASSERT_EQ(cb.size(), 0); + ASSERT_EQ(cb.pop_samples(read_data, 10), 0); +} + +TEST(CircularBuffer, overflow) +{ + ring_buffer cb(10); + cf_t data[15]; + for (int i = 0; i < 15; i++) { + data[i] = {(float)i, (float)(i + 1)}; + } + ASSERT_EQ(cb.push_samples(data, 15), 5); + cf_t read_data[10]; + ASSERT_EQ(cb.pop_samples(read_data, 10), 10); + // The first 5 samples are lost due to overflow + for (int i = 0; i < 10; i++) { + ASSERT_EQ(read_data[i].r, i + 5); + ASSERT_EQ(read_data[i].i, i + 6); + } +} + +TEST(CircularBuffer, partialPop) +{ + ring_buffer cb(10); + cf_t data[10]; + for (int i = 0; i < 10; i++) { + data[i] = {(float)i, (float)(i + 1)}; + } + ASSERT_EQ(cb.push_samples(data, 10), 0); + cf_t read_data[5]; + ASSERT_EQ(cb.pop_samples(read_data, 5), 5); + for (int i = 0; i < 5; i++) { + ASSERT_EQ(read_data[i].r, i); + ASSERT_EQ(read_data[i].i, i + 1); + } + ASSERT_EQ(cb.size(), 5); + ASSERT_EQ(cb.pop_samples(read_data, 5), 5); + for (int i = 0; i < 5; i++) { + ASSERT_EQ(read_data[i].r, i + 5); + ASSERT_EQ(read_data[i].i, i + 6); + } +} + +TEST(CircularBuffer, popMore) +{ + ring_buffer cb(10); + cf_t data[10]; + for (int i = 0; i < 10; i++) { + data[i] = {(float)i, (float)(i + 1)}; + } + cb.push_samples(data, 10); + cf_t read_data[15]; + ASSERT_EQ(cb.pop_samples(read_data, 15), 10); + for (int i = 0; i < 10; i++) { + ASSERT_EQ(read_data[i].r, i); + ASSERT_EQ(read_data[i].i, i + 1); + } +} + +TEST(CircularBuffer, newOverflow) +{ + ring_buffer cb(10); + cf_t data1[8]; + for (int i = 0; i < 8; i++) { + data1[i] = {(float)i, (float)(i + 1)}; + } + ASSERT_EQ(cb.push_samples(data1, 8), 0); + + cf_t data2[4]; + for (int i = 0; i < 4; i++) { + data2[i] = {(float)(i + 8), (float)(i + 9)}; + } + ASSERT_EQ(cb.push_samples(data2, 4), 2); + + cf_t read_data[10]; + ASSERT_EQ(cb.pop_samples(read_data, 10), 10); + // The first 2 samples are lost due to overflow + for (int i = 0; i < 8; i++) { + ASSERT_EQ(read_data[i].r, i + 2); + ASSERT_EQ(read_data[i].i, i + 3); + } + ASSERT_EQ(read_data[8].r, 10); + ASSERT_EQ(read_data[8].i, 11); + ASSERT_EQ(read_data[9].r, 11); + ASSERT_EQ(read_data[9].i, 12); +} + +TEST(overflow_buffer, simplePushPop) +{ + overflow_buffer pb(10); + cf_t data[10]; + for (int i = 0; i < 10; i++) { + data[i] = {(float)i, (float)(i + 1)}; + } + pb.push_samples(data, 10); + ASSERT_EQ(pb.size(), 10); + cf_t read_data[10]; + ASSERT_EQ(pb.pop_samples(read_data, 10), 10); + for (int i = 0; i < 10; i++) { + ASSERT_EQ(read_data[i].r, i); + ASSERT_EQ(read_data[i].i, i + 1); + } +} + +TEST(overflow_buffer, zeros_to_send) +{ + overflow_buffer pb(10); + pb.push_zeros(10); + cf_t data[5]; + for (int i = 0; i < 5; i++) { + data[i] = {(float)i, (float)(i + 1)}; + } + pb.push_samples(data, 5); + ASSERT_EQ(pb.size(), 15); + cf_t read_data[10]; + ASSERT_EQ(pb.pop_samples(read_data, 10), 10); + for (int i = 0; i < 10; i++) { + ASSERT_EQ(read_data[i].r, 0); + ASSERT_EQ(read_data[i].i, 0); + } + ASSERT_EQ(pb.pop_samples(read_data, 10), 5); + for (int i = 0; i < 5; i++) { + ASSERT_EQ(read_data[i].r, i); + ASSERT_EQ(read_data[i].i, i + 1); + } +} + +TEST(overflow_buffer, no_zeros_to_send) +{ + overflow_buffer pb(10); + cf_t data[10]; + for (int i = 0; i < 10; i++) { + data[i] = {(float)i, (float)(i + 1)}; + } + pb.push_samples(data, 10); + ASSERT_EQ(pb.size(), 10); + cf_t read_data[10]; + ASSERT_EQ(pb.pop_samples(read_data, 10), 10); + for (int i = 0; i < 10; i++) { + ASSERT_EQ(read_data[i].r, i); + ASSERT_EQ(read_data[i].i, i + 1); + } +} + +TEST(overflow_buffer, overflow) +{ + overflow_buffer pb(10); + cf_t data[15]; + for (int i = 0; i < 15; i++) { + data[i] = {(float)i, (float)(i + 1)}; + } + pb.push_samples(data, 15); + ASSERT_EQ(pb.size(), 15); + cf_t read_data[15]; + ASSERT_EQ(pb.pop_samples(read_data, 15), 15); + // The first 5 samples were lost due to overflow. + int i = 0; + for (; i < 5; i++) { + ASSERT_EQ(read_data[i].r, 0); + ASSERT_EQ(read_data[i].i, 0); + } + for (; i < 15; i++) { + ASSERT_EQ(read_data[i].r, i); + ASSERT_EQ(read_data[i].i, i + 1); + } + ASSERT_EQ(pb.size(), 0); +} + +TEST(overflow_buffer, push_zeros) +{ + overflow_buffer pb(10); + pb.push_zeros(5); + ASSERT_EQ(pb.size(), 5); + cf_t data[5]; + for (int i = 0; i < 5; i++) { + data[i] = {(float)i, (float)(i + 1)}; + } + pb.push_samples(data, 5); + ASSERT_EQ(pb.size(), 10); + cf_t read_data[10]; + ASSERT_EQ(pb.pop_samples(read_data, 10), 10); + for (int i = 0; i < 10; i++) { + if (i < 5) { + ASSERT_EQ(read_data[i].r, 0); + ASSERT_EQ(read_data[i].i, 0); + } else { + ASSERT_EQ(read_data[i].r, data[i - 5].r); + ASSERT_EQ(read_data[i].i, data[i - 5].i); + } + } + ASSERT_EQ(pb.size(), 0); +} + +int main(int argc, char **argv) +{ + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} diff --git a/radio/zmq/tests/test_zmq_radio.cpp b/radio/zmq/tests/test_zmq_radio.cpp new file mode 100644 index 0000000000..78e8e6b85c --- /dev/null +++ b/radio/zmq/tests/test_zmq_radio.cpp @@ -0,0 +1,162 @@ +/* + * SPDX-License-Identifier: LicenseRef-CSSL-1.0 + */ + +#include "common_lib.h" +#include +#include "common/config/config_userapi.h" +#include +extern "C" { +#include "common/config/config_userapi.h" +#include "openair1/SIMULATION/TOOLS/sim.h" +extern int device_init(openair0_device_t *device, openair0_config_t *openair0_cfg); +static softmodem_params_t softmodem_params; +softmodem_params_t *get_softmodem_params(void) +{ + return &softmodem_params; +} +} +#include "common/platform_types.h" +#include + +configmodule_interface_t *uniqCfg = NULL; + +extern "C" void exit_function(const char *file, const char *function, const int line, const char *s, const int assert) +{ + fprintf(stderr, "FATAL: %s at %s:%s:%d\n", s, file, function, line); + exit(EXIT_FAILURE); +} + +class ZMQTest : public ::testing::Test { + protected: + configmodule_interface_t *cfg1 = nullptr; + configmodule_interface_t *cfg2 = nullptr; + openair0_device_t device1 = {0}; + openair0_device_t device2 = {0}; + openair0_config_t config1 = {0}; + openair0_config_t config2 = {0}; + char *argv[4]; + + void SetUp() override + { + argv[0] = strdup("--zmq.[0].tx_channels"); + argv[1] = strdup("tcp://127.0.0.1:5555"); + argv[2] = strdup("--zmq.[0].rx_channels"); + argv[3] = strdup("tcp://127.0.0.1:5556"); + + cfg1 = load_configmodule(sizeofArray(argv), argv, CONFIG_ENABLECMDLINEONLY); + uniqCfg = cfg1; + // 2. Initialize the ZMQ device + config1.tx_num_channels = 1; + config1.rx_num_channels = 1; + config1.sample_rate = 30; + ASSERT_EQ(device_init(&device1, &config1), 0); + ASSERT_EQ(device1.trx_start_func(&device1), 0); + + // Swap the RX with TX for second device + char *tmp = argv[0]; + argv[0] = argv[2]; + argv[2] = tmp; + cfg2 = load_configmodule(sizeofArray(argv), argv, CONFIG_ENABLECMDLINEONLY); + uniqCfg = cfg2; + config2.tx_num_channels = 1; + config2.rx_num_channels = 1; + config2.sample_rate = 1500; + ASSERT_EQ(device_init(&device2, &config2), 0); + ASSERT_EQ(device2.trx_start_func(&device2), 0); + } + + void TearDown() override + { + if (device1.trx_end_func) { + device1.trx_end_func(&device1); + } + if (device2.trx_end_func) { + device2.trx_end_func(&device2); + } + if (cfg1) { + end_configmodule(cfg1); + } + if (cfg2) { + end_configmodule(cfg2); + } + for (auto i = 0U; i < sizeofArray(argv); i++) { + free(argv[i]); + } + } +}; + +TEST_F(ZMQTest, RXSamples) +{ + std::thread t1([this]() { + c16_t rx_samples[10]; + openair0_timestamp_t rx_timestamp; + void *samples[1] = {rx_samples}; + ASSERT_EQ(device1.trx_read_func(&device1, &rx_timestamp, samples, 10, 1), 10); + for (int i = 0; i < 10; i++) { + ASSERT_EQ(rx_samples[i].r, 0); + ASSERT_EQ(rx_samples[i].i, 0); + } + }); + std::thread t2([this]() { + c16_t rx_samples[10]; + openair0_timestamp_t rx_timestamp; + void *samples[1] = {rx_samples}; + ASSERT_EQ(device2.trx_read_func(&device2, &rx_timestamp, samples, 10, 1), 10); + for (int i = 0; i < 10; i++) { + ASSERT_EQ(rx_samples[i].r, 0); + ASSERT_EQ(rx_samples[i].i, 0); + } + }); + t1.join(); + t2.join(); +} + +TEST_F(ZMQTest, TxRxSamples) +{ + std::thread t1([this]() { + c16_t rx_samples[10]; + openair0_timestamp_t rx_timestamp; + void *samples[1] = {rx_samples}; + ASSERT_EQ(device1.trx_read_func(&device1, &rx_timestamp, samples, 10, 1), 10); + for (int i = 0; i < 10; i++) { + ASSERT_EQ(rx_samples[i].r, 0); + ASSERT_EQ(rx_samples[i].i, 0); + } + c16_t tx_samples[10]; + openair0_timestamp_t tx_timestamp = rx_timestamp + 10; + for (int i = 0; i < 10; i++) { + tx_samples[i].r = i; + tx_samples[i].i = i + 1; + } + samples[0] = tx_samples; + ASSERT_EQ(device1.trx_write_func(&device1, tx_timestamp, samples, 10, 1, 0), 10); + }); + std::thread t2([this]() { + c16_t rx_samples[10]; + openair0_timestamp_t rx_timestamp; + void *samples[1] = {rx_samples}; + ASSERT_EQ(device2.trx_read_func(&device2, &rx_timestamp, samples, 10, 1), 10); + for (int i = 0; i < 10; i++) { + ASSERT_EQ(rx_samples[i].r, 0); + ASSERT_EQ(rx_samples[i].i, 0); + } + openair0_timestamp_t rx_timestamp2; + ASSERT_EQ(device2.trx_read_func(&device2, &rx_timestamp2, samples, 10, 1), 10); + for (int i = 0; i < 10; i++) { + ASSERT_EQ(rx_samples[i].r, i); + ASSERT_EQ(rx_samples[i].i, i + 1); + } + ASSERT_EQ(rx_timestamp + 10, rx_timestamp2); + }); + t1.join(); + t2.join(); +} + +int main(int argc, char **argv) +{ + logInit(); + g_log->log_component[HW].level = OAILOG_DEBUG; + testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} diff --git a/radio/zmq/zmq_imported.cpp b/radio/zmq/zmq_imported.cpp new file mode 100644 index 0000000000..19bf542098 --- /dev/null +++ b/radio/zmq/zmq_imported.cpp @@ -0,0 +1,131 @@ +/* + * SPDX-License-Identifier: BSD-3-Clause-Open-MPI + * Based on zmq library in OCUDU project: ocudu/lib/radio/zmq + * Refer to https://gitlab.com/ocudu/ocudu/-/raw/dev/LICENSE?ref_type=heads + */ + +#include "zmq_imported.h" +#include "log.h" + +const float c16_t_to_cf_t_factor = std::numeric_limits::max(); +static constexpr std::chrono::milliseconds TRANSMIT_TS_ALIGN_TIMEOUT = std::chrono::milliseconds(0); +static constexpr std::chrono::milliseconds RECEIVE_TS_ALIGN_TIMEOUT = std::chrono::milliseconds(100); + +void zmq_tx_channel::transmit(c16_t *samples, size_t nsamps, uint64_t timestamp) +{ + std::scoped_lock lock(transmit_alignment_mutex_); + size_t overflow = 0; + if (timestamp > sample_count_) { + overflow += buffer_.push_zeros(timestamp - sample_count_); + sample_count_ = timestamp; + } + cf_t samples_float[nsamps]; + for (size_t i = 0; i < nsamps; i++) { + samples_float[i].r = samples[i].r / c16_t_to_cf_t_factor; + samples_float[i].i = samples[i].i / c16_t_to_cf_t_factor; + } + overflow += buffer_.push_samples(samples_float, nsamps); + sample_count_ += nsamps; + if (overflow) { + LOG_W(HW, "Overflow on ZMQ channel by %lu samples\n", overflow); + } + is_tx_enabled_ = true; + transmit_alignment_cvar_.notify_all(); +} + +void zmq_tx_channel::start(uint64_t init_time) +{ + sample_count_ = init_time; +} + +bool zmq_tx_channel::align(uint64_t timestamp, std::chrono::milliseconds timeout) +{ + if (sample_count_ >= timestamp) { + return sample_count_ > timestamp; + } + std::unique_lock lock(transmit_alignment_mutex_); + if (is_tx_enabled_ && (timeout.count() != 0)) { + bool is_not_timeout = + transmit_alignment_cvar_.wait_for(lock, timeout, [this, timestamp]() { return sample_count_ >= timestamp; }); + if (is_not_timeout) { + return sample_count_ > timestamp; + } + LOG_W(HW, "Timeout waiting for TX path to align samples\n"); + is_tx_enabled_ = false; + } + if (sample_count_ < timestamp) { + buffer_.push_zeros(timestamp - sample_count_); + sample_count_ = timestamp; + } + return false; +} + +void zmq_rx_channel::receive(c16_t *samples, size_t nsamps) +{ + size_t samples_popped = 0; + cf_t samples_float[nsamps]; + while (samples_popped < (size_t)nsamps && !stopped_) { + size_t popped_now = buffer_.pop_samples(samples_float + samples_popped, nsamps - samples_popped); + samples_popped += popped_now; + if (popped_now == 0) { + usleep(100); // wait for more samples to arrive + } + } + for (size_t i = 0; i < nsamps; i++) { + samples[i].r = samples_float[i].r * c16_t_to_cf_t_factor + 0.5; + samples[i].i = samples_float[i].i * c16_t_to_cf_t_factor + 0.5; + } +} +void zmq_rx_channel::stop() +{ + stopped_ = true; +} + +void zmq_tx_stream::start(uint64_t init_time) +{ + for (auto &chan : channels_) { + chan->start(init_time); + } +} +bool zmq_tx_stream::align(uint64_t timestamp, std::chrono::milliseconds timeout) +{ + bool timestamp_passed = false; + for (auto &chan : channels_) { + timestamp_passed = timestamp_passed || chan->align(timestamp, timeout); + } + return timestamp_passed; +} +void zmq_tx_stream::transmit(c16_t **samples, size_t nsamps, uint64_t timestamp) +{ + bool timestamp_passed = align(timestamp, TRANSMIT_TS_ALIGN_TIMEOUT); + if (timestamp_passed) { + LOG_W(HW, "Error, channel timeout\n"); + return; + } + int i = 0; + for (auto chan : channels_) { + chan->transmit(samples[i++], nsamps, timestamp); + } +} + +void zmq_rx_stream::start(uint64_t init_time) +{ + sample_count_ = init_time; +} +void zmq_rx_stream::stop() +{ + for (auto &chan : channels_) { + chan->stop(); + } +} +void zmq_rx_stream::receive(c16_t **samples, size_t nsamps, uint64_t *timestamp) +{ + *timestamp = sample_count_; + uint64_t passed_timestamp = sample_count_ + nsamps; + tx_stream_->align(passed_timestamp, RECEIVE_TS_ALIGN_TIMEOUT); + int i = 0; + for (auto chan : channels_) { + chan->receive(samples[i++], nsamps); + } + sample_count_ += nsamps; +} diff --git a/radio/zmq/zmq_imported.h b/radio/zmq/zmq_imported.h new file mode 100644 index 0000000000..7262ea0605 --- /dev/null +++ b/radio/zmq/zmq_imported.h @@ -0,0 +1,71 @@ +/* + * SPDX-License-Identifier: BSD-3-Clause-Open-MPI + * Based on zmq library in OCUDU project: ocudu/lib/radio/zmq + * Refer to https://gitlab.com/ocudu/ocudu/-/raw/dev/LICENSE?ref_type=heads + */ + +#ifndef ZMQ_IMPORTED_H +#define ZMQ_IMPORTED_H + +#include +#include "ring_buffer.h" +#include +#include +#include +#include + +class zmq_tx_channel { + public: + void *socket_; + overflow_buffer buffer_; + std::atomic sample_count_ = 0; + std::atomic is_tx_enabled_ = false; + std::mutex transmit_alignment_mutex_; + std::condition_variable transmit_alignment_cvar_; + + zmq_tx_channel(void *s, uint64_t buffer_size) : socket_(s), buffer_(buffer_size) + { + } + + void transmit(c16_t *samples, size_t nsamps, uint64_t timestamp); + + void start(uint64_t init_time); + + bool align(uint64_t timestamp, std::chrono::milliseconds timeout); +}; + +class zmq_rx_channel { + public: + void *socket_; + overflow_buffer buffer_; + bool request_sent_; + std::atomic stopped_; + zmq_rx_channel(void *s, uint64_t buffer_size) : socket_(s), buffer_(buffer_size), stopped_(false) + { + } + void receive(c16_t *samples, size_t nsamps); + void stop(); +}; + +class zmq_tx_stream { + public: + std::vector channels_; + void start(uint64_t init_time); + bool align(uint64_t timestamp, std::chrono::milliseconds timeout); + void transmit(c16_t **samples, size_t nsamps, uint64_t timestamp); +}; + +class zmq_rx_stream { + public: + std::vector channels_; + zmq_tx_stream *tx_stream_; + uint64_t sample_count_ = 0; + zmq_rx_stream() : sample_count_(0) + { + } + void start(uint64_t init_time); + void stop(); + void receive(c16_t **samples, size_t nsamps, uint64_t *timestamp); +}; + +#endif diff --git a/radio/zmq/zmq_radio.cpp b/radio/zmq/zmq_radio.cpp new file mode 100644 index 0000000000..93774bb2c3 --- /dev/null +++ b/radio/zmq/zmq_radio.cpp @@ -0,0 +1,345 @@ +/* + * SPDX-License-Identifier: LicenseRef-CSSL-1.0 + */ + +#include "PHY/TOOLS/tools_defs.h" +#include "PHY/defs_common.h" +#include "common/platform_types.h" +#include "softmodem-common.h" +#include "utils.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include "common_lib.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include "zmq_imported.h" + +#define ZMQ_SECTION "zmq" +#define ZMQ_TX_CHANNELS "tx_channels" +#define ZMQ_RX_CHANNELS "rx_channels" + +#define ZMQ_PARAMS_DESC \ + { \ + STRINGLISTPARAM(ZMQ_TX_CHANNELS, "list of zmq addresses represeting tx channels_\n", PARAMFLAG_MANDATORY, nullptr, nullptr), \ + STRINGLISTPARAM(ZMQ_RX_CHANNELS, "list of zmq addresses represeting rx channels_\n", PARAMFLAG_MANDATORY, nullptr, nullptr), \ + }; + +const size_t sample_size = sizeof(cf_t); +const size_t rx_buffer_size = sample_size * 300000; + +typedef struct { + void *context; + zmq_tx_stream tx_stream; + zmq_rx_stream rx_stream; + std::thread poll_thread; + std::atomic poll_thread_running; + bool stopped = false; + double sample_rate; +} zmq_state_t; + +static void poll_thread(zmq_state_t *s) +{ + s->poll_thread_running = true; + unsigned char *rx_buffer = static_cast(malloc(rx_buffer_size)); + const auto num_tx_channels = s->tx_stream.channels_.size(); + const auto num_rx_channels = s->rx_stream.channels_.size(); + std::vector items(num_tx_channels + num_rx_channels); + std::vector reply_requested(num_tx_channels); + for (size_t i = 0; i < num_tx_channels; ++i) { + items[i] = {s->tx_stream.channels_[i]->socket_, 0, ZMQ_POLLIN, 0}; + // wait for REQ + reply_requested[i] = false; + } + for (size_t i = 0; i < num_rx_channels; i++) { + items[i + num_tx_channels] = {s->rx_stream.channels_[i]->socket_, 0, ZMQ_POLLIN, 0}; + } + + const auto num_channels = num_tx_channels + num_rx_channels; + while (s->poll_thread_running) { + for (size_t i = 0; i < num_tx_channels; i++) { + auto chan = s->tx_stream.channels_[i]; + if (!reply_requested[i]) { + continue; + } + std::vector samples(1024); + size_t num_popped = chan->buffer_.pop_samples(samples.data(), 1024); + if (num_popped == 0) { + continue; + } + int rc = zmq_send(chan->socket_, samples.data(), num_popped * sizeof(cf_t), 0); + if (rc < 0) { + LOG_E(HW, "[ZMQ] poll_thread zmq_send for TX antenna %d failed: %s\n", (int)i, zmq_strerror(errno)); + } + reply_requested[i] = false; + } + + int rc = zmq_poll(items.data(), num_channels, 10); // 10ms timeout + if (rc < 0) { + if (errno == EINTR) + continue; + LOG_E(HW, "[ZMQ] poll_thread zmq_poll failed: %s\n", zmq_strerror(errno)); + break; + } + if (rc == 0) { + continue; // timeout + } + + // --- TX Sockets (ZMQ_REP) --- + for (size_t i = 0; i < num_tx_channels; i++) { + if (items[i].revents & ZMQ_POLLIN) { + auto chan = s->tx_stream.channels_[i]; + char dummy; + rc = zmq_recv(chan->socket_, &dummy, 1, 0); + if (rc < 0) { + LOG_E(HW, "[ZMQ] poll_thread zmq_recv for TX antenna %d failed: %s\n", (int)i, zmq_strerror(errno)); + continue; + } + if (reply_requested[i]) { + LOG_E(HW, "[ZMQ] Error, unexpected REQ before REP on TX antenna %d\n", (int)i); + } + reply_requested[i] = true; + } + } + + // --- RX Sockets (ZMQ_REQ) --- + for (size_t i = 0; i < num_rx_channels; i++) { + if (items[i + num_tx_channels].revents & ZMQ_POLLIN) { + auto chan = s->rx_stream.channels_[i]; + rc = zmq_recv(chan->socket_, rx_buffer, rx_buffer_size, 0); + if (rc < 0) { + LOG_E(HW, "[ZMQ] poll_thread zmq_recv for RX antenna %d failed: %s\n", (int)i, zmq_strerror(errno)); + } else { + size_t received_bytes = rc; + if (rx_buffer_size < received_bytes) { + LOG_W(HW, + "[ZMQ] the RX buffer is too small! The received message size is %lu while the buffer is %lu. Message truncated\n", + received_bytes, + rx_buffer_size); + } + size_t num_samples_received = std::min(received_bytes, rx_buffer_size) / sizeof(cf_t); + cf_t *samples = reinterpret_cast(rx_buffer); + size_t overflow = chan->buffer_.push_samples(samples, num_samples_received); + if (rx_buffer_size < received_bytes) { + overflow += chan->buffer_.push_zeros((received_bytes - rx_buffer_size) / sizeof(cf_t)); + } + if (overflow) { + LOG_W(HW, "Overflow on receive\n"); + } + // After receiving, send next request to keep the stream flowing + char dummy = 0; + if (zmq_send(chan->socket_, &dummy, 1, 0) != 1) { + LOG_E(HW, "[ZMQ] poll_thread zmq_send for RX antenna %d failed: %s\n", (int)i, zmq_strerror(errno)); + } + } + } + } + } + free(rx_buffer); +} + +static int zmq_write(openair0_device_t *device, openair0_timestamp_t timestamp, void **buff, int nsamps, int cc, int flags) +{ + zmq_state_t *s = static_cast(device->priv); + AssertFatal((uint)cc == s->tx_stream.channels_.size(), + "Request to write on more antennas (%d) than configured (%d)", + cc, + (int)s->tx_stream.channels_.size()); + + s->tx_stream.transmit((c16_t **)buff, nsamps, timestamp); + + return nsamps; +} + +static int zmq_read(openair0_device_t *device, openair0_timestamp_t *ptimestamp, void **samplesVoid, int nsamps, int nbAnt) +{ + zmq_state_t *s = static_cast(device->priv); + AssertFatal((uint)nbAnt == s->rx_stream.channels_.size(), + "Request to read on more antennas (%d) than configured (%d)", + nbAnt, + (int)s->rx_stream.channels_.size()); + uint64_t timestamp; + s->rx_stream.receive((c16_t **)samplesVoid, nsamps, ×tamp); + *ptimestamp = timestamp; + return nsamps; +} + +static int zmq_get_stats(openair0_device_t *device) +{ + return 0; +} +static int zmq_reset_stats(openair0_device_t *device) +{ + return 0; +} +static void zmq_end(openair0_device_t *device) +{ + zmq_state_t *s = static_cast(device->priv); + if (s) { + if (s->poll_thread_running) { + s->poll_thread_running = false; + if (s->poll_thread.joinable()) { + s->poll_thread.join(); + } + } + for (auto &chan : s->tx_stream.channels_) { + if (chan->socket_) + zmq_close(chan->socket_); + delete chan; + } + s->tx_stream.channels_.clear(); + + for (auto &chan : s->rx_stream.channels_) { + if (chan->socket_) + zmq_close(chan->socket_); + delete chan; + } + s->rx_stream.channels_.clear(); + + if (s->context) + zmq_ctx_destroy(s->context); + delete s; + } +} + +static int zmq_start(openair0_device_t *device) +{ + zmq_state_t *s = static_cast(device->priv); + s->rx_stream.start(s->sample_rate / 100); + s->tx_stream.start(s->sample_rate / 100); + for (size_t i = 0; i < s->rx_stream.channels_.size(); i++) { + auto channel = s->rx_stream.channels_[i]; + // Send initial request to start data flow + char dummy = 0; + if (zmq_send(channel->socket_, &dummy, 1, 0) != 1) { + LOG_E(HW, "[ZMQ] zmq_send for initial RX request failed for antenna %lu: %s\n", i, zmq_strerror(errno)); + return -1; + } + } + s->poll_thread = std::thread(poll_thread, s); + return 0; +} + +static int zmq_stop(openair0_device_t *device) +{ + zmq_state_t *s = static_cast(device->priv); + s->rx_stream.stop(); + return 0; +} + +static int zmq_set_freq(openair0_device_t *device, openair0_config_t *openair0_cfg) +{ + return 0; +} +static int zmq_set_gains(openair0_device_t *device, openair0_config_t *openair0_cfg) +{ + return 0; +} +static int zmq_write_init(openair0_device_t *device) +{ + return 0; +} + +extern "C" __attribute__((__visibility__("default"))) int device_init(openair0_device_t *device, openair0_config_t *openair0_cfg) +{ + auto *zmq_state = new zmq_state_t(); + zmq_state->context = zmq_ctx_new(); + AssertFatal(zmq_state->context != NULL, "zmq_ctx_new failed"); + + LOG_I(HW, "[ZMQ] tx_antennas: %d, rx_antennas: %d\n", openair0_cfg->tx_num_channels, openair0_cfg->rx_num_channels); + configmodule_interface_t *cfg = config_get_if(); + paramdef_t param_desc[] = ZMQ_PARAMS_DESC; + std::string zmq_section = std::string(ZMQ_SECTION); + int ru_id = openair0_cfg->ru_id; + std::string zmq_array_section = std::string(ZMQ_SECTION) + ".[" + std::to_string(ru_id) + "]"; + int ret = config_get(cfg, param_desc, sizeofArray(param_desc), zmq_array_section.c_str()); + AssertFatal(ret >= 0, "configuration couldn't be performed\n"); + int num_configured_tx_channels = gpd(param_desc, sizeofArray(param_desc), ZMQ_TX_CHANNELS)->numelt; + AssertFatal(num_configured_tx_channels == openair0_cfg->tx_num_channels, + "Incorrect configuration: Number of zmq tx channels (%d) != number of configured tx channels (%d)\n", + num_configured_tx_channels, + openair0_cfg->tx_num_channels); + int num_configured_rx_channels = gpd(param_desc, sizeofArray(param_desc), ZMQ_RX_CHANNELS)->numelt; + AssertFatal(num_configured_rx_channels == openair0_cfg->rx_num_channels, + "Incorrect configuration: Number of zmq rx channels (%d) != number of configured rx channels (%d)\n", + num_configured_rx_channels, + openair0_cfg->rx_num_channels); + char **tx_channels = gpd(param_desc, sizeofArray(param_desc), ZMQ_TX_CHANNELS)->strlistptr; + char **rx_channels = gpd(param_desc, sizeofArray(param_desc), ZMQ_RX_CHANNELS)->strlistptr; + + // Setup TX sockets (one per antenna) + if (openair0_cfg->tx_num_channels > 0) { + zmq_state->tx_stream.channels_.resize(openair0_cfg->tx_num_channels); + for (int i = 0; i < openair0_cfg->tx_num_channels; i++) { + void *socket = zmq_socket(zmq_state->context, ZMQ_REP); + AssertFatal(socket != NULL, "zmq_socket(ZMQ_REP) for TX antenna %d failed", i); + int linger = 0; + zmq_setsockopt(socket, ZMQ_LINGER, &linger, sizeof(linger)); + AssertFatal(zmq_bind(socket, tx_channels[i]) == 0, "zmq_bind for TX antenna %d failed on %s", i, tx_channels[i]); + auto channel = new zmq_tx_channel(socket, openair0_cfg->sample_rate); + LOG_I(HW, "[ZMQ] TX socket for antenna %d bound to %s\n", i, tx_channels[i]); + zmq_state->tx_stream.channels_[i] = channel; + } + } + zmq_state->sample_rate = openair0_cfg->sample_rate; + + // Setup RX sockets (one per antenna) + if (openair0_cfg->rx_num_channels > 0) { + zmq_state->rx_stream.channels_.resize(openair0_cfg->rx_num_channels); + for (int i = 0; i < openair0_cfg->rx_num_channels; i++) { + void *socket = zmq_socket(zmq_state->context, ZMQ_REQ); + AssertFatal(socket != NULL, "zmq_socket(ZMQ_REQ) for RX antenna %d failed", i); + int linger = 0; + zmq_setsockopt(socket, ZMQ_LINGER, &linger, sizeof(linger)); + AssertFatal(zmq_connect(socket, rx_channels[i]) == 0, "zmq_connect for RX antenna %d failed on %s", i, rx_channels[i]); + auto channel = new zmq_rx_channel(socket, openair0_cfg->sample_rate); + LOG_I(HW, "[ZMQ] RX socket for antenna %d connected to %s\n", i, rx_channels[i]); + zmq_state->rx_stream.channels_[i] = channel; + } + zmq_state->rx_stream.tx_stream_ = &zmq_state->tx_stream; + } + + device->trx_start_func = zmq_start; + device->trx_get_stats_func = zmq_get_stats; + device->trx_reset_stats_func = zmq_reset_stats; + device->trx_end_func = zmq_end; + device->trx_stop_func = zmq_stop; + device->trx_set_freq_func = zmq_set_freq; + device->trx_set_gains_func = zmq_set_gains; + device->trx_write_func = zmq_write; + device->trx_read_func = zmq_read; + device->type = RFSIMULATOR; + IS_SOFTMODEM_RFSIM = 1U; + openair0_cfg->rx_gain[0] = 0; + device->openair0_cfg = openair0_cfg; + device->priv = zmq_state; + device->trx_write_init = zmq_write_init; + + return 0; +}