Replace PAD fifo by UNIX socket

4 files changed, 209 insertions, 43 deletions

diff --git a/Makefile.am b/Makefile.am
index 872543e..e54ed95 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -102,6 +102,8 @@ odr_audioenc_CXXFLAGS    = $(GST_CFLAGS) $(GITVERSION_FLAGS) \
 						   -Wall -ggdb -O2 -Isrc -Icontrib
 
 odr_audioenc_SOURCES     = src/odr-audioenc.cpp \
+						   src/PadInterface.cpp \
+						   src/PadInterface.h \
 						   src/FileInput.cpp \
 						   src/FileInput.h \
 						   src/AlsaInput.cpp \
diff --git a/src/PadInterface.cpp b/src/PadInterface.cpp
new file mode 100644
index 0000000..7e8a6ee
--- /dev/null
+++ b/src/PadInterface.cpp
@@ -0,0 +1,138 @@
+/* ------------------------------------------------------------------
+ * Copyright (C) 2020 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 "PadInterface.h"
+#include <stdexcept>
+#include <sstream>
+#include <cstring>
+#include <cerrno>
+#include <cassert>
+#include <sys/socket.h>
+#include <sys/un.h>
+#include <unistd.h>
+
+#define MESSAGE_REQUEST 1
+#define MESSAGE_PAD_DATA 2
+
+using namespace std;
+
+void PadInterface::open(const std::string &pad_ident)
+{
+    m_pad_ident = pad_ident;
+
+    m_sock = socket(AF_UNIX, SOCK_DGRAM | SOCK_NONBLOCK, 0);
+    if (m_sock == -1) {
+        throw runtime_error("PAD 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/%s.audioenc", m_pad_ident.c_str());
+    if (unlink(claddr.sun_path) == -1 and errno != ENOENT) {
+        fprintf(stderr, "Unlinking of socket %s failed: %s\n", claddr.sun_path, strerror(errno));
+    }
+
+    int ret = bind(m_sock, (const struct sockaddr *) &claddr, sizeof(struct sockaddr_un));
+    if (ret == -1) {
+        throw runtime_error("PAD socket bind failed " + string(strerror(errno)));
+    }
+}
+
+vector<uint8_t> PadInterface::request(uint8_t padlen)
+{
+    if (m_pad_ident.empty()) {
+        throw logic_error("Uninitialised PadInterface::request() called");
+    }
+
+    // Sending requests allows the PadEnc to know both the padlen, but also
+    // will allow proper timing.
+
+    uint8_t packet[2];
+
+    packet[0] = MESSAGE_REQUEST; // Message type, to allow future expansion
+    packet[1] = padlen;
+
+    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/%s.padenc", m_pad_ident.c_str());
+
+    ssize_t ret = sendto(m_sock, packet, sizeof(packet), 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_padenc_reachable) {
+                fprintf(stderr, "ODR-PadEnc at %s not reachable\n", claddr.sun_path);
+                m_padenc_reachable = false;
+            }
+        }
+        else {
+            fprintf(stderr, "PAD request send failed: %s\n", strerror(errno));
+        }
+    }
+    else if (ret != sizeof(packet)) {
+        fprintf(stderr, "PAD request incorrect length sent: %zu bytes of %zu transmitted\n",
+                ret, sizeof(packet));
+    }
+    else if (not m_padenc_reachable) {
+        fprintf(stderr, "ODR-PadEnc is now reachable at %s\n", claddr.sun_path);
+        m_padenc_reachable = true;
+    }
+
+    vector<uint8_t> buffer(2048);
+
+    while (true) {
+        ret = recvfrom(m_sock, buffer.data(), buffer.size(), 0, nullptr, nullptr);
+
+        if (ret == -1) {
+            // This suppresses the -Wlogical-op warning
+#if EAGAIN == EWOULDBLOCK
+            if (errno != EAGAIN)
+#else
+            if (not (errno == EAGAIN or errno == EWOULDBLOCK))
+#endif
+            {
+                throw runtime_error(string("Can't receive data: ") + strerror(errno));
+            }
+
+            return {};
+        }
+        else if (ret > 0) {
+            buffer.resize(ret);
+
+            // We could check where the data comes from, but since we're using UNIX sockets
+            // the source is anyway local to the machine.
+
+            if (buffer[0] == MESSAGE_PAD_DATA) {
+                vector<uint8_t> pad_data(buffer.size() - 1);
+                copy(buffer.begin() + 1, buffer.end(), pad_data.begin());
+                return pad_data;
+            }
+            else {
+                continue;
+            }
+        }
+    }
+}
diff --git a/src/PadInterface.h b/src/PadInterface.h
new file mode 100644
index 0000000..9787d06
--- /dev/null
+++ b/src/PadInterface.h
@@ -0,0 +1,44 @@
+/* ------------------------------------------------------------------
+ * Copyright (C) 2020 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 <vector>
+#include <cstdint>
+#include <cstddef>
+#include <cstdio>
+
+/*! \file PadInterface.h
+ *
+ * Handles communication with ODR-PadEnc using a socket
+ */
+
+class PadInterface {
+    public:
+        /*! Create a new PAD data interface that binds to /tmp/pad_ident.audioenc and
+         * communicates with ODR-PadEnc at /tmp/pad_ident.padenc
+         */
+        void open(const std::string &pad_ident);
+
+        std::vector<uint8_t> request(uint8_t padlen);
+
+    private:
+        std::string m_pad_ident;
+        int m_sock = -1;
+        bool m_padenc_reachable = true;
+};
diff --git a/src/odr-audioenc.cpp b/src/odr-audioenc.cpp
index c6887da..5b0facf 100644
--- a/src/odr-audioenc.cpp
+++ b/src/odr-audioenc.cpp
@@ -50,6 +50,7 @@
  */
 
 #include "config.h"
