aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--doc/STATS.md7
-rwxr-xr-xdoc/show_dabmux_stats.py22
-rw-r--r--lib/Socket.cpp22
-rw-r--r--lib/Socket.h10
-rw-r--r--lib/ThreadsafeQueue.h38
-rw-r--r--lib/edioutput/Transport.cpp16
-rw-r--r--lib/edioutput/Transport.h11
-rw-r--r--src/DabMultiplexer.cpp7
-rw-r--r--src/ManagementServer.cpp75
-rw-r--r--src/ManagementServer.h20
10 files changed, 181 insertions, 47 deletions
diff --git a/doc/STATS.md b/doc/STATS.md
index 385d41e..435a92e 100644
--- a/doc/STATS.md
+++ b/doc/STATS.md
@@ -4,12 +4,13 @@ Stats available through Management Server
Interface
---------
-The management server makes statistics about the inputs available through a ZMQ request/reply socket.
+The management server makes statistics about the inputs and EDI/TCP outputs
+available through a ZMQ request/reply socket.
The `show_dabmux_stats.py` illustrates how to access this information.
-Meaning of values
------------------
+Meaning of values for inputs
+----------------------------
`max` and `min` indicate input buffer fullness in bytes.
diff --git a/doc/show_dabmux_stats.py b/doc/show_dabmux_stats.py
index 7ea60f7..3b6d869 100755
--- a/doc/show_dabmux_stats.py
+++ b/doc/show_dabmux_stats.py
@@ -46,6 +46,7 @@ if len(sys.argv) == 1:
data = sock.recv().decode("utf-8")
values = json.loads(data)['values']
+ print("## INPUT STATS")
tmpl = "{ident:20}{maxfill:>8}{minfill:>8}{under:>8}{over:>8}{audioleft:>8}{audioright:>8}{peakleft:>8}{peakright:>8}{state:>16}{version:>48}{uptime:>8}{offset:>8}"
print(tmpl.format(
ident="id",
@@ -89,6 +90,27 @@ if len(sys.argv) == 1:
uptime=v['uptime'],
offset=v['last_tist_offset']))
+ sock.send(b"output_values")
+
+ poller = zmq.Poller()
+ poller.register(sock, zmq.POLLIN)
+
+ socks = dict(poller.poll(1000))
+ if socks:
+ if socks.get(sock) == zmq.POLLIN:
+ print()
+ print("## OUTPUT STATS")
+ data = sock.recv().decode("utf-8")
+ values = json.loads(data)['output_values']
+ for identifier in values:
+ if identifier.startswith("edi_tcp_"):
+ listen_port = identifier.rsplit("_", 1)[-1]
+ num_connections = values[identifier]["num_connections"]
+ print(f"EDI TCP on port {listen_port}: {num_connections} connections")
+ else:
+ print(f"Unknown output type: {identifier}")
+
+
elif len(sys.argv) == 2 and sys.argv[1] == "config":
sock = connect()
diff --git a/lib/Socket.cpp b/lib/Socket.cpp
index 938b573..5c920d7 100644
--- a/lib/Socket.cpp
+++ b/lib/Socket.cpp
@@ -24,6 +24,7 @@
#include "Socket.h"
+#include <numeric>
#include <stdexcept>
#include <cstdio>
#include <cstring>
@@ -1063,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),
@@ -1136,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)
{
diff --git a/lib/Socket.h b/lib/Socket.h
index 7709145..29b618a 100644
--- a/lib/Socket.h
+++ b/lib/Socket.h
@@ -213,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);
@@ -254,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;
@@ -276,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/lib/ThreadsafeQueue.h b/lib/ThreadsafeQueue.h
index 8b385d6..13bc19e 100644
--- a/lib/ThreadsafeQueue.h
+++ b/lib/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/lib/edioutput/Transport.cpp b/lib/edioutput/Transport.cpp
index 4979e93..a5e0bc3 100644
--- a/lib/edioutput/Transport.cpp
+++ b/lib/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/lib/edioutput/Transport.h b/lib/edioutput/Transport.h
index c62545c..2ca638e 100644
--- a/lib/edioutput/Transport.h
+++ b/lib/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();
diff --git a/src/DabMultiplexer.cpp b/src/DabMultiplexer.cpp
index a68f09a..b9575fc 100644
--- a/src/DabMultiplexer.cpp
+++ b/src/DabMultiplexer.cpp
@@ -28,6 +28,7 @@
#include <memory>
#include "DabMultiplexer.h"
#include "ConfigParser.h"
+#include "ManagementServer.h"
#include "crc.h"
#include "utils.h"
@@ -795,6 +796,12 @@ void DabMultiplexer::mux_frame(std::vector<std::shared_ptr<DabOutput> >& outputs
}
edi_sender->write(edi_tagpacket);
+
+ for (const auto& stat : edi_sender->get_tcp_server_stats()) {
+ get_mgmt_server().update_edi_tcp_output_stat(
+ stat.listen_port,
+ stat.stats.size());
+ }
}
#if _DEBUG
diff --git a/src/ManagementServer.cpp b/src/ManagementServer.cpp
index 568e80e..dff093a 100644
--- a/src/ManagementServer.cpp
+++ b/src/ManagementServer.cpp
@@ -2,7 +2,7 @@
Copyright (C) 2009 Her Majesty the Queen in Right of Canada (Communications
Research Center Canada)
- Copyright (C) 2018
+ Copyright (C) 2025
Matthias P. Braendli, matthias.braendli@mpb.li
http://www.opendigitalradio.org
@@ -28,13 +28,12 @@
along with ODR-DabMux. If not, see <http://www.gnu.org/licenses/>.
*/
-#include <errno.h>
-#include <string.h>
-#include <math.h>
-#include <stdint.h>
-#include <limits>
#include <sstream>
#include <algorithm>
+#include <cstring>
+#include <cmath>
+#include <cstdint>
+#include <limits>
#include <boost/version.hpp>
#include "ManagementServer.h"
#include "Log.h"
@@ -127,37 +126,42 @@ ManagementServer& get_mgmt_server()
*/
}
-void ManagementServer::registerInput(InputStat* is)
+void ManagementServer::register_input(InputStat* is)
{
unique_lock<mutex> lock(m_statsmutex);
std::string id(is->get_name());
- if (m_inputStats.count(id) == 1) {
+ if (m_input_stats.count(id) == 1) {
etiLog.level(error) <<
"Double registration in MGMT Server with id '" <<
id << "'";
return;
}
- m_inputStats[id] = is;
+ m_input_stats[id] = is;
}
-void ManagementServer::unregisterInput(std::string id)
+void ManagementServer::unregister_input(std::string id)
{
unique_lock<mutex> lock(m_statsmutex);
- if (m_inputStats.count(id) == 1) {
- m_inputStats.erase(id);
+ if (m_input_stats.count(id) == 1) {
+ m_input_stats.erase(id);
}
}
+// outputs will never disappear, no need to have a "remove" logic
+void ManagementServer::update_edi_tcp_output_stat(uint16_t listen_port, size_t num_connections)
+{
+ m_output_stats[listen_port] = num_connections;
+}
bool ManagementServer::isInputRegistered(std::string& id)
{
unique_lock<mutex> lock(m_statsmutex);
- if (m_inputStats.count(id) == 0) {
+ if (m_input_stats.count(id) == 0) {
etiLog.level(error) <<
"Management Server: id '" <<
id << "' does was not registered";
@@ -166,7 +170,7 @@ bool ManagementServer::isInputRegistered(std::string& id)
return true;
}
-std::string ManagementServer::getStatConfigJSON()
+std::string ManagementServer::get_input_config_json()
{
unique_lock<mutex> lock(m_statsmutex);
@@ -175,7 +179,7 @@ std::string ManagementServer::getStatConfigJSON()
std::map<std::string,InputStat*>::iterator iter;
int i = 0;
- for(iter = m_inputStats.begin(); iter != m_inputStats.end();
+ for (iter = m_input_stats.begin(); iter != m_input_stats.end();
++iter, i++)
{
std::string id = iter->first;
@@ -192,16 +196,15 @@ std::string ManagementServer::getStatConfigJSON()
return ss.str();
}
-std::string ManagementServer::getValuesJSON()
+std::string ManagementServer::get_input_values_json()
{
unique_lock<mutex> lock(m_statsmutex);
std::ostringstream ss;
ss << "{ \"values\" : {\n";
- std::map<std::string,InputStat*>::iterator iter;
int i = 0;
- for(iter = m_inputStats.begin(); iter != m_inputStats.end();
+ for (auto iter = m_input_stats.begin(); iter != m_input_stats.end();
++iter, i++)
{
const std::string& id = iter->first;
@@ -220,6 +223,31 @@ std::string ManagementServer::getValuesJSON()
return ss.str();
}
+std::string ManagementServer::get_output_values_json()
+{
+ unique_lock<mutex> lock(m_statsmutex);
+
+ std::ostringstream ss;
+ ss << "{ \"output_values\" : {\n";
+
+ int i = 0;
+ for (auto iter = m_output_stats.begin(); iter != m_output_stats.end();
+ ++iter, i++)
+ {
+ auto listen_port = iter->first;
+ auto num_connections = iter->second;
+ if (i > 0) {
+ ss << " ,\n";
+ }
+ ss << " \"edi_tcp_" << listen_port << "\" : { \"num_connections\": " <<
+ num_connections << "} ";
+ }
+
+ ss << "}\n}\n";
+
+ return ss.str();
+}
+
ManagementServer::ManagementServer() :
m_zmq_context(),
m_zmq_sock(m_zmq_context, ZMQ_REP),
@@ -323,10 +351,13 @@ void ManagementServer::handle_message(zmq::message_t& zmq_message)
<< "}\n";
}
else if (data == "config") {
- answer << getStatConfigJSON();
+ answer << get_input_config_json();
}
else if (data == "values") {
- answer << getValuesJSON();
+ answer << get_input_values_json();
+ }
+ else if (data == "output_values") {
+ answer << get_output_values_json();
}
else if (data == "getptree") {
unique_lock<mutex> lock(m_configmutex);
@@ -366,12 +397,12 @@ InputStat::InputStat(const std::string& name) :
InputStat::~InputStat()
{
- get_mgmt_server().unregisterInput(m_name);
+ get_mgmt_server().unregister_input(m_name);
}
void InputStat::registerAtServer()
{
- get_mgmt_server().registerInput(this);
+ get_mgmt_server().register_input(this);
}
void InputStat::notifyBuffer(long bufsize)
diff --git a/src/ManagementServer.h b/src/ManagementServer.h
index 6e39922..c7a4222 100644
--- a/src/ManagementServer.h
+++ b/src/ManagementServer.h
@@ -2,7 +2,7 @@
Copyright (C) 2009 Her Majesty the Queen in Right of Canada (Communications
Research Center Canada)
- Copyright (C) 2018
+ Copyright (C) 2025
Matthias P. Braendli, matthias.braendli@mpb.li
http://www.opendigitalradio.org
@@ -50,6 +50,7 @@
# include "config.h"
#endif
+#include "Socket.h"
#include "zmq.hpp"
#include <string>
#include <map>
@@ -167,8 +168,10 @@ class ManagementServer
void open(int listenport);
/* Un-/Register a statistics data source */
- void registerInput(InputStat* is);
- void unregisterInput(std::string id);
+ void register_input(InputStat* is);
+ void unregister_input(std::string id);
+
+ void update_edi_tcp_output_stat(uint16_t listen_port, size_t num_connections);
/* Load a ptree given by the management server.
*
@@ -205,20 +208,25 @@ class ManagementServer
std::thread m_restarter_thread;
/******* Statistics Data ********/
- std::map<std::string, InputStat*> m_inputStats;
+ std::map<std::string, InputStat*> m_input_stats;
+
+ // Holds information about EDI/TCP outputs
+ std::map<uint16_t /* port */, size_t /* num_connections */> m_output_stats;
/* Return a description of the configuration that will
* allow to define what graphs to be created
*
* returns: a JSON encoded configuration
*/
- std::string getStatConfigJSON();
+ std::string get_input_config_json();
/* Return the values for the statistics as defined in the configuration
*
* returns: JSON encoded statistics
*/
- std::string getValuesJSON();
+ std::string get_input_values_json();
+
+ std::string get_output_values_json();
// mutex for accessing the map
std::mutex m_statsmutex;
generated by cgit v1.2.3 (git 2.39.1) at 2025-03-14 05:16:19 +0000