Implement TCP input auto-reconnect

5 files changed, 346 insertions, 196 deletions

diff --git a/src/DabMod.cpp b/src/DabMod.cpp
index 90751f6..80dc6b4 100644
--- a/src/DabMod.cpp
+++ b/src/DabMod.cpp
@@ -528,8 +528,12 @@ static run_modulator_state_t run_modulator(modulator_data& m)
                      */
                 }
 #endif // defined(HAVE_ZEROMQ)
-                // No need to handle the TCP input in a special way to get SIGINT working,
-                // because recv() will return with EINTR.
+                else if (dynamic_pointer_cast<InputTcpReader>(m.inputReader)) {
+                    /* Same as for ZeroMQ */
+                }
+                else {
+                    throw logic_error("Unhandled framesize==0!");
+                }
             }
             else {
                 etiLog.level(error) << "Input read error.";
diff --git a/src/InputReader.h b/src/InputReader.h
index 5b9e80a..84f6835 100644
--- a/src/InputReader.h
+++ b/src/InputReader.h
@@ -40,6 +40,7 @@
 #  include "ThreadsafeQueue.h"
 #endif
 #include "Log.h"
+#include "Socket.h"
 #include <unistd.h>
 #define INVALID_SOCKET   -1
 
@@ -129,11 +130,6 @@ class InputFileReader : public InputReader
 class InputTcpReader : public InputReader
 {
     public:
-        InputTcpReader();
-        InputTcpReader(const InputTcpReader& other) = delete;
-        InputTcpReader& operator=(const InputTcpReader& other) = delete;
-        virtual ~InputTcpReader();
-
         // Endpoint is either host:port or tcp://host:port
         void Open(const std::string& endpoint);
 
@@ -146,7 +142,7 @@ class InputTcpReader : public InputReader
         virtual void PrintInfo() const;
 
     private:
-        int m_sock = INVALID_SOCKET;
+        TCPClient m_tcpclient;
         std::string m_uri;
 };
 
diff --git a/src/InputTcpReader.cpp b/src/InputTcpReader.cpp
index 9a93ad1..e48e258 100644
--- a/src/InputTcpReader.cpp
+++ b/src/InputTcpReader.cpp
@@ -3,7 +3,7 @@
    Her Majesty the Queen in Right of Canada (Communications Research
    Center Canada)
 
-   Copyright (C) 2016
+   Copyright (C) 2018
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://www.opendigitalradio.org
@@ -34,24 +34,6 @@
 #include "Utils.h"
 #include <unistd.h>
 #include <errno.h>
-#include <sys/socket.h>
-#include <netinet/in.h>
-#include <netdb.h>
-#include <arpa/inet.h>
-
-InputTcpReader::InputTcpReader()
-{
-    if ((m_sock = socket(PF_INET, SOCK_STREAM, 0)) == INVALID_SOCKET) {
-        throw std::runtime_error("Can't create TCP socket");
-    }
-}
-
-InputTcpReader::~InputTcpReader()
-{
-    if (m_sock != INVALID_SOCKET) {
-        close(m_sock);
-    }
-}
 
 void InputTcpReader::Open(const std::string& endpoint)
 {
@@ -79,35 +61,7 @@ void InputTcpReader::Open(const std::string& endpoint)
 
     hostname = hostname.substr(0, colon_pos);
 
-    struct sockaddr_in addr;
-    addr.sin_family = PF_INET;
-    addr.sin_addr.s_addr = htons(INADDR_ANY);
-    addr.sin_port = htons(port);
-
-    hostent *host = gethostbyname(hostname.c_str());
-    if (host) {
-        addr.sin_addr = *(in_addr *)(host->h_addr);
-    }
-    else {
-        std::stringstream ss;
-        ss << "Could not resolve hostname " << hostname << ": " << strerror(errno);
-        throw std::runtime_error(ss.str());
-    }
-
-    /* Set recv timeout of 30s, see socket(7) for details */
-    struct timeval tv = {0};
-    tv.tv_sec = 30;
-    if (setsockopt(m_sock, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(struct timeval)) == -1) {
-        std::stringstream ss;
-        ss << "Could not set socket recv timeout: " << strerror(errno);
-        throw std::runtime_error(ss.str());
-    }
-
-    if (connect(m_sock, (struct sockaddr*)&addr, sizeof(addr)) == -1) {
-        std::stringstream ss;
-        ss << "Could not connect to " << hostname << ":" << port << " :" << strerror(errno);
-        throw std::runtime_error(ss.str());
-    }
+    m_tcpclient.connect(hostname, port);
 
     m_uri = endpoint;
 }
@@ -117,16 +71,16 @@ int InputTcpReader::GetNextFrame(void* buffer)
     uint8_t* buf = (uint8_t*)buffer;
 
     const size_t framesize = 6144;
+    const int timeout_ms = 8000;
 
-    ssize_t r = recv(m_sock, buf, framesize, MSG_WAITALL);
+    ssize_t ret = m_tcpclient.recv(buf, framesize, MSG_WAITALL, timeout_ms);
 
-    if (r == -1) {
-        std::stringstream ss;
-        ss << "Could not receive from socket :" << strerror(errno);
-        throw std::runtime_error(ss.str());
+    if (ret == 0) {
+        etiLog.level(debug) << "TCP input auto reconnect";
+        std::this_thread::sleep_for(std::chrono::seconds(1));
     }
 
-    return r;
+    return ret;
 }
 
 void InputTcpReader::PrintInfo() const
diff --git a/src/Socket.cpp b/src/Socket.cpp
new file mode 100644
index 0000000..08cda68
--- /dev/null
+++ b/src/Socket.cpp
@@ -0,0 +1,275 @@
+/*
+   Copyright (C) 2005, 2006, 2007, 2008, 2009, 2010 Her Majesty the
+   Queen in Right of Canada (Communications Research Center Canada)
+
+   Copyright (C) 2018
+   Matthias P. Braendli, matthias.braendli@mpb.li
+
+    http://opendigitalradio.org
+*/
+
+/*
+   This file is part of ODR-DabMod.
+
+   ODR-DabMod is free software: you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as
+   published by the Free Software Foundation, either version 3 of the
+   License, or (at your option) any later version.
+
+   ODR-DabMod is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with ODR-DabMod.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "Socket.h"
+#include "Log.h"
+#include <fcntl.h>
+
+TCPSocket::TCPSocket()
+{
+    if ((m_sock = socket(PF_INET, SOCK_STREAM, 0)) < 0) {
+        throw std::runtime_error("Can't create TCP socket");
+    }
+
+#if defined(HAVE_SO_NOSIGPIPE)
+    int val = 1;
+    if (setsockopt(m_sock, SOL_SOCKET, SO_NOSIGPIPE,
+                &val, sizeof(val)) < 0) {
+        throw std::runtime_error("Can't set SO_NOSIGPIPE");
+    }
+#endif
+}
+
+TCPSocket::~TCPSocket()
+{
+    if (m_sock != -1) {
+        ::close(m_sock);
+    }
+}
+
+TCPSocket::TCPSocket(TCPSocket&& other)
+{
+    m_sock = other.m_sock;
+
+    if (other.m_sock != -1) {
+        other.m_sock = -1;
+    }
+}
+
+TCPSocket& TCPSocket::operator=(TCPSocket&& other)
+{
+    m_sock = other.m_sock;
+
+    if (other.m_sock != -1) {
+        other.m_sock = -1;
+    }
+
+    return *this;
+}
+
+bool TCPSocket::valid() const
+{
+    return m_sock != -1;
+}
+
+void TCPSocket::connect(const std::string& hostname, int port)
+{
+    struct sockaddr_in addr;
+    addr.sin_family = PF_INET;
+    addr.sin_addr.s_addr = htons(INADDR_ANY);
+    addr.sin_port = htons(port);
+
+    hostent *host = gethostbyname(hostname.c_str());
+    if (host) {
+        addr.sin_addr = *(in_addr *)(host->h_addr);
+    }
+    else {
+        std::string errstr(strerror(errno));
+        throw std::runtime_error(
+                "could not resolve hostname " +
+                hostname + ":" + std::to_string(port) +
+                " : " + errstr);
+    }
+
+    int ret = ::connect(m_sock, (struct sockaddr*)&addr, sizeof(addr));
+    if (ret == -1 and errno != EINPROGRESS) {
+        std::string errstr(strerror(errno));
+        throw std::runtime_error(
+                "could not connect to " +
+                hostname + ":" + std::to_string(port) +
+                " : " + errstr);
+    }
+}
+
+void TCPSocket::listen(int port)
+{
+    struct sockaddr_in addr;
+    addr.sin_family = AF_INET;
+    addr.sin_port = htons(port);
+    addr.sin_addr.s_addr = htonl(INADDR_ANY);
+
+    const int reuse = 1;
+    if (setsockopt(m_sock, SOL_SOCKET, SO_REUSEADDR,
+                &reuse, sizeof(reuse)) < 0) {
+        throw std::runtime_error("Can't reuse address for TCP socket");
+    }
+
+    if (::bind(m_sock, (struct sockaddr*)&addr, sizeof(addr)) < 0) {
+        close();
+        throw std::runtime_error("Can't bind TCP socket");
+    }
+
+    if (::listen(m_sock, 1) < 0) {
+        close();
+        m_sock = -1;
+        throw std::runtime_error("Can't listen TCP socket");
+    }
+
+}
+
+void TCPSocket::close()
+{
+    ::close(m_sock);
+    m_sock = -1;
+}
+
+TCPSocket TCPSocket::accept_with_timeout(int timeout_ms, struct sockaddr_in *client)
+{
+    struct pollfd fds[1];
+    fds[0].fd = m_sock;
+    fds[0].events = POLLIN;
+
+    int retval = poll(fds, 1, timeout_ms);
+
+    if (retval == -1) {
+        std::string errstr(strerror(errno));
+        throw std::runtime_error("TCP Socket accept error: " + errstr);
+    }
+    else if (retval > 0) {
+        socklen_t client_len = sizeof(struct sockaddr_in);
+        int sockfd = accept(m_sock, (struct sockaddr*)&client, &client_len);
+        TCPSocket s(sockfd);
+        return s;
+    }
+    else {
+        TCPSocket s(-1);
+        return s;
+    }
+}
+
+ssize_t TCPSocket::sendall(const void *buffer, size_t buflen)
+{
+    uint8_t *buf = (uint8_t*)buffer;
+    while (buflen > 0) {
+        /* On Linux, the MSG_NOSIGNAL flag ensures that the process
+         * would not receive a SIGPIPE and die.
+         * Other systems have SO_NOSIGPIPE set on the socket for the
+         * same effect. */
+#if defined(HAVE_MSG_NOSIGNAL)
+        const int flags = MSG_NOSIGNAL;
+#else
+        const int flags = 0;
+#endif
+        ssize_t sent = ::send(m_sock, buf, buflen, flags);
+        if (sent < 0) {
+            return -1;
+        }
+        else {
+            buf += sent;
+            buflen -= sent;
+        }
+    }
+    return buflen;
+}
+
+ssize_t TCPSocket::recv(void *buffer, size_t length, int flags)
+{
+    ssize_t ret = ::recv(m_sock, buffer, length, flags);
+    if (ret == -1) {
+        std::string errstr(strerror(errno));
+        throw std::runtime_error("TCP receive error: " + errstr);
+    }
+    return ret;
+}
+
+ssize_t TCPSocket::recv(void *buffer, size_t length, int flags, int timeout_ms)
+{
+    struct pollfd fds[1];
+    fds[0].fd = m_sock;
+    fds[0].events = POLLIN;
+
+    int retval = poll(fds, 1, timeout_ms);
+
+    if (retval == -1 and errno == EINTR) {
+        throw Interrupted();
+    }
+    else if (retval == -1) {
+        std::string errstr(strerror(errno));
+        throw std::runtime_error("TCP receive with poll() error: " + errstr);
+    }
+    else if (retval > 0 and (fds[0].revents | POLLIN)) {
+        ssize_t ret = ::recv(m_sock, buffer, length, flags);
+        if (ret == -1) {
+            if (errno == ECONNREFUSED) {
+                return 0;
+            }
+            std::string errstr(strerror(errno));
+            throw std::runtime_error("TCP receive after poll() error: " + errstr);
+        }
+        return ret;
+    }
+    else {
+        throw Timeout();
+    }
+}
+
+TCPSocket::TCPSocket(int sockfd) {
+    m_sock = sockfd;
+}
+
+void TCPClient::connect(const std::string& hostname, int port)
+{
+    m_hostname = hostname;
+    m_port = port;
+    reconnect();
+}
+
+ssize_t TCPClient::recv(void *buffer, size_t length, int flags, int timeout_ms)
+{
+    try {
+        ssize_t ret = m_sock.recv(buffer, length, flags, timeout_ms);
+
+        if (ret == 0) {
+            m_sock.close();
+
+            TCPSocket newsock;
+            m_sock = std::move(newsock);
+            reconnect();
+        }
+
+        return ret;
+    }
+    catch (const TCPSocket::Interrupted&) {
+        return -1;
+    }
+    catch (const TCPSocket::Timeout&) {
+        return 0;
+    }
+
+    return 0;
+}
+
+void TCPClient::reconnect()
+{
+    int flags = fcntl(m_sock.m_sock, F_GETFL);
+    if (fcntl(m_sock.m_sock, F_SETFL, flags | O_NONBLOCK) == -1) {
+        std::string errstr(strerror(errno));
+        throw std::runtime_error("TCP: Could not set O_NONBLOCK: " + errstr);
+    }
+
+    m_sock.connect(m_hostname, m_port);
+}
diff --git a/src/Socket.h b/src/Socket.h
index 392e758..14c5cbe 100644
--- a/src/Socket.h
+++ b/src/Socket.h
@@ -2,7 +2,7 @@
    Copyright (C) 2005, 2006, 2007, 2008, 2009, 2010 Her Majesty the
    Queen in Right of Canada (Communications Research Center Canada)
 
