iproto: introduce graceful shutdown protocol

This commit adds the graceful shutdown feature to the IPROTO protocol.
Net.box is patched by the next commit, which also adds tests.

Part of #5924

@TarantoolBot document
Title: Document IPROTO graceful shutdown

A new IPROTO request type was introduced - `IPROTO_SHUTDOWN`, code 77.
When asked to shut down (`os.exit()` is called or `SIGTERM` signal is
received), a server stops accepting new connections and sends a packet
of this type to each of its clients that support the graceful shutdown
feature (see below how a server figures out if a client supports the
feature). The server won't exit until all the clients that were sent the
packets close connections. If all the clients don't close connections
within the shutdown timeout, the server will exit anyway. The default
shutdown timeout is 3 seconds, and it can be configured with
`box.ctl.set_on_shutdown_timeout()`, which also determines the timeout
of `box.ctl.on_shutdown()` triggers.

An `IPROTO_SHUTDOWN` packet doesn't have any keys in its headers (not
even sync number or schema version) nor a body. A client isn't supposed
to reply to an `IPROTO_SHUTDOWN` packet. Instead it's supposed to close
its connection as soon as possible. A client may wait for pending
requests to complete and even send new requests after receiving an
`IPROTO_SHUTDOWN` packet. The server will serve them as usual until it
exits on timeout.

Clients that support the graceful shutdown feature are supposed to set
the `IPROTO_FEATURE_GRACEFUL_SHUTDOWN` feature (bit 4) when sending an
`IPROTO_ID` request to a server. Servers that support the feature set
the same bit in reply to an `IPROTO_ID` request. Introduction of this
feature bumped the IPROTO protocol version up to 4.

Author: locker

src/box/iproto.cc

@@ -41,6 +41,7 @@
 
 #include "version.h"
 #include "fiber.h"
+#include "fiber_cond.h"
 #include "cbus.h"
 #include "say.h"
 #include "sio.h"
@@ -68,6 +69,7 @@
 #include "salad/stailq.h"
 #include "assoc.h"
 #include "txn.h"
+#include "on_shutdown.h"
 
 enum {
 	IPROTO_SALT_SIZE = 32,
@@ -255,6 +257,23 @@ unsigned iproto_readahead = 16320;
 /* The maximal number of iproto messages in fly. */
 static int iproto_msg_max = IPROTO_MSG_MAX_MIN;
 
+/**
+ * List of connections blocking shutdown in tx, linked by
+ * iproto_connection::in_shutdown_list. Accessed and modified
+ * only from the tx thread.
+ *
+ * The shutdown trigger callback won't return until it's empty.
+ * We add a connection to the list if it supports the graceful
+ * shutdown feature, and remove it when it's closed.
+ */
+static RLIST_HEAD(shutdown_list);
+
+/** Signalled when shutdown_list becomes empty. */
+static struct fiber_cond shutdown_list_empty_cond;
+
+/** This flag is set when the shutdown trigger is called. */
+static bool shutdown_is_inprogress;
+
 int
 iproto_addr_count(void)
 {
@@ -645,6 +664,8 @@ struct iproto_connection
 		 * return.
 		 */
 		bool is_push_pending;
+		/** Link in shutdown_list. */
+		struct rlist in_shutdown_list;
 	} tx;
 	/** Authentication salt. */
 	char salt[IPROTO_SALT_SIZE];
@@ -1433,6 +1454,7 @@ iproto_connection_new(struct iproto_thread *iproto_thread)
 	con->state = IPROTO_CONNECTION_ALIVE;
 	con->tx.is_push_pending = false;
 	con->tx.is_push_sent = false;
+	rlist_create(&con->tx.in_shutdown_list);
 	rmean_collect(iproto_thread->rmean, IPROTO_CONNECTIONS, 1);
 	return con;
 }
@@ -1445,6 +1467,7 @@ iproto_connection_delete(struct iproto_connection *con)
 	assert(!iostream_is_initialized(&con->io));
 	assert(con->session == NULL);
 	assert(con->state == IPROTO_CONNECTION_DESTROYED);
