detectors/slsDetectorPackage
25
0
Fork 0
mirror of https://github.com/slsdetectorgroup/slsDetectorPackage.git synced 2025-06-09 03:20:42 +02:00
Code Issues Actions Packages Releases Activity

removed copy, fixed receive to external buffer

Browse Source
This commit is contained in:
Erik Frojdh 2020-02-18 15:02:40 +01:00
parent fc27cfd663
commit 53a3656d5e
2 changed files with 63 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

66
slsSupportLib/include/UdpRxSocket.h
View File

@ -9,12 +9,14 @@ this might be deprecated in the future
*/ */
#include "container_utils.h"
#include "genericSocket.h" #include "genericSocket.h"
#include "network_utils.h" #include "network_utils.h"
#include "sls_detector_exceptions.h" #include "sls_detector_exceptions.h"
#include <cstdint> #include <cstdint>
#include <errno.h> #include <errno.h>
#include <iostream> #include <iostream>
#include <memory>
#include <netdb.h> #include <netdb.h>
#include <netinet/in.h> #include <netinet/in.h>
#include <string.h> #include <string.h>
@ -26,12 +28,12 @@ namespace sls {
class UdpRxSocket { class UdpRxSocket {
const ssize_t packet_size; const ssize_t packet_size;
char *buff; std::unique_ptr<char[]> buff;
int fd = -1; int sockfd = -1;
public: public:
UdpRxSocket(int port, ssize_t packet_size, const char *hostname = nullptr, UdpRxSocket(int port, ssize_t packet_size, const char *hostname = nullptr,
ssize_t buffer_size = 0) ssize_t kernel_rbuffer_size = 0)
: packet_size(packet_size) { : packet_size(packet_size) {
/* hostname = nullptr -> wildcard */ /* hostname = nullptr -> wildcard */
@ -48,11 +50,11 @@ class UdpRxSocket {
throw RuntimeError("Failed at getaddrinfo with " + throw RuntimeError("Failed at getaddrinfo with " +
std::string(hostname)); std::string(hostname));
} }
fd = socket(res->ai_family, res->ai_socktype, res->ai_protocol); sockfd = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
if (fd == -1) { if (sockfd == -1) {
throw RuntimeError("Failed to create UDP RX socket"); throw RuntimeError("Failed to create UDP RX socket");
} }
if (bind(fd, res->ai_addr, res->ai_addrlen) == -1) { if (bind(sockfd, res->ai_addr, res->ai_addrlen) == -1) {
throw RuntimeError("Failed to bind UDP RX socket"); throw RuntimeError("Failed to bind UDP RX socket");
} }
freeaddrinfo(res); freeaddrinfo(res);
@ -60,19 +62,19 @@ class UdpRxSocket {
// If we get a specified buffer size that is larger than the set one // 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 // we set it. Otherwise we leave it there since it could have been
// set by the rx_udpsocksize command // set by the rx_udpsocksize command
if (buffer_size) { if (kernel_rbuffer_size) {
auto current = getBufferSize() / 2; auto current = getBufferSize() / 2;
if (current < buffer_size) { if (current < kernel_rbuffer_size) {
setBufferSize(buffer_size); setBufferSize(kernel_rbuffer_size);
if (getBufferSize() / 2 < buffer_size) { if (getBufferSize() / 2 < kernel_rbuffer_size) {
FILE_LOG(logWARNING) FILE_LOG(logWARNING)
<< "Could not set buffer size. Got: " << "Could not set buffer size. Got: "
<< getBufferSize() / 2 << " instead of " << buffer_size; << getBufferSize() / 2 << " instead of "
<< kernel_rbuffer_size;
} }
} }
} }
// Allocate at the end to avoid memory leak if we throw buff = sls::make_unique<char[]>(packet_size);
buff = new char[packet_size];
} }
// Delegating constructor to allow drop in replacement for old socket class // Delegating constructor to allow drop in replacement for old socket class
@ -85,34 +87,38 @@ class UdpRxSocket {
buf_size) {} buf_size) {}
~UdpRxSocket() { ~UdpRxSocket() {
delete[] buff; if (sockfd >= 0)
Shutdown(); close(sockfd);
} }
const char *LastPacket() const noexcept { return buff; } UdpRxSocket(const UdpRxSocket &) = delete;
UdpRxSocket(UdpRxSocket &&) = delete;
const char *LastPacket() const noexcept { return buff.get(); }
constexpr ssize_t getPacketSize() const noexcept { return packet_size; } constexpr ssize_t getPacketSize() const noexcept { return packet_size; }
bool ReceivePacket() noexcept { bool ReceivePacket() noexcept {
auto bytes_received = return ReceivePacket(buff.get());
recvfrom(fd, buff, packet_size, 0, nullptr, nullptr); // auto bytes_received =
return bytes_received == packet_size; // recvfrom(sockfd, buff.get(), packet_size, 0, nullptr, nullptr);
// return bytes_received == packet_size;
} }
bool ReceivePacket(char *dst) noexcept { bool ReceivePacket(char *dst) noexcept {
auto bytes_received = auto bytes_received =
recvfrom(fd, buff, packet_size, 0, nullptr, nullptr); recvfrom(sockfd, dst, packet_size, 0, nullptr, nullptr);
return bytes_received == packet_size; return bytes_received == packet_size;
} }
// Only for backwards compatibility this function will be removed during // Only for backwards compatibility this function will be removed during
// refactoring of the receiver // refactoring of the receiver
ssize_t ReceiveDataOnly(char *dst) { ssize_t ReceiveDataOnly(char *dst) {
auto r = recvfrom(fd, dst, packet_size, 0, nullptr, nullptr); auto r = recvfrom(sockfd, dst, packet_size, 0, nullptr, nullptr);
constexpr ssize_t eiger_header_packet = constexpr ssize_t eiger_header_packet =
40; // only detector that has this 40; // only detector that has this
if (r == eiger_header_packet) { if (r == eiger_header_packet) {
FILE_LOG(logWARNING) << "Got header pkg"; FILE_LOG(logWARNING) << "Got header pkg";
r = recvfrom(fd, dst, packet_size, 0, nullptr, nullptr); r = recvfrom(sockfd, dst, packet_size, 0, nullptr, nullptr);
} }
return r; return r;
} }
@ -120,7 +126,7 @@ class UdpRxSocket {
ssize_t getBufferSize() const { ssize_t getBufferSize() const {
uint64_t ret_size = 0; uint64_t ret_size = 0;
socklen_t optlen = sizeof(uint64_t); socklen_t optlen = sizeof(uint64_t);
if (getsockopt(fd, SOL_SOCKET, SO_RCVBUF, &ret_size, &optlen) == -1) if (getsockopt(sockfd, SOL_SOCKET, SO_RCVBUF, &ret_size, &optlen) == -1)
return -1; return -1;
else else
return ret_size; return ret_size;
@ -130,20 +136,20 @@ class UdpRxSocket {
ssize_t getActualUDPSocketBufferSize() const { return getBufferSize(); } ssize_t getActualUDPSocketBufferSize() const { return getBufferSize(); }
// Only for backwards compatibility will be removed // Only for backwards compatibility will be removed
void ShutDownSocket() { Shutdown(); } void ShutDownSocket() { Close(); }
void setBufferSize(ssize_t size) { void setBufferSize(ssize_t size) {
socklen_t optlen = sizeof(size); socklen_t optlen = sizeof(size);
if (setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &size, optlen)) { if (setsockopt(sockfd, SOL_SOCKET, SO_RCVBUF, &size, optlen)) {
throw RuntimeError("Could not set socket buffer size"); throw RuntimeError("Could not set socket buffer size");
} }
} }
void Shutdown() { // Do we need this function or can we rely on scope?
shutdown(fd, SHUT_RDWR); void Close() {
if (fd >= 0) { if (sockfd >= 0) {
close(fd); close(sockfd);
fd = -1; sockfd = -1;
} }
} }
}; };

34
slsSupportLib/tests/test-UdpRxSocket.cpp
View File

@ -5,6 +5,8 @@
#include <thread> #include <thread>
#include <vector> #include <vector>
constexpr int default_port = 50001;
int open_socket(int port) { int open_socket(int port) {
const char *host = nullptr; // localhost const char *host = nullptr; // localhost
@ -34,9 +36,8 @@ int open_socket(int port) {
return fd; return fd;
} }
TEST_CASE("Receive a packet on localhost") { TEST_CASE("Receive data on localhost") {
constexpr int port = 50001; constexpr int port = 50001;
std::vector<int> data_to_send{4, 5, 3, 2, 5, 7, 2, 3}; std::vector<int> data_to_send{4, 5, 3, 2, 5, 7, 2, 3};
ssize_t packet_size = ssize_t packet_size =
sizeof(decltype(data_to_send)::value_type) * data_to_send.size(); sizeof(decltype(data_to_send)::value_type) * data_to_send.size();
@ -44,9 +45,6 @@ TEST_CASE("Receive a packet on localhost") {
int fd = open_socket(port); 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); auto n = write(fd, data_to_send.data(), packet_size);
CHECK(n == packet_size); CHECK(n == packet_size);
CHECK(udpsock.ReceivePacket()); CHECK(udpsock.ReceivePacket());
@ -60,7 +58,7 @@ TEST_CASE("Receive a packet on localhost") {
} }
} }
TEST_CASE("Shutdown socket without hanging") { TEST_CASE("Shutdown socket without hanging when waiting for data") {
constexpr int port = 50001; constexpr int port = 50001;
constexpr ssize_t packet_size = 8000; constexpr ssize_t packet_size = 8000;
sls::UdpRxSocket s{port, packet_size}; sls::UdpRxSocket s{port, packet_size};
@ -72,7 +70,7 @@ TEST_CASE("Shutdown socket without hanging") {
&sls::UdpRxSocket::ReceivePacket), &sls::UdpRxSocket::ReceivePacket),
&s); &s);
s.Shutdown(); s.Close();
auto r = ret.get(); auto r = ret.get();
CHECK(r == false); // since we didn't get the packet CHECK(r == false); // since we didn't get the packet
@ -86,4 +84,26 @@ TEST_CASE("Too small packet"){
write(fd, &val, sizeof(val)); write(fd, &val, sizeof(val));
CHECK(s.ReceivePacket() == false); CHECK(s.ReceivePacket() == false);
close(fd); close(fd);
}
TEST_CASE("Receive an int to internal buffer"){
int to_send = 5;
int received = -1;
auto fd = open_socket(default_port);
sls::UdpRxSocket s(default_port, sizeof(int));
write(fd, &to_send, sizeof(to_send));
CHECK(s.ReceivePacket());
memcpy(&received, s.LastPacket(), sizeof(int));
CHECK(received == to_send);
}
TEST_CASE("Receive an int to an external buffer"){
int to_send = 5;
int received = -1;
auto fd = open_socket(default_port);
sls::UdpRxSocket s(default_port, sizeof(int));
write(fd, &to_send, sizeof(to_send));
CHECK(s.ReceivePacket(reinterpret_cast<char*>(&received)));
CHECK(received == to_send);
} }