MEDIUM: servers: Add strict-maxconn.

Maxconn is a bit of a misnomer when it comes to servers, as it doesn't
control the maximum number of connections we establish to a server, but
the maximum number of simultaneous requests. So add "strict-maxconn",
that will make it so we will never establish more connections than
maxconn.
It extends the meaning of the "restricted" setting of
tune.takeover-other-tg-connections, as it will also attempt to get idle
connections from other thread groups if strict-maxconn is set.

doc/configuration.txt

@@ -4591,8 +4591,9 @@ tune.takeover-other-tg-connections <value>
   default, if used, no attempt will be made to use idle connections from
   other thread groups, "restricted" where we will only attempt to get an idle
   connection from another thread if we're using protocols that can't create
-  new connections, such as reverse http, and "full" where we will always look
-  in other thread groups for idle connections.
+  new connections, such as reverse http, as well as when using strict-maxconn,
+  and "full" where we will always look in other thread groups for idle
+  connections.
   Note that using connections from other thread groups can occur performance
   penalties, so it should not be used unless really needed.
 
@@ -19106,6 +19107,20 @@ stick
   It may also be used as "default-server" setting to reset any previous
   "default-server" "non-stick" setting.
 
+strict-maxconn
+  May be used in the following contexts: tcp, http
+
+  maxconn to servers is a bit of a misnomer, it actually configure the maximum
+  number of requests we send to a server, but with idle connections, we may
+  have more total connections to the server. If a strict limit of connections
+  to a server is required, then adding strict-maxconn can be used. We will
+  then never establish more connections to a server than maxconn, and try to
+  reuse or kill connections if needed. Please note, however, than it may lead
+  to failed requests in case we can't establish a new connection, and no
+  idle connection is available. This can happen when 'private" connections
+  are established, connections tied only to a session, because authentication
+  happened.
+
 socks4 <addr>:<port>
   May be used in the following contexts: tcp, http, log, peers, ring
 

include/haproxy/server-t.h

