Add stats sender to UNIX datagram socket

5 files changed, 269 insertions, 1 deletions

diff --git a/Makefile.am b/Makefile.am
index aea695c..05136ba 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -95,6 +95,8 @@ odr_audioenc_SOURCES     = src/odr-audioenc.cpp \
 						   src/AACDecoder.cpp \
 						   src/AACDecoder.h \
 						   src/SampleQueue.h \
+						   src/StatsPublish.cpp \
+						   src/StatsPublish.h \
 						   src/encryption.c \
 						   src/encryption.h \
 						   src/zmq.hpp \
diff --git a/example_stats_receiver.py b/example_stats_receiver.py
new file mode 100755
index 0000000..99ce199
--- /dev/null
+++ b/example_stats_receiver.py
@@ -0,0 +1,38 @@
+#!/usr/bin/env python3
+
+import logging
+logging.basicConfig(level=logging.DEBUG)
+import sys
+import os
+import os.path
+import socket
+import argparse
+import yaml
+
+parser = argparse.ArgumentParser(
+    description="Example Stats UNIX Datagram Socket Receiver")
+parser.add_argument('-s', '--socket', default="/tmp/stats", type=str,
+        help='Full path of the socket',
+        required=False)
+
+cli_args = parser.parse_args()
+
+sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
+
+if os.path.exists(cli_args.socket):
+    try:
+        os.unlink(cli_args.socket)
+    except OSError:
+        logging.warning("Could not unlink socket %s", cli_args.socket)
+
+sock.bind(cli_args.socket)
+
+logging.info("Starting receiver using socket '{}'".format(cli_args.socket))
+
+
+while True:
+    data, addr = sock.recvfrom(256)
+
+    logging.info("RX from {}". format(addr))
+    data = yaml.load(data)
+    print(data)
diff --git a/src/StatsPublish.cpp b/src/StatsPublish.cpp
new file mode 100644
index 0000000..ccf2bb4
--- /dev/null
+++ b/src/StatsPublish.cpp
@@ -0,0 +1,123 @@
+/* ------------------------------------------------------------------
+ * Copyright (C) 2019 Matthias P. Braendli
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied.
+ * See the License for the specific language governing permissions
+ * and limitations under the License.
+ * -------------------------------------------------------------------
+ */
+
+#include "config.h"
+#include "StatsPublish.h"
+#include <stdexcept>
+#include <sstream>
+#include <cstring>
+#include <cerrno>
+#include <cassert>
+#include <sys/socket.h>
+#include <sys/un.h>
+#include <unistd.h>
+
+using namespace std;
+
+StatsPublisher::StatsPublisher(const string& socket_path) :
+    m_socket_path(socket_path)
+{
+    // The client socket binds to a socket whose name depends on PID, and connects to
+    // `socket_path`
+
+    m_sock = socket(AF_UNIX, SOCK_DGRAM | SOCK_NONBLOCK, 0);
+    if (m_sock == -1) {
+        throw runtime_error("Stats socket creation failed: " + string(strerror(errno)));
+    }
+
+    struct sockaddr_un claddr;
+    memset(&claddr, 0, sizeof(struct sockaddr_un));
+    claddr.sun_family = AF_UNIX;
+    snprintf(claddr.sun_path, sizeof(claddr.sun_path), "/tmp/odr-audioenc.%ld", (long) getpid());
+
+    int ret = bind(m_sock, (const struct sockaddr *) &claddr, sizeof(struct sockaddr_un));
+    if (ret == -1) {
+        throw runtime_error("Stats socket bind failed " + string(strerror(errno)));
+    }
+}
+
+void StatsPublisher::update_audio_levels(int16_t audiolevel_left, int16_t audiolevel_right)
+{
+    m_audio_left = audiolevel_left;
+    m_audio_right = audiolevel_right;
+}
+
+void StatsPublisher::notify_underrun()
+{
+    m_num_underruns++;
+}
+
+void StatsPublisher::notify_overrun()
+{
+    m_num_overruns++;
+}
+
+void StatsPublisher::send_stats()
+{
+    // Manually build YAML, as it's quite easy.
+    stringstream yaml;
+    yaml << "---\n";
+    yaml << "program: " << PACKAGE_NAME << "\n";
+    yaml << "version: " <<
+#if defined(GITVERSION)
+            GITVERSION
+#else
+            PACKAGE_VERSION
+#endif
+            << "\n";
+    yaml << "audiolevels: { left: " << m_audio_left << ", right: " << m_audio_right << "}\n";
+    yaml << "driftcompensation: { underruns: " << m_num_underruns << ", overruns: " << m_num_overruns << "}\n";
+
+    const auto yamlstr = yaml.str();
+
+    struct sockaddr_un claddr;
+    memset(&claddr, 0, sizeof(struct sockaddr_un));
+    claddr.sun_family = AF_UNIX;
+    snprintf(claddr.sun_path, sizeof(claddr.sun_path), "%s", m_socket_path.c_str());
+
+    int ret = sendto(m_sock, yamlstr.data(), yamlstr.size(), 0,
+            (struct sockaddr *) &claddr, sizeof(struct sockaddr_un));
+    if (ret == -1) {
+        // This suppresses the -Wlogical-op warning
+        if (errno == EAGAIN
+#if EAGAIN != EWOULDBLOCK
+                or errno == EWOULDBLOCK
+#endif
+                or errno == ECONNREFUSED
+                or errno == ENOENT) {
+            if (m_destination_available) {
+                fprintf(stderr, "Stats destination not available at %s\n", m_socket_path.c_str());
+                m_destination_available = false;
+            }
+        }
+        else {
+            fprintf(stderr, "Statistics send failed: %s\n", strerror(errno));
+        }
+    }
+    else if (ret != yamlstr.size()) {
+        fprintf(stderr, "Statistics send incorrect length: %d bytes of %zu transmitted\n",
+                ret, yamlstr.size());
+    }
+    else if (not m_destination_available) {
+        fprintf(stderr, "Stats destination is now available at %s\n", m_socket_path.c_str());
+        m_destination_available = true;
+    }
+
+    m_audio_left = 0;
+    m_audio_right = 0;
+}
diff --git a/src/StatsPublish.h b/src/StatsPublish.h
new file mode 100644
index 0000000..7702f66
--- /dev/null
+++ b/src/StatsPublish.h
@@ -0,0 +1,66 @@
+/* ------------------------------------------------------------------
+ * Copyright (C) 2019 Matthias P. Braendli
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied.
+ * See the License for the specific language governing permissions
+ * and limitations under the License.
+ * -------------------------------------------------------------------
+ */
+
+#pragma once
+#include <string>
+#include <cstdint>
+#include <cstddef>
+#include <cstdio>
+#include "common.h"
+
+/*! \file StatsPublish.h
+ *
+ * Collects and sends some stats to a UNIX DGRAM socket so that an external tool
+ * like ODR-EncoderManager can display it.
+ *
+ * Currently, only audio levels are collected.
+ *
+ * Output is formatted in YAML
+ */
+class StatsPublisher {
+    public:
+        StatsPublisher(const std::string& socket_path);
+
+        /*! Update peak audio level information */
+        void update_audio_levels(int16_t audiolevel_left, int16_t audiolevel_right);
+
+        /*! Increments the underrun counter */
+        void notify_underrun();
+
+        /*! Increments the overrun counter */
+        void notify_overrun();
+
+        /*! Send the collected stats to the socket, doesn't block. If the socket is
+         * not connected, the data is lost.
+         *
+         * Clears the collected data.  */
+        void send_stats();
+
+    private:
+        std::string m_socket_path;
+        int m_sock = -1;
+
+        int16_t m_audio_left = 0;
+        int16_t m_audio_right = 0;
+
+        size_t m_num_underruns = 0;
+        size_t m_num_overruns = 0;
+
+        bool m_destination_available = true;
+};
+
diff --git a/src/odr-audioenc.cpp b/src/odr-audioenc.cpp
index 6347c90..09dceb5 100644
--- a/src/odr-audioenc.cpp
+++ b/src/odr-audioenc.cpp
@@ -55,6 +55,7 @@
 #include "VLCInput.h"
 #include "SampleQueue.h"
 #include "AACDecoder.h"