+	assert(rlist_empty(&con->tx.in_shutdown_list));
 	/*
 	 * The output buffers must have been deleted
 	 * in tx thread.
@@ -1711,6 +1734,9 @@ tx_process_disconnect(struct cmsg *m)
 			session_run_on_disconnect_triggers(con->session);
 		}
 	}
+	rlist_del_entry(con, tx.in_shutdown_list);
+	if (rlist_empty(&shutdown_list))
+		fiber_cond_broadcast(&shutdown_list_empty_cond);
 }
 
 static void
@@ -2154,6 +2180,28 @@ tx_process_call(struct cmsg *m)
 	tx_end_msg(msg);
 }
 
+static void
+tx_process_id(struct iproto_connection *con, const struct id_request *id)
+{
+	con->session->meta.features = id->features;
+	if (iproto_features_test(&id->features,
+				 IPROTO_FEATURE_GRACEFUL_SHUTDOWN)) {
+		rlist_add_entry(&shutdown_list, con, tx.in_shutdown_list);
+		if (shutdown_is_inprogress) {
+			/*
+			 * IPROTO_ID enabling graceful shutdown was received
+			 * after we've sent shutdown packets. Sent a packet
+			 * to this one as well so that it doesn't block
+			 * shutdown.
+			 */
+			if (iproto_send_shutdown(con->tx.p_obuf) != 0) {
+				diag_log();
+				rlist_del_entry(con, tx.in_shutdown_list);
+			}
+		}
+	}
+}
+
 static void
 iproto_session_notify(struct session *session,
 		      const char *key, size_t key_len,
@@ -2182,7 +2230,7 @@ tx_process_misc(struct cmsg *m)
 					   ::schema_version);
 			break;
 		case IPROTO_ID:
-			con->session->meta.features = msg->id.features;
+			tx_process_id(con, &msg->id);
 			iproto_reply_id_xc(out, msg->header.sync,
 					   ::schema_version);
 			break;
