replaced udp socket

slsSupportLib/CMakeLists.txt

@@ -58,6 +58,7 @@ set_target_properties(slsSupportLib PROPERTIES
 target_link_libraries(slsSupportLib 
     slsProjectOptions 
     slsProjectWarnings 
+    ${ZeroMQ_LIBRARIES}
     rapidjson)
 
 if (SLS_USE_TESTS)

slsSupportLib/include/UdpRxSocket.h

@@ -0,0 +1,151 @@
+
+/*
+UdpRxSocket provies socket control to receive
+data on a udp socket.
+
+It provides a drop in replacement for
+genericSocket. But please be careful since
+this might be deprecated in the future
+
+*/
+
+#include "genericSocket.h"
+#include "network_utils.h"
+#include "sls_detector_exceptions.h"
+#include <cstdint>
+#include <errno.h>
+#include <iostream>
+#include <netdb.h>
+#include <netinet/in.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <unistd.h>
+#include <vector>
+
+namespace sls {
+
+class UdpRxSocket {
+    const ssize_t packet_size;
+    char *buff;
+    int fd = -1;
+
+  public:
+    UdpRxSocket(int port, ssize_t packet_size, const char *hostname = nullptr,
+                ssize_t buffer_size = 0)
+        : packet_size(packet_size) {
+        /* hostname = nullptr -> wildcard */
+
+        struct addrinfo hints;
+        memset(&hints, 0, sizeof(hints));
+        hints.ai_family = AF_UNSPEC;
+        hints.ai_socktype = SOCK_DGRAM;
+        hints.ai_protocol = 0;
+        hints.ai_flags = AI_PASSIVE | AI_ADDRCONFIG;
+        struct addrinfo *res = 0;
+
+        const std::string portname = std::to_string(port);
+        if (getaddrinfo(hostname, portname.c_str(), &hints, &res)) {
+            throw RuntimeError("Failed at getaddrinfo with " +
+                               std::string(hostname));
+        }
+        fd = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
+        if (fd == -1) {
+            throw RuntimeError("Failed to create UDP RX socket");
+        }
+        if (bind(fd, res->ai_addr, res->ai_addrlen) == -1) {
+            throw RuntimeError("Failed to bind UDP RX socket");
+        }
+        freeaddrinfo(res);
+
+        // If we get a specified buffer size that is larger than the set one
+        // we set it. Otherwise we leave it there since it could have been
+        // set by the rx_udpsocksize command
+        if (buffer_size) {
+            auto current = getBufferSize() / 2;
+            if (current < buffer_size) {
+                setBufferSize(buffer_size);
+                if (getBufferSize() / 2 < buffer_size) {
+                    FILE_LOG(logWARNING)
+                        << "Could not set buffer size. Got: "
+                        << getBufferSize() / 2 << " instead of " << buffer_size;
+                }
+            }
+        }
+        // Allocate at the end to avoid memory leak if we throw
+        buff = new char[packet_size];
+    }
+
+    // Delegating constructor to allow drop in replacement for old socket class
+    // This one might be removed in the future
+    UdpRxSocket(unsigned short int const port_number,
+                genericSocket::communicationProtocol p,
+                int ps = DEFAULT_PACKET_SIZE, const char *eth = NULL,
+                int hsize = 0, uint64_t buf_size = SOCKET_BUFFER_SIZE)
+        : UdpRxSocket(port_number, ps, InterfaceNameToIp(eth).str().c_str(),
+                      buf_size) {}
+
+    ~UdpRxSocket() {
+        delete[] buff;
+        Shutdown();
+    }
+
+    const char *LastPacket() const noexcept { return buff; }
+    constexpr ssize_t getPacketSize() const noexcept { return packet_size; }
+
+    bool ReceivePacket() noexcept {
+        auto bytes_received =
+            recvfrom(fd, buff, packet_size, 0, nullptr, nullptr);
+        return bytes_received == packet_size;
+    }
+
+    bool ReceivePacket(char *dst) noexcept {
+        auto bytes_received =
+            recvfrom(fd, buff, packet_size, 0, nullptr, nullptr);
+        return bytes_received == packet_size;
+    }
+
+    // Only for backwards compatibility this function will be removed during
+    // refactoring of the receiver
+    ssize_t ReceiveDataOnly(char *dst) {
+        auto r = recvfrom(fd, dst, packet_size, 0, nullptr, nullptr);
+        constexpr ssize_t eiger_header_packet =
+            40; // only detector that has this
+        if (r == eiger_header_packet) {
+            FILE_LOG(logWARNING) << "Got header pkg";
+            r = recvfrom(fd, dst, packet_size, 0, nullptr, nullptr);
+        }
+        return r;
+    }
+
+    ssize_t getBufferSize() const {
+        uint64_t ret_size = 0;
+        socklen_t optlen = sizeof(uint64_t);
+        if (getsockopt(fd, SOL_SOCKET, SO_RCVBUF, &ret_size, &optlen) == -1)
+            return -1;
+        else
+            return ret_size;
+    }
+
+    // Only for backwards compatibility will be removed
+    ssize_t getActualUDPSocketBufferSize() const { return getBufferSize(); }
+
+    // Only for backwards compatibility will be removed
+    void ShutDownSocket() { Shutdown(); }
+
+    void setBufferSize(ssize_t size) {
+        socklen_t optlen = sizeof(size);
+        if (setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &size, optlen)) {
+            throw RuntimeError("Could not set socket buffer size");
+        }
+    }
+
+    void Shutdown() {
+        shutdown(fd, SHUT_RDWR);
+        if (fd >= 0) {
+            close(fd);
+            fd = -1;
+        }
+    }
+};
+
+} // namespace sls

