Extend listen-socket to epoll and kqueue transports

The `existing-channel` option added in #748 works only with NIO
transport (Aleph default). When trying to use it with epoll, we would
get a Netty error: `incompatible event loop type:
io.netty.channel.epoll.EpollEventLoop`.

This is not surprising, since the server-channel class to use depends
on the transport, as per
[`transport-server-channel-class`](https://github.com/clj-commons/aleph/blob/5a0c6976aa42f30ca3786cef64f9ba70cf037ae1/src/aleph/netty.clj#L1327)
and the `existing-channel` handling was always creating a NIO
server-channel regardless of transport.

It’s not difficult to support most of the other transports. We just
have to use the server-channel class as per the aforementioned fn.
Their constructors in case of epoll and kqueue take an FD number
rather than Java channel. There is no suitable constructor for
io-uring, so it is not supported. The validation has been updated
accordingly.

The full integration test is still limited to NIO, because obtaining
an FD of a server-socket opened in Java requires deep reflection
(the `--add-opens` flag), which would add some complexity (perhaps a
dedicated lein profile) just to test this. The validation test checks
that the FD route is attempted for the other transports, but it stops
at `Bad file descriptor` error.

The option is being renamed from `existing-channel` to
`listen-socket`, because this seems more fitting when we are accepting
not just channels but also FDs. The `existing-channel` was merged very
recently and hasn’t yet been released in a maven version.

Author: PawelStroinski

src/aleph/http.clj

@@ -51,7 +51,7 @@
    | ---                               | --- |
    | `port`                            | The port the server will bind to. If `0`, the server will bind to a random port. |
    | `socket-address`                  | A `java.net.SocketAddress` specifying both the port and interface to bind to. |
-   | `existing-channel`                | A pre-bound `java.nio.channels.ServerSocketChannel` for the server to use rather than opening and binding its own server-socket. It won't be closed by the server. Possibly obtained from `System/inheritedChannel`. |
+   | `listen-socket`                   | A pre-bound server-socket as a `java.nio.channels.ServerSocketChannel` (NIO transport) or a `long` file descriptor (epoll and kqueue transports) for the server to use rather than opening and binding its own server-socket. It won't be closed by the server. Possibly obtained from `System/inheritedChannel`. |
    | `bootstrap-transform`             | A function that takes an `io.netty.bootstrap.ServerBootstrap` object, which represents the server, and modifies it. |
    | `http-versions`                   | An optional vector of allowable HTTP versions to negotiate via ALPN, in preference order. Defaults to `[:http1]`. |
    | `ssl-context`                     | An `io.netty.handler.ssl.SslContext` object or a map of SSL context options (see `aleph.netty/ssl-server-context` for more details) if an SSL connection is desired. When passing an `io.netty.handler.ssl.SslContext` object, it must have an ALPN config matching the `http-versions` option (see `aleph.netty/ssl-server-context` and `aleph.netty/application-protocol-config`). If only HTTP/1.1 is desired, ALPN config is optional.

src/aleph/http/server.clj

@@ -737,7 +737,7 @@
   [handler
    {:keys [port
            socket-address
-           existing-channel
+           listen-socket
            executor
            http-versions
            ssl-context
@@ -789,8 +789,8 @@
        :bootstrap-transform bootstrap-transform
        :socket-address      (cond
                               socket-address socket-address
-                              (nil? existing-channel) (InetSocketAddress. port))
-       :existing-channel    existing-channel
+                              (nil? listen-socket) (InetSocketAddress. port))
+       :listen-socket       listen-socket
        :on-close            (when (and shutdown-executor?
                                        (or (instance? ExecutorService executor)
                                            (instance? ExecutorService continue-executor)))

src/aleph/netty.clj

@@ -1331,6 +1331,39 @@
     :io-uring IOUringServerSocketChannel
     :nio NioServerSocketChannel))
 
+(defn- wrapping-channel-factory
+  ^ChannelFactory [listen-socket transport]
+  (proxy [ChannelFactory] []
+    (newChannel []
+      (case transport
+        :epoll (EpollServerSocketChannel. ^long listen-socket)
+        :kqueue (KQueueServerSocketChannel. ^long listen-socket)
+        :nio (NioServerSocketChannel. ^java.nio.channels.ServerSocketChannel listen-socket)))))
+
+(defn- validate-listen-socket
+  [listen-socket transport]
+  (when (some? listen-socket)
+    (case transport
+      (:epoll :kqueue) (when-not (int? listen-socket)
+                         (throw (IllegalArgumentException.
+                                 (str "With epoll and kqueue transports, only a numeric file descriptor "
+                                      "is supported as listen-socket, but received: "
+                                      (pr-str listen-socket)))))
+
+      :nio (cond (not (instance? java.nio.channels.ServerSocketChannel listen-socket))
+                 (throw (IllegalArgumentException.
+                         (str "With NIO transport, only a java.nio.channels.ServerSocketChannel "
+                              "is supported as listen-socket, but received: "
+                              (pr-str listen-socket))))
+
+                 (nil? (.getLocalAddress ^java.nio.channels.ServerSocketChannel listen-socket))
+                 (throw (IllegalArgumentException.
+                         (str "The listen-socket is not bound: " (pr-str listen-socket)))))
+
+      (throw (IllegalArgumentException.
+              (str "The listen-socket option is not supported with this transport: "
+                   (pr-str transport)))))))
+
 (defn ^:no-doc convert-address-types [address-types]
   (case address-types
     :ipv4-only ResolvedAddressTypes/IPV4_ONLY
@@ -1668,22 +1701,6 @@
     (.addFirst pipeline "channel-tracker" ^ChannelHandler (channel-tracking-handler chan-group))
     (pipeline-builder pipeline)))
 
-(defn- validate-existing-channel
-  [existing-channel]
-  (when (some? existing-channel)
-    (when-not (instance? java.nio.channels.ServerSocketChannel existing-channel)
-      (throw (IllegalArgumentException.
-               (str "The existing-channel type is not supported: " (pr-str existing-channel)))))
-    (when (nil? (.getLocalAddress ^java.nio.channels.ServerSocketChannel existing-channel))
-      (throw (IllegalArgumentException.
-               (str "The existing-channel is not bound: " (pr-str existing-channel)))))))
-
-(defn- wrapping-channel-factory
-  ^ChannelFactory [^java.nio.channels.ServerSocketChannel channel]
-  (proxy [ChannelFactory] []
-    (newChannel []
-      (NioServerSocketChannel. channel))))
-
 (defn ^:no-doc start-server
   ([pipeline-builder
     ssl-context
@@ -1702,13 +1719,13 @@
             bootstrap-transform
             on-close
             ^SocketAddress socket-address
-            existing-channel
+            listen-socket
             transport
             shutdown-timeout]
      :or   {shutdown-timeout default-shutdown-timeout}
      :as   opts}]
    (ensure-transport-available! transport)
-   (validate-existing-channel existing-channel)
+   (validate-listen-socket listen-socket transport)
    (let [num-cores (.availableProcessors (Runtime/getRuntime))
          num-threads (* 2 num-cores)
          thread-factory (enumerating-thread-factory "aleph-server-pool" false)
@@ -1734,16 +1751,16 @@
                      (.option ChannelOption/SO_REUSEADDR true)
                      (.option ChannelOption/MAX_MESSAGES_PER_READ Integer/MAX_VALUE)
                      (.group group)
-                     (cond-> (nil? existing-channel) (.channel channel-class))
-                     (cond-> (some? existing-channel) (.channelFactory (wrapping-channel-factory existing-channel)))
+                     (cond-> (nil? listen-socket) (.channel channel-class))
+                     (cond-> (some? listen-socket) (.channelFactory (wrapping-channel-factory listen-socket transport)))
                      ;;TODO: add a server (.handler) call to the bootstrap, for logging or something
                      (.childHandler (pipeline-initializer pipeline-builder))
                      (.childOption ChannelOption/SO_REUSEADDR true)
                      (.childOption ChannelOption/MAX_MESSAGES_PER_READ Integer/MAX_VALUE)
                      bootstrap-transform)
 
              ^ServerSocketChannel
-             ch (-> (if (nil? existing-channel)
+             ch (-> (if (nil? listen-socket)
                       (.bind b socket-address)
                       (.register b))
                     .sync
@@ -1755,7 +1772,7 @@
              (when (compare-and-set! closed? false true)
                ;; This is the three step closing sequence:
                ;; 1. Stop listening to incoming requests
-               (if (nil? existing-channel)
+               (if (nil? listen-socket)
                  (-> ch .close .sync)
                  (-> ch .deregister .sync))
                (-> (if (pos? shutdown-timeout)
@@ -1785,7 +1802,7 @@
            (port [_]
              (-> ch .localAddress .getPort))
            (wait-for-close [_]
-             (when (nil? existing-channel)
+             (when (nil? listen-socket)
                (-> ch .closeFuture .await))
              (-> group .terminationFuture .await)
              nil)))

src/aleph/tcp.clj

@@ -74,15 +74,15 @@
    | ---                   | ---
    | `port`                | the port the server will bind to.  If `0`, the server will bind to a random port.
    | `socket-address`      | a `java.net.SocketAddress` specifying both the port and interface to bind to.
-   | `existing-channel`    | a pre-bound `java.nio.channels.ServerSocketChannel` for the server to use rather than opening and binding its own server-socket. It won't be closed by the server. Possibly obtained from `System/inheritedChannel`. |
+   | `listen-socket`       | a pre-bound server-socket as a `java.nio.channels.ServerSocketChannel` (NIO transport) or a `long` file descriptor (epoll and kqueue transports) for the server to use rather than opening and binding its own server-socket. It won't be closed by the server. Possibly obtained from `System/inheritedChannel`.
    | `ssl-context`         | an `io.netty.handler.ssl.SslContext` object or a map of SSL context options (see `aleph.netty/ssl-server-context` for more details). If given, the server will only accept SSL connections and call the handler once the SSL session has been successfully established. If a self-signed certificate is all that's required, `(aleph.netty/self-signed-ssl-context)` will suffice.
    | `bootstrap-transform` | a function that takes an `io.netty.bootstrap.ServerBootstrap` object, which represents the server, and modifies it.
    | `pipeline-transform`  | a function that takes an `io.netty.channel.ChannelPipeline` object, which represents a connection, and modifies it.
    | `raw-stream?`         | if true, messages from the stream will be `io.netty.buffer.ByteBuf` objects rather than byte-arrays.  This will minimize copying, but means that care must be taken with Netty's buffer reference counting.  Only recommended for advanced users.
    | `shutdown-timeout`    | interval in seconds within which in-flight requests must be processed, defaults to 15 seconds. A value of 0 bypasses waiting entirely.
    | `transport`           | the transport to use, one of `:nio`, `:epoll`, `:kqueue` or `:io-uring` (defaults to `:nio`)."
   [handler
-   {:keys [port socket-address existing-channel ssl-context bootstrap-transform pipeline-transform epoll?
+   {:keys [port socket-address listen-socket ssl-context bootstrap-transform pipeline-transform epoll?
            shutdown-timeout transport]
     :or {bootstrap-transform identity
          pipeline-transform identity
@@ -104,8 +104,8 @@
       :bootstrap-transform bootstrap-transform
       :socket-address (cond
                         socket-address socket-address
-                        (nil? existing-channel) (InetSocketAddress. port))
-      :existing-channel existing-channel
+                        (nil? listen-socket) (InetSocketAddress. port))
+      :listen-socket listen-socket
       :transport (netty/determine-transport transport epoll?)
       :shutdown-timeout shutdown-timeout})))
 

test/aleph/http_test.clj

@@ -30,10 +30,11 @@
       IPool
       Pool)
     (io.netty.channel
-      ChannelHandlerContext
-      ChannelOutboundHandlerAdapter
-      ChannelPipeline
-      ChannelPromise)
+     ChannelException
+     ChannelHandlerContext
+     ChannelOutboundHandlerAdapter
+     ChannelPipeline
+     ChannelPromise)
     (io.netty.handler.codec TooLongFrameException)
     (io.netty.handler.codec.compression
       CompressionOptions
@@ -1774,15 +1775,50 @@
             (catch Exception e
               (is (instance? ProxyConnectException e)))))))))
 
-(deftest test-existing-channel
-  (testing "validation"
-    (is (thrown-with-msg? Exception #"existing-channel"
-                          (with-http-servers basic-handler {:existing-channel "a string"})))
-    (with-open [unbound-channel (ServerSocketChannel/open)]
-      (is (thrown-with-msg? Exception #"existing-channel"
-                            (with-http-servers basic-handler {:existing-channel unbound-channel})))))
-
-  (testing "with a bound server-socket channel"
+(deftest test-listen-socket
+  (testing "validation when transport is"
+    (let [file-descriptor Integer/MAX_VALUE]
+      (testing "nio"
+        (is (thrown-with-msg? IllegalArgumentException #"listen-socket"
+                              (with-http-servers basic-handler {:listen-socket file-descriptor})))
+        (with-open [unbound-channel (ServerSocketChannel/open)]
+          (is (thrown-with-msg? IllegalArgumentException #"listen-socket"
+                                (with-http-servers basic-handler {:listen-socket unbound-channel})))))
+
+      (testing "epoll"
+        (when (netty/epoll-available?)
+          (testing "we should pass our validation and get an error due to the fake file descriptor"
+            (is (thrown-with-msg? ChannelException #"Bad file descriptor"
+                                  (with-http-servers basic-handler {:transport :epoll
+                                                                    :listen-socket file-descriptor}))))
+          (with-open [channel (bound-channel port)]
+            (is (thrown-with-msg? IllegalArgumentException #"listen-socket"
+                                  (with-http-servers basic-handler {:transport :epoll
+                                                                    :listen-socket channel}))))))
+
+      (testing "kqueue"
+        (when (netty/kqueue-available?)
+          (testing "we should pass our validation and get an error due to the fake file descriptor"
+            (is (thrown-with-msg? ChannelException #"Bad file descriptor"
+                                  (with-http-servers basic-handler {:transport :kqueue
+                                                                    :listen-socket file-descriptor}))))
+          (with-open [channel (bound-channel port)]
+            (is (thrown-with-msg? IllegalArgumentException #"listen-socket"
+                                  (with-http-servers basic-handler {:transport :kqueue
+                                                                    :listen-socket channel}))))))
+
+      (testing "io-uring"
+        (when (netty/io-uring-available?)
+          (testing "which is not supported"
+            (is (thrown-with-msg? IllegalArgumentException #"listen-socket"
+                                  (with-http-servers basic-handler {:transport :io-uring
+                                                                    :listen-socket file-descriptor}))))
+          (with-open [channel (bound-channel port)]
+            (is (thrown-with-msg? IllegalArgumentException #"listen-socket"
+                                  (with-http-servers basic-handler {:transport :io-uring
+                                                                    :listen-socket channel}))))))))
+
+  (testing "with nio transport and a bound server-socket channel"
     (testing "- unknown to the server"
       (with-open [_ (bound-channel port)]
         ;; The port is already bound by a server-socket channel, but
@@ -1796,16 +1832,16 @@
         ;; This time, we shouldn't get a BindException, because we are
         ;; telling Aleph to use an existing server-socket channel,
         ;; which should be already bound, so Aleph doesn't try to bind.
-        (with-http-servers basic-handler {:existing-channel channel}
+        (with-http-servers basic-handler {:listen-socket channel}
           (is (= 200 (:status @(http-get "/string")))))
         ;; The existing channel should not be closed on a server shutdown,
         ;; because that channel is not owned by the server.
         (is (.isOpen channel)))))
 
-  (testing "the :port option is not required when :existing-channel is passed"
+  (testing "the :port option is not required when :listen-socket is passed"
     (with-redefs [http-server-options (dissoc http-server-options :port)]
       (with-open [channel (bound-channel port)]
-        (with-http1-server basic-handler {:existing-channel channel}
+        (with-http1-server basic-handler {:listen-socket channel}
           (is (= 200 (:status @(http-get "/string")))))))))
 
 (deftest ^:leak test-leak-in-raw-stream-handler

test/aleph/tcp_test.clj

@@ -56,11 +56,11 @@
       (catch Exception _
         (is (not (netty/io-uring-available?)))))))
 
-(deftest test-existing-channel
-  (testing "the :port option is not required when :existing-channel is passed"
+(deftest test-listen-socket
+  (testing "the :port option is not required when :listen-socket is passed"
     (let [port 8083]
       (with-open [channel (bound-channel port)]
-        (with-server (tcp/start-server echo-handler {:existing-channel channel :shutdown-timeout 0})
+        (with-server (tcp/start-server echo-handler {:listen-socket channel :shutdown-timeout 0})
           (let [c @(tcp/client {:host "localhost" :port port})]
             (s/put! c "foo")
             (is (= "foo" (bs/to-string @(s/take! c))))))))))