@@ -2758,6 +2806,40 @@ iproto_session_notify(struct session *session,
 
 /** }}} */
 
+static void
+iproto_send_stop_msg(void);
+
+/**
+ * Shutdown trigger callback.
+ *  1. Stops accepting new connections.
+ *  2. Sends an IPROTO_SHUTDOWN packet to all clients that support
+ *     the graceful shutdown feature.
+ *  3. Waits for the clients to close connections.
+ */
+static int
+iproto_on_shutdown_f(void *arg)
+{
+	(void)arg;
+	assert(!shutdown_is_inprogress);
+	shutdown_is_inprogress = true;
+	fiber_set_name(fiber_self(), "iproto.shutdown");
+	iproto_send_stop_msg();
+	evio_service_stop(&tx_binary);
+	struct iproto_connection *con, *next_con;
+	rlist_foreach_entry_safe(con, &shutdown_list, tx.in_shutdown_list,
+				 next_con) {
+		if (iproto_send_shutdown(con->tx.p_obuf) != 0) {
+			diag_log();
+			rlist_del_entry(con, tx.in_shutdown_list);
+			continue;
+		}
+		tx_push(con);
+	}
+	while (!rlist_empty(&shutdown_list))
+		fiber_cond_wait(&shutdown_list_empty_cond);
+	return 0;
+}
+
 static inline void
 iproto_thread_init_routes(struct iproto_thread *iproto_thread)
 {
@@ -2914,6 +2996,10 @@ iproto_init(int threads_count)
 	}
 
 	session_vtab_registry[SESSION_TYPE_BINARY] = iproto_session_vtab;
+
+	fiber_cond_create(&shutdown_list_empty_cond);
+	if (box_on_shutdown(NULL, iproto_on_shutdown_f, NULL) != 0)
+		panic("failed to set iproto shutdown trigger");
 	return;
 
 fail:
@@ -3063,7 +3149,7 @@ iproto_do_cfg_crit(struct iproto_thread *iproto_thread,
 	assert(rc == 0);
 }
 
-static inline void
+static void
 iproto_send_stop_msg(void)
 {
 	struct iproto_cfg_msg cfg_msg;
@@ -3072,7 +3158,7 @@ iproto_send_stop_msg(void)
 		iproto_do_cfg_crit(&iproto_threads[i], &cfg_msg);
 }
 
-static inline int
+static int
 iproto_send_listen_msg(struct evio_service *binary)
 {
 	struct iproto_cfg_msg cfg_msg;

src/box/iproto_constants.h

@@ -282,6 +282,17 @@ enum iproto_type {
 	IPROTO_WATCH = 74,
 	IPROTO_UNWATCH = 75,
 	IPROTO_EVENT = 76,
+	/**
+	 * Special request type sent by a server to all clients that support
+	 * the graceful shutdown feature to notify them that the server is
+	 * about to be shut down. The server won't exit until all the clients
+	 * close connections or a timeout occurs.
+	 *
+	 * The request type is asynchronous (doesn't have a sync number):
+	 * a client isn't supposed to reply to it; instead it's expected to
+	 * close the connection as soon as possible.
+	 */
+	IPROTO_SHUTDOWN = 77,
 
 	/** Vinyl run info stored in .index file */
 	VY_INDEX_RUN_INFO = 100,

src/box/iproto_features.c

@@ -66,4 +66,6 @@ iproto_features_init(void)
 			    IPROTO_FEATURE_ERROR_EXTENSION);
 	iproto_features_set(&IPROTO_CURRENT_FEATURES,
 			    IPROTO_FEATURE_WATCHERS);
+	iproto_features_set(&IPROTO_CURRENT_FEATURES,
+			    IPROTO_FEATURE_GRACEFUL_SHUTDOWN);
 }

src/box/iproto_features.h

@@ -45,6 +45,15 @@ enum iproto_feature_id {
 	 * IPROTO_WATCH, IPROTO_UNWATCH, IPROTO_EVENT commands.
 	 */
 	IPROTO_FEATURE_WATCHERS = 3,
+	/**
+	 * Graceful shutdown support: IPROTO_SHUTDOWN command.
+	 *
+	 * Upon receiving a shutdown request (os.exit or SIGTERM), a server
+	 * will send an IPROTO_SHUTDOWN packet to all clients that support
+	 * the graceful shutdown feature and won't exit until all of them
+	 * close connections or a timeout occurs.
+	 */
+	IPROTO_FEATURE_GRACEFUL_SHUTDOWN = 4,
 	iproto_feature_id_MAX,
 };
 
@@ -60,7 +69,7 @@ struct iproto_features {
  * It should be incremented every time a new feature is added or removed.
  */
 enum {
-	IPROTO_CURRENT_VERSION = 3,
+	IPROTO_CURRENT_VERSION = 4,
 };
 
 /**

src/box/lua/net_box.lua

@@ -53,6 +53,7 @@ local IPROTO_FEATURE_NAMES = {
     [1]     = 'transactions',
     [2]     = 'error_extension',
     [3]     = 'watchers',
+    [4]     = 'graceful_shutdown',
 }
 
 -- Given an array of IPROTO feature ids, returns a map {feature_name: bool}.

src/box/xrow.c

@@ -798,6 +798,37 @@ iproto_send_event(struct obuf *out, const char *key, size_t key_len,
 	return 0;
 }
 
+int
+iproto_send_shutdown(struct obuf *out)
+{
+	/*
+	 * Calculate the packet size.
+	 * Note: no body; no sync or schema version.
+	 */
+	size_t size = 5;
+	size += mp_sizeof_map(1);
+	size += mp_sizeof_uint(IPROTO_REQUEST_TYPE);
+	size += mp_sizeof_uint(IPROTO_SHUTDOWN);
+	/* Encode the packet. */
+	char *buf = obuf_alloc(out, size);
+	if (buf == NULL) {
+		diag_set(OutOfMemory, size, "obuf_alloc", "buf");
+		return -1;
+	}
+	char *p = buf;
+	/* Fix header. */
+	*(p++) = 0xce;
+	mp_store_u32(p, size - 5);
+	p += 4;
+	/* Packet header. */
+	p = mp_encode_map(p, 1);
+	p = mp_encode_uint(p, IPROTO_REQUEST_TYPE);
+	p = mp_encode_uint(p, IPROTO_SHUTDOWN);
+	assert(size == (size_t)(p - buf));
+	(void)p;
+	return 0;
+}
+
 int
 xrow_decode_dml(struct xrow_header *row, struct request *request,
 		uint64_t key_map)

src/box/xrow.h

@@ -803,6 +803,13 @@ int
 iproto_send_event(struct obuf *out, const char *key, size_t key_len,
 		  const char *data, const char *data_end);
 
+/**
+ * Encode IPROTO_SHUTDOWN packet.
+ * @param out Encode to.
+ */
+int
+iproto_send_shutdown(struct obuf *out);
+
 /** Write error directly to a socket. */
 void
 iproto_do_write_error(struct iostream *io, const struct error *e,

src/lib/core/evio.c

@@ -485,6 +485,8 @@ evio_service_attach(struct evio_service *dst, const struct evio_service *src)
 void
 evio_service_detach(struct evio_service *service)
 {
+	if (service->entries == NULL)
+		return;
 	for (int i = 0; i < service->entry_count; i++)
 		evio_service_entry_detach(&service->entries[i]);
 	free(service->entries);
@@ -515,6 +517,8 @@ evio_service_listen(struct evio_service *service)
 void
 evio_service_stop(struct evio_service *service)
 {
+	if (service->entries == NULL)
+		return;
 	say_info("%s: stopped", evio_service_name(service));
 	for (int i = 0; i < service->entry_count; i++)
 		evio_service_entry_stop(&service->entries[i]);

test/box-py/iproto.result

@@ -208,9 +208,9 @@ Invalid MsgPack - request body
 # Invalid features
 Invalid MsgPack - request body
 # Empty request body
-version=3, features=[0, 1, 2, 3]
+version=4, features=[0, 1, 2, 3, 4]
 # Unknown version and features
-version=3, features=[0, 1, 2, 3]
+version=4, features=[0, 1, 2, 3, 4]
 
 #
 # gh-6257 Watchers

test/box/net.box_iproto_id.result

@@ -15,14 +15,15 @@ c = net.connect(box.cfg.listen)
  | ...
 c.peer_protocol_version
  | ---
- | - 3
+ | - 4
  | ...
 c.peer_protocol_features
  | ---
  | - transactions: true
  |   watchers: true
  |   error_extension: true
  |   streams: true
+ |   graceful_shutdown: true
  | ...
 c:close()
  | ---
@@ -50,6 +51,7 @@ c.peer_protocol_features
  |   watchers: false
  |   error_extension: false
  |   streams: false
+ |   graceful_shutdown: false
  | ...
 errinj.set('ERRINJ_IPROTO_DISABLE_ID', false)
  | ---
@@ -98,6 +100,7 @@ c.peer_protocol_features
  |   watchers: true
  |   error_extension: true
  |   streams: true
+ |   graceful_shutdown: true
  | ...
 c:close()
  | ---
@@ -143,14 +146,15 @@ c.error -- error
  | ...
 c.peer_protocol_version
  | ---
- | - 3
+ | - 4
  | ...
 c.peer_protocol_features
  | ---
  | - transactions: false
  |   watchers: true
  |   error_extension: true
  |   streams: true
+ |   graceful_shutdown: true
  | ...
 c:close()
  | ---
@@ -169,14 +173,15 @@ c.error -- error
  | ...
 c.peer_protocol_version
  | ---
- | - 3
+ | - 4
  | ...
 c.peer_protocol_features
  | ---
  | - transactions: true
  |   watchers: true
  |   error_extension: true
  |   streams: true
+ |   graceful_shutdown: true
  | ...
 c:close()
  | ---