+#include "PadInterface.h"
 #include "AlsaInput.h"
 #include "FileInput.h"
 #include "JackInput.h"
@@ -121,7 +122,7 @@ static void usage(const char* name)
     "artifacts.\n"
     "\n"
     "This encoder is able to insert PAD (DLS and MOT Slideshow)\n"
-    "generated by ODR-PadEnc.\n"
+    "generated by ODR-PadEnc, and communicates using a UNIX socket.\n"
     "\nUsage:\n"
     "%s [INPUT SELECTION] [OPTION...]\n",
 #if defined(GITVERSION)
@@ -200,16 +201,13 @@ static void usage(const char* name)
     "                                          add a delay (in milliseconds) to the timestamps carried in EDI\n"
     "     -k, --secret-key=FILE                Enable ZMQ encryption with the given secret key.\n"
     "     -p, --pad=BYTES                      Enable PAD insertion and set PAD size in bytes.\n"
-    "     -P, --pad-fifo=FILENAME              Set PAD data input fifo name"
-    "                                          (default: /tmp/pad.fifo).\n"
+    "     -P, --pad-socket=IDENTIFIER          Use the given identifier to communicate with ODR-PadEnc."
     "     -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"
     "         --version                        Show version and quit.\n"
     "\n"
-    "Only the tcp:// zeromq transport has been tested until now,\n"
-    " but epgm://, pgm:// and ipc:// are also accepted\n"
     );
 
 }
@@ -469,8 +467,8 @@ public:
     int measured_silence_ms = 0;
 
     /* For MOT Slideshow and DLS insertion */
-    const char* pad_fifo = "/tmp/pad.fifo";
-    int pad_fd = -1;
+    string pad_ident = "";
+    PadInterface pad_intf;
     int padlen = 0;
 
     /* Encoder status, see the above STATUS macros */
@@ -557,7 +555,7 @@ int AudioEnc::run()
         }
     }
 
-    if (padlen < 0) {
+    if (padlen < 0 or padlen > 255) {
         fprintf(stderr, "Invalid PAD length specified\n");
         return 1;
     }
@@ -637,27 +635,11 @@ int AudioEnc::run()
         edi_output.set_odr_version_tag(ss.str());
     }
 
