aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--contrib/Socket.cpp41
-rw-r--r--contrib/Socket.h16
-rw-r--r--contrib/ThreadsafeQueue.h38
-rw-r--r--contrib/edioutput/Transport.cpp16
-rw-r--r--contrib/edioutput/Transport.h11
5 files changed, 101 insertions, 21 deletions
diff --git a/contrib/Socket.cpp b/contrib/Socket.cpp
index a85b98b..5c920d7 100644
--- a/contrib/Socket.cpp
+++ b/contrib/Socket.cpp
@@ -24,6 +24,7 @@
#include "Socket.h"
+#include <numeric>
#include <stdexcept>
#include <cstdio>
#include <cstring>
@@ -478,7 +479,7 @@ TCPSocket::~TCPSocket()
TCPSocket::TCPSocket(TCPSocket&& other) :
m_sock(other.m_sock),
- m_remote_address(move(other.m_remote_address))
+ m_remote_address(std::move(other.m_remote_address))
{
if (other.m_sock != -1) {
other.m_sock = -1;
@@ -967,12 +968,22 @@ ssize_t TCPClient::recv(void *buffer, size_t length, int flags, int timeout_ms)
reconnect();
}
+ m_last_received_packet_ts = chrono::steady_clock::now();
+
return ret;
}
catch (const TCPSocket::Interrupted&) {
return -1;
}
catch (const TCPSocket::Timeout&) {
+ const auto timeout = chrono::milliseconds(timeout_ms * 5);
+ if (m_last_received_packet_ts.has_value() and
+ chrono::steady_clock::now() - *m_last_received_packet_ts > timeout)
+ {
+ // This is to catch half-closed TCP connections
+ reconnect();
+ }
+
return 0;
}
@@ -983,6 +994,7 @@ void TCPClient::reconnect()
{
TCPSocket newsock;
m_sock = std::move(newsock);
+ m_last_received_packet_ts = nullopt;
m_sock.connect(m_hostname, m_port, true);
}
@@ -990,7 +1002,7 @@ TCPConnection::TCPConnection(TCPSocket&& sock) :
queue(),
m_running(true),
m_sender_thread(),
- m_sock(move(sock))
+ m_sock(std::move(sock))
{
#if MISSING_OWN_ADDR
auto own_addr = m_sock.getOwnAddress();
@@ -1052,6 +1064,17 @@ void TCPConnection::process()
#endif
}
+TCPConnection::stats_t TCPConnection::get_stats() const
+{
+ TCPConnection::stats_t s;
+ const vector<size_t> buffer_sizes = queue.map<size_t>(
+ [](const vector<uint8_t>& vec) { return vec.size(); }
+ );
+
+ s.buffer_fullness = std::accumulate(buffer_sizes.cbegin(), buffer_sizes.cend(), 0);
+ s.remote_address = m_sock.get_remote_address();
+ return s;
+}
TCPDataDispatcher::TCPDataDispatcher(size_t max_queue_size, size_t buffers_to_preroll) :
m_max_queue_size(max_queue_size),
@@ -1109,7 +1132,7 @@ void TCPDataDispatcher::process()
auto sock = m_listener_socket.accept(timeout_ms);
if (sock.valid()) {
auto lock = unique_lock<mutex>(m_mutex);
- m_connections.emplace(m_connections.begin(), move(sock));
+ m_connections.emplace(m_connections.begin(), std::move(sock));
if (m_buffers_to_preroll > 0) {
for (const auto& buf : m_preroll_queue) {
@@ -1125,6 +1148,16 @@ void TCPDataDispatcher::process()
}
}
+
+std::vector<TCPConnection::stats_t> TCPDataDispatcher::get_stats() const
+{
+ std::vector<TCPConnection::stats_t> s;
+ for (const auto& conn : m_connections) {
+ s.push_back(conn.get_stats());
+ }
+ return s;
+}
+
TCPReceiveServer::TCPReceiveServer(size_t blocksize) :
m_blocksize(blocksize)
{
@@ -1181,7 +1214,7 @@ void TCPReceiveServer::process()
}
else {
buf.resize(r);
- m_queue.push(make_shared<TCPReceiveMessageData>(move(buf)));
+ m_queue.push(make_shared<TCPReceiveMessageData>(std::move(buf)));
}
}
catch (const TCPSocket::Interrupted&) {
diff --git a/contrib/Socket.h b/contrib/Socket.h
index ab2a14a..29b618a 100644
--- a/contrib/Socket.h
+++ b/contrib/Socket.h
@@ -31,9 +31,11 @@
#include "ThreadsafeQueue.h"
#include <cstdlib>
#include <atomic>
-#include <string>
+#include <chrono>
#include <list>
#include <memory>
+#include <optional>
+#include <string>
#include <thread>
#include <vector>
@@ -211,6 +213,8 @@ class TCPSocket {
SOCKET get_sockfd() const { return m_sock; }
+ InetAddress get_remote_address() const { return m_remote_address; }
+
private:
explicit TCPSocket(int sockfd);
explicit TCPSocket(int sockfd, InetAddress remote_address);
@@ -236,6 +240,8 @@ class TCPClient {
TCPSocket m_sock;
std::string m_hostname;
int m_port;
+
+ std::optional<std::chrono::steady_clock::time_point> m_last_received_packet_ts;
};
/* Helper class for TCPDataDispatcher, contains a queue of pending data and
@@ -250,6 +256,12 @@ class TCPConnection
ThreadsafeQueue<std::vector<uint8_t> > queue;
+ struct stats_t {
+ size_t buffer_fullness = 0;
+ InetAddress remote_address;
+ };
+ stats_t get_stats() const;
+
private:
std::atomic<bool> m_running;
std::thread m_sender_thread;
@@ -272,6 +284,8 @@ class TCPDataDispatcher
void start(int port, const std::string& address);
void write(const std::vector<uint8_t>& data);
+ std::vector<TCPConnection::stats_t> get_stats() const;
+
private:
void process();
diff --git a/contrib/ThreadsafeQueue.h b/contrib/ThreadsafeQueue.h
index 8b385d6..13bc19e 100644
--- a/contrib/ThreadsafeQueue.h
+++ b/contrib/ThreadsafeQueue.h
@@ -2,7 +2,7 @@
Copyright (C) 2007, 2008, 2009, 2010, 2011 Her Majesty the Queen in
Right of Canada (Communications Research Center Canada)
- Copyright (C) 2023
+ Copyright (C) 2025
Matthias P. Braendli, matthias.braendli@mpb.li
An implementation for a threadsafe queue, depends on C++11
@@ -28,6 +28,7 @@
#pragma once
+#include <functional>
#include <mutex>
#include <condition_variable>
#include <queue>
@@ -63,10 +64,10 @@ public:
std::unique_lock<std::mutex> lock(the_mutex);
size_t queue_size_before = the_queue.size();
if (max_size == 0) {
- the_queue.push(val);
+ the_queue.push_back(val);
}
else if (queue_size_before < max_size) {
- the_queue.push(val);
+ the_queue.push_back(val);
}
size_t queue_size = the_queue.size();
lock.unlock();
@@ -80,10 +81,10 @@ public:
std::unique_lock<std::mutex> lock(the_mutex);
size_t queue_size_before = the_queue.size();
if (max_size == 0) {
- the_queue.emplace(std::move(val));
+ the_queue.emplace_back(std::move(val));
}
else if (queue_size_before < max_size) {
- the_queue.emplace(std::move(val));
+ the_queue.emplace_back(std::move(val));
}
size_t queue_size = the_queue.size();
lock.unlock();
@@ -110,9 +111,9 @@ public:
bool overflow = false;
while (the_queue.size() >= max_size) {
overflow = true;
- the_queue.pop();
+ the_queue.pop_front();
}
- the_queue.push(val);
+ the_queue.push_back(val);
const size_t queue_size = the_queue.size();
lock.unlock();
@@ -129,9 +130,9 @@ public:
bool overflow = false;
while (the_queue.size() >= max_size) {
overflow = true;
- the_queue.pop();
+ the_queue.pop_front();
}
- the_queue.emplace(std::move(val));
+ the_queue.emplace_back(std::move(val));
const size_t queue_size = the_queue.size();
lock.unlock();
@@ -152,7 +153,7 @@ public:
while (the_queue.size() >= threshold) {
the_tx_notification.wait(lock);
}
- the_queue.push(val);
+ the_queue.push_back(val);
size_t queue_size = the_queue.size();
lock.unlock();
@@ -198,7 +199,7 @@ public:
}
popped_value = the_queue.front();
- the_queue.pop();
+ the_queue.pop_front();
lock.unlock();
the_tx_notification.notify_one();
@@ -220,15 +221,26 @@ public:
}
else {
std::swap(popped_value, the_queue.front());
- the_queue.pop();
+ the_queue.pop_front();
lock.unlock();
the_tx_notification.notify_one();
}
}
+ template<typename R>
+ std::vector<R> map(std::function<R(const T&)> func) const
+ {
+ std::vector<R> result;
+ std::unique_lock<std::mutex> lock(the_mutex);
+ for (const T& elem : the_queue) {
+ result.push_back(func(elem));
+ }
+ return result;
+ }
+
private:
- std::queue<T> the_queue;
+ std::deque<T> the_queue;
mutable std::mutex the_mutex;
std::condition_variable the_rx_notification;
std::condition_variable the_tx_notification;
diff --git a/contrib/edioutput/Transport.cpp b/contrib/edioutput/Transport.cpp
index 4979e93..a5e0bc3 100644
--- a/contrib/edioutput/Transport.cpp
+++ b/contrib/edioutput/Transport.cpp
@@ -1,5 +1,5 @@
/*
- Copyright (C) 2022
+ Copyright (C) 2025
Matthias P. Braendli, matthias.braendli@mpb.li
http://www.opendigitalradio.org
@@ -220,6 +220,20 @@ void Sender::override_pft_sequence(uint16_t pseq)
edi_pft.OverridePSeq(pseq);
}
+std::vector<Sender::stats_t> Sender::get_tcp_server_stats() const
+{
+ std::vector<Sender::stats_t> stats;
+
+ for (auto& el : tcp_dispatchers) {
+ Sender::stats_t s;
+ s.listen_port = el.first->listen_port;
+ s.stats = el.second->get_stats();
+ stats.push_back(s);
+ }
+
+ return stats;
+}
+
void Sender::run()
{
while (m_running) {
diff --git a/contrib/edioutput/Transport.h b/contrib/edioutput/Transport.h
index c62545c..2ca638e 100644
--- a/contrib/edioutput/Transport.h
+++ b/contrib/edioutput/Transport.h
@@ -1,5 +1,5 @@
/*
- Copyright (C) 2022
+ Copyright (C) 2025
Matthias P. Braendli, matthias.braendli@mpb.li
http://www.opendigitalradio.org
@@ -38,10 +38,11 @@
#include <cstdint>
#include <thread>
#include <mutex>
+#include <vector>
namespace edi {
-/** STI sender for EDI output */
+/** ETI/STI sender for EDI output */
class Sender {
public:
@@ -64,6 +65,12 @@ class Sender {
void override_af_sequence(uint16_t seq);
void override_pft_sequence(uint16_t pseq);
+ struct stats_t {
+ uint16_t listen_port;
+ std::vector<Socket::TCPConnection::stats_t> stats;
+ };
+ std::vector<stats_t> get_tcp_server_stats() const;
+
private:
void run();
generated by cgit v1.2.3 (git 2.39.1) at 2025-07-03 00:53:45 +0000