Repository: clj-commons/aleph
Pull request #642 (2022-11-09 17:10:01)

netty: configurable server graceful shutdowns
The recently closed #641 and previous #639 were both attempting to improve the server shutdown procedure.

The intent of both these PRs is to make it easily viable to provide the classic blue-green deployment mode where:

- Servers stop listening for incoming requests
- In-flight requests are still processed for a configurable grace period
- All resources are then cleaned up

After more digging (my Netty-foo being a bit rusty of late :-)) an approach relying on
Netty's [*ChannelGroup*](https://netty.io/4.1/api/io/netty/channel/group/ChannelGroup.html) functionality seems to be the right mechanism.

Of course, not everyone will want to keep a set of all active connections, including its memory and latency impact.

This revised PR modifies the previous approach in the following way:

- Allow for hooking into the shutdown process by way of the `shutdown-hook` parameter (accepted by `http/start-server`,` `tcp/start-server`, and `netty/start-server`).
- Provide a maximum time interval for `shutdown-hook` to run in, defaulting to 15 seconds
- ~Walk back configurability of `shutdownGracefully` parameters as they do not give a reliable way to control shutdown.~

In addition to this, a few utilities are provided to handle channel groups, as well as a ready made channel handler to populate a channel group, since this will be the most common use of `shutdown-hook`.

If this PR goes through, I am happy to open a subsequent one with updated docs with a walk-through on how to achieve these stop semantics.

(useful disclosure, @arnaudgeiser and I work together so take his approval seal with a grain of salt :smile:)

arnaudgeiser: Funny... I just added a commit on top of this PR and the tests ran....
pyr: @KingMob I agree with you, I still think hooking into the process could be important so I left the possibility open but now defaulted to creating a sensible hook.

The parameters need rewording and the docstrings need to be updated, but before going further I want to know if we agree on the flow:

- by default, behave as before
- if a `shutdown-hook-timeout` is provided but no hook, track requests and let them be processed before shutting down within the supplied interval
- if both shutdown-hook-timeout and shutdown-hook are provided, run the provided hook

I suppose the parameters could be named: `graceful-shutdown-period` and `graceful-shutdown-hook`, would this work?
KingMob: For future reference, please don't force-push in the middle of reviewing a large PR. I personally love to rewrite history, but it has to be done carefully.

It makes it impossible to see only new changes since the last review. I only force push when a) nobody's started reviewing, b) everybody's done reviewing, c) _all_ the code needs to be re-reviewed, or d) nobody's going to review the changes.
KingMob: Let's take a step back and think about the usage and the API. 

Will users _ever_ want/need NON-graceful shutdown? (I.e., giving channels time to wrap up)  

If not, we should wrap their hook in one that automatically takes care of the graceful details, and eliminate fns they can mess with, either by inlining them, declaring them private, or at a bare minimum, apply `^:no-doc` metadata. In this scenario, everyone always wants a graceful shutdown.

If they _do_ need immediate shutdown, can we still use the graceful shutdown method (channel group, pipeline xform, etc), but tell users to set the timeout to 0 if they want it closed ASAP? It's intuitive and doesn't require two code paths.

If we _are_ maintaining a separate code path, should there be a default, which should it be, and what should that look like?
       - `:graceful-shutdown?` (maybe defaulting to true)
       - `:immediate-shutdown?` (maybe defaulting to false)
       - `:wait-for-channels-during-shutdown?`
       - etc.

I would prefer to avoid separate code paths if avoidable. Conditionally making channel groups, altering the pipeline, and maintaining different hooks is really unappealing to me if we don't have to. My preference is to make it all graceful, and let people set the timeout to 0 if they're in a hurry.
pyr: > Will users ever want/need NON-graceful shutdown? (I.e., giving channels time to wrap up)

Yes, high traffic proxies will not want to pay the memory and latency cost of maintaining the channel set.
pyr: > If we _are_ maintaining a separate code path, should there be a default, which should it be, and what should that look like? 
> - `:graceful-shutdown?` (maybe defaulting to true)
> - `:immediate-shutdown?` (maybe defaulting to false)
> - `:wait-for-channels-during-shutdown?` 
> - etc

Figuring this out was the intent of my last comment. 

My initial hunch was that we cannot change the default behavior since aleph updates are
usually relatively silent. If you feel this is warranted and provided we make it clear in
the release notes, why not. I leave the judgment of what you feel is OK to you. If the
behavior change is OK, I would indeed make it the default as it is the sanest thing to do.

The ideal option from my PoV:

- Graceful shutdown by default with a 15 second timeout interval
- No waiting on the event loop group (i.e: `(.shutdownGracefully 0 0 TimeUnit/SECONDS)`)

Control-hooks:

- `graceful-shutdown-fn`: change the hook if something else, additional 
- `graceful-shutdown?`: 
- `graceful-shutdown-timeout`: timeout period for the channel group close future, defaults to 15 seconds
- `group-shutdown-timeout`: control the parameters to `.shutdownGracefully group`, quiet period would be `(min 2 timeout)` . Defaulting to 0.

If we want to preserve the existing behavior, `graceful-shutdown?` would need to be `false` by default, and `group-shutdown-timeout` set to 15.

So in essence, I guess the two good couple of defaults most people would want to use are:

- `[false 15]`
- `[true 0]`



KingMob: It's getting late in my TZ, so I have to go, but on one point:

> we cannot change the default behavior since aleph updates are usually relatively silent.

You're right, this is usually true, but I don't think we've ever promised any behavior in regards to shutdown, so I think we have some extra leeway here.
arnaudgeiser: So, if we agree to change the current shutdown behaviour, we can pivot only around `shutdown-timeout` which will default to 15 seconds.

So I totally concur with this:

> The ideal option from my PoV:
Graceful shutdown by default with a 15 second timeout interval
No waiting on the event loop group (i.e: (.shutdownGracefully 0 0 TimeUnit/SECONDS))


DerGuteMoritz: > Will users ever want/need NON-graceful shutdown? (I.e., giving channels time to wrap up)

Just to throw my 2 cents into the hat: I wanted this in the past for use in integration test suites. When there are a lot of tests which start a server at the beginning and stop it again at the end, you want to await their proper shutdown so that different tests don't collide (e.g. because they always bind to the same listening port). Having each test wait for at least 2 seconds adds up quickly then. This is a case where stopping the server as god intended doesn't work either :smile: 
arnaudgeiser: With the approach that has been discussed, it will still be supported @DerGuteMoritz.
KingMob: Graceful by default sounds good to me. I think we have a solution.

Thanks for reminding me about test suites. Yeah, we definitely want a way to kill the server ASAP in that situation.

Having `graceful-shutdown-timeout` and `eventloop-shutdown-timeout` is OK, but the naming is confusing in regards to their relationship. Can we come up with better names? Failing that, we need some good docstrings.

I don't think we need to remove a user-supplied `shutdown-hook` option, as long as the user doesn't have to orchestrate "graceful-ness", and the docs are clear on _when_ their hook is called. OTOH, if nobody needs the hook, we can skip it until someone asks for it. I leave that up to you.
arnaudgeiser: > I don't think we need to remove a user-supplied shutdown-hook option, as long as the user doesn't have to orchestrate "graceful-ness", and the docs are clear on when their hook is called. OTOH, if nobody needs the hook, we can skip it until someone asks for it. I leave that up to you.

Agreed, but depending on where the `shutdown-hook` is placed, it should not be an aleph responsibility, but the responsibility of your favorite dependency management library (component, integrant, mount etc...)

- Whether the `shutdown-hook` is placed between [1] and [2], it definitely has this place on Aleph
- Whether the `shutdown-hook` is placed between [2] and [3], it should be the responsibility of a dependency management library

[1] : Stop listening to incoming requests
[2] : Run the shutdown hook within the supplied timeout interval when provided. This can be used to finish processing in-flight requests for instance.
[3] : At this stage, stop the EventLoopGroup
pyr: > Agreed, but depending on where the `shutdown-hook` is placed, it should not be an aleph responsibility, but the responsibility of your favorite dependency management library (component, integrant, mount etc...)

I agree with this, and it would speak against the current place where `on-close` gets ran. `on-close` is necessary internally but maybe it shouldn't have been exposed since it can be fully orchestrated outside of aleph.

If we were to consider changing the placement of the hook point I think it would make sense. Otherwise, after having given it more thought I think it's too many knobs to tweak for library consumers.

> Having `graceful-shutdown-timeout` and `eventloop-shutdown-timeout` is OK, but the naming is confusing in regards to their relationship. Can we come up with better names? Failing that, we need some good docstrings.

I agree with the poor naming, thinking about it, the knob that most people will want to tweak is the graceful one, so I would go with `shutdown-timeout` for this one (the existing one hasn't made it to a release yet).

After a bit of thought and looking at the feedback here, it seems is that the default parameters to `EventLoopGroup::shutdownGracefully` will work out for everyone **except** in odd scenarios like test ones, maybe we can go with `fast-eventloop-shutdown?` being a boolean. 

So to sum it up one possible strategy would be:

- `shutdown-timeout`: Controlling the amount of time the channel group
- `fast-eventloop-shutdown?`: Would set the parameters of `EventLoopGroup::shutdownGracefully` to `0, 0, TimeUnit/SECONDS`. Defaults to `false`

If everyone wants to consider the change of location of the hook I would make the current `on-close` parameter internal and use the same name for the hook that can run between server channel closing and EventLoopGroup shutdown.




arnaudgeiser: > I agree with this, and it would speak against the current place where on-close gets ran. on-close is necessary internally but maybe it shouldn't have been exposed since it can be fully orchestrated outside of aleph.

Exactly, it should never have been exposed. But what has been done cannot be undone... :(

> I agree with the poor naming, thinking about it, the knob that most people will want to tweak is the graceful one, so I would go with shutdown-timeout for this one (the existing one hasn't made it to a release yet).

Fine for me!

> fast-eventloop-shutdown?: Would set the parameters of EventLoopGroup::shutdownGracefully to 0, 0, TimeUnit/SECONDS. Defaults to false

What about :

```clojure
(if (zero? shutdown-timeout)
  (.shutdownGracefully group 0 0)
  (.shutdownGracefully group))
```

> If everyone wants to consider the change of location of the hook I would make the current on-close parameter internal and use the same name for the hook that can run between server channel closing and EventLoopGroup shutdown.

To me, that's the way to go (if we are talking about putting the `on-close` parameter between [1] and [2] - what I understood).

[1] : Stop listening to incoming requests
[2] : Run the shutdown hook within the supplied timeout interval when provided. This can be used to finish processing in-flight requests for instance.
[3] : At this stage, stop the EventLoopGroup
pyr: Your proposal kills the opportunity to have fast test shutdowns unfortunately. For the rest we agree.
arnaudgeiser: And why not `(.shutdownGracefully group 0 0)` in all cases?
Whether waiting for active connections is the default, it's acceptable to have this behavior as non-parameterizable, no?
pyr: I adapted to the proposed approach, taking into account @arnaudgeiser's comment
pyr: > Approving but would be nice to do the adaptations on the other tests to make them fast again.
> Furthermore, the title of the PR has to be changed to reflect what has been done.

I will do now that the direction seems to work for everyone, once I get a second approval I will also do a last force push to have proper commit messages
pyr: All comments have been addressed except for @KingMob's regarding `shutdown-quiet-period` which I think doesn't apply. Once we find an agreement on this last one I'll rework the commit messages and push a cleaned rebased set of commits.

@arnaudgeiser tests are indeed way faster now :-)

Thanks for the thorough reviews everyone!
pyr: @KingMob The code is ready for review, I will leave the rebase + force-push for after your review/approval

Patch:
diff --git a/src/aleph/http.clj b/src/aleph/http.clj
index 984ffce..80db96a 100644
--- a/src/aleph/http.clj
+++ b/src/aleph/http.clj
@@ -52,8 +52,7 @@
    | `idle-timeout` | when set, connections are closed after not having performed any I/O operations for the given duration, in milliseconds. Defaults to `0` (infinite idle time).
    | `continue-handler` | optional handler which is invoked when header sends \"Except: 100-continue\" header to test whether the request should be accepted or rejected. Handler should return `true`, `false`, ring responseo to be used as a reject response or deferred that yields one of those.
    | `continue-executor` | optional `java.util.concurrent.Executor` which is used to handle requests passed to :continue-handler.  To avoid this indirection you may specify `:none`, but in this case extreme care must be taken to avoid blocking operations on the handler's thread.
-   | `shutdown-quiet-period` | optional period in seconds for which new connections will still be serviced after a scheduled shutdown via `java.io.Closeable#close`. Defaults to 2 seconds.
-   | `shutdown-timeout` | optional grace period in seconds on which to wait for the event loop group to empty during a scheduled shutdown. Defaults to 15 seconds."
+   | `shutdown-timeout` | interval in seconds within which in-flight requests must be processed, defaults to 15 seconds. A value of 0 bypasses waiting entirely."
   [handler options]
   (server/start-server handler options))
 
