Adds configuration options for TCP sockets, tuning for Hub app. (#759)

Adds several tuning parameters for TCP sockets:
- SO_SNDBUF
- SO_RCVBUF
- TCP_NOTSENT_LOWAT

Refactors where the fd socket options are set to a single place.
Applies some reasonable (but not great) values for the Hub application.
Switches the Hub application to use select(). Tunes the parameters to limit buffer size in the hub.
Fixes a crashing bug where a GridconnectBridge was deleted too early when packets owned by its pool were still in flight.

===

* Adds constants for TCP socket options fo SO_SNDBUF and SO_RCVBUF

* Optimizes the memory and buffer use of the hub application.

* gc-tcp-hub: applies SO_SNDBUF and SO_RCVBUF options to sockets.

* client connection util: exposes the file descriptor when available.
Applies the gridconnect use select link option.
Applies the SO_SNDBUF/RCVBUF link option.

* Ensures that the ClientCOnnection.cxx is built and linked.

* Sorts lines in sources.

* Refactors the fd optimization code to FdUtils class.

* Makes tcp_lowat also configurable.

* Fix compile error.

* Adds comment.

* Switch gridconnect hub to LimitedPool instead of FixedPool.
This needs no dedicated memory, but uses the regular mainBufferPool.

Add a check that all outgoing packets are released before deleting the pool.
This solves hub crashing when a client disconnects.

* Tune TCP options.

applications/hub/main.cxx

@@ -56,6 +56,13 @@ CanHubFlow can_hub0(&g_service);
 OVERRIDE_CONST(gc_generate_newlines, 1);
 OVERRIDE_CONST(gridconnect_buffer_size, 1300);
 OVERRIDE_CONST(gridconnect_buffer_delay_usec, 2000);
+OVERRIDE_CONST(gridconnect_bridge_max_incoming_packets, 5);
+OVERRIDE_CONST(gridconnect_bridge_max_outgoing_packets, 5);
+OVERRIDE_CONST(gridconnect_tcp_snd_buffer_size, 8192);
+OVERRIDE_CONST(gridconnect_tcp_rcv_buffer_size, 8192);
+OVERRIDE_CONST(gridconnect_tcp_notsent_lowat_buffer_size, 1024);
+
+OVERRIDE_CONST_TRUE(gridconnect_tcp_use_select);
 
 
 int port = 12021;

include/nmranet_config.h

@@ -104,6 +104,18 @@ DECLARE_CONST(gridconnect_bridge_max_incoming_packets);
 /// output socket cannot send the data fast enough.
 DECLARE_CONST(gridconnect_bridge_max_outgoing_packets);
 
+/// TCP receive buffer size in bytes for gridconnect hubs. Used via
+/// setsockopt(SO_RCVBUF). Set to 1 (default) to not bound it.
+DECLARE_CONST(gridconnect_tcp_rcv_buffer_size);
+
+/// TCP send buffer size in bytes for gridconnect hubs. Used via
+/// setsockopt(SO_SENDBUF). Set to 1 (default) to not bound it.
+DECLARE_CONST(gridconnect_tcp_snd_buffer_size);
+
+/// TCP_NOTSENT_LOWAT kernel parameter (in bytes) for TCP links. Used via
+/// setsockopt. Set to 1 (default) to not bound it.
+DECLARE_CONST(gridconnect_tcp_notsent_lowat_buffer_size);
+
 /** Number of bytes of gridconnect data to buffer before sending off the
  * lowlevel system (such as TCP socket). */
 DECLARE_CONST(gridconnect_buffer_size);

src/utils/ClientConnection.cxx

@@ -0,0 +1,55 @@
+/** \copyright
+ * Copyright (c) 2023, Balazs Racz
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are  permitted provided that the following conditions are met:
+ *
+ *  - Redistributions of source code must retain the above copyright notice,
+ *    this list of conditions and the following disclaimer.
+ *
+ *  - Redistributions in binary form must reproduce the above copyright notice,
+ *    this list of conditions and the following disclaimer in the documentation
+ *    and/or other materials provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ *
+ * \file ClientConnection.cxx
+ *
+ * Utilities for managing can-hub connections as a client application.
+ *
+ * @author Balazs Racz
+ * @date 27 Dec 2023
+ */
+
+#include "utils/ClientConnection.hxx"
+
+#include "netinet/in.h"
+#include "netinet/tcp.h"
+#include "nmranet_config.h"
+
+#include "utils/FdUtils.hxx"
+
+/// Callback from try_connect to donate the file descriptor.
+/// @param fd is the file destriptor of the connection freshly opened.
+void GCFdConnectionClient::connection_complete(int fd)
+{
+    const bool use_select =
+        (config_gridconnect_tcp_use_select() == CONSTANT_TRUE);
+
+    // Applies kernel parameters like socket options.
+    FdUtils::optimize_fd(fd);
+
+    fd_ = fd;
+    create_gc_port_for_can_hub(hub_, fd, &closedNotify_, use_select);
+}

src/utils/ClientConnection.hxx

@@ -36,11 +36,12 @@
 #define _UTILS_CLIENTCONNECTION_HXX_
 
 #include <stdio.h>
-#include <termios.h> /* tc* functions */
 #include <unistd.h>
+#include <fcntl.h>
 
 #include "utils/GridConnectHub.hxx"
 #include "utils/socket_listener.hxx"
+#include "utils/FdUtils.hxx"
 
 /// Abstract base class for the Hub's connections.
 class ConnectionClient
@@ -50,6 +51,10 @@ public:
      * is dead. @returns true if there is a live connection. */
     virtual bool ping() = 0;
 
+    /// @return the file descriptor, if this connection has one, or -1 if the
+    /// connection is down or doesn't have an fd.
+    virtual int fd() { return -1; }
+
     virtual ~ConnectionClient()
     {
     }
@@ -113,18 +118,19 @@ public:
         return fd_ >= 0;
     }
 
