cool#9833: Implement MaxConnection Limit (2b)

Sven Göthel · Sven Göthel · commit 013278f58917 · 2024-10-30T16:45:37.000+01:00
Reimplementation of commit 80246f7 (post revert). Adding TCP connection limit to server-side TCP IPv4/IPv6 Sockets - Validated and registered at StreamSocket::create<T>() - Only limits TCP connections for server-side IPv4 or IPv6 TCP connections. - If rejected (limit reached), the accepted FD gets released immediately and the TSocket ctor only receives a -1 FD, i.e. !isOpen(). This also signals `ServerSocket::handlePoll` to dismiss the instance. - If counted against max TCP connections, Socket::_isCounted is set true (see below) - Unregistered at Socket dtor post socket closing - If counted against max TCP connections (_isCounted) `decrTCPConnCount` is called after socket ::close (returned to system). TODO: - revise net::Defaults.maxTCPConnections to match actual system settings Signed-off-by: Sven Göthel <sven.gothel@collabora.com> Change-Id: Ib9f0ac17c05ffe65c2370490f68f581fa76730e7
diff --git a/net/ServerSocket.hpp b/net/ServerSocket.hpp
@@ -108,9 +108,11 @@ class ServerSocket : public Socket
                 const std::string msg = "Failed to accept. (errno: ";
                 throw std::runtime_error(msg + std::strerror(errno) + ')');
             }
-
-            LOG_TRC("Accepted client #" << clientSocket->getFD());
-            _clientPoller.insertNewSocket(std::move(clientSocket));
+            if( clientSocket->isOpen() )
+            {
+                LOG_TRC("Accepted client #" << clientSocket->getFD());
+                _clientPoller.insertNewSocket(std::move(clientSocket));
+            } // else intentionally cancelled accepted connection (e.g. connection limiter)
         }
     }
 
diff --git a/net/Socket.cpp b/net/Socket.cpp
@@ -65,6 +65,9 @@ std::atomic<bool> Socket::InhibitThreadChecks(false);
 
 std::unique_ptr<Watchdog> SocketPoll::PollWatchdog;
 
