diff --git a/receiver/host/CMakeLists.txt b/receiver/host/CMakeLists.txt index 1087988e9..b6500387c 100644 --- a/receiver/host/CMakeLists.txt +++ b/receiver/host/CMakeLists.txt @@ -20,10 +20,13 @@ IF(IBVERBS) TARGET_LINK_LIBRARIES(JungfraujochHost ${IBVERBS}) MESSAGE(STATUS "JFJochReceiver compiled with IBVerbs support") - ADD_EXECUTABLE(MlxRawEthRcv MlxRawEthRcv.cpp) - TARGET_LINK_LIBRARIES(MlxRawEthRcv JungfraujochHost) + ADD_EXECUTABLE(jfjoch_mlx_test jfjoch_mlx_test.cpp) + TARGET_LINK_LIBRARIES(jfjoch_mlx_test JungfraujochHost) ENDIF() +ADD_EXECUTABLE(jfjoch_lxsocket_test jfjoch_lxsocket_test.cpp) +TARGET_LINK_LIBRARIES(jfjoch_lxsocket_test JungfraujochHost) + IF(HAS_NUMAIF AND HAS_NUMA_H AND NUMA_LIBRARY) TARGET_COMPILE_DEFINITIONS(JungfraujochHost PUBLIC -DJFJOCH_USE_NUMA) TARGET_LINK_LIBRARIES(JungfraujochHost ${NUMA_LIBRARY}) diff --git a/receiver/host/LinuxSocketDevice.cpp b/receiver/host/LinuxSocketDevice.cpp index a42858657..d15e34453 100644 --- a/receiver/host/LinuxSocketDevice.cpp +++ b/receiver/host/LinuxSocketDevice.cpp @@ -14,60 +14,47 @@ LinuxSocketDevice::LinuxSocketDevice(uint32_t in_ipv4_addr, uint16_t in_udp_port, uint16_t data_stream, size_t in_frame_buffer_size_modules, - int16_t in_numa_node) : + int32_t in_rcv_buf_size, int16_t in_numa_node) : AcquisitionDevice(data_stream), ipv4_addr(in_ipv4_addr), udp_port(in_udp_port), - numa_node(in_numa_node), process(completion_queue, wr_queue, MAX_MODULES) { + numa_node(in_numa_node), rcv_buf_size(in_rcv_buf_size) { max_modules = 16; MapBuffersStandard(in_frame_buffer_size_modules, 1, numa_node); + mac_addr = 0; FindMACAddress(); } -void LinuxSocketDevice::MeasureThread() { - int fd; +void LinuxSocketDevice::MeasureThread(int fd) { + jf_udp_payload jf{}; + + uint64_t packet_count = 0; + + completion_queue.Put(Completion{ + .type = Completion::Type::Start + }); try { - fd = socket(AF_INET, SOCK_DGRAM, 0); - if (fd < 0) - throw JFJochException(JFJochExceptionCategory::UDPError, "Cannot create UDP socket"); - - sockaddr_in server_addr{ - .sin_family = AF_INET, - .sin_port = htons(udp_port), - .sin_addr = {.s_addr = ipv4_addr} - }; - - timeval timeout{ - .tv_sec = 0, - .tv_usec = 10000 // 10 ms - }; - - if (setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)) <= 0) - throw JFJochException(JFJochExceptionCategory::UDPError, "Cannot set socket timeout"); - - if (bind(fd, (struct sockaddr *) &server_addr, sizeof(server_addr)) <= 0) - throw JFJochException(JFJochExceptionCategory::UDPError, "Cannot bind to UDP port"); - - char buffer[9000]; + ProcessJFPacket process(completion_queue, wr_queue, max_modules); while (!cancel) { - auto count = recv(fd, buffer, sizeof(buffer), 0); + auto count = recv(fd, &jf, sizeof(jf_udp_payload), 0); - if (count == sizeof(jf_udp_payload)) - process.ProcessPacket((jf_udp_payload *) buffer); - else if ((errno != EAGAIN) && (errno != EWOULDBLOCK) && (count == -1)) + if (count == sizeof(jf_udp_payload)) { + process.ProcessPacket(&jf); + packet_count++; + } else if ((count == -1) && (errno != EAGAIN) && (errno != EWOULDBLOCK)) throw JFJochException(JFJochExceptionCategory::UDPError, "Error in UDP receiving"); } - - close(fd); - - idle = true; } catch (const JFJochException &e) { - cancel = true; - idle = true; - if (fd > 0) - close(fd); - throw e; + if (logger) + logger->ErrorException(e); } + // End message should be sent always + completion_queue.Put(Completion{ + .type = Completion::Type::End, + .frame_number = packet_count + }); + close(fd); + idle = true; } Completion LinuxSocketDevice::ReadCompletion() { @@ -87,9 +74,35 @@ void LinuxSocketDevice::HW_StartAction() { throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "Conversion on CPU flag has to be enabled for Raw Ethernet device"); + int fd = socket(AF_INET, SOCK_DGRAM, 0); + if (fd < 0) + throw JFJochException(JFJochExceptionCategory::UDPError, "Cannot create UDP socket"); + + sockaddr_in server_addr{ + .sin_family = AF_INET, + .sin_port = htons(udp_port), + .sin_addr = {.s_addr = ipv4_addr} + }; + + timeval timeout{ + .tv_sec = 0, + .tv_usec = 10000 // 10 ms + }; + + if (setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)) != 0) + throw JFJochException(JFJochExceptionCategory::UDPError, "Cannot set socket timeout"); + + if (rcv_buf_size > 0) { + if (setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &rcv_buf_size, sizeof(rcv_buf_size)) != 0) + throw JFJochException(JFJochExceptionCategory::UDPError, "Cannot set receive buffer size"); + } + if (bind(fd, (struct sockaddr *) &server_addr, sizeof(server_addr)) != 0) + throw JFJochException(JFJochExceptionCategory::UDPError, "Cannot bind to UDP port"); + cancel = false; idle = false; - measure = std::async(std::launch::async, &LinuxSocketDevice::MeasureThread, this); + + measure = std::async(std::launch::async, &LinuxSocketDevice::MeasureThread, this, fd); } bool LinuxSocketDevice::HW_IsIdle() const { diff --git a/receiver/host/LinuxSocketDevice.h b/receiver/host/LinuxSocketDevice.h index 6de0c5657..57eedcdea 100644 --- a/receiver/host/LinuxSocketDevice.h +++ b/receiver/host/LinuxSocketDevice.h @@ -12,7 +12,8 @@ class LinuxSocketDevice : public AcquisitionDevice { ThreadSafeFIFO completion_queue; ThreadSafeFIFO wr_queue; - ProcessJFPacket process; + + int32_t rcv_buf_size; uint64_t mac_addr; uint32_t ipv4_addr; @@ -37,13 +38,13 @@ class LinuxSocketDevice : public AcquisitionDevice { uint32_t HW_GetIPv4Address() const override; void HW_EndAction() override; void CopyInternalPacketGenFrameToDeviceBuffer() override; - void MeasureThread(); + void MeasureThread(int fd); void FindMACAddress(); public: - LinuxSocketDevice(uint32_t in_ipv4_addr, uint16_t in_udp_port, - uint16_t data_stream, size_t in_frame_buffer_size_modules, - int16_t in_numa_node); - ~LinuxSocketDevice() override; + LinuxSocketDevice(uint32_t ipv4_addr, uint16_t udp_port, + uint16_t data_stream, size_t frame_buffer_size_modules, + int32_t rcv_buf_size = -1, int16_t in_numa_node = -1); + ~LinuxSocketDevice() override = default; void InitializeCalibration(const DiffractionExperiment &experiment, const JFCalibration &calib) override; int32_t GetNUMANode() const override; diff --git a/receiver/host/MlxRawEthDevice.cpp b/receiver/host/MlxRawEthDevice.cpp index 567685770..8bdaaccd3 100644 --- a/receiver/host/MlxRawEthDevice.cpp +++ b/receiver/host/MlxRawEthDevice.cpp @@ -42,6 +42,13 @@ void MlxRawEthDevice::HW_ReadActionRegister(ActionConfig *job) { } void MlxRawEthDevice::MeasureThread() { + + uint64_t packet_count = 0; + + completion_queue.Put(Completion{ + .type = Completion::Type::Start + }); + try { IBProtectionDomain pd(context); IBCompletionQueue cq(context, BUFFER_COUNT+2); @@ -58,10 +65,6 @@ void MlxRawEthDevice::MeasureThread() { for (int i = 0; i < BUFFER_COUNT-1; i++) qp.PostReceiveWR(*buffer.GetMemoryRegion(), i, buffer.GetLocation(i), BUFFER_SIZE); - completion_queue.Put(Completion{ - .type = Completion::Type::Start - }); - auto cq_poll_future = std::async(std::launch::async, &MlxRawEthDevice::PollCQ, this, std::ref(buffer), @@ -74,20 +77,19 @@ void MlxRawEthDevice::MeasureThread() { std::ref(buffer), std::ref(qp)); - uint64_t packet_count = cq_poll_future.get(); - + packet_count = cq_poll_future.get(); arp_future.get(); - completion_queue.Put(Completion{ - .type = Completion::Type::End, - .frame_number = packet_count - }); - } catch (const JFJochException &e) { cancel = true; - idle = true; - throw e; + if (logger) + logger->ErrorException(e); } + + completion_queue.Put(Completion{ + .type = Completion::Type::End, + .frame_number = packet_count + }); idle = true; } diff --git a/receiver/host/jfjoch_lxsocket_test.cpp b/receiver/host/jfjoch_lxsocket_test.cpp new file mode 100644 index 000000000..650552bcd --- /dev/null +++ b/receiver/host/jfjoch_lxsocket_test.cpp @@ -0,0 +1,40 @@ +// Copyright (2019-2022) Paul Scherrer Institute +// SPDX-License-Identifier: GPL-3.0-or-later + +#include "LinuxSocketDevice.h" +#include "../common/NetworkAddressConvert.h" + +int main(int argc, char **argv) { + DiffractionExperiment experiment(DetectorGeometry(4)); + experiment.Mode(DetectorMode::Raw); + experiment.ImagesPerTrigger(1000); + + Logger logger("jfjoch_lxsocket_test"); + logger.Verbose(true); + + if (argc != 3) { + logger.Error("Usage ./jfjoch_lxsocket_test "); + exit(EXIT_FAILURE); + } + + uint32_t ipv4 = IPv4AddressFromStr(argv[1]); + uint16_t udp = std::strtol(argv[2], nullptr, 10); + + try { + LinuxSocketDevice dev(ipv4, udp, 0, 4096); + logger.Info("Mac addr {}", dev.GetMACAddress()); + dev.EnableLogging(&logger); + dev.StartAction(experiment); + dev.WaitForActionComplete(); + logger.Info("Bytes received {}", dev.GetBytesReceived()); + + JFJochProtoBuf::AcquisitionDeviceStatistics statistics; + + dev.SaveStatistics(experiment, statistics); + logger.Info("{} {}", statistics.efficiency(), statistics.packets_received_per_module(0)); + + } catch (std::exception &e) { + logger.ErrorException(e); + exit(EXIT_FAILURE); + } +} \ No newline at end of file diff --git a/receiver/host/MlxRawEthRcv.cpp b/receiver/host/jfjoch_mlx_test.cpp similarity index 89% rename from receiver/host/MlxRawEthRcv.cpp rename to receiver/host/jfjoch_mlx_test.cpp index b77fd9840..899127f55 100644 --- a/receiver/host/MlxRawEthRcv.cpp +++ b/receiver/host/jfjoch_mlx_test.cpp @@ -2,10 +2,10 @@ // SPDX-License-Identifier: GPL-3.0-or-later #include "MlxRawEthDevice.h" -#include "../common/DiffractionExperiment.h" +#include "../common/NetworkAddressConvert.h" int main(int argc, char **argv) { - DiffractionExperiment experiment(2, {4,4}, 8, 36); + DiffractionExperiment experiment(DetectorGeometry(4)); experiment.Mode(DetectorMode::Raw); experiment.ImagesPerTrigger(1000); @@ -29,4 +29,5 @@ int main(int argc, char **argv) { logger.ErrorException(e); exit(EXIT_FAILURE); } + } \ No newline at end of file diff --git a/tools/UDPSimulator.cpp b/tools/UDPSimulator.cpp index a5ab5fa1a..1419154d2 100644 --- a/tools/UDPSimulator.cpp +++ b/tools/UDPSimulator.cpp @@ -29,13 +29,14 @@ UDPSimulator::~UDPSimulator() { } void UDPSimulator::SendImage(const std::string &ipv4_dest_addr, + uint16_t udp_port, uint64_t frame_number, uint16_t module_number) { std::unique_lock ul(m); struct sockaddr_in addr { .sin_family = AF_INET, - .sin_port = htons(8192 + module_number * 2), + .sin_port = htons(udp_port), .sin_addr = { .s_addr = inet_addr(ipv4_dest_addr.c_str()) } @@ -44,8 +45,10 @@ void UDPSimulator::SendImage(const std::string &ipv4_dest_addr, if (inet_pton(AF_INET, ipv4_dest_addr.c_str(), &addr.sin_addr.s_addr) <= 0) throw JFJochException(JFJochExceptionCategory::UDPError, "Cannot parse address " + ipv4_dest_addr); + uint16_t half_module_number = 2 * module_number; jf_udp_payload packet{ - .framenum = frame_number + .framenum = frame_number, + .xCoord = half_module_number }; for (int i = 0; i < 128; i++) { diff --git a/tools/UDPSimulator.h b/tools/UDPSimulator.h index 676c32e3b..95751bf2d 100644 --- a/tools/UDPSimulator.h +++ b/tools/UDPSimulator.h @@ -15,7 +15,7 @@ class UDPSimulator { public: UDPSimulator(const std::vector &image); ~UDPSimulator(); - void SendImage(const std::string &ipv4_dest_addr, uint64_t frame_number, uint16_t module_number); + void SendImage(const std::string &ipv4_dest_addr, uint16_t udp_port, uint64_t frame_number, uint16_t module_number); }; diff --git a/tools/jfjoch_udp_simulator.cpp b/tools/jfjoch_udp_simulator.cpp index 61bc5db13..60068d7a3 100644 --- a/tools/jfjoch_udp_simulator.cpp +++ b/tools/jfjoch_udp_simulator.cpp @@ -13,23 +13,27 @@ int main(int argc, char **argv) { Logger logger("jfjoch_udp_simulator"); - if ((argc < 2) || (argc > 5)) { - logger.Error("Usage ./jfjoch_udp_simulator {<# of frames> <# of modules> }"); + if ((argc < 3) || (argc > 6)) { + logger.Error("Usage ./jfjoch_udp_simulator {<# of frames> }"); exit(EXIT_FAILURE); } const std::string ipv4_addr(argv[1]); - logger.Info("IPv4 to send: {}", ipv4_addr); + + + const uint16_t udp_port = std::strtol(argv[2], nullptr, 10); + + logger.Info("Address to send: {}:{}", ipv4_addr, udp_port); uint64_t nframes = 1; - if (argc == 3) - nframes = std::strtol(argv[2], nullptr, 10); + if (argc == 4) + nframes = std::strtol(argv[3], nullptr, 10); logger.Info("Frames to send: {}", nframes); const std::vector image(RAW_MODULE_SIZE, 0); - if (argc == 4) { - std::fstream file(argv[3], std::fstream::in | std::fstream::binary); + if (argc == 5) { + std::fstream file(argv[4], std::fstream::in | std::fstream::binary); if (file.is_open()) file.read((char *) image.data(), RAW_MODULE_SIZE * sizeof(uint16_t)); else { @@ -45,7 +49,7 @@ int main(int argc, char **argv) { UDPSimulator simulator(image); for (int i = 0 ; i < nframes; i++) - simulator.SendImage(ipv4_addr, i + 1, 0); + simulator.SendImage(ipv4_addr, udp_port, i + 1, 0); auto end_time = std::chrono::system_clock::now(); logger.Info(" ... done");