posix-stack: fix memory leak with shutdown/accept race

When `posix_server_socket_impl::accept()` runs it may start a cross-core
background fiber that inserts a pending connection into the thread local
container posix_ap_server_socket_impl::conn_q.

However, the continuation that enqueues the pending connection may not
aactually run until after the target core calls abort_accept() (e.g.
parallel shutdown via a seastar::sharded<server>::stop).

This can leave an entry in the conn_q container that is destroyed when
the reactor thread exits. Unfortunately the conn_q container holds
conntrack::handle type that schedules additional work in its destructor.

```
   class handle {
       foreign_ptr<lw_shared_ptr<load_balancer>> _lb;
       ~handle() {
           (void)smp::submit_to(_host_cpu, [cpu = _target_cpu, lb = std::move(_lb)] {
               lb->closed_cpu(cpu);
           });
       }
       ...
```

When this race occurs and the destructor runs the reactor is no longer
available, leading to the following memory leak in which the continuation that
is scheduled onto the reactor is leaked:

Direct leak of 88 byte(s) in 1 object(s) allocated from:
    #0 0x557c91ca5b7d in operator new(unsigned long) /v/llvm/llvm/src/compiler-rt/lib/asan/asan_new_delete.cpp:95:3

    #1 0x557ca3e3cc08 in void seastar::future<void>::schedule<seastar::internal::promise_ba...
    ...
    // the unordered map here is conn_q
    scylladb#19 0x557ca47034d8 in std::__1::unordered_multimap<std::__1::tuple<int, seastar::socket...
    scylladb#20 0x7f98dcaf238e in __call_tls_dtors (/lib64/libc.so.6+0x4038e) (BuildId: 6e3c087aca9...

fixes: scylladb#738

Signed-off-by: Noah Watkins <noahwatkins@gmail.com>

include/seastar/net/posix-stack.hh

@@ -141,6 +141,7 @@ class posix_ap_server_socket_impl : public server_socket_impl {
     using sockets_map_t = std::unordered_map<protocol_and_socket_address, promise<accept_result>>;
     using conn_map_t = std::unordered_multimap<protocol_and_socket_address, connection>;
     static thread_local sockets_map_t sockets;
+    friend class posix_network_stack;
     static thread_local conn_map_t conn_q;
     int _protocol;
     socket_address _sa;
@@ -204,6 +205,7 @@ public:
     virtual bool has_per_core_namespace() override { return _reuseport; };
     bool supports_ipv6() const override;
     std::vector<network_interface> network_interfaces() override;
+    virtual future<> initialize() override;
 };
 
 class posix_ap_network_stack : public posix_network_stack {

src/net/posix-stack.cc

@@ -29,6 +29,7 @@
 
 #include <seastar/core/loop.hh>
 #include <seastar/core/reactor.hh>
+#include <seastar/core/gate.hh>
 #include <seastar/net/posix-stack.hh>
 #include <seastar/net/net.hh>
 #include <seastar/net/packet.hh>
@@ -96,6 +97,7 @@ class posix_connected_socket_operations {
 
 thread_local posix_ap_server_socket_impl::sockets_map_t posix_ap_server_socket_impl::sockets{};
 thread_local posix_ap_server_socket_impl::conn_map_t posix_ap_server_socket_impl::conn_q{};
+static thread_local gate exiting_gate;
 
 class posix_tcp_connected_socket_operations : public posix_connected_socket_operations {
 public:
@@ -498,9 +500,12 @@ posix_server_socket_impl::accept() {
             return make_ready_future<accept_result>(
                     accept_result{connected_socket(std::move(csi)), sa});
         } else {
-            // FIXME: future is discarded
-            (void)smp::submit_to(cpu, [protocol = _protocol, ssa = _sa, fd = std::move(fd.get_file_desc()), sa, cth = std::move(cth), allocator = _allocator] () mutable {
-                posix_ap_server_socket_impl::move_connected_socket(protocol, ssa, pollable_fd(std::move(fd)), sa, std::move(cth), allocator);
+            (void)with_gate(exiting_gate, [cpu, protocol = _protocol, ssa = _sa, fd = std::move(fd.get_file_desc()), sa, cth = std::move(cth), allocator = _allocator] () mutable {
+                return smp::submit_to(cpu, [protocol, ssa, fd = std::move(fd), sa, cth = std::move(cth), allocator] () mutable {
+                    posix_ap_server_socket_impl::move_connected_socket(protocol, ssa, pollable_fd(std::move(fd)), sa, std::move(cth), allocator);
+                });
+            }).handle_exception([] (const std::exception_ptr&) {
+                // intentionally ignored since we are exiting
             });
             return accept();
         }
@@ -584,7 +589,7 @@ posix_ap_server_socket_impl::move_connected_socket(int protocol, socket_address
             i->second.set_exception(std::current_exception());
         }
         sockets.erase(i);
-    } else {
+    } else if (!exiting_gate.is_closed()) {
         conn_q.emplace(std::piecewise_construct, std::make_tuple(t_sa), std::make_tuple(std::move(fd), std::move(addr), std::move(cth)));
     }
 }
@@ -1093,6 +1098,15 @@ std::vector<network_interface> posix_network_stack::network_interfaces() {
     return std::vector<network_interface>(thread_local_interfaces.begin(), thread_local_interfaces.end());
 }
 
+future<> posix_network_stack::initialize() {
+    engine().at_exit([] {
+        return exiting_gate.close().then([] {
+            posix_ap_server_socket_impl::conn_q.clear();
+        });
+    });
+    return make_ready_future<>();
+}
+
 }
 
 }