diff --git a/src/aleph/http/server.clj b/src/aleph/http/server.clj
index 41bbe67..929ed37 100644
--- a/src/aleph/http/server.clj
+++ b/src/aleph/http/server.clj
@@ -578,14 +578,12 @@
            compression?
            continue-handler
            continue-executor
-           shutdown-quiet-period
            shutdown-timeout]
     :or {bootstrap-transform identity
          pipeline-transform identity
          shutdown-executor? true
          epoll? false
          compression? false
-         shutdown-quiet-period netty/default-shutdown-quiet-period
          shutdown-timeout netty/default-shutdown-timeout}
     :as options}]
   (let [executor (cond
@@ -642,7 +640,6 @@
                      (when (instance? ExecutorService continue-executor)
                        (.shutdown ^ExecutorService continue-executor))))
       :transport (if epoll? :epoll :nio)
-      :shutdown-quiet-period shutdown-quiet-period
       :shutdown-timeout shutdown-timeout})))
 
 ;;;
diff --git a/src/aleph/netty.clj b/src/aleph/netty.clj
index 44347cd..33145ac 100644
--- a/src/aleph/netty.clj
+++ b/src/aleph/netty.clj
@@ -22,6 +22,8 @@
      ChannelOutboundHandler
      ChannelHandlerContext
      ChannelInitializer]