-    if (padlen != 0) {
-        int flags;
-        if (mkfifo(pad_fifo, S_IWUSR | S_IRUSR | S_IRGRP | S_IROTH) != 0) {
-            if (errno != EEXIST) {
-                fprintf(stderr, "Can't create pad file: %d!\n", errno);
-                return 1;
-            }
-        }
-        pad_fd = open(pad_fifo, O_RDONLY | O_NONBLOCK);
-        if (pad_fd == -1) {
-            fprintf(stderr, "Can't open pad file!\n");
-            return 1;
-        }
-        flags = fcntl(pad_fd, F_GETFL, 0);
-        if (fcntl(pad_fd, F_SETFL, flags | O_NONBLOCK)) {
-            fprintf(stderr, "Can't set non-blocking mode in pad file!\n");
-            return 1;
-        }
+    if (padlen != 0 and not pad_ident.empty()) {
+        pad_intf.open(pad_ident);
+        fprintf(stderr, "PAD socket opened\n");
     }
 
-
     vec_u8 input_buf;
 
     if (selected_encoder == encoder_selection_t::fdk_dabplus) {
@@ -806,7 +788,7 @@ int AudioEnc::run()
             break;
     }
 
-    unsigned char pad_buf[padlen + 1];
+    vector<uint8_t> pad_buf(padlen + 1);
 
     if (restart_on_fault) {
         fprintf(stderr, "Autorestart has been deprecated and will be removed in the future!\n");
@@ -822,18 +804,17 @@ int AudioEnc::run()
     int calls = 0; // for checking
     ssize_t read_bytes = 0;
     do {
-        // --------------- Read data from the PAD fifo
+        // --------------- Read data from the PAD socket
         int calculated_padlen = 0;
 
         if (padlen != 0) {
-            ssize_t pad_ret = read(pad_fd, pad_buf, padlen + 1);
+            vector<uint8_t> pad_data = pad_intf.request(padlen);
 
-            if ((pad_ret < 0 && errno == EAGAIN) || pad_ret == 0) {
-                // If this condition passes, there is no data to be read
+            if (pad_data.empty()) {
+                /* no PAD available */
             }
-            else if (pad_ret == padlen + 1) {
-                // Otherwise, you're good to go and buffer should contain "count" bytes.
-                calculated_padlen = pad_buf[padlen];
+            else if (pad_data.size() == pad_buf.size()) {
+                calculated_padlen = pad_data[padlen];
 
                 if (calculated_padlen < 2) {
                     throw runtime_error("Invalid X-PAD length " + to_string(calculated_padlen));
@@ -844,14 +825,15 @@ int AudioEnc::run()
                  */
                 if (    selected_encoder == encoder_selection_t::fdk_dabplus &&
                         calculated_padlen == 2 &&
-                        pad_buf[padlen - 2] == 0x00 &&
-                        pad_buf[padlen - 1] == 0x00 ) {
+                        pad_data[padlen - 2] == 0x00 &&
+                        pad_data[padlen - 1] == 0x00 ) {
                     calculated_padlen = 0;
                 }
+
+                copy(pad_data.begin(), pad_data.end(), pad_buf.begin());
             }
             else {
-                // Some other error occurred during read.
-                fprintf(stderr, "Unable to read from PAD!\n");
+                fprintf(stderr, "Incorrect PAD length received: %zu expected %d\n", pad_data.size(), padlen + 1);
                 break;
             }
         }
@@ -1072,7 +1054,7 @@ int AudioEnc::run()
             int out_size, out_elem_size;
 
             in_ptr[0] = input_buf.data();
-            in_ptr[1] = pad_buf + (padlen - calculated_padlen); // offset due to unused PAD bytes
+            in_ptr[1] = pad_buf.data() + (padlen - calculated_padlen); // offset due to unused PAD bytes
             in_size[0] = read_bytes;
             in_size[1] = calculated_padlen;
             in_elem_size[0] = BYTES_PER_SAMPLE;
@@ -1133,7 +1115,7 @@ int AudioEnc::run()
             }
 
             if (read_bytes) {
-                numOutBytes = toolame_encode_frame(input_buffers, pad_buf, calculated_padlen, outbuf.data(), outbuf.size());
+                numOutBytes = toolame_encode_frame(input_buffers, pad_buf.data(), calculated_padlen, outbuf.data(), outbuf.size());
             }
             else {
                 numOutBytes = toolame_finish(outbuf.data(), outbuf.size());
@@ -1375,7 +1357,7 @@ int main(int argc, char *argv[])
         {"jack",                   required_argument,  0, 'j'},
         {"output",                 required_argument,  0, 'o'},
         {"pad",                    required_argument,  0, 'p'},
-        {"pad-fifo",               required_argument,  0, 'P'},
+        {"pad-socket",             required_argument,  0, 'P'},
         {"rate",                   required_argument,  0, 'r'},
         {"secret-key",             required_argument,  0, 'k'},
         {"silence",                required_argument,  0, 's'},
@@ -1543,7 +1525,7 @@ int main(int argc, char *argv[])
             audio_enc.padlen = std::stoi(optarg);
             break;
         case 'P':
-            audio_enc.pad_fifo = optarg;
+            audio_enc.pad_ident = optarg;
             break;
         case 'r':
             audio_enc.sample_rate = std::stoi(optarg);