+    int fd() override
+    {
+        return fd_;
+    }
+
 protected:
     /// Abstrct base function to attempt to connect (or open device) to the
     /// destination.
     virtual void try_connect() = 0;
 
     /** Callback from try_connect to donate the file descriptor. @param fd is
      * the file destriptor of the connection freshly opened.  */
-    void connection_complete(int fd)
-    {
-        fd_ = fd;
-        create_gc_port_for_can_hub(hub_, fd, &closedNotify_);
-    }
+    void connection_complete(int fd);
 
 private:
     /// Will be called when the descriptor experiences an error (typivcally
@@ -165,18 +171,9 @@ private:
         int fd = ::open(dev_.c_str(), O_RDWR);
         if (fd >= 0)
         {
-            // Sets up the terminal in raw mode. Otherwise linux might echo
-            // characters coming in from the device and that will make
-            // packets go back to where they came from.
-            HASSERT(!tcflush(fd, TCIOFLUSH));
-            struct termios settings;
-            HASSERT(!tcgetattr(fd, &settings));
-            cfmakeraw(&settings);
-            cfsetspeed(&settings, B115200);
-            HASSERT(!tcsetattr(fd, TCSANOW, &settings));
+            FdUtils::optimize_tty_fd(fd);
             LOG(INFO, "Opened device %s.\n", dev_.c_str());
             connection_complete(fd);
-            //
         }
         else
         {

src/utils/FdUtils.cxx

@@ -0,0 +1,132 @@
+/** \copyright
+ * Copyright (c) 2023, Balazs Racz
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are  permitted provided that the following conditions are met:
+ *
+ *  - Redistributions of source code must retain the above copyright notice,
+ *    this list of conditions and the following disclaimer.
+ *
+ *  - Redistributions in binary form must reproduce the above copyright notice,
+ *    this list of conditions and the following disclaimer in the documentation
+ *    and/or other materials provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ *
+ * \file FdUtils.cxx
+ *
+ * Helper functions for dealing with posix fds.
+ *
+ * @author Balazs Racz
+ * @date 29 Dec 2023
+ */
+
+#include "FdUtils.hxx"
+
+#include <netinet/in.h>
+#include <netinet/tcp.h>
+#include <sys/stat.h>
+#include <termios.h> /* tc* functions */
+
+#include "nmranet_config.h"
+
+/// Performs a system call on an fd. If an error is returned, prints the error
+/// using the log mechanism, but otherwise ignores it.
+/// @param where user-readable text printed with the error, e.g. "setsockopt"
+/// @param callfn system call, like ::setsockopt
+/// @param fd file descriptor (int)
+/// @param args... all other arguments to callfn
+#define PCALL_LOGERR(where, callfn, fd, args...)                               \
+    do                                                                         \
+    {                                                                          \
+        int ret = callfn(fd, args);                                            \
+        if (ret < 0)                                                           \
+        {                                                                      \
+            char buf[256];                                                     \
+            strerror_r(errno, buf, sizeof(buf));                               \
+            LOG_ERROR("fd %d %s: %s", fd, where, buf);                         \
+        }                                                                      \
+    } while (0)
+
+/// Optimizes the kernel settings like socket and TCP options for an fd
+/// that is an outgoing TCP socket.
+/// @param fd socket file descriptor.
+void FdUtils::optimize_socket_fd(int fd)
+{
+#ifdef __linux__
+    const int rcvbuf = config_gridconnect_tcp_rcv_buffer_size();
+    if (rcvbuf > 1)
+    {
+        PCALL_LOGERR("setsockopt SO_RCVBUF", ::setsockopt, fd, SOL_SOCKET,
+            SO_RCVBUF, &rcvbuf, sizeof(rcvbuf));
+    }
+    const int sndbuf = config_gridconnect_tcp_snd_buffer_size();
+    if (sndbuf > 1)
+    {
+        PCALL_LOGERR("setsockopt SO_SNDBUF", ::setsockopt, fd, SOL_SOCKET,
+            SO_SNDBUF, &sndbuf, sizeof(sndbuf));
+        int ret = 0;
+        socklen_t retsize = sizeof(ret);
+        ::getsockopt(fd, SOL_SOCKET, SO_SNDBUF, &ret, &retsize);
+        LOG(ALWAYS, "fd %d sndbuf %d", fd, ret);
+    }
+    const int lowat = config_gridconnect_tcp_notsent_lowat_buffer_size();
+    if (lowat > 1)
+    {
+        PCALL_LOGERR("setsockopt tcp_notsent_lowat", ::setsockopt, fd,
+            IPPROTO_TCP, TCP_NOTSENT_LOWAT, &lowat, sizeof(lowat));
+        int ret = 0;
+        socklen_t retsize = sizeof(ret);
+        ::getsockopt(fd, IPPROTO_TCP, TCP_NOTSENT_LOWAT, &ret, &retsize);
+        LOG(ALWAYS, "fd %d lowat %d", fd, ret);
+    }
+#endif
+}
+
+/// Sets the kernel settings like queuing and terminal settings for an fd
+/// that is an outgoing tty.
+/// @param fd tty file descriptor.
+void FdUtils::optimize_tty_fd(int fd)
+{
+#ifdef __linux__
+    // Sets up the terminal in raw mode. Otherwise linux might echo
+    // characters coming in from the device and that will make
+    // packets go back to where they came from.
+    HASSERT(!tcflush(fd, TCIOFLUSH));
+    struct termios settings;
+    HASSERT(!tcgetattr(fd, &settings));
+    cfmakeraw(&settings);
+    cfsetspeed(&settings, B115200);
+    HASSERT(!tcsetattr(fd, TCSANOW, &settings));
+#endif
+}
+
+/// For an fd that is an outgoing link, detects what kind of file
+/// descriptor this is and calls the appropriate optimize call for it.
+void FdUtils::optimize_fd(int fd)
+{
+#ifdef __linux__
+    struct stat statbuf;
+    fstat(fd, &statbuf);
+
+    if (S_ISSOCK(statbuf.st_mode))
+    {
+        optimize_socket_fd(fd);
+    }
+    else if (isatty(fd))
+    {
+        optimize_tty_fd(fd);
+    }
+#endif
+}

src/utils/FdUtils.hxx

@@ -35,6 +35,8 @@
 #ifndef _UTILS_FD_UTILS_HXX_
 #define _UTILS_FD_UTILS_HXX_
 
+#include <unistd.h>
+
 #include "utils/logging.h"
 #include "utils/macros.h"
 
@@ -81,6 +83,20 @@ struct FdUtils
             dst += ret;
         }
     }