+#include "StatsPublish.h"
 #include "Outputs.h"
 #include "common.h"
 #include "wavfile.h"
@@ -192,6 +193,8 @@ void usage(const char* name)
     "     -P, --pad-fifo=FILENAME              Set PAD data input fifo name"
     "                                          (default: /tmp/pad.fifo).\n"
     "     -l, --level                          Show peak audio level indication.\n"
+    "     -S, --stats=SOCKET_NAME              Connect to the specified UNIX Datagram socket and send statistics.\n"
+    "                                          This allows external tools to collect audio and drift compensation stats.\n"
     "     -s, --silence=TIMEOUT                Abort encoding after TIMEOUT seconds of silence.\n"
     "\n"
     "Only the tcp:// zeromq transport has been tested until now,\n"
@@ -460,6 +463,9 @@ public:
     /* Whether to show the 'sox'-like measurement */
     int show_level = 0;
 
+    /* If not empty, send stats over UNIX DGRAM socket */
+    string send_stats_to = "";
+
     /* Data for ZMQ CURVE authentication */
     char* keyfile = nullptr;
     char secretkey[CURVE_KEYLEN+1];
@@ -468,6 +474,7 @@ public:
 
     HANDLE_AACENCODER encoder;
     unique_ptr<AACDecoder> decoder;