+    [io.netty.channel.group
+     ChannelGroup DefaultChannelGroup]
     [io.netty.channel.epoll Epoll EpollEventLoopGroup
      EpollServerSocketChannel
      EpollSocketChannel
@@ -57,7 +59,8 @@
      ResourceLeakDetector$Level]
     [java.net URI SocketAddress InetSocketAddress]
     [io.netty.util.concurrent
-     AbstractEventExecutor FastThreadLocalThread GenericFutureListener Future]
+     AbstractEventExecutor FastThreadLocalThread GenericFutureListener Future
+     GlobalEventExecutor]
     [java.io InputStream File]
     [java.nio ByteBuffer]
     [java.nio.channels ClosedChannelException]
@@ -121,15 +124,11 @@
 (defn channel-remote-address [^Channel ch]
   (some-> ch ^InetSocketAddress (.remoteAddress) .getAddress .getHostAddress))
 
-;;;
-
-(def ^:const default-shutdown-quiet-period
-  "Default quiet period in seconds when shutting down an eventloop group"
-  2) ;; As in https://github.com/netty/netty/blob/b61d7d40f40e3a797e8a60cd567f849a9799c771/common/src/main/java/io/netty/util/concurrent/AbstractEventExecutor.java#L40
+;;; Defaults defined here since they are not publically exposed by Netty
 
 (def ^:const default-shutdown-timeout
-  "Default timeout in seconds when shutting down an eventloop group"
-  15) ;; As in https://github.com/netty/netty/blob/b61d7d40f40e3a797e8a60cd567f849a9799c771/common/src/main/java/io/netty/util/concurrent/AbstractEventExecutor.java#L41
+  "Default timeout in seconds to wait for graceful shutdown complete"
+  15)
 
 (def ^:const array-class (class (clojure.core/byte-array 0)))
 