slsSupportLib/include/network_utils.h

@@ -62,7 +62,7 @@ class MacAddr {
 IpAddr HostnameToIp(const char *hostname);
 std::string IpToInterfaceName(const std::string& ip);
 MacAddr InterfaceNameToMac(const std::string& inf);
-
+IpAddr InterfaceNameToIp(const std::string& ifn);
 std::ostream &operator<<(std::ostream &out, const IpAddr &addr);
 std::ostream &operator<<(std::ostream &out, const MacAddr &addr);
 

slsSupportLib/src/network_utils.cpp

@@ -1,36 +1,33 @@
 #include "sls_detector_exceptions.h"
 
 #include <algorithm>
+#include <arpa/inet.h>
 #include <cassert>
 #include <cstdlib>
 #include <cstring>
+#include <ifaddrs.h>
 #include <iomanip>
-#include <sstream>
-#include <sys/prctl.h> 
-#include <arpa/inet.h>
+#include <net/if.h>
 #include <netdb.h>
+#include <sstream>
+#include <sys/ioctl.h>
+#include <sys/prctl.h>
 #include <sys/socket.h>
 #include <sys/types.h>
-#include <sys/ioctl.h>
-#include <ifaddrs.h>
-#include <net/if.h>
 
 #include "network_utils.h"
 
 namespace sls {
 
-
 IpAddr::IpAddr(const std::string &address) {
     inet_pton(AF_INET, address.c_str(), &addr_);
 }
 
 IpAddr::IpAddr(const char *address) { inet_pton(AF_INET, address, &addr_); }
 
-std::string IpAddr::str() const {
-    return arr().data();
-}
+std::string IpAddr::str() const { return arr().data(); }
 
-std::array<char, INET_ADDRSTRLEN> IpAddr::arr() const{
+std::array<char, INET_ADDRSTRLEN> IpAddr::arr() const {
     std::array<char, INET_ADDRSTRLEN> ipstring{};
     inet_ntop(AF_INET, &addr_, ipstring.data(), INET_ADDRSTRLEN);
     return ipstring;
@@ -96,7 +93,7 @@ IpAddr HostnameToIp(const char *hostname) {
 }
 
 std::string IpToInterfaceName(const std::string &ip) {
-    //TODO! Copied from genericSocket needs to be refactored!
+    // TODO! Copied from genericSocket needs to be refactored!
     struct ifaddrs *addrs, *iap;
     struct sockaddr_in *sa;
 
@@ -122,33 +119,61 @@ std::string IpToInterfaceName(const std::string &ip) {
     return std::string(buf);
 }
 
-MacAddr InterfaceNameToMac(const std::string& inf) {
-        //TODO! Copied from genericSocket needs to be refactored!
-		struct ifreq ifr;
-		char mac[32];
-		const int mac_len = sizeof(mac);
-		memset(mac,0,mac_len);
+IpAddr InterfaceNameToIp(const std::string &ifn) {
+    struct ifaddrs *ifaddr, *ifa;
+    // int family, s;
+    char host[NI_MAXHOST];
 
-		int sock=socket(PF_INET, SOCK_STREAM, 0);
-		strncpy(ifr.ifr_name,inf.c_str(),sizeof(ifr.ifr_name)-1);
-		ifr.ifr_name[sizeof(ifr.ifr_name)-1]='\0';
+    if (getifaddrs(&ifaddr) == -1) {
+        return {};
+    }
 
+    for (ifa = ifaddr; ifa != NULL; ifa = ifa->ifa_next) {
+        if (ifa->ifa_addr == NULL)
+            continue;
 
-		if (-1==ioctl(sock, SIOCGIFHWADDR, &ifr)) {
-			perror("ioctl(SIOCGIFHWADDR) ");
-			return MacAddr{};
-		}
-		for (int j=0, k=0; j<6; j++) {
-			k+=snprintf(mac+k, mac_len-k-1, j ? ":%02X" : "%02X",
-					(int)(unsigned int)(unsigned char)ifr.ifr_hwaddr.sa_data[j]);
-		}
-		mac[mac_len-1]='\0';
+        auto s = getnameinfo(ifa->ifa_addr, sizeof(struct sockaddr_in), host,
+                        NI_MAXHOST, NULL, 0, NI_NUMERICHOST);
 
-		if(sock!=1){
-			close(sock);
-		}
-		return MacAddr(mac);
+        if ((strcmp(ifa->ifa_name, ifn.c_str()) == 0) &&
+            (ifa->ifa_addr->sa_family == AF_INET)) {
+            if (s != 0) {
+                return {};
+            }
+            break;
+        }
+    }
 
-	}
+    freeifaddrs(ifaddr);
+    return IpAddr{host};
+}
+
+MacAddr InterfaceNameToMac(const std::string &inf) {
+    // TODO! Copied from genericSocket needs to be refactored!
+    struct ifreq ifr;
+    char mac[32];
+    const int mac_len = sizeof(mac);
+    memset(mac, 0, mac_len);
+
+    int sock = socket(PF_INET, SOCK_STREAM, 0);
+    strncpy(ifr.ifr_name, inf.c_str(), sizeof(ifr.ifr_name) - 1);
+    ifr.ifr_name[sizeof(ifr.ifr_name) - 1] = '\0';
+
+    if (-1 == ioctl(sock, SIOCGIFHWADDR, &ifr)) {
+        perror("ioctl(SIOCGIFHWADDR) ");
+        return MacAddr{};
+    }
+    for (int j = 0, k = 0; j < 6; j++) {
+        k += snprintf(
+            mac + k, mac_len - k - 1, j ? ":%02X" : "%02X",
+            (int)(unsigned int)(unsigned char)ifr.ifr_hwaddr.sa_data[j]);
+    }
+    mac[mac_len - 1] = '\0';
+
+    if (sock != 1) {
+        close(sock);
+    }
+    return MacAddr(mac);
+}
 
 } // namespace sls

slsSupportLib/tests/CMakeLists.txt

@@ -8,4 +8,5 @@ target_sources(tests PRIVATE
                 ${CMAKE_CURRENT_SOURCE_DIR}/test-FixedCapacityContainer.cpp
                 ${CMAKE_CURRENT_SOURCE_DIR}/test-ToString.cpp
                 ${CMAKE_CURRENT_SOURCE_DIR}/test-TypeTraits.cpp
+                ${CMAKE_CURRENT_SOURCE_DIR}/test-UdpRxSocket.cpp
                 )

slsSupportLib/tests/test-UdpRxSocket.cpp

@@ -0,0 +1,89 @@
+#include "UdpRxSocket.h"
+#include "catch.hpp"
+#include "sls_detector_exceptions.h"
+#include <future>
+#include <thread>
+#include <vector>
+
+int open_socket(int port) {
+    const char *host = nullptr; // localhost
+
+    // Create a socket for sending
+    struct addrinfo hints;
+    memset(&hints, 0, sizeof(hints));
+    hints.ai_family = AF_UNSPEC;
+    hints.ai_socktype = SOCK_DGRAM;
+    hints.ai_protocol = 0;
+    hints.ai_flags = AI_PASSIVE | AI_ADDRCONFIG;
+    struct addrinfo *res = 0;
+
+    const std::string portname = std::to_string(port);
+    if (getaddrinfo(host, portname.c_str(), &hints, &res)) {
+        throw sls::RuntimeError("Failed at getaddrinfo with " +
+                                std::string(host));
+    }
+    int fd = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
+    if (fd == -1) {
+        throw sls::RuntimeError("Failed to create UDP RX socket");
+    }
+
+    if (connect(fd, res->ai_addr, res->ai_addrlen)){
+        throw sls::RuntimeError("Failed to connect socket");
+    }
+    freeaddrinfo(res);
+    return fd;
+}
+
+TEST_CASE("Receive a packet on localhost") {
+    constexpr int port = 50001;
+
+    std::vector<int> data_to_send{4, 5, 3, 2, 5, 7, 2, 3};
+    ssize_t packet_size =
+        sizeof(decltype(data_to_send)::value_type) * data_to_send.size();
+    sls::UdpRxSocket udpsock{port, packet_size};
+
+
+    int fd = open_socket(port);
+    // int n = sendto(fd, data_to_send.data(), packet_size, 0, res->ai_addr,
+    //                res->ai_addrlen);
+
+    auto n = write(fd, data_to_send.data(), packet_size);
+    CHECK(n == packet_size);
+    CHECK(udpsock.ReceivePacket());
+    close(fd);
+    // Copy data from buffer and compare values
+    std::vector<int> data_received(data_to_send.size());
+    memcpy(data_received.data(), udpsock.LastPacket(), udpsock.getPacketSize());
+    CHECK(data_received.size() == data_to_send.size()); // sanity check
+    for (size_t i = 0; i != data_to_send.size(); ++i) {
+        CHECK(data_to_send[i] == data_received[i]);
+    }
+}
+
+TEST_CASE("Shutdown socket without hanging") {
+    constexpr int port = 50001;
+    constexpr ssize_t packet_size = 8000;
+    sls::UdpRxSocket s{port, packet_size};
+
+    // Start a thread and wait for package
+    // if the socket is left open we would block
+    std::future<bool> ret =
+        std::async(static_cast<bool (sls::UdpRxSocket::*)()>(
+                       &sls::UdpRxSocket::ReceivePacket),
+                   &s);
+
+    s.Shutdown();
+    auto r = ret.get();
+
+    CHECK(r == false); // since we didn't get the packet
+}
+
+TEST_CASE("Too small packet"){
+    constexpr int port = 50001;
+    sls::UdpRxSocket s(port, 2*sizeof(uint32_t));
+    auto fd = open_socket(port);
+    uint32_t val = 10;
+    write(fd, &val, sizeof(val));
+    CHECK(s.ReceivePacket() == false);
+    close(fd);
+}