Add nghttp2_submit_shutdown_notice() to start graceful shutdown

nghttp2_submit_shutdown_notice() is used to notify the client that
graceful shutdown is started.  We expect that after this call, the
server application should send another GOAWAY using
nghttp2_submit_goaway() with appropriate last_stream_id.  In this
commit, we also added nghttp2_session_get_last_proc_stream_id(), which
can be used as last_stream_id parameter.

This commit implements graceful shutdown in nghttpx.  The integration
test for graceful shutdown is also added.

integration-tests/nghttpx_test.go

@@ -9,6 +9,7 @@ import (
 	"io"
 	"io/ioutil"
 	"net/http"
+	"syscall"
 	"testing"
 )
 
@@ -391,6 +392,82 @@ func TestH2H1ConnectFailure(t *testing.T) {
 	}
 }
 
+func TestH2H1GracefulShutdown(t *testing.T) {
+	st := newServerTester(nil, t, noopHandler)
+	defer st.Close()
+
+	fmt.Fprint(st.conn, "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n")
+	if err := st.fr.WriteSettings(); err != nil {
+		t.Fatalf("st.fr.WriteSettings(): %v", err)
+	}
+
+	header := []hpack.HeaderField{
+		pair(":method", "GET"),
+		pair(":scheme", "http"),
+		pair(":authority", st.authority),
+		pair(":path", "/"),
+	}
+
+	for _, h := range header {
+		_ = st.enc.WriteField(h)
+	}
+
+	if err := st.fr.WriteHeaders(http2.HeadersFrameParam{
+		StreamID:      1,
+		EndStream:     false,
+		EndHeaders:    true,
+		BlockFragment: st.headerBlkBuf.Bytes(),
+	}); err != nil {
+		t.Fatalf("st.fr.WriteHeaders(): %v", err)
+	}
+
+	// send SIGQUIT signal to nghttpx to perform graceful shutdown
+	st.cmd.Process.Signal(syscall.SIGQUIT)
+
+	// after signal, finish request body
+	if err := st.fr.WriteData(1, true, nil); err != nil {
+		t.Fatalf("st.fr.WriteData(): %v", err)
+	}
+
+	numGoAway := 0
+
+	for {
+		fr, err := st.readFrame()
+		if err != nil {
+			if err == io.EOF {
+				want := 2
+				if got := numGoAway; got != want {
+					t.Fatalf("numGoAway: %v; want %v", got, want)
+				}
+				return
+			}
+			t.Fatalf("st.readFrame(): %v", err)
+		}
+		switch f := fr.(type) {
+		case *http2.GoAwayFrame:
+			numGoAway += 1
+			want := http2.ErrCodeNo
+			if got := f.ErrCode; got != want {
+				t.Fatalf("f.ErrCode(%v): %v; want %v", numGoAway, got, want)
+			}
+			switch numGoAway {
+			case 1:
+				want := (uint32(1) << 31) - 1
+				if got := f.LastStreamID; got != want {
+					t.Fatalf("f.LastStreamID(%v): %v; want %v", numGoAway, got, want)
+				}
+			case 2:
+				want := uint32(1)
+				if got := f.LastStreamID; got != want {
+					t.Fatalf("f.LastStreamID(%v): %v; want %v", numGoAway, got, want)
+				}
+			case 3:
+				t.Fatalf("too many GOAWAYs received")
+			}
+		}
+	}
+}
+
 func TestH2H2MultipleResponseCL(t *testing.T) {
 	st := newServerTester([]string{"--http2-bridge"}, t, func(w http.ResponseWriter, r *http.Request) {
 		w.Header().Add("content-length", "1")

integration-tests/server_tester.go

@@ -206,7 +206,7 @@ func (st *serverTester) readFrame() (http2.Frame, error) {
 		return f, nil
 	case err := <-st.errCh:
 		return nil, err
-	case <-time.After(2 * time.Second):
+	case <-time.After(5 * time.Second):
 		return nil, errors.New("timeout waiting for frame")
 	}
 }
@@ -446,7 +446,7 @@ func (st *serverTester) http2(rp requestParam) (*serverResponse, error) {
 	_ = st.enc.WriteField(pair("test-case", rp.name))
 
 	for _, h := range rp.header {
-		_ = st.enc.WriteField(pair(strings.ToLower(h.Name), h.Value))
+		_ = st.enc.WriteField(h)
 	}
 
 	err := st.fr.WriteHeaders(http2.HeadersFrameParam{

lib/includes/nghttp2/nghttp2.h

@@ -2462,6 +2462,43 @@ int nghttp2_session_terminate_session2(nghttp2_session *session,
                                        int32_t last_stream_id,
                                        uint32_t error_code);
 
+/**
+ * @function
+ *
+ * Signals to the client that the server started graceful shutdown
+ * procedure.
+ *
+ * This function is only usable for server.  If this function is
+ * called with client side session, this function returns
+ * :enum:`NGHTTP2_ERR_INVALID_STATE`.
+ *
+ * To gracefully shutdown HTTP/2 session, server should call this
+ * function to send GOAWAY with last_stream_id (1u << 31) - 1.  And
+ * after some delay (e.g., 1 RTT), send another GOAWAY with the stream
+ * ID that the server has some processing using
+ * `nghttp2_submit_goaway()`.  See also
+ * `nghttp2_session_get_last_proc_stream_id()`.
+ *
+ * Unlike `nghttp2_submit_goaway()`, this function just sends GOAWAY
+ * and does nothing more.  This is a mere indication to the client
+ * that session shutdown is imminent.  The application should call
+ * `nghttp2_submit_goaway()` with appropriate last_stream_id after
+ * this call.
+ *
+ * If one or more GOAWAY frame have been already sent by either
+ * `nghttp2_submit_goaway()` or `nghttp2_session_terminate_session()`,
+ * this function has no effect.
+ *
+ * This function returns 0 if it succeeds, or one of the following
+ * negative error codes:
+ *
+ * :enum:`NGHTTP2_ERR_NOMEM`
+ *     Out of memory.
+ * :enum:`NGHTTP2_ERR_INVALID_STATE`
+ *     The |session| is initialized as client.
+ */
+int nghttp2_submit_shutdown_notice(nghttp2_session *session);
+
 /**
  * @function
  *
@@ -3061,6 +3098,13 @@ int nghttp2_submit_ping(nghttp2_session *session, uint8_t flags,
  * keep this memory after the return of this function.  If the
  * |opaque_data_len| is 0, the |opaque_data| could be ``NULL``.
  *
+ * After successful transmission of GOAWAY, following things happen.
+ * All incoming streams having strictly more than |last_stream_id| are
+ * closed.  All incoming HEADERS which starts new stream are simply
+ * ignored.  After all active streams are handled, both
+ * `nghttp2_session_want_read()` and `nghttp2_session_want_write()`
+ * return 0 and the application can close session.
+ *
  * This function returns 0 if it succeeds, or one of the following
  * negative error codes:
  *
@@ -3074,6 +3118,19 @@ int nghttp2_submit_goaway(nghttp2_session *session, uint8_t flags,
                           int32_t last_stream_id, uint32_t error_code,
                           const uint8_t *opaque_data, size_t opaque_data_len);
 
+/**
+ * @function
+ *
+ * Returns the last stream ID of a stream for which
+ * :type:`nghttp2_on_frame_recv_callback` was invoked most recently.
+ * The returned value can be used as last_stream_id parameter for
+ * `nghttp2_submit_goaway()` and
+ * `nghttp2_session_terminate_session2()`.
+ *
+ * This function always succeeds.
+ */
+int32_t nghttp2_session_get_last_proc_stream_id(nghttp2_session *session);
+
 /**
  * @function
  *

lib/nghttp2_outbound_item.h

@@ -70,11 +70,21 @@ typedef struct {
   uint8_t eof;
 } nghttp2_data_aux_data;
 
+typedef enum {
+  NGHTTP2_GOAWAY_AUX_NONE = 0x0,
+  /* indicates that session should be terminated after the
+     transmission of this frame. */
+  NGHTTP2_GOAWAY_AUX_TERM_ON_SEND = 0x1,
+  /* indicates that this GOAWAY is just a notification for graceful
+     shutdown.  No nghttp2_session.goaway_flags should be updated on
+     the reaction to this frame. */
+  NGHTTP2_GOAWAY_AUX_SHUTDOWN_NOTICE = 0x2,
+} nghttp2_goaway_aux_flag;
+
 /* struct used for GOAWAY frame */
 typedef struct {
-  /* nonzero if session should be terminated after the transmission of
-     this frame. */
-  int terminate_on_send;
+  /* bitwise-OR of one or more of nghttp2_goaway_aux_flag. */
+  uint8_t flags;
 } nghttp2_goaway_aux_data;
 
 /* Additional data which cannot be stored in nghttp2_frame struct */

lib/nghttp2_session.c

@@ -136,7 +136,8 @@ static int session_terminate_session(nghttp2_session *session,
   }
 
   rv = nghttp2_session_add_goaway(session, last_stream_id, error_code,
-                                  debug_data, debug_datalen, 1);
+                                  debug_data, debug_datalen,
+                                  NGHTTP2_GOAWAY_AUX_TERM_ON_SEND);
 
   if (rv != 0) {
     return rv;
@@ -2351,17 +2352,20 @@ static int session_after_frame_sent1(nghttp2_session *session) {
 
       aux_data = &item->aux_data.goaway;
 
-      if (aux_data->terminate_on_send) {
-        session->goaway_flags |= NGHTTP2_GOAWAY_TERM_SENT;
-      }
+      if ((aux_data->flags & NGHTTP2_GOAWAY_AUX_SHUTDOWN_NOTICE) == 0) {
 
-      session->goaway_flags |= NGHTTP2_GOAWAY_SENT;
+        if (aux_data->flags & NGHTTP2_GOAWAY_AUX_TERM_ON_SEND) {
+          session->goaway_flags |= NGHTTP2_GOAWAY_TERM_SENT;
+        }
 
-      rv = session_close_stream_on_goaway(session, frame->goaway.last_stream_id,
-                                          1);
+        session->goaway_flags |= NGHTTP2_GOAWAY_SENT;
 
-      if (nghttp2_is_fatal(rv)) {
-        return rv;
+        rv = session_close_stream_on_goaway(session,
+                                            frame->goaway.last_stream_id, 1);
+
+        if (nghttp2_is_fatal(rv)) {
+          return rv;
+        }
       }
 
       break;
@@ -5721,7 +5725,7 @@ int nghttp2_session_add_ping(nghttp2_session *session, uint8_t flags,
 
 int nghttp2_session_add_goaway(nghttp2_session *session, int32_t last_stream_id,
                                uint32_t error_code, const uint8_t *opaque_data,
-                               size_t opaque_data_len, int terminate_on_send) {
+                               size_t opaque_data_len, uint8_t aux_flags) {
   int rv;
   nghttp2_outbound_item *item;
   nghttp2_frame *frame;
@@ -5764,7 +5768,7 @@ int nghttp2_session_add_goaway(nghttp2_session *session, int32_t last_stream_id,
                             opaque_data_copy, opaque_data_len);
 
   aux_data = &item->aux_data.goaway;
-  aux_data->terminate_on_send = terminate_on_send;
+  aux_data->flags = aux_flags;
 
   rv = nghttp2_session_add_item(session, item);
   if (rv != 0) {
@@ -6265,3 +6269,7 @@ int nghttp2_session_set_next_stream_id(nghttp2_session *session,
 uint32_t nghttp2_session_get_next_stream_id(nghttp2_session *session) {
   return session->next_stream_id;
 }
+
+int32_t nghttp2_session_get_last_proc_stream_id(nghttp2_session *session) {
+  return session->last_proc_stream_id;
+}

lib/nghttp2_session.h

@@ -347,7 +347,9 @@ int nghttp2_session_add_ping(nghttp2_session *session, uint8_t flags,
 /*
  * Adds GOAWAY frame with the last-stream-ID |last_stream_id| and the
  * error code |error_code|. This is a convenient function built on top
- * of nghttp2_session_add_frame() to add GOAWAY easily.
+ * of nghttp2_session_add_frame() to add GOAWAY easily.  The
+ * |aux_flags| are bitwise-OR of one or more of
+ * nghttp2_goaway_aux_flag.
  *
  * This function returns 0 if it succeeds, or one of the following
  * negative error codes:
@@ -359,7 +361,7 @@ int nghttp2_session_add_ping(nghttp2_session *session, uint8_t flags,
  */
 int nghttp2_session_add_goaway(nghttp2_session *session, int32_t last_stream_id,
                                uint32_t error_code, const uint8_t *opaque_data,
-                               size_t opaque_data_len, int terminate_on_send);
+                               size_t opaque_data_len, uint8_t aux_flags);
 
 /*
  * Adds WINDOW_UPDATE frame with stream ID |stream_id| and

lib/nghttp2_submit.c

@@ -236,8 +236,24 @@ int nghttp2_submit_rst_stream(nghttp2_session *session, uint8_t flags _U_,
 int nghttp2_submit_goaway(nghttp2_session *session, uint8_t flags _U_,
                           int32_t last_stream_id, uint32_t error_code,
                           const uint8_t *opaque_data, size_t opaque_data_len) {
+  if (session->goaway_flags & NGHTTP2_GOAWAY_TERM_ON_SEND) {
+    return 0;
+  }
   return nghttp2_session_add_goaway(session, last_stream_id, error_code,
-                                    opaque_data, opaque_data_len, 0);
+                                    opaque_data, opaque_data_len,
+                                    NGHTTP2_GOAWAY_AUX_NONE);
+}
+
+int nghttp2_submit_shutdown_notice(nghttp2_session *session) {
+  if (!session->server) {
+    return NGHTTP2_ERR_INVALID_STATE;
+  }
+  if (session->goaway_flags) {
+    return 0;
+  }
+  return nghttp2_session_add_goaway(session, (1u << 31) - 1, NGHTTP2_NO_ERROR,
+                                    NULL, 0,
+                                    NGHTTP2_GOAWAY_AUX_SHUTDOWN_NOTICE);
 }
 
 int nghttp2_submit_settings(nghttp2_session *session, uint8_t flags _U_,

src/shrpx_http2_upstream.cc

@@ -600,6 +600,46 @@ void settings_timeout_cb(struct ev_loop *loop, ev_timer *w, int revents) {
 }
 } // namespace
 
+namespace {
+void shutdown_timeout_cb(struct ev_loop *loop, ev_timer *w, int revents) {
+  auto upstream = static_cast<Http2Upstream *>(w->data);
+  auto handler = upstream->get_client_handler();
+  upstream->submit_goaway();
+  handler->signal_write();
+}
+} // namespace
+
+namespace {
+void prepare_cb(struct ev_loop *loop, ev_prepare *w, int revents) {
+  auto upstream = static_cast<Http2Upstream *>(w->data);
+  upstream->check_shutdown();
+}
+} // namespace
+
+void Http2Upstream::submit_goaway() {
+  auto last_stream_id = nghttp2_session_get_last_proc_stream_id(session_);
+  nghttp2_submit_goaway(session_, NGHTTP2_FLAG_NONE, last_stream_id,
+                        NGHTTP2_NO_ERROR, nullptr, 0);
+}
+
+void Http2Upstream::check_shutdown() {
+  int rv;
+  if (shutdown_handled_) {
+    return;
+  }
+  if (worker_config->graceful_shutdown) {
+    shutdown_handled_ = true;
+    rv = nghttp2_submit_shutdown_notice(session_);
+    if (rv != 0) {
+      ULOG(FATAL, this) << "nghttp2_submit_shutdown_notice() failed: "
+                        << nghttp2_strerror(rv);
+      return;
+    }
+    handler_->signal_write();
+    ev_timer_start(handler_->get_loop(), &shutdown_timer_);
+  }
+}
+
 Http2Upstream::Http2Upstream(ClientHandler *handler)
     : downstream_queue_(
           get_config()->http2_proxy
@@ -609,7 +649,7 @@ Http2Upstream::Http2Upstream(ClientHandler *handler)
                     : 0,
           !get_config()->http2_proxy),
       handler_(handler), session_(nullptr), data_pending_(nullptr),
-      data_pendinglen_(0) {
+      data_pendinglen_(0), shutdown_handled_(false) {
 
   int rv;
 
@@ -703,6 +743,15 @@ Http2Upstream::Http2Upstream(ClientHandler *handler)
 
   settings_timer_.data = this;
 
+  // timer for 2nd GOAWAY.  HTTP/2 spec recommend 1 RTT.  We wait for
+  // 2 seconds.
+  ev_timer_init(&shutdown_timer_, shutdown_timeout_cb, 2., 0);
+  shutdown_timer_.data = this;
+
+  ev_prepare_init(&prep_, prepare_cb);
+  prep_.data = this;
+  ev_prepare_start(handler_->get_loop(), &prep_);
+
   handler_->reset_upstream_read_timeout(
       get_config()->http2_upstream_read_timeout);
 
@@ -711,6 +760,8 @@ Http2Upstream::Http2Upstream(ClientHandler *handler)
 
 Http2Upstream::~Http2Upstream() {
   nghttp2_session_del(session_);
+  ev_prepare_stop(handler_->get_loop(), &prep_);
+  ev_timer_stop(handler_->get_loop(), &shutdown_timer_);
   ev_timer_stop(handler_->get_loop(), &settings_timer_);
 }
 

src/shrpx_http2_upstream.h

@@ -97,17 +97,23 @@ public:
   void start_downstream(Downstream *downstream);
   void initiate_downstream(std::unique_ptr<Downstream> downstream);
 
+  void submit_goaway();
+  void check_shutdown();
+
 private:
   // must be put before downstream_queue_
   std::unique_ptr<HttpsUpstream> pre_upstream_;
   MemchunkPool mcpool_;
   DownstreamQueue downstream_queue_;
   ev_timer settings_timer_;
+  ev_timer shutdown_timer_;
+  ev_prepare prep_;
   ClientHandler *handler_;
   nghttp2_session *session_;
   const uint8_t *data_pending_;
   size_t data_pendinglen_;
   bool flow_control_;
+  bool shutdown_handled_;
 };
 
 } // namespace shrpx

tests/main.c

@@ -174,6 +174,8 @@ int main(int argc _U_, char *argv[] _U_) {
                    test_nghttp2_submit_window_update) ||
       !CU_add_test(pSuite, "submit_window_update_local_window_size",
                    test_nghttp2_submit_window_update_local_window_size) ||
+      !CU_add_test(pSuite, "submit_shutdown_notice",
+                   test_nghttp2_submit_shutdown_notice) ||
       !CU_add_test(pSuite, "submit_invalid_nv",
                    test_nghttp2_submit_invalid_nv) ||
       !CU_add_test(pSuite, "session_open_stream",

tests/nghttp2_session_test.c

@@ -3996,6 +3996,62 @@ void test_nghttp2_submit_window_update_local_window_size(void) {
   nghttp2_session_del(session);
 }
 
+void test_nghttp2_submit_shutdown_notice(void) {
+  nghttp2_session *session;
+  nghttp2_session_callbacks callbacks;
+  my_user_data ud;
+
+  memset(&callbacks, 0, sizeof(nghttp2_session_callbacks));
+  callbacks.send_callback = null_send_callback;
+  callbacks.on_frame_send_callback = on_frame_send_callback;
+  callbacks.on_frame_not_send_callback = on_frame_not_send_callback;
+
+  nghttp2_session_server_new(&session, &callbacks, &ud);
+
+  CU_ASSERT(0 == nghttp2_submit_shutdown_notice(session));
+
+  ud.frame_send_cb_called = 0;
+
+  nghttp2_session_send(session);
+
+  CU_ASSERT(1 == ud.frame_send_cb_called);
+  CU_ASSERT(NGHTTP2_GOAWAY == ud.sent_frame_type);
+  CU_ASSERT((1u << 31) - 1 == session->local_last_stream_id);
+
+  /* After another GOAWAY, nghttp2_submit_shutdown_notice() is
+     noop. */
+  CU_ASSERT(0 == nghttp2_session_terminate_session(session, NGHTTP2_NO_ERROR));
+
+  ud.frame_send_cb_called = 0;
+
+  nghttp2_session_send(session);
+
+  CU_ASSERT(1 == ud.frame_send_cb_called);
+  CU_ASSERT(NGHTTP2_GOAWAY == ud.sent_frame_type);
+  CU_ASSERT(0 == session->local_last_stream_id);
+
+  CU_ASSERT(0 == nghttp2_submit_shutdown_notice(session));
+
+  ud.frame_send_cb_called = 0;
+  ud.frame_not_send_cb_called = 0;
+
+  nghttp2_session_send(session);
+
+  CU_ASSERT(0 == ud.frame_send_cb_called);
+  CU_ASSERT(0 == ud.frame_not_send_cb_called);
+
+  nghttp2_session_del(session);
+
+  /* Using nghttp2_submit_shutdown_notice() with client side session
+     is error */
+  nghttp2_session_client_new(&session, &callbacks, NULL);
+
+  CU_ASSERT(NGHTTP2_ERR_INVALID_STATE ==
+            nghttp2_submit_shutdown_notice(session));
+
+  nghttp2_session_del(session);
+}
+
 void test_nghttp2_submit_invalid_nv(void) {
   nghttp2_session *session;
   nghttp2_session_callbacks callbacks;
@@ -6360,14 +6416,12 @@ void test_nghttp2_session_graceful_shutdown(void) {
 
   open_stream(session, 301);
   open_stream(session, 302);
+  open_stream(session, 309);
   open_stream(session, 311);
   open_stream(session, 319);
 
-  CU_ASSERT(0 == nghttp2_submit_goaway(session, NGHTTP2_FLAG_NONE,
-                                       (1u << 31) - 1, NGHTTP2_NO_ERROR, NULL,
-                                       0));
+  CU_ASSERT(0 == nghttp2_submit_shutdown_notice(session));
 
-  ud.block_count = 1;
   ud.frame_send_cb_called = 0;
 
   CU_ASSERT(0 == nghttp2_session_send(session));
@@ -6375,6 +6429,19 @@ void test_nghttp2_session_graceful_shutdown(void) {
   CU_ASSERT(1 == ud.frame_send_cb_called);
   CU_ASSERT((1u << 31) - 1 == session->local_last_stream_id);
 
+  CU_ASSERT(0 == nghttp2_submit_goaway(session, NGHTTP2_FLAG_NONE, 311,
+                                       NGHTTP2_NO_ERROR, NULL, 0));
+
+  ud.block_count = 1;
+  ud.frame_send_cb_called = 0;
+  ud.stream_close_cb_called = 0;
+
+  CU_ASSERT(0 == nghttp2_session_send(session));
+
+  CU_ASSERT(1 == ud.frame_send_cb_called);
+  CU_ASSERT(311 == session->local_last_stream_id);
+  CU_ASSERT(1 == ud.stream_close_cb_called);
+
   CU_ASSERT(0 ==
             nghttp2_session_terminate_session2(session, 301, NGHTTP2_NO_ERROR));
 
@@ -6390,6 +6457,7 @@ void test_nghttp2_session_graceful_shutdown(void) {
 
   CU_ASSERT(NULL != nghttp2_session_get_stream(session, 301));
   CU_ASSERT(NULL != nghttp2_session_get_stream(session, 302));
+  CU_ASSERT(NULL == nghttp2_session_get_stream(session, 309));
   CU_ASSERT(NULL == nghttp2_session_get_stream(session, 311));
   CU_ASSERT(NULL == nghttp2_session_get_stream(session, 319));
 

tests/nghttp2_session_test.h

@@ -80,6 +80,7 @@ void test_nghttp2_submit_settings_update_local_window_size(void);
 void test_nghttp2_submit_push_promise(void);
 void test_nghttp2_submit_window_update(void);
 void test_nghttp2_submit_window_update_local_window_size(void);
+void test_nghttp2_submit_shutdown_notice(void);
 void test_nghttp2_submit_invalid_nv(void);
 void test_nghttp2_session_open_stream(void);
 void test_nghttp2_session_open_stream_with_idle_stream_dep(void);