+    unique_ptr<StatsPublisher> stats_publisher;
 
     AudioEnc() : queue(BYTES_PER_SAMPLE, channels, 0, drift_compensation) { }
     AudioEnc(const AudioEnc&) = delete;
@@ -696,6 +703,21 @@ int AudioEnc::run()
         }
     }
 
+    if (not send_stats_to.empty()) {
+        StatsPublisher *s = nullptr;
+        try {
+            s = new StatsPublisher(send_stats_to);
+            stats_publisher.reset(s);
+        }
+        catch (const runtime_error& e) {
+            fprintf(stderr, "Failed to initialise Stats Publisher: %s", e.what());
+            if (s != nullptr) {
+                delete s;
+            }
+            return 1;
+        }
+    }
+
     /* We assume that we need to call the encoder
      * enc_calls_per_output before it gives us one encoded audio
      * frame. This information is used when the alsa drift compensation
@@ -867,10 +889,16 @@ int AudioEnc::run()
 
             if (bytes_from_queue != input_buf.size()) {
                 status |= STATUS_UNDERRUN;
+                if (stats_publisher) {
+                    stats_publisher->notify_underrun();
+                }
             }
 
             if (overruns) {
                 status |= STATUS_OVERRUN;
+                if (stats_publisher) {
+                    stats_publisher->notify_overrun();
+                }
             }
         }
         else {
@@ -943,6 +971,10 @@ int AudioEnc::run()
             peak_right = MAX(peak_right, r);
         }
 
+        if (stats_publisher) {
+            stats_publisher->update_audio_levels(peak_left, peak_right);
+        }
+
         /*! \section SilenceDetection
          * Silence detection looks at the audio level and is
          * only useful if the connection dropped, or if no data is available. It is not
@@ -1155,7 +1187,10 @@ int AudioEnc::run()
                 if (status & STATUS_UNDERRUN) {
                     fprintf(stderr, "U");
                 }
+            }
 
+            if (stats_publisher) {
+                stats_publisher->send_stats();
             }
 
             peak_right = 0;
@@ -1282,6 +1317,7 @@ int main(int argc, char *argv[])
         {"rate",                   required_argument,  0, 'r'},
         {"secret-key",             required_argument,  0, 'k'},
         {"silence",                required_argument,  0, 's'},
+        {"stats",                  required_argument,  0, 'S'},
         {"vlc-cache",              required_argument,  0, 'C'},
         {"vlc-gain",               required_argument,  0, 'g'},
         {"vlc-uri",                required_argument,  0, 'v'},
@@ -1323,7 +1359,7 @@ int main(int argc, char *argv[])
     int ch=0;
     int index;
     while(ch != -1) {
-        ch = getopt_long(argc, argv, "aAhDlRVb:B:c:e:f:i:j:k:L:o:r:d:p:P:s:v:w:Wg:C:", longopts, &index);
+        ch = getopt_long(argc, argv, "aAhDlRVb:B:c:e:f:i:j:k:L:o:r:d:p:P:s:S:v:w:Wg:C:", longopts, &index);
         switch (ch) {
         case 0: // AAC-LC
             audio_enc.aot = AOT_DABPLUS_AAC_LC;
@@ -1432,6 +1468,9 @@ int main(int argc, char *argv[])
             }
 
             break;
+        case 'S':
+            audio_enc.send_stats_to = optarg;
+            break;
 #ifdef HAVE_VLC
         case 'v':
             audio_enc.vlc_uri = optarg;