@@ -170,6 +170,7 @@ enum srv_init_state {
 #define SRV_F_NON_PURGEABLE 0x2000       /* this server cannot be removed at runtime */
 #define SRV_F_DEFSRV_USE_SSL 0x4000      /* default-server uses SSL */
 #define SRV_F_DELETED 0x8000             /* srv is deleted but not yet purged */
+#define SRV_F_STRICT_MAXCONN 0x10000     /* maxconn is to be strictly enforced, as a limit of outbound connections */
 
 /* configured server options for send-proxy (server->pp_opts) */
 #define SRV_PP_V1               0x0001   /* proxy protocol version 1 */
@@ -366,6 +367,7 @@ struct server {
 	unsigned int curr_idle_nb;              /* Current number of connections in the idle list */
 	unsigned int curr_safe_nb;              /* Current number of connections in the safe list */
 	unsigned int curr_used_conns;           /* Current number of used connections */
+	unsigned int curr_total_conns;          /* Current number of total connections to the server, used or idle, only calculated if strict-maxconn is used */
 	unsigned int max_used_conns;            /* Max number of used connections (the counter is reset at each connection purges */
 	unsigned int est_need_conns;            /* Estimate on the number of needed connections (max of curr and previous max_used) */
 

src/backend.c

@@ -1362,7 +1362,7 @@ struct connection *conn_backend_get(struct stream *s, struct server *srv, int is
 
 	if (!found && (global.tune.tg_takeover == FULL_THREADGROUP_TAKEOVER ||
 	    (global.tune.tg_takeover == RESTRICTED_THREADGROUP_TAKEOVER &&
-	    srv->flags & SRV_F_RHTTP))) {
+	    srv->flags & (SRV_F_RHTTP | SRV_F_STRICT_MAXCONN)))) {
 		curtgid = curtgid + 1;
 		if (curtgid == global.nbtgroups + 1)
 			curtgid = 1;
@@ -1439,6 +1439,83 @@ static int do_connect_server(struct stream *s, struct connection *conn)
 	return ret;
 }
 
+/*
+ * Returns the first connection from a tree we managed to take over,
+ * if any.
+ */
+static struct connection *
+takeover_random_idle_conn(struct eb_root *root, int curtid)
+{
+	struct conn_hash_node *hash_node;
+	struct connection *conn = NULL;
+	struct eb64_node *node = eb64_first(root);
+
+	while (node) {
+		hash_node = eb64_entry(node, struct conn_hash_node, node);
+		conn = hash_node->conn;
+		if (conn && conn->mux->takeover && conn->mux->takeover(conn, curtid, 0) == 0) {
+			conn_delete_from_tree(conn);
+			return conn;
+		}
+		node = eb64_next(node);
+	}
+
+	return NULL;
+}
+
+/*
+ * Kills an idle connection, any idle connection we can get a hold on.
+ * The goal is just to free a connection in case we reached the max and
+ * have to establish a new one.
+ * Returns -1 if there is no idle connection to kill, 0 if there are some
+ * available but we failed to get one, and 1 if we successfully killed one.
+ */
+static int
+kill_random_idle_conn(struct server *srv)
+{
+	struct connection *conn = NULL;
+	int i;
+	int curtid;
+	/* No idle conn, then there is nothing we can do at this point */
+
+	if (srv->curr_idle_conns == 0)
+		return -1;
+	for (i = 0; i < global.nbthread; i++) {
+		curtid = (i + tid) % global.nbthread;
+
+		if (HA_SPIN_TRYLOCK(IDLE_CONNS_LOCK, &idle_conns[curtid].idle_conns_lock) != 0)
+			continue;
+		conn = takeover_random_idle_conn(&srv->per_thr[curtid].idle_conns, curtid);
+		if (!conn)
+			conn = takeover_random_idle_conn(&srv->per_thr[curtid].safe_conns, curtid);
+		HA_SPIN_UNLOCK(IDLE_CONNS_LOCK, &idle_conns[curtid].idle_conns_lock);
+		if (conn)
+			break;
+	}
+	if (conn) {
+		/*
+		 * We have to manually decrement counters, as srv_release_conn()
+		 * will attempt to access the current tid's counters, while
+		 * we may have taken the connection from a different thread.
+		 */
+		if (conn->flags & CO_FL_LIST_MASK) {
+			_HA_ATOMIC_DEC(&srv->curr_idle_conns);
+			_HA_ATOMIC_DEC(conn->flags & CO_FL_SAFE_LIST ? &srv->curr_safe_nb : &srv->curr_idle_nb);
+			_HA_ATOMIC_DEC(&srv->curr_idle_thr[curtid]);
+			conn->flags &= ~CO_FL_LIST_MASK;
+			/*
+			 * If we have no list flag then srv_release_conn()
+			 * will consider the connection is used, so let's
+			 * pretend it is.
+			 */
+			_HA_ATOMIC_INC(&srv->curr_used_conns);
+		}
+		conn->mux->destroy(conn->ctx);
+		return 1;
+	}
+	return 0;
+}
+
 /*
  * This function initiates a connection to the server assigned to this stream
  * (s->target, (s->scb)->addr.to). It will assign a server if none
@@ -1728,12 +1805,49 @@ int connect_server(struct stream *s)
 skip_reuse:
 	/* no reuse or failed to reuse the connection above, pick a new one */
 	if (!srv_conn) {
+		unsigned int total_conns;
+
 		if (srv && (srv->flags & SRV_F_RHTTP)) {
 			DBG_TRACE_USER("cannot open a new connection for reverse server", STRM_EV_STRM_PROC|STRM_EV_CS_ST, s);
 			s->conn_err_type = STRM_ET_CONN_ERR;
 			return SF_ERR_INTERNAL;
 		}
 
+		if (srv && (srv->flags & SRV_F_STRICT_MAXCONN)) {
+			int kill_tries = 0;
+			/*
+			 * Before creating a new connection, make sure we still
+			 * have a slot for that
+			 */
+			total_conns = srv->curr_total_conns;
+
+			while (1) {
+				if (total_conns < srv->maxconn) {
+					if (_HA_ATOMIC_CAS(&srv->curr_total_conns,
+					    &total_conns, total_conns + 1))
+						break;
+					__ha_cpu_relax();
+				} else {
+					int ret = kill_random_idle_conn(srv);
+
+					/*
+					 * There is no idle connection to kill
+					 * so there is nothing we can do at
+					 * that point but to report an
+					 * error.
+					 */
+					if (ret == -1)
+						return SF_ERR_RESOURCE;
+					kill_tries++;
+					/*
+					 * We tried 3 times to kill an idle
+					 * connection, we failed, give up now.
+					 */
+					if (ret == 0 && kill_tries == 3)
+						return SF_ERR_RESOURCE;
+				}
+			}
+		}
 		srv_conn = conn_new(s->target);
 		if (srv_conn) {
 			DBG_TRACE_STATE("alloc new be connection", STRM_EV_STRM_PROC|STRM_EV_CS_ST, s);
@@ -1762,7 +1876,8 @@ int connect_server(struct stream *s)
 			}
 
 			srv_conn->hash_node->node.key = hash;
-		}
+		} else if (srv && (srv->flags & SRV_F_STRICT_MAXCONN))
+			_HA_ATOMIC_DEC(&srv->curr_total_conns);
 	}
 
 	/* if bind_addr is non NULL free it */

src/connection.c

@@ -502,10 +502,15 @@ static void conn_backend_deinit(struct connection *conn)
 	if (LIST_INLIST(&conn->sess_el))
 		session_unown_conn(conn->owner, conn);
 
