MINOR: servers: Introduce src_manage_queues()

Instead of having a common pattern of
	if (may_dequeue_tasks())
		process_srv_queue()
introduce an inline function to do that, and use it where appropriate
(mostly everywhere where the pattern was used).
Later on this function will be responsible for figuring out if, after
the queue has been processed, the server is full or not.

include/haproxy/server.h

@@ -30,6 +30,7 @@
 #include <haproxy/applet-t.h>
 #include <haproxy/arg-t.h>
 #include <haproxy/freq_ctr.h>
+#include <haproxy/queue.h>
 #include <haproxy/proxy-t.h>
 #include <haproxy/resolvers-t.h>
 #include <haproxy/sample-t.h>
@@ -390,6 +391,19 @@ static inline int srv_is_quic(const struct server *srv)
 	       srv->addr_type.xprt_type == PROTO_TYPE_STREAM;
 }
 
+/*
+ * Returns 1 if the server is full, 0 if it is not, and -1 if unknown
+ */
+static inline int srv_manage_queues(struct server *srv, struct proxy *px)
+{
+	int full = -1;
+
+	if (may_dequeue_tasks(srv, px))
+		full = process_srv_queue(srv);
+
+	return full;
+}
+
 #endif /* _HAPROXY_SERVER_H */
 
 /*

include/haproxy/stream.h

@@ -354,8 +354,7 @@ static inline void stream_choose_redispatch(struct stream *s)
 	     (!(s->flags & SF_DIRECT) && s->be->srv_act > 1 &&
 	      ((s->be->lbprm.algo & BE_LB_KIND) != BE_LB_KIND_HI)))) {
 		sess_change_server(s, NULL);
-		if (may_dequeue_tasks(objt_server(s->target), s->be))
-			process_srv_queue(objt_server(s->target));
+		srv_manage_queues(objt_server(s->target), s->be);
 
 		sockaddr_free(&s->scb->dst);
 		s->flags &= ~(SF_DIRECT | SF_ASSIGNED);

src/backend.c

@@ -860,8 +860,7 @@ out_ok:
 		if (conn_slot == srv) {
 			sess_change_server(s, srv);
 		} else {
-			if (may_dequeue_tasks(conn_slot, s->be))
-				process_srv_queue(conn_slot);
+			srv_manage_queues(conn_slot, s->be);
 		}
 	}
 
@@ -2400,8 +2399,8 @@ int srv_redispatch_connect(struct stream *s)
 		 * Not needed for backend queues, already handled in
 		 * assign_server_and_queue().
 		 */
-		if (unlikely(srv && may_dequeue_tasks(srv, s->be)))
-			process_srv_queue(srv);
+		if (unlikely(srv))
+		       srv_manage_queues(srv, s->be);
 
 		return 1;
 
@@ -2421,8 +2420,7 @@ int srv_redispatch_connect(struct stream *s)
 			_HA_ATOMIC_INC(&s->be_tgcounters->failed_conns);
 
 		/* release other streams waiting for this server */
-		if (may_dequeue_tasks(srv, s->be))
-			process_srv_queue(srv);
+		srv_manage_queues(srv, s->be);
 		return 1;
 	}
 	/* if we get here, it's because we got SRV_STATUS_OK, which also
@@ -2498,8 +2496,7 @@ void back_try_conn_req(struct stream *s)
 
 			/* release other streams waiting for this server */
 			sess_change_server(s, NULL);
-			if (may_dequeue_tasks(srv, s->be))
-				process_srv_queue(srv);
+			srv_manage_queues(srv, s->be);
 
 			/* Failed and not retryable. */
 			sc_abort(sc);
@@ -2816,8 +2813,7 @@ void back_handle_st_cer(struct stream *s)
 		if (s->be_tgcounters)
 			_HA_ATOMIC_INC(&s->be_tgcounters->failed_conns);
 		sess_change_server(s, NULL);
-		if (may_dequeue_tasks(objt_server(s->target), s->be))
-			process_srv_queue(objt_server(s->target));
+		srv_manage_queues(objt_server(s->target), s->be);
 
 		/* shutw is enough to stop a connecting socket */
 		sc_shutdown(sc);
@@ -2850,8 +2846,7 @@ void back_handle_st_cer(struct stream *s)
 		if (s->be_tgcounters)
 			_HA_ATOMIC_INC(&s->be_tgcounters->internal_errors);
 		sess_change_server(s, NULL);
-		if (may_dequeue_tasks(objt_server(s->target), s->be))
-			process_srv_queue(objt_server(s->target));
+		srv_manage_queues(objt_server(s->target), s->be);
 
 		/* shutw is enough to stop a connecting socket */
 		sc_shutdown(sc);

src/cli.c

@@ -3534,8 +3534,7 @@ int pcli_wait_for_response(struct stream *s, struct channel *rep, int an_bit)
 
 		stream_del_srv_conn(s);
 		if (objt_server(s->target)) {
-			if (may_dequeue_tasks(__objt_server(s->target), be))
-				process_srv_queue(__objt_server(s->target));
+			srv_manage_queues(__objt_server(s->target), be);
 		}
 
 		s->target = NULL;

src/server.c

@@ -2597,9 +2597,7 @@ const char *server_parse_maxconn_change_request(struct server *sv,
 		sv->maxconn = v;
 	}
 
-	if (may_dequeue_tasks(sv, sv->proxy))
-		process_srv_queue(sv);
-
+	srv_manage_queues(sv, sv->proxy);
 	return NULL;
 }
 

src/stream.c

@@ -630,8 +630,7 @@ void stream_free(struct stream *s)
 	pendconn_free(s);
 
 	if (objt_server(s->target)) { /* there may be requests left pending in queue */
-		if (may_dequeue_tasks(__objt_server(s->target), s->be))
-			process_srv_queue(__objt_server(s->target));
+		srv_manage_queues(__objt_server(s->target), s->be);
 	}
 
 	if (unlikely(s->srv_conn)) {
@@ -647,8 +646,7 @@ void stream_free(struct stream *s)
 		 */
 		if (!(oldsrv->flags & SRV_F_STRICT_MAXCONN)) {
 			sess_change_server(s, NULL);
-			if (may_dequeue_tasks(oldsrv, s->be))
-				process_srv_queue(oldsrv);
+			srv_manage_queues(oldsrv, s->be);
 		}
 	}
 
@@ -758,8 +756,7 @@ void stream_free(struct stream *s)
 
 		if ((oldsrv->flags & SRV_F_STRICT_MAXCONN)) {
 			sess_change_server(s, NULL);
-			if (may_dequeue_tasks(oldsrv, s->be))
-				process_srv_queue(oldsrv);
+			srv_manage_queues(oldsrv, s->be);
 		}
 	}
 
@@ -2060,8 +2057,7 @@ struct task *process_stream(struct task *t, void *context, unsigned int state)
 			 */
 			if (!(srv->flags & SRV_F_STRICT_MAXCONN)) {
 				sess_change_server(s, NULL);
-				if (may_dequeue_tasks(srv, s->be))
-					process_srv_queue(srv);
+				srv_manage_queues(srv, s->be);
 			}
 		}
 
@@ -2975,8 +2971,7 @@ void stream_shutdown_self(struct stream *stream, int why)
 
 	if (objt_server(stream->target)) {
 		sess_change_server(stream, NULL);
-		if (may_dequeue_tasks(objt_server(stream->target), stream->be))
-			process_srv_queue(objt_server(stream->target));
+		srv_manage_queues(__objt_server(stream->target), stream->be);
 	}
 
 	/* shutw is enough to stop a connecting socket */