+
+    /// Optimizes the kernel settings like socket and TCP options for an fd
+    /// that is an outgoing TCP socket.
+    /// @param fd socket file descriptor.
+    static void optimize_socket_fd(int fd);
+
+    /// Sets the kernel settings like queuing and terminal settings for an fd
+    /// that is an outgoing tty.
+    /// @param fd tty file descriptor.
+    static void optimize_tty_fd(int fd);
+
+    /// For an fd that is an outgoing link, detects what kind of file
+    /// descriptor this is and calls the appropriate optimize call for it.
+    static void optimize_fd(int fd);
 };
 
 #endif // _UTILS_FD_UTILS_HXX_

src/utils/GcTcpHub.cxx

@@ -31,12 +31,14 @@
  * @date 26 Apr 2014
  */
 
-#include <memory>
-
 #include "utils/GcTcpHub.hxx"
 
+#include <memory>
+#include <sys/socket.h>
+
 #include "nmranet_config.h"
 #include "utils/GridConnectHub.hxx"
+#include "utils/FdUtils.hxx"
 
 void GcTcpHub::on_new_connection(int fd)
 {
@@ -46,6 +48,8 @@ void GcTcpHub::on_new_connection(int fd)
         AtomicHolder h(this);
         numClients_++;
     }
+    // Applies kernel parameters like socket options.
+    FdUtils::optimize_socket_fd(fd);
     create_gc_port_for_can_hub(canHub_, fd, this, use_select);
 }
 