-	/* If the connection is not private, it is accounted by the server. */
-	if (!(conn->flags & CO_FL_PRIVATE)) {
-		if (obj_type(conn->target) == OBJ_TYPE_SERVER)
-			srv_release_conn(__objt_server(conn->target), conn);
+	if (obj_type(conn->target) == OBJ_TYPE_SERVER) {
+		struct server *srv = __objt_server(conn->target);
+
+		/* If the connection is not private, it is accounted by the server. */
+		if (!(conn->flags & CO_FL_PRIVATE)) {
+			srv_release_conn(srv, conn);
+		}
+		if (srv->flags & SRV_F_STRICT_MAXCONN)
+			_HA_ATOMIC_DEC(&srv->curr_total_conns);
 	}
 
 	/* Make sure the connection is not left in the idle connection tree */

src/server.c

@@ -1997,6 +1997,13 @@ static int srv_parse_weight(char **args, int *cur_arg, struct proxy *px, struct
 	return 0;
 }
 
+static int srv_parse_strict_maxconn(char **args, int *cur_arg, struct proxy *px, struct server *newsrv, char **err)
+{
+
+	newsrv->flags |= SRV_F_STRICT_MAXCONN;
+
+	return 0;
+}
 /* Returns 1 if the server has streams pointing to it, and 0 otherwise.
  *
  * Must be called with the server lock held.
@@ -2397,6 +2404,7 @@ static struct srv_kw_list srv_kws = { "ALL", { }, {
 	{ "slowstart",            srv_parse_slowstart,            1,  1,  1 }, /* Set the warm-up timer for a previously failed server */
 	{ "source",               srv_parse_source,              -1,  1,  1 }, /* Set the source address to be used to connect to the server */
 	{ "stick",                srv_parse_stick,                0,  1,  0 }, /* Enable stick-table persistence */
+	{ "strict-maxconn",       srv_parse_strict_maxconn,       0,  1,  1 }, /* Strictly enforces maxconn */
 	{ "tfo",                  srv_parse_tfo,                  0,  1,  1 }, /* enable TCP Fast Open of server */
 	{ "track",                srv_parse_track,                1,  1,  1 }, /* Set the current state of the server, tracking another one */
 	{ "socks4",               srv_parse_socks4,               1,  1,  0 }, /* Set the socks4 proxy of the server*/

src/stream.c

@@ -634,9 +634,16 @@ void stream_free(struct stream *s)
 		 * it should normally be only the same as the one above,
 		 * so this should not happen in fact.
 		 */
-		sess_change_server(s, NULL);
-		if (may_dequeue_tasks(oldsrv, s->be))
-			process_srv_queue(oldsrv);
+		/*
+		 * We don't want to release the slot just yet
+		 * if we're using strict-maxconn, we want to
+		 * free the connection before.
+		 */
+		if (!(oldsrv->flags & SRV_F_STRICT_MAXCONN)) {
+			sess_change_server(s, NULL);
+			if (may_dequeue_tasks(oldsrv, s->be))
+				process_srv_queue(oldsrv);
+		}
 	}
 
 	/* We may still be present in the buffer wait queue */
@@ -729,6 +736,20 @@ void stream_free(struct stream *s)
 
 	sc_destroy(s->scb);
 	sc_destroy(s->scf);
+	/*
+	 * Now we've free'd the connection, so if we're running with
+	 * strict-maxconn, now is a good time to free the slot, and see
+	 * if we can dequeue anything.
+	 */
+	if (s->srv_conn && (s->srv_conn->flags & SRV_F_STRICT_MAXCONN)) {
+		struct server *oldsrv = s->srv_conn;
+
+		if ((oldsrv->flags & SRV_F_STRICT_MAXCONN)) {
+			sess_change_server(s, NULL);
+			if (may_dequeue_tasks(oldsrv, s->be))
+				process_srv_queue(oldsrv);
+		}
+	}
 
 	pool_free(pool_head_stream, s);
 
@@ -1929,9 +1950,16 @@ struct task *process_stream(struct task *t, void *context, unsigned int state)
 				s->flags &= ~SF_CURR_SESS;
 				_HA_ATOMIC_DEC(&srv->cur_sess);
 			}
-			sess_change_server(s, NULL);
-			if (may_dequeue_tasks(srv, s->be))
-				process_srv_queue(srv);
+			/*
+			 * We don't want to release the slot just yet
+			 * if we're using strict-maxconn, we want to
+			 * free the connection before.
+			 */
+			if (!(srv->flags & SRV_F_STRICT_MAXCONN)) {
+				sess_change_server(s, NULL);
+				if (may_dequeue_tasks(srv, s->be))
+					process_srv_queue(srv);
+			}
 		}
 
 		/* This is needed only when debugging is enabled, to indicate