+std::mutex Socket::maxTCPConnMutex;
+size_t Socket::maxTCPConnCount = 0;
+
 net::DefaultValues net::Defaults = { .inactivityTimeout = std::chrono::seconds(3600),
                                      .wsPingAvgTimeout = std::chrono::seconds(12),
                                      .wsPingInterval = std::chrono::seconds(18),
@@ -146,8 +149,10 @@ std::string Socket::getStatsString(const std::chrono::steady_clock::time_point &
 std::ostream& Socket::streamImpl(std::ostream& os) const
 {
     os << "Socket[#" << getFD()
-       << ", " << toString(type())
-       << " @ ";
+       << ", " << toString(type());
+    if (isCounted())
+       os << ", counted";
+    os << " @ ";
     if (Type::IPv6 == type())
     {
         os << "[" << clientAddress() << "]:" << clientPort();
@@ -1188,9 +1193,9 @@ std::shared_ptr<Socket> ServerSocket::accept()
             ::inet_ntop(clientInfo.sin6_family, inAddr, addrstr, sizeof(addrstr));
             _socket->setClientAddress(addrstr, clientInfo.sin6_port);
 
-            LOG_TRC("Accepted socket #" << _socket->getFD() << " has family "
-                                        << clientInfo.sin6_family << " address "
-                                        << _socket->clientAddress());
+            LOG_TRC((_socket->isOpen() ? "Accepted":"TCP Limiter rejected")
+                    << " socket #" << rc << " has family "
+                    << clientInfo.sin6_family << ", " << *_socket);
 #else
             std::shared_ptr<Socket> _socket = createSocketFromAccept(rc, Socket::Type::Unix);
 #endif
@@ -1393,8 +1398,10 @@ std::ostream& StreamSocket::stream(std::ostream& os) const
 {
     os << "StreamSocket[#" << getFD()
        << ", " << toStringShort(_wsState)
-       << ", " << Socket::toString(type())
-       << " @ ";
+       << ", " << Socket::toString(type());
+    if (isCounted())
+       os << ", counted";
+    os << " @ ";
     if (Type::IPv6 == type())
     {
         os << "[" << clientAddress() << "]:" << clientPort();
diff --git a/net/Socket.hpp b/net/Socket.hpp
@@ -33,6 +33,7 @@
 
 #include <common/StateEnum.hpp>
 #include "Log.hpp"
+#include "NetUtil.hpp"
 #include "Util.hpp"
 #include "Buffer.hpp"
 #include "SigUtil.hpp"
@@ -129,6 +130,8 @@ class SocketDisposition final
     std::shared_ptr<Socket> _socket;
 };
 
+class StreamSocket; // fwd
+
 /// A non-blocking, streaming socket.
 class Socket
 {
@@ -151,6 +154,7 @@ class Socket
         , _lastSeenTime(_creationTime)
         , _bytesSent(0)
         , _bytesRcvd(0)
+        , _isCounted(false)
     {
         init();
     }
@@ -162,19 +166,25 @@ class Socket
         // Doesn't block on sockets; no error handling needed.
         if constexpr (!Util::isMobileApp())
         {
-            ::close(_fd);
+            if (::close(_fd))
+                LOG_SYS("Ignored error closing socket #" << _fd);
             LOG_DBG("Closed socket " << toStringImpl());
         }
         else
-        {
             fakeSocketClose(_fd);
+        if (_isCounted)
+        {
+            _isCounted = false;
+            decrTCPConnCount(_fd);
         }
     }
 
     /// Returns true if this socket is open, i.e. allowed to be polled and not shutdown
     bool isOpen() const { return _open; }
     /// Returns true if this socket has been closed, i.e. rejected from polling and potentially shutdown
     bool isClosed() const { return !_open; }
+    /// Returns true if this socket is counted against free maximum TCP connections
+    bool isCounted() const { return _isCounted; }
 
     constexpr Type type() const { return _type; }
     constexpr bool isIPType() const { return Type::IPv4 == _type || Type::IPv6 == _type; }
@@ -249,7 +259,7 @@ class Socket
         if constexpr (!Util::isMobileApp())
         {
             const int val = 1;
-            if (::setsockopt(_fd, IPPROTO_TCP, TCP_NODELAY, (char*)&val, sizeof(val)) == -1)
+            if (isOpen() && ::setsockopt(_fd, IPPROTO_TCP, TCP_NODELAY, (char*)&val, sizeof(val)) == -1)
             {
                 static std::once_flag once;
                 std::call_once(once,
@@ -409,6 +419,14 @@ class Socket
         LOG_TRC("Ignore further input on socket.");
         _ignoreInput = true;
     }
+
+    /// Returns connection count
+    static size_t getConnectionCount()
+    {
+        std::lock_guard<std::mutex> lock(maxTCPConnMutex);
+        return maxTCPConnCount;
+    }
+
 protected:
     /// Construct based on an existing socket fd.
     /// Used by accept() only.
@@ -422,6 +440,7 @@ class Socket
         , _lastSeenTime(_creationTime)
         , _bytesSent(0)
         , _bytesRcvd(0)
+        , _isCounted(false)
     {
         init();
     }
@@ -493,6 +512,53 @@ class Socket
 
     /// We check the owner even in the release builds, needs to be always correct.
     std::thread::id _owner;
+
+    /// true if counted against net::Defaults.maxTCPConnections and dtor must call `decrTCPConnCount`
+    /// post socket ::close where the socket has been returned to the system.
+    bool _isCounted;
+
+    /// Decrements global connection count
+    /// Call this after socket ::close, where it has been returned to the system.
+    /// @param fd the related file descriptor for logging
+    static void decrTCPConnCount(int fd)
+    {
+        std::lock_guard<std::mutex> lock(maxTCPConnMutex);
+        const size_t u = maxTCPConnCount;
+        auto logPrefix = [fd](std::ostream& os) { os << '#' << fd << ": "; };
+        if (u > 0)
+        {
+            const size_t v = --maxTCPConnCount;
+            LOG_TRC("TCP Limiter: Count decremented: " << v);
+        }
+        else
+            LOG_WRN("TCP Limiter: Count decrement underflow: " << u);
+    }
+
+    /// Increments global TCP connection counter if not exceeding net::DefaultValue::maxTCPConnections.
+    /// Returns true if free connections were available, otherwise false.
+    /// No limitation is applied if net::DefaultValue::maxTCPConnections == 0.
+    /// @param fd the related file descriptor for logging
+    static bool checkAndIncrTCPConnCount(int fd)
+    {
+        std::lock_guard<std::mutex> lock(maxTCPConnMutex);
+        const size_t u = maxTCPConnCount;
+        auto logPrefix = [fd](std::ostream& os) { os << '#' << fd << ": "; };
+        if (net::Defaults.maxTCPConnections == 0 || u < net::Defaults.maxTCPConnections)
+        {
+            const size_t v = ++maxTCPConnCount;
+            LOG_TRC("TCP Limiter: Count incremented: " << v);
+            return true;
+        }
+        else
+        {
+            LOG_WRN("TCP Limiter: Limit reached: " << u);
+            return false;
+        }
+    }
+    friend class StreamSocket; // allow `checkAndIncrConnectionCount`
+
+    static std::mutex maxTCPConnMutex;
+    static size_t maxTCPConnCount; // accepted TCP IPv4/IPv6 socket count
 };
 
 inline std::ostream& operator<<(std::ostream& os, const Socket &s) { return s.stream(os); }
@@ -1322,11 +1388,12 @@ class StreamSocket : public Socket,
         _socketHandler.reset();
     }
 
-    /// Create a socket of type TSocket given an FD and a handler.
+    /// Create a socket of type TSocket derived from StreamSocket given an FD and a handler.
     /// We need this helper since the handler needs a shared_ptr to the socket
     /// but we can't have a shared_ptr in the ctor.
-    template <typename TSocket>
-    static std::shared_ptr<TSocket> create(std::string hostname, const int fd, Type type,
+    template <typename TSocket,
+              std::enable_if_t<  std::is_base_of_v<StreamSocket, TSocket>, bool> = true>
+    static std::shared_ptr<TSocket> create(std::string hostname, int fd, Type type,
                                            bool isClient, HostType hostType,
                                            std::shared_ptr<ProtocolHandlerInterface> handler,
                                            ReadType readType = ReadType::NormalRead,
@@ -1337,7 +1404,31 @@ class StreamSocket : public Socket,
             throw std::runtime_error("StreamSocket " + std::to_string(fd) +
                                      " expects a valid SocketHandler instance.");
 
-        auto socket = std::make_shared<TSocket>(std::move(hostname), fd, type, isClient, hostType, readType, creationTime);
+        bool isCounted = false;
+        if (!isClient && fd >= 0 && (type == Type::IPv4 || type == Type::IPv6))
+        {
+            /// Only limit TCP connections for server-side IPv4 or IPv6 TCP connections.
+            ///
+            /// If limited, the accepted FD gets released immediately
+            /// and the TSocket ctor only receives a -1 FD, i.e. !isOpen(),
+            /// which also signals ServerSocket::handlePoll to dismiss the instance.
+            if (!Socket::checkAndIncrTCPConnCount(fd))
+            {
+                if constexpr (!Util::isMobileApp())
+                {
+                    auto logPrefix = [fd](std::ostream& os) { os << '#' << fd << ": "; };
+                    if (::close(fd))
+                        LOG_SYS("Ignored error closing socket #" << fd);
+                }
+                else
+                    fakeSocketClose(fd);
+                fd = -1; // closed!
+            } else
+                isCounted = true;
+        }
+        auto socket = std::make_shared<TSocket>(std::move(hostname), fd, type, isClient, hostType,
+                                                readType, creationTime);
+        socket->_isCounted = isCounted;
         socket->setHandler(std::move(handler));
 
         return socket;
diff --git a/test/UnitTimeoutBase.hpp b/test/UnitTimeoutBase.hpp
@@ -204,6 +204,8 @@ inline UnitBase::TestResult UnitTimeoutBase1::testHttp(const size_t connectionLi
     sessions.clear();
     TST_LOG("Clearing Poller: " << testname);
     socketPollers.clear();
+    // TCP Connection Count: Just an estimation, no locking on server side
+    TST_LOG("TCP Connection Count: " << testname << ", " << Socket::getConnectionCount() << " / " << net::Defaults.maxTCPConnections);
     TST_LOG("Ending Test: " << testname);
     return TestResult::Ok;
 }
@@ -302,6 +304,8 @@ inline UnitBase::TestResult UnitTimeoutBase1::testWSPing(const size_t connection
     sessions.clear();
     TST_LOG("Clearing Poller: " << testname);
     socketPollers.clear();
+    // TCP Connection Count: Just an estimation, no locking on server side
+    TST_LOG("TCP Connection Count: " << testname << ", " << Socket::getConnectionCount() << " / " << net::Defaults.maxTCPConnections);
     TST_LOG("Ending Test: " << testname);
     return TestResult::Ok;
 }
@@ -397,6 +401,8 @@ inline UnitBase::TestResult UnitTimeoutBase1::testWSDChatPing(const size_t conne
     sessions.clear();
     TST_LOG("Clearing Poller: " << testname);
     socketPollers.clear();
+    // TCP Connection Count: Just an estimation, no locking on server side
+    TST_LOG("TCP Connection Count: " << testname << ", " << Socket::getConnectionCount() << " / " << net::Defaults.maxTCPConnections);
     TST_LOG("Ending Test: " << testname);
     return TestResult::Ok;
 }

Original file line number	Diff line number	Diff line change
`@@ -108,9 +108,11 @@ class ServerSocket : public Socket`
`108`	`108`	`const std::string msg = "Failed to accept. (errno: ";`
`109`	`109`	`throw std::runtime_error(msg + std::strerror(errno) + ')');`
`110`	`110`	`}`
`111`		`-`
`112`		`- LOG_TRC("Accepted client #" << clientSocket->getFD());`
`113`		`- _clientPoller.insertNewSocket(std::move(clientSocket));`
	`111`	`+ if( clientSocket->isOpen() )`
	`112`	`+ {`
	`113`	`+ LOG_TRC("Accepted client #" << clientSocket->getFD());`
	`114`	`+ _clientPoller.insertNewSocket(std::move(clientSocket));`
	`115`	`+ } // else intentionally cancelled accepted connection (e.g. connection limiter)`
`114`	`116`	`}`
`115`	`117`	`}`
`116`	`118`