src/utils/GridConnectHub.cxx

@@ -109,7 +109,8 @@ class GCAdapter : public GCAdapterBase
     bool shutdown() OVERRIDE
     {
         unregister();
-        return formatter_.shutdown() && parser_.is_waiting() && formatter_.is_waiting();
+        return formatter_.shutdown() && parser_.shutdown_ready() &&
+            formatter_.is_waiting();
     }
 
     /// HubPort (on a CAN-typed hub) that turns a binary CAN packet into a
@@ -250,11 +251,28 @@ class GCAdapter : public GCAdapterBase
             int max_frames_to_parse =
                 config_gridconnect_bridge_max_incoming_packets();
             if (max_frames_to_parse > 1) {
-                frameAllocator_.reset(new FixedPool(
+                frameAllocator_.reset(new LimitedPool(
                     sizeof(CanHubFlow::buffer_type), max_frames_to_parse));
             }
         }
 
+        /// @return true when this object can be deleted.  This is typically
+        /// once all outgoing packets are released back to the pool, and there
+        /// is no incoming data processing happening.
+        bool shutdown_ready()
+        {
+            int max_frames_to_parse =
+                config_gridconnect_bridge_max_incoming_packets();
+            if (max_frames_to_parse > 1)
+            {
+                if (frameAllocator_->free_items() < (size_t)max_frames_to_parse)
+                {
+                    return false;
+                }
+            }
+            return is_waiting();
+        }
+
         /// @return the destination to write data to.
         CanHubFlow *destination()
         {
@@ -317,7 +335,7 @@ class GCAdapter : public GCAdapterBase
 
         // Allocator to get the frame from. If NULL, the target's default
         // buffer pool will be used.
-        std::unique_ptr<FixedPool> frameAllocator_;
+        std::unique_ptr<LimitedPool> frameAllocator_;
 
         // ==== static data ====