Add Stats output through UNIX DGRAM socket

4 files changed, 223 insertions, 0 deletions

diff --git a/Makefile.am b/Makefile.am
index aff4694..1bac77d 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -14,6 +14,7 @@ odr_sourcecompanion_SOURCES     = src/odr-sourcecompanion.cpp \
 								  src/AACDecoder.h src/AACDecoder.cpp \
 								  src/AVTInput.h src/AVTInput.cpp \
 								  src/OrderedQueue.h src/OrderedQueue.cpp \
+								  src/StatsPublish.h src/StatsPublish.cpp \
 								  src/crc.h src/crc.c \
 								  src/encryption.h src/encryption.c \
 								  src/utils.h src/utils.c \
diff --git a/src/StatsPublish.cpp b/src/StatsPublish.cpp
new file mode 100644
index 0000000..0bad833
--- /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 != (ssize_t)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..f593c7c
--- /dev/null
+++ b/src/StatsPublish.h
@@ -0,0 +1,65 @@
+/* ------------------------------------------------------------------
+ * 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>
+
+/*! \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-sourcecompanion.cpp b/src/odr-sourcecompanion.cpp
index df8feac..a69f705 100644
--- a/src/odr-sourcecompanion.cpp
+++ b/src/odr-sourcecompanion.cpp
@@ -28,6 +28,7 @@
 
 #include "AVTInput.h"
 #include "AACDecoder.h"
+#include "StatsPublish.h"
 #include <sys/time.h>
 #include <sys/types.h>
 #include <unistd.h>
@@ -97,6 +98,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"
     "\n"
     "Only the tcp:// zeromq transport has been tested until now,\n"
     " but epgm:// and pgm:// are also accepted\n"
@@ -120,6 +123,7 @@ int main(int argc, char *argv[])
     std::vector<std::string> output_uris;
 
     AACDecoder decoder;
+    unique_ptr<StatsPublisher> stats_publisher;
 
     /* For MOT Slideshow and DLS insertion */
     const char* pad_fifo = "/tmp/pad.fifo";
@@ -129,6 +133,9 @@ int main(int argc, char *argv[])
     /* Whether to show the 'sox'-like measurement */
     bool show_level = false;
 
+    /* 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];
@@ -140,6 +147,7 @@ int main(int argc, char *argv[])
         {"pad",                    required_argument,  0, 'p'},
         {"pad-fifo",               required_argument,  0, 'P'},
         {"rate",                   required_argument,  0, 'r'},
+        {"stats",                  required_argument,  0, 'S'},
         {"secret-key",             required_argument,  0, 'k'},
         {"input-uri",              required_argument,  0, 'I'},
         {"control-uri",            required_argument,  0,  6 },
@@ -219,6 +227,9 @@ int main(int argc, char *argv[])
         case 'r':
             sample_rate = atoi(optarg);
             break;
+        case 'S':
+            send_stats_to = optarg;
+            break;
         case 'I':
             avt_input_uri = optarg;
             fprintf(stderr, "AVT Encoder Mode\n");
@@ -323,6 +334,21 @@ int main(int argc, char *argv[])
         return 1;
     }
 
+    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;
+        }
+    }
+
     int outbuf_size;
     std::vector<uint8_t> zmqframebuf;
     std::vector<uint8_t> outbuf;
@@ -414,6 +440,10 @@ int main(int argc, char *argv[])
                 peak_left = 0;
                 peak_right = 0;
             }
+
+            if (stats_publisher) {
+                stats_publisher->update_audio_levels(peak_left, peak_right);
+            }
         }
 
         read_bytes = numOutBytes;
@@ -461,6 +491,10 @@ int main(int argc, char *argv[])
 
             peak_right = 0;
             peak_left = 0;
+
+            if (stats_publisher) {
+                stats_publisher->send_stats();
+            }
         }
     } while (read_bytes > 0);