MEDIUM: sink: "max-reuse" support for sink servers

Thanks to the previous commit, it is now possible to know how many events
were processed for a given sft/server sink pair. As mentioned in commit
c454296 ("OPTIM: sink: balance applets accross threads"), let's provide
the ability to restart a server connection when a certain amount of events
were processed to help better balance the load over multiple threads.

For this, we make use the of "max-reuse" server keyword which was only
relevant under "http" context so far. Under sink context, "max-reuse"
corresponds to the number of times the tcp connection can be reused
for sending messages, which in fact means that "max-reuse + 1" is the
number of events (ie: messages) that are allowed to be sent using the
same tcp server connection: when this threshold is met, the connection
will be destroyed and a new one will be created on a random thread.
The value is not strict: it is the minimum value above which the
connection may be destroyed since the value is checked after
ring_dispatch_messages() which may process multiple messages at once.

By default, no limit is enforced (the connection will be reused for as
long as it is available).

The documentation was updated accordingly.

doc/configuration.txt

@@ -17449,7 +17449,9 @@ maxqueue <maxqueue>
   and "balance leastconn".
 
 max-reuse <count>
-  May be used in the following contexts: http
+  May be used in the following contexts: http, ring
+
+  When used under http context:
 
   The "max-reuse" argument indicates the HTTP connection processors that they
   should not reuse a server connection more than this number of times to send
@@ -17460,6 +17462,32 @@ max-reuse <count>
   layers as there might be technical limitations making it impossible to
   enforce. At least HTTP/2 connections to servers will respect it.
 
+  When used under ring context:
+
+  The "max-reuse" argument indicates that the sink TCP connection processors
+  that they should not reuse a server connection more than this number of
+  times to send messages. It means that the connection to the server will be
+  forcefully destroyed once at least "max-reuse + 1" messages were handled on
+  the same connection. The connection to the server will then be automatically
+  re-created. When dealing with a large amount of messages in multithreading
+  context, this can help to better distribute the ring's load over multiple
+  threads. Indeed, each connection is bound to the same CPU thread for its
+  entire duration: unlike HTTP, there is no thing like syslog transaction, so
+  the server connection could live indefinitely as long as the server doesn't
+  close the connection or no network error occurs. By destroying connections
+  from time to time we give the opportunity to other threads to pick-up some
+  messages in turn. It may also help gracefully rotate log servers in contexts
+  where there is an extra load-balancing layer between haproxy and the log
+  servers. However, keep in mind that each connection recycling will leave an
+  outgoing port in TIME_WAIT state that will not be reusable for around one
+  minute on modern operating systems, and that as such, one must be careful
+  not to use too low values to prevent rapid source port exhaustion. As a rule
+  of thumb, make sure never to close more than a few times per second, and
+  preferably much less often. Permitted values are -1 (the default), which
+  disables this limit, or any positive value. Unlike under HTTP context, when
+  used with sink servers "max-reuse" is a best-effort: ring messages are
+  batched, so the limit is checked between each batch.
+
 minconn <minconn>
   May be used in the following contexts: tcp, http
 

src/sink.c

@@ -387,6 +387,24 @@ static void _sink_forward_io_handler(struct appctx *appctx,
 	                             msg_handler, &processed);
 	sft->e_processed += processed;
 
+	/* if server's max-reuse is set (>= 0), destroy the applet once the
+	 * connection has been reused at least 'max-reuse' times, which means
+	 * it has processed at least 'max-reuse + 1' events (applet will
+	 * perform a new connection attempt)
+	 */
+	if (sft->srv->max_reuse >= 0) {
+		uint max_reuse = sft->srv->max_reuse + 1;
+
+		if (max_reuse < sft->srv->max_reuse)
+			max_reuse = sft->srv->max_reuse; // overflow, cap to max value
+
+		if (sft->e_processed / max_reuse !=
+		    (sft->e_processed - processed) / max_reuse) {
+			HA_SPIN_UNLOCK(SFT_LOCK, &sft->lock);
+			goto soft_close;
+		}
+	}
+
 	if (ret) {
 		/* let's be woken up once new data arrive */
 		MT_LIST_APPEND(&ring->waiters, &appctx->wait_entry);