mirror of
https://github.com/slsdetectorgroup/slsDetectorPackage.git
synced 2025-06-09 11:30:43 +02:00
Udpsocket (#507)
* udp socket refactor from reuss: closing socket before throwing for bind error in constructor, closing socket at destructor and not at shutdown to allow other thread to read with a -1, so the object can still be accessed * check for size of packet for every detector * nullptr to unique ptr to call its destructor, when deallocating resources moved out of shutdown * minor
This commit is contained in:
parent
07ff28e9e5
commit
e0207cfac1
@ -85,6 +85,7 @@ This document describes the differences between v7.0.0 and v6.x.x
|
|||||||
- number of storage cells is not updated in teh receiver. done. and also allowing it to be modified in running status
|
- number of storage cells is not updated in teh receiver. done. and also allowing it to be modified in running status
|
||||||
- refactored memory structure in receiver and listener code (maybe resolves stuck issue, need to check)
|
- refactored memory structure in receiver and listener code (maybe resolves stuck issue, need to check)
|
||||||
- callback modified to have rx header and not rx header pointer
|
- callback modified to have rx header and not rx header pointer
|
||||||
|
- rx udp socket refactored (maybe resolves getting stuck?)remove check for eiger header and isntead checks for malformed packets for every detector
|
||||||
-help should not create a new object
|
-help should not create a new object
|
||||||
|
|
||||||
|
|
||||||
|
@ -151,14 +151,13 @@ void Listener::CreateUDPSocket(int &actualSize) {
|
|||||||
if (disabledPort) {
|
if (disabledPort) {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
ShutDownUDPSocket();
|
|
||||||
|
|
||||||
uint32_t packetSize = generalData->packetSize;
|
uint32_t packetSize = generalData->packetSize;
|
||||||
if (generalData->detType == GOTTHARD2 && index != 0) {
|
if (generalData->detType == GOTTHARD2 && index != 0) {
|
||||||
packetSize = generalData->vetoPacketSize;
|
packetSize = generalData->vetoPacketSize;
|
||||||
}
|
}
|
||||||
|
|
||||||
try {
|
try {
|
||||||
|
udpSocket = nullptr;
|
||||||
udpSocket = make_unique<UdpRxSocket>(
|
udpSocket = make_unique<UdpRxSocket>(
|
||||||
udpPortNumber, packetSize,
|
udpPortNumber, packetSize,
|
||||||
(eth.length() ? InterfaceNameToIp(eth).str().c_str() : nullptr),
|
(eth.length() ? InterfaceNameToIp(eth).str().c_str() : nullptr),
|
||||||
@ -340,18 +339,13 @@ uint32_t Listener::ListenToAnImage(sls_receiver_header &dstHeader,
|
|||||||
// never entering this loop)
|
// never entering this loop)
|
||||||
while (numpackets < pperFrame) {
|
while (numpackets < pperFrame) {
|
||||||
// listen to new packet
|
// listen to new packet
|
||||||
int rc = 0;
|
if (!udpSocketAlive || !udpSocket->ReceivePacket(&listeningPacket[0])) {
|
||||||
if (udpSocketAlive) {
|
// end of acquisition
|
||||||
rc = udpSocket->ReceiveDataOnly(&listeningPacket[0]);
|
|
||||||
}
|
|
||||||
// end of acquisition
|
|
||||||
if (rc <= 0) {
|
|
||||||
if (numpackets == 0)
|
if (numpackets == 0)
|
||||||
return 0;
|
return 0;
|
||||||
return HandleFuturePacket(true, numpackets, fnum, isHeaderEmpty,
|
return HandleFuturePacket(true, numpackets, fnum, isHeaderEmpty,
|
||||||
imageSize, dstHeader);
|
imageSize, dstHeader);
|
||||||
}
|
}
|
||||||
|
|
||||||
numPacketsCaught++;
|
numPacketsCaught++;
|
||||||
numPacketsStatistic++;
|
numPacketsStatistic++;
|
||||||
GetPacketIndices(fnum, pnum, bnum, standardHeader,
|
GetPacketIndices(fnum, pnum, bnum, standardHeader,
|
||||||
|
@ -23,10 +23,6 @@ class UdpRxSocket {
|
|||||||
void setBufferSize(int size);
|
void setBufferSize(int size);
|
||||||
ssize_t getPacketSize() const noexcept;
|
ssize_t getPacketSize() const noexcept;
|
||||||
void Shutdown();
|
void Shutdown();
|
||||||
|
|
||||||
// Only for backwards compatibility, this drops the EIGER small pkt, may be
|
|
||||||
// removed
|
|
||||||
ssize_t ReceiveDataOnly(char *dst) noexcept;
|
|
||||||
};
|
};
|
||||||
|
|
||||||
} // namespace sls
|
} // namespace sls
|
||||||
|
@ -18,13 +18,11 @@ namespace sls {
|
|||||||
UdpRxSocket::UdpRxSocket(int port, ssize_t packet_size, const char *hostname,
|
UdpRxSocket::UdpRxSocket(int port, ssize_t packet_size, const char *hostname,
|
||||||
int kernel_buffer_size)
|
int kernel_buffer_size)
|
||||||
: packet_size_(packet_size) {
|
: packet_size_(packet_size) {
|
||||||
struct addrinfo hints;
|
struct addrinfo hints {};
|
||||||
memset(&hints, 0, sizeof(hints));
|
|
||||||
hints.ai_family = AF_UNSPEC;
|
hints.ai_family = AF_UNSPEC;
|
||||||
hints.ai_socktype = SOCK_DGRAM;
|
hints.ai_socktype = SOCK_DGRAM;
|
||||||
hints.ai_protocol = 0;
|
|
||||||
hints.ai_flags = AI_PASSIVE | AI_ADDRCONFIG;
|
hints.ai_flags = AI_PASSIVE | AI_ADDRCONFIG;
|
||||||
struct addrinfo *res = nullptr;
|
struct addrinfo *res{nullptr};
|
||||||
|
|
||||||
const std::string portname = std::to_string(port);
|
const std::string portname = std::to_string(port);
|
||||||
if (getaddrinfo(hostname, portname.c_str(), &hints, &res)) {
|
if (getaddrinfo(hostname, portname.c_str(), &hints, &res)) {
|
||||||
@ -36,6 +34,7 @@ UdpRxSocket::UdpRxSocket(int port, ssize_t packet_size, const char *hostname,
|
|||||||
throw RuntimeError("Failed to create UDP RX socket");
|
throw RuntimeError("Failed to create UDP RX socket");
|
||||||
}
|
}
|
||||||
if (bind(sockfd_, res->ai_addr, res->ai_addrlen) == -1) {
|
if (bind(sockfd_, res->ai_addr, res->ai_addrlen) == -1) {
|
||||||
|
close(sockfd_);
|
||||||
throw RuntimeError("Failed to bind UDP RX socket");
|
throw RuntimeError("Failed to bind UDP RX socket");
|
||||||
}
|
}
|
||||||
freeaddrinfo(res);
|
freeaddrinfo(res);
|
||||||
@ -56,29 +55,19 @@ UdpRxSocket::UdpRxSocket(int port, ssize_t packet_size, const char *hostname,
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
UdpRxSocket::~UdpRxSocket() { Shutdown(); }
|
UdpRxSocket::~UdpRxSocket() {
|
||||||
|
Shutdown();
|
||||||
|
close(sockfd_);
|
||||||
|
sockfd_ = -1;
|
||||||
|
}
|
||||||
|
|
||||||
ssize_t UdpRxSocket::getPacketSize() const noexcept { return packet_size_; }
|
ssize_t UdpRxSocket::getPacketSize() const noexcept { return packet_size_; }
|
||||||
|
|
||||||
bool UdpRxSocket::ReceivePacket(char *dst) noexcept {
|
bool UdpRxSocket::ReceivePacket(char *dst) noexcept {
|
||||||
auto bytes_received =
|
auto bytes_received =
|
||||||
recvfrom(sockfd_, dst, packet_size_, 0, nullptr, nullptr);
|
recvfrom(sockfd_, dst, packet_size_, 0, nullptr, nullptr);
|
||||||
return bytes_received == packet_size_;
|
|
||||||
}
|
|
||||||
|
|
||||||
ssize_t UdpRxSocket::ReceiveDataOnly(char *dst) noexcept {
|
return bytes_received == packet_size_;
|
||||||
auto r = recvfrom(sockfd_, dst, packet_size_, 0, nullptr, nullptr);
|
|
||||||
constexpr ssize_t eiger_header_packet = 40; // only detector that has this
|
|
||||||
if (r == eiger_header_packet) {
|
|
||||||
LOG(logWARNING) << "Got header pkg";
|
|
||||||
r = recvfrom(sockfd_, dst, packet_size_, 0, nullptr, nullptr);
|
|
||||||
}
|
|
||||||
// temporary workaround for Eiger firmware (stop sends bad packets of size 8
|
|
||||||
// bytes)
|
|
||||||
if (r == 8) {
|
|
||||||
LOG(logWARNING) << "Ignoring bad packet of size 8 bytes";
|
|
||||||
r = recvfrom(sockfd_, dst, packet_size_, 0, nullptr, nullptr);
|
|
||||||
}
|
|
||||||
return r;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
int UdpRxSocket::getBufferSize() const {
|
int UdpRxSocket::getBufferSize() const {
|
||||||
@ -95,10 +84,7 @@ void UdpRxSocket::setBufferSize(int size) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
void UdpRxSocket::Shutdown() {
|
void UdpRxSocket::Shutdown() {
|
||||||
|
// not closing yet on purpose, but read gives -1
|
||||||
shutdown(sockfd_, SHUT_RDWR);
|
shutdown(sockfd_, SHUT_RDWR);
|
||||||
if (sockfd_ >= 0) {
|
|
||||||
close(sockfd_);
|
|
||||||
sockfd_ = -1;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
} // namespace sls
|
} // namespace sls
|
Loading…
x
Reference in New Issue
Block a user