-   Copyright (C) 2017
+   Copyright (C) 2018
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://opendigitalradio.org
@@ -34,150 +34,71 @@ DESCRIPTION:
 #   include <config.h>
 #endif
 
-#include <unistd.h>
-#include <cstdint>
 #include <stdexcept>
 #include <string>
-#include <sys/socket.h>
-#include <netinet/ip.h>
+#include <cstdint>
+#include <cstring>
+#include <unistd.h>
 #include <errno.h>
 #include <poll.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <netdb.h>
+#include <arpa/inet.h>
 
 class TCPSocket {
     public:
-        TCPSocket() {
-            if ((m_sock = socket(PF_INET, SOCK_STREAM, 0)) < 0) {
-                throw std::runtime_error("Can't create TCP socket");
-            }
-
-#if defined(HAVE_SO_NOSIGPIPE)
-            int val = 1;
-            if (setsockopt(m_sock, SOL_SOCKET, SO_NOSIGPIPE,
-                        &val, sizeof(val)) < 0) {
-                throw std::runtime_error("Can't set SO_NOSIGPIPE");
-            }
-#endif
-        }
-
-        ~TCPSocket() {
-            if (m_sock != -1) {
-                ::close(m_sock);
-            }
-        }
-
+        TCPSocket();
+        ~TCPSocket();
         TCPSocket(const TCPSocket& other) = delete;
         TCPSocket& operator=(const TCPSocket& other) = delete;
-        TCPSocket(TCPSocket&& other) {
-            m_sock = other.m_sock;
-
-            if (other.m_sock != -1) {
-                other.m_sock = -1;
-            }
-        }
-
-        TCPSocket& operator=(TCPSocket&& other)
-        {
-            m_sock = other.m_sock;
-
-            if (other.m_sock != -1) {
-                other.m_sock = -1;
-            }
-
-            return *this;
-        }
-
-        bool valid(void) const {
-            return m_sock != -1;
-        }
-
-        void listen(int port) {
-            struct sockaddr_in addr;
-            addr.sin_family = AF_INET;
-            addr.sin_port = htons(port);
-            addr.sin_addr.s_addr = htonl(INADDR_ANY);
-
-            const int reuse = 1;
-            if (setsockopt(m_sock, SOL_SOCKET, SO_REUSEADDR,
-                        &reuse, sizeof(reuse)) < 0) {
-                throw std::runtime_error("Can't reuse address for TCP socket");
-            }
-
-            if (::bind(m_sock, (struct sockaddr*)&addr, sizeof(addr)) < 0) {
-                close();
-                throw std::runtime_error("Can't bind TCP socket");
-            }
-
-            if (::listen(m_sock, 1) < 0) {
-                close();
-                m_sock = -1;
-                throw std::runtime_error("Can't listen TCP socket");
-            }
-
-        }
-
-        void close(void) {
-            ::close(m_sock);
-            m_sock = -1;
-        }
-
-        TCPSocket accept_with_timeout(int timeout_ms, struct sockaddr_in *client)
-        {
-            struct pollfd fds[1];
-            fds[0].fd = m_sock;
-            fds[0].events = POLLIN | POLLOUT;
-
-            int retval = poll(fds, 1, timeout_ms);
-
-            if (retval == -1) {
-                throw std::runtime_error("TCP Socket accept error: " + std::to_string(errno));
-            }
-            else if (retval) {
-                socklen_t client_len = sizeof(struct sockaddr_in);
-                int sockfd = accept(m_sock, (struct sockaddr*)&client, &client_len);
-                TCPSocket s(sockfd);
-                return s;
-            }
-            else {
-                TCPSocket s(-1);
-                return s;
-            }
-        }
-
-        ssize_t sendall(const void *buffer, size_t buflen)
-        {
-            uint8_t *buf = (uint8_t*)buffer;
-            while (buflen > 0) {
-                /* On Linux, the MSG_NOSIGNAL flag ensures that the process
-                 * would not receive a SIGPIPE and die.
-                 * Other systems have SO_NOSIGPIPE set on the socket for the
-                 * same effect. */
-#if defined(HAVE_MSG_NOSIGNAL)
-                const int flags = MSG_NOSIGNAL;
-#else
-                const int flags = 0;
-#endif
-                ssize_t sent = ::send(m_sock, buf, buflen, flags);
-                if (sent < 0) {
-                    return -1;
-                }
-                else {
-                    buf += sent;
-                    buflen -= sent;
-                }
-            }
-            return buflen;
-        }
-
-        ssize_t recv(void *buffer, size_t length, int flags)
-        {
-            return ::recv(m_sock, buffer, length, flags);
-        }
+        TCPSocket(TCPSocket&& other);
+        TCPSocket& operator=(TCPSocket&& other);
 
-    private:
-        explicit TCPSocket(int sockfd) {
-            m_sock = sockfd;
-        }
+        bool valid(void) const;
+        void connect(const std::string& hostname, int port);
+        void listen(int port);
+        void close(void);
+
+        /* throws a runtime_error on failure, an invalid socket on timeout */
+        TCPSocket accept_with_timeout(int timeout_ms, struct sockaddr_in *client);
 
+        /* returns -1 on error */
+        ssize_t sendall(const void *buffer, size_t buflen);
+
+        /* Returns number of bytes read, 0 on disconnect. Throws a
+         * runtime_error on error */
+        ssize_t recv(void *buffer, size_t length, int flags);
+
+        class Timeout {};
+        class Interrupted {};
+        /* Returns number of bytes read, 0 on disconnect or refused connection.
+         * Throws a Timeout on timeout, Interrupted on EINTR, a runtime_error
+         * on error
+         */
+        ssize_t recv(void *buffer, size_t length, int flags, int timeout_ms);
+
+    private:
+        explicit TCPSocket(int sockfd);
         int m_sock = -1;
+
+        friend class TCPClient;
+};
+
+/* Implement a TCP receiver that auto-reconnects on errors */
+class TCPClient {
+    public:
+        void connect(const std::string& hostname, int port);
+
+        /* Returns numer of bytes read, 0 on auto-reconnect, -1
+         * on interruption.
+         * Throws a runtime_error on error */
+        ssize_t recv(void *buffer, size_t length, int flags, int timeout_ms);
+
+    private:
+        void reconnect(void);
+        TCPSocket m_sock;
+        std::string m_hostname;
+        int m_port;
 };