MXS-1951: Make listener fd worker-local

By storing the file descriptor inside a worker-local variable, it is
possible to handle both unique file descriptors (created with
SO_REUSEPORT) and shared file descriptors with the same code. The way in
which the file descriptor is stored in the rworker_local object determines
the way the listener behaves.

include/maxscale/listener.hh

@@ -203,7 +203,7 @@ private:
     MXS_PROTOCOL      m_proto_func;     /**< Preloaded protocol functions */
     MXS_AUTHENTICATOR m_auth_func;      /**< Preloaded authenticator functions */
 
-    int m_fd;     /**< File descriptor the listener listens on */
+    mxs::rworker_local<int> m_fd {-1};      /**< File descriptor the listener listens on */
 
     /** A shared pointer to the listener itself that is passed as the argument to
      * the protocol's accept function. This allows client connections to live
@@ -238,11 +238,15 @@ private:
     /**
      * Listen on a file descriptor shared between all workers
      *
-     *
      * @return True if the listening was started successfully
      */
     bool listen_shared();
 
+    /**
+     * Close all opened file descriptors for this listener
+     */
+    void close_all_fds();
+
     /**
      * Accept a single client connection
      *
@@ -253,11 +257,11 @@ private:
     /**
      * The file descriptor for accepting new connections
      *
-     * When SO_REUSEPORT is in use, each worker has a separate file descriptor that they accept on.
+     * @return The worker-local file descriptor
      */
     int fd() const
     {
-        return m_fd;
+        return *m_fd;
     }
 
     // Handler for EPOLL_IN events

server/core/listener.cc

@@ -128,14 +128,27 @@ SListener Listener::create(SERVICE* service,
     return listener;
 }
 
+void Listener::close_all_fds()
+{
+    // Shared fds all have the same value. Unique fds each have a unique value. By sorting the values,
+    // removing duplicates and skipping negative values, both cases work and use the same code.
+    auto values = m_fd.values();
+    std::sort(values.begin(), values.end());
+    auto end = std::unique(values.begin(), values.end());
+    auto start = std::upper_bound(values.begin(), end, -1);
+    std::for_each(start, end, close);
+
+    // Make sure we don't accidentally use a closed fd
+    m_fd.assign(-1);
+}
+
 void Listener::destroy(const SListener& listener)
 {
     // Remove the listener from all workers. This makes sure that there's no concurrent access while we're
     // closing things up.
     listener->stop();
 
-    close(listener->m_fd);
+    listener->close_all_fds();
-    listener->m_fd = -1;
     listener->m_state = DESTROYED;
 
     std::lock_guard<std::mutex> guard(listener_lock);
@@ -949,7 +962,8 @@ bool Listener::listen_shared()
     {
         if (mxs::RoutingWorker::add_shared_fd(fd, EPOLLIN, this))
         {
-            m_fd = fd;
+            // All workers share the same fd, assign it here
+            m_fd.assign(fd);
             rval = true;
             m_state = STARTED;
         }