@@ -294,6 +293,12 @@
     x
     (.channel ^ChannelHandlerContext x)))
 
+(defn ^ChannelGroup make-channel-group
+  "Create a channel group which can be used to perform channel operations on
+   several channels at once."
+  []
+  (DefaultChannelGroup. GlobalEventExecutor/INSTANCE))
+
 (defmacro safe-execute
   "Executes the body on the event-loop (an executor service) associated with the Netty channel.
 
@@ -434,8 +439,6 @@
   (put [this msg blocking? timeout timeout-value]
     (.put this msg blocking?)))
 
-
-
 (defn sink
   ([ch]
     (sink ch true identity (fn [])))
@@ -693,6 +696,17 @@
             (.readableBytes ^ByteBuf msg)))
         (.write ctx msg promise)))))
 
+(defn ^ChannelHandler channel-tracking-handler
+  "Yields an inbound handler, ready to be added to a pipeline,
+   which keeps track of requests in a provided channel group.
+   The channel-group can be created via `make-channel-group`."
+  [^ChannelGroup group]
+  (channel-inbound-handler
+   :channel-active
+   ([_ ctx]
+    (.add group (channel ctx))
+    (.fireChannelActive ctx))))
+
 (defn pipeline-initializer [pipeline-builder]
   (proxy [ChannelInitializer] []
     (initChannel [^Channel ch]
@@ -703,6 +717,16 @@
     (.remove pipeline handler))
   pipeline)
 
+(defn append-handler-to-pipeline
+  "Convenience function to add a handler to the tail of a netty pipeline"
+  [^ChannelPipeline pipeline handler-id ^ChannelHandler handler]
+  (.addLast pipeline (str handler-id) handler))
+
+(defn prepend-handler-to-pipeline
+  "Convenience function to add a handler to the head of a netty pipeline"
+  [^ChannelPipeline pipeline handler-id ^ChannelHandler handler]
+  (.addFirst pipeline (str handler-id) handler))
+
 (defn instrument!
   [stream]
   (if-let [^Channel ch (->> stream meta :aleph/channel)]
@@ -1226,6 +1250,21 @@ initialize an DnsAddressResolverGroup instance.
               (let [ch (.channel ^ChannelFuture f)]
                 (maybe-ssl-handshake-future ch)))))))))
 
+(defn- add-ssl-handler
+  [pipeline-builder ssl-ctx]
+  (fn [^ChannelPipeline pipeline]
+    (append-handler-to-pipeline
+     pipeline "ssl-handler"
+     (.newHandler ^SslContext ssl-ctx (-> pipeline .channel .alloc)))
+    (pipeline-builder pipeline)))
+
+(defn- add-channel-tracker-handler
+  [pipeline-builder chan-group]
+  (fn [pipeline]
+    (prepend-handler-to-pipeline pipeline "channel-tracker"
+                                (channel-tracking-handler chan-group))
+    (pipeline-builder pipeline)))
+
 (defn start-server
   ([pipeline-builder
     ssl-context
@@ -1233,26 +1272,27 @@ initialize an DnsAddressResolverGroup instance.
     on-close
     socket-address
     epoll?]
-   (start-server {:pipeline-builder pipeline-builder
-                  :ssl-context ssl-context
+   (start-server {:pipeline-builder    pipeline-builder
+                  :ssl-context         ssl-context
                   :bootstrap-transform bootstrap-transform
-                  :on-close on-close
-                  :socket-address socket-address
-                  :transport (if epoll? :epoll :nio)}))
+                  :on-close            on-close
+                  :socket-address      socket-address
+                  :transport           (if epoll? :epoll :nio)}))
   ([{:keys [pipeline-builder
             ssl-context
             bootstrap-transform
             on-close
             ^SocketAddress socket-address
             transport
-            shutdown-quiet-period
-            shutdown-timeout]}]
+            shutdown-timeout]
+     :or {shutdown-timeout default-shutdown-timeout}}]
    (when (= transport :epoll)
      (ensure-epoll-available!))
    (let [num-cores      (.availableProcessors (Runtime/getRuntime))
          num-threads    (* 2 num-cores)
          thread-factory (enumerating-thread-factory "aleph-netty-server-event-pool" false)
          closed?        (atom false)
+         chan-group     (make-channel-group)
 
          ^EventLoopGroup group
          (condp = transport
@@ -1272,13 +1312,8 @@ initialize an DnsAddressResolverGroup instance.
                        (coerce-ssl-server-context ssl-context))
 
          pipeline-builder
-         (if ssl-context
-           (fn [^ChannelPipeline p]
-             (.addLast p "ssl-handler"
-                       (.newHandler ssl-context
-                                    (-> p .channel .alloc)))
-             (pipeline-builder p))
-           pipeline-builder)]
+         (cond-> (add-channel-tracker-handler pipeline-builder chan-group)
+           (some? ssl-context) (add-ssl-handler ssl-context))]
 
      (try
        (let [b (doto (ServerBootstrap.)
@@ -1298,10 +1333,25 @@ initialize an DnsAddressResolverGroup instance.
            java.io.Closeable
            (close [_]
              (when (compare-and-set! closed? false true)
+               ;; This is the three step closing sequence:
+               ;; 1. Stop listening to incoming requests
                (-> ch .close .sync)
-               (-> group (.shutdownGracefully (long shutdown-quiet-period)
-                                              (long shutdown-timeout)
-                                              TimeUnit/SECONDS))
+               (-> (if (pos? shutdown-timeout)
+                     ;; 2. Wait for in-flight requests to stop processing within the supplied timeout
+                     ;;    interval.
+                     (-> (wrap-future (.newCloseFuture chan-group))
+                         (d/timeout! (* shutdown-timeout 1000) ::timeout))
+                     (d/success-deferred ::noop))
+                   (d/chain'
+                    (fn [shutdown-output]
+                      (when (= shutdown-output ::timeout)
+                        (log/error
+                         (format "Timeout while waiting for requests to close (exceeded: %ss)"
+                                 shutdown-timeout)))))
+                   (d/finally'
+                    ;; 3. At this stage, stop the EventLoopGroup, this will cancel any
+                    ;;    in flight pending requests.
+                    #(.shutdownGracefully group 0 0 TimeUnit/SECONDS)))
                (when on-close
                  (d/chain'
                   (wrap-future (.terminationFuture group))
diff --git a/src/aleph/tcp.clj b/src/aleph/tcp.clj
index a48e011..f9b576c 100644
--- a/src/aleph/tcp.clj
+++ b/src/aleph/tcp.clj
@@ -80,15 +80,13 @@
    | `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-quiet-period` | optional period in seconds for which new connections will still be serviced after a scheduled shutdown via `java.io.Closeable::close`. Defaults to 2 seconds.
-   | `shutdown-timeout` | optional grace period in seconds on which to wait for the event loop group to empty during a scheduled shutdown. Defaults to 15 seconds.  "
+   | `shutdown-timeout` | interval in seconds within which in-flight requests must be processed, defaults to 15 seconds. A value of 0 bypasses waiting entirely."
   [handler
    {:keys [port socket-address ssl-context bootstrap-transform pipeline-transform epoll?
-           shutdown-quiet-period shutdown-timeout]
+           shutdown-timeout]
     :or {bootstrap-transform identity
          pipeline-transform identity
          epoll? false
-         shutdown-quiet-period netty/default-shutdown-quiet-period
          shutdown-timeout netty/default-shutdown-timeout}
     :as options}]
   (netty/start-server
@@ -103,7 +101,6 @@
                       socket-address
                       (InetSocketAddress. port))
     :transport (if epoll? :epoll :nio)
-    :shutdown-quiet-period shutdown-quiet-period
     :shutdown-timeout shutdown-timeout}))
 
 (defn- ^ChannelHandler client-channel-handler
@@ -161,7 +158,7 @@
    | `port` | the port of the server.
    | `remote-address` | a `java.net.SocketAddress` specifying the server's address.
    | `local-address` | a `java.net.SocketAddress` specifying the local network interface to use.
-   | `ssl-context` | an explicit `io.netty.handler.ssl.SslHandler` or a map of SSL context options (see `aleph.netty/ssl-client-context` for more details) to use. Defers to `ssl?` and `insecure?` configuration if omitted.
+   | `ssl-context` | an explicit `io.netty.handler.ssl.SslHandler` or a map of SSL context options (see `aleph.netty/ssl-server-context` for more details) to use. Defers to `ssl?` and `insecure?` configuration if omitted.
    | `ssl?` | if true, the client attempts to establish a secure connection with the server.
    | `insecure?` | if true, the client will ignore the server's certificate.
    | `bootstrap-transform` | a function that takes an `io.netty.bootstrap.Bootstrap` object, which represents the client, and modifies it.


Test patch:
diff --git a/test/aleph/classloader_test.clj b/test/aleph/classloader_test.clj
index 8370e29..2e515e3 100644
--- a/test/aleph/classloader_test.clj
+++ b/test/aleph/classloader_test.clj
@@ -44,7 +44,7 @@
       (with-dynamic-redefs [netty/operation-complete (partial operation-complete result)]
         (let [server (http/start-server
                       (constantly {:body "ok"})
-                      {:port 9999})]
+                      {:port 9999 :shutdown-timeout 0})]
           (on-signal :int
                      (bound-fn [_] (.close ^java.io.Closeable server)))
           (.exec (Runtime/getRuntime) (format "kill -SIGINT %s" (pid)))
diff --git a/test/aleph/http/multipart_test.clj b/test/aleph/http/multipart_test.clj
index 348d37d..3aa50f7 100644
--- a/test/aleph/http/multipart_test.clj
+++ b/test/aleph/http/multipart_test.clj
@@ -163,7 +163,7 @@
    :body body})
 
 (deftest test-send-multipart-request
-  (let [s (http/start-server echo-handler {:port port1})
+  (let [s (http/start-server echo-handler {:port port1 :shutdown-timeout 0})
         ^String resp @(d/chain'
                        (http/post url1 {:multipart parts})
                        :body
@@ -202,6 +202,7 @@
 
 (defn- test-decoder [port url raw-stream?]
   (let [s (http/start-server decode-handler {:port port
+                                             :shutdown-timeout 0
                                              :raw-stream? raw-stream?})
         chunks (-> (http/post url {:multipart parts})
                    (deref 1e3 {:body "timeout"})
diff --git a/test/aleph/http_continue_test.clj b/test/aleph/http_continue_test.clj
index 3c69877..4af946a 100644
--- a/test/aleph/http_continue_test.clj
+++ b/test/aleph/http_continue_test.clj
@@ -36,7 +36,7 @@
 (defn- test-accept [server-options]
   (with-server (http/start-server ok-handler (merge
                                               server-options
-                                              {:port port}))
+                                              {:port port :shutdown-timeout 0}))
     (let [c @(tcp/client {:host "localhost" :port port})]
       @(s/put! c (pack-lines ["PUT /file HTTP/1.1"
                               "Host: localhost"
@@ -69,7 +69,7 @@
   (let [resp (or resp "417 Expectation Failed")]
     (with-server (http/start-server ok-handler (merge
                                                 server-options
-                                                {:port port}))
+                                                {:port port :shutdown-timeout 0}))
       (let [c @(tcp/client {:host "localhost" :port port})]
         @(s/put! c (pack-lines ["PUT /file HTTP/1.1"
                                 "Host: localhost"
diff --git a/test/aleph/http_test.clj b/test/aleph/http_test.clj
index a0e3d3a..37fb502 100644
--- a/test/aleph/http_test.clj
+++ b/test/aleph/http_test.clj
@@ -180,25 +180,25 @@
            (.shutdown *pool*))))))
 
 (defmacro with-handler [handler & body]
-  `(with-server (http/start-server ~handler {:port port})
+  `(with-server (http/start-server ~handler {:port port :shutdown-timeout 0})
      ~@body))
 
 (defmacro with-compressed-handler [handler & body]
   `(do
-     (with-server (http/start-server ~handler {:port port :compression? true})
+     (with-server (http/start-server ~handler {:port port :compression? true :shutdown-timeout 0})
        ~@body)
-     (with-server (http/start-server ~handler {:port port :compression-level 3})
+     (with-server (http/start-server ~handler {:port port :compression-level 3 :shutdown-timeout 0})
        ~@body)))
 
 (def default-ssl-options {:port port, :ssl-context (netty/self-signed-ssl-context)})
 
 (defmacro with-handler-options
-   [handler options & body]
-   `(with-server (http/start-server ~handler ~options)
-      ~@body))
+  [handler options & body]
+  `(with-server (http/start-server ~handler (assoc ~options :shutdown-timeout 0))
+     ~@body))
 
 (defmacro with-raw-handler [handler & body]
-  `(with-server (http/start-server ~handler {:port port, :raw-stream? true})
+  `(with-server (http/start-server ~handler {:port port, :raw-stream? true :shutdown-timeout 0})
      ~@body))
 
 (defmacro with-both-handlers [handler & body]
@@ -474,6 +474,7 @@
   (testing "custom error handler"
     (with-server (http/start-server invalid-handler
                                     {:port port
+                                     :shutdown-timeout 0
                                      :error-handler (fn [_]
                                                       {:status 500
                                                        :body "Internal error"})})
@@ -589,7 +590,7 @@
 (deftest ^:benchmark benchmark-http
   (println "starting HTTP benchmark server on 8080")
   (netty/leak-detector-level! :disabled)
-  (let [server (http/start-server hello-handler {:port 8080})]
+  (let [server (http/start-server hello-handler {:port 8080 :shutdown-timeout 0})]
     (Thread/sleep (* 1000 60))
     (println "stopping server")
     (.close ^java.io.Closeable server)))
@@ -601,7 +602,8 @@
                  (fn [req]
                    (d/let-flow [s (http/websocket-connection req)]
                      (s/consume #(s/put! s %) s)))
-                 {:port 8080})]
+                 {:port 8080
+                  :shutdown-timeout 0})]
     (Thread/sleep (* 1000 60))
     (println "stopping server")
     (.close ^java.io.Closeable server)))
@@ -613,6 +615,7 @@
     (with-server (http/start-server
                    basic-handler
                    {:port port
+                    :shutdown-timeout 0
                     :pipeline-transform
                     (fn [^ChannelPipeline pipeline]
                       (.addBefore pipeline
@@ -642,6 +645,7 @@
   (with-server (http/start-server
                 basic-handler
                 {:port port
+                 :shutdown-timeout 0
                  :pipeline-transform add-http-object-aggregator})
     (let [rsp @(http-put (str "http://localhost:" port "/echo")
                          {:body "hello"})]
@@ -657,6 +661,7 @@
   (with-server (http/start-server
                 basic-handler
                 {:port port
+                 :shutdown-timeout 0
                  :raw-stream? true
                  :pipeline-transform add-http-object-aggregator})
     (let [rsp @(http-put (str "http://localhost:" port "/echo")
diff --git a/test/aleph/http_timeout_test.clj b/test/aleph/http_timeout_test.clj
new file mode 100644
index 0000000..729b4ed
--- /dev/null
+++ b/test/aleph/http_timeout_test.clj
@@ -0,0 +1,97 @@
+(ns aleph.http-timeout-test
+  (:require [aleph.http   :as http]
+            [clj-commons.byte-streams :as bs]
+            [clojure.test :refer [deftest testing is]]
+            [manifold.stream :as s])
+  (:import java.lang.AutoCloseable))
+
+(def ^:private default-options
+  {:throw-exceptions? false
+   :pool (http/connection-pool {:connection-options {:keep-alive? false}})})
+
+(def ^:private port 8082)
+
+(defn- http-get
+  [& {:as options}]
+  (http/get (str "http://localhost:" port) (merge default-options options)))
+
+(defn- streaming-handler
+  [_]
+  (let [body (s/stream)]
+    (future
+      (dotimes [i 10]
+        (s/put! body (str i "."))
+        (Thread/sleep 200))
+      (s/close! body))
+    {:status 200
+     :body body}))
+
+(defn- waiting-handler
+  [_]
+  (Thread/sleep 2000)
+  {:status 200
+   :body   "waited"})
+
+(defn- server-options
+  [& {:as options}]
+  (merge {:port 8082 :shutdown-timeout 4}
+         options))
+
+(deftest test-shutdown-timeout-1
+  (testing "shutdown and wait for streaming in-flight requests to finish."
+    (let [server (http/start-server streaming-handler (server-options))]
+      (try
+        (let [resp (http-get)]
+          (Thread/sleep 50) ;; wait a bit for the request to be initiated
+          (.close ^AutoCloseable server)
+          (is (= 200 (:status @resp)))
+          (is (= "0.1.2.3.4.5.6.7.8.9." (bs/to-string (:body @resp)))))
+        (finally
+          (.close ^AutoCloseable server)))))
+
+  (testing "shutdown and wait for blocking in-flight requests to finish."
+    (let [server (http/start-server waiting-handler (server-options :shutdown-timeout 4))]
+      (try
+        (let [resp (http-get)]
+          (Thread/sleep 50) ;; wait a bit for the request to be initiated
+          (.close ^AutoCloseable server)
+          (is (= 200 (:status @resp)))
+          (is (= "waited" (bs/to-string (:body @resp)))))
+        (finally
+          (.close ^AutoCloseable server))))))
+
+(deftest test-shutdown-timeout-2
+  (testing "shutdown with a timeout of 0 second and no grace for in-flight requests"
+    (let [server (http/start-server waiting-handler (server-options :shutdown-timeout 0))]
+      (try
+        (let [resp (http-get)]
+          (Thread/sleep 50) ;; wait a bit for the request to be initiated
+          (.close ^AutoCloseable server)
+          (is (thrown-with-msg? Exception #"connection was closed after 0\." @resp)))
+        (finally
+          (.close ^AutoCloseable server))))))
+
+(deftest test-shutdown-timeout-3
+  (testing "shutdown with a timeout of 1 seconds with streaming body"
+    (let [server (http/start-server streaming-handler (server-options :shutdown-timeout 1))]
+      (try
+        (let [resp (http-get)]
+          (Thread/sleep 50) ;; wait a bit for the request to be initiated
+          (.close ^AutoCloseable server)
+          (is (= 200 (:status @resp)))
+          (is (= "0.1.2.3.4.5." (bs/to-string (:body @resp)))))
+        (finally
+          (.close ^AutoCloseable server))))))
+
+
+(deftest test-shutdown-timeout-4
+  (testing "shutdown with a timeout of 1 seconds while waiting for body"
+    (let [server (http/start-server waiting-handler (server-options :shutdown-timeout 1))]
+      (try
+        (let [resp (http-get)]
+          (Thread/sleep 50) ;; wait a bit for the request to be initiated
+          (.close ^AutoCloseable server)
+          (is (thrown-with-msg? Exception #"connection was closed after 1\." @resp)))
+        (finally
+          (.close ^AutoCloseable server))))))
+
diff --git a/test/aleph/ring_test.clj b/test/aleph/ring_test.clj
index 5cb17bd..f819010 100644
--- a/test/aleph/ring_test.clj
+++ b/test/aleph/ring_test.clj
@@ -46,7 +46,7 @@
      :body (prn-str (get-request-value request keys))}))
 
 (defmacro with-server [keys & body]
-  `(let [server# (http/start-server (request-callback ~keys) {:port 8080})]
+  `(let [server# (http/start-server (request-callback ~keys) {:port 8080 :shutdown-timeout 0})]
      (try
        ~@body
        (finally
diff --git a/test/aleph/tcp_ssl_test.clj b/test/aleph/tcp_ssl_test.clj
index 8cd8225..1029ff3 100644
--- a/test/aleph/tcp_ssl_test.clj
+++ b/test/aleph/tcp_ssl_test.clj
@@ -25,6 +25,7 @@
   (let [ssl-session (atom nil)]
     (with-server (tcp/start-server (ssl-echo-handler ssl-session)
                                    {:port 10001
+                                    :shutdown-timeout 0
                                     :ssl-context ssl/server-ssl-context})
       (let [c @(tcp/client {:host "localhost"
                             :port 10001
@@ -39,6 +40,7 @@
   (let [ssl-session (atom nil)]
     (with-server (tcp/start-server (ssl-echo-handler ssl-session)
                                    {:port 10001
+                                    :shutdown-timeout 0
                                     :ssl-context ssl/server-ssl-context-opts})
       (let [c @(tcp/client {:host "localhost"
                             :port 10001
@@ -53,6 +55,7 @@
   (let [ssl-session (atom nil)]
     (with-server (tcp/start-server (ssl-echo-handler ssl-session)
                                    {:port 10001
+                                    :shutdown-timeout 0
                                     :ssl-context ssl/server-ssl-context})
       (let [c @(tcp/client {:host "localhost"
                             :port 10001
@@ -74,6 +77,7 @@
                                                        (.fireChannelInactive ctx)))]
     (with-server (tcp/start-server (ssl-echo-handler ssl-session)
                                    {:port 10001
+                                    :shutdown-timeout 0
                                     :ssl-context ssl/server-ssl-context
                                     :pipeline-transform (fn [p]
                                                           (.addLast p notify-connection-closed))})
@@ -95,6 +99,7 @@
                                                        (.fireChannelActive ctx)))]
     (with-server (tcp/start-server (fn [s _])
                                    {:port 10001
+                                    :shutdown-timeout 0
                                     :ssl-context ssl/server-ssl-context})
       (let [c (tcp/client {:host "localhost"
                            :port 10001
diff --git a/test/aleph/tcp_test.clj b/test/aleph/tcp_test.clj
index ace8775..b34c7d2 100644
--- a/test/aleph/tcp_test.clj
+++ b/test/aleph/tcp_test.clj
@@ -19,7 +19,7 @@
          (.close ^java.io.Closeable server#)))))
 
 (deftest test-echo
-  (with-server (tcp/start-server echo-handler {:port 10001})
+  (with-server (tcp/start-server echo-handler {:port 10001 :shutdown-timeout 0})
     (let [c @(tcp/client {:host "localhost", :port 10001})]
       (s/put! c "foo")
       (is (= "foo" (bs/to-string @(s/take! c)))))))
diff --git a/test/aleph/websocket_test.clj b/test/aleph/websocket_test.clj
index 303fd04..24d9fec 100644
--- a/test/aleph/websocket_test.clj
+++ b/test/aleph/websocket_test.clj
@@ -22,15 +22,15 @@
          (netty/wait-for-close server#)))))
 
 (defmacro with-handler [handler & body]
-  `(with-server (http/start-server ~handler {:port 8080})
+  `(with-server (http/start-server ~handler {:port 8080 :shutdown-timeout 0})
      ~@body))
 
 (defmacro with-raw-handler [handler & body]
-  `(with-server (http/start-server ~handler {:port 8081, :raw-stream? true})
+  `(with-server (http/start-server ~handler {:port 8081, :raw-stream? true :shutdown-timeout 0})
      ~@body))
 
 (defmacro with-compressing-handler [handler & body]
-  `(with-server (http/start-server ~handler {:port 8080, :compression? true})
+  `(with-server (http/start-server ~handler {:port 8080, :compression? true :shutdown-timeout 0})
      ~@body))
 
 (defn connection-handler