OF-2505: BOSH: add back-pressure to network IO when processing can't keep up

i18n/src/main/resources/openfire_i18n.properties

@@ -1769,6 +1769,8 @@ system_property.xmpp.httpbind.worker.threads-min=Minimum amount of threads used
 system_property.xmpp.httpbind.worker.threads=Maximum number of threads used to process incoming BOSH data. Defaults to the same amount of threads as what's used for non-BOSH/TCP-connected XMPP clients. Note: Apart from the processing threads configured in this class, the server also uses a thread pool to perform the network IO (see property 'httpbind.client.processing.threads'). BOSH installations expecting heavy loads may want to allocate additional threads to this worker pool to ensure timely delivery of inbound packets.
 system_property.xmpp.httpbind.worker.timeout=Duration that unused, surplus threads that once processed BOSH data are kept alive. See also property 'httpbind.client.processing.threads-timeout'
 system_property.xmpp.httpbind.worker.cleanupcheck=Interval in which a check is executed that will cleanup unused/inactive BOSH sessions.
+system_property.xmpp.httpbind.worker.backpressure-enabled=When BOSH/websocket network IO outpaces processing of data, enabling back pressure will cause the network IO to block when the processing queue is full.
+system_property.xmpp.httpbind.worker.queue-capacity=Maximum size of the queue that holds tasks to be processed.
 
 system_property.xmpp.jmx.enabled=Enables / disables JMX support in Openfire
 system_property.xmpp.jmx.secure=Controls if the JMX connector is configured to require Openfire admin credentials.

xmppserver/src/main/java/org/jivesoftware/openfire/http/HttpSessionManager.java

@@ -27,10 +27,7 @@
 import org.jivesoftware.openfire.mbean.ThreadPoolExecutorDelegate;
 import org.jivesoftware.openfire.mbean.ThreadPoolExecutorDelegateMBean;
 import org.jivesoftware.openfire.session.ConnectionSettings;
-import org.jivesoftware.util.JiveGlobals;
-import org.jivesoftware.util.NamedThreadFactory;
-import org.jivesoftware.util.SystemProperty;
-import org.jivesoftware.util.TaskEngine;
+import org.jivesoftware.util.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -44,10 +41,7 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.TimerTask;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 
 /**
  * Manages sessions for all users connecting to Openfire using the HTTP binding protocol,
@@ -95,9 +89,17 @@ public void start() {
 
         this.sessionManager = SessionManager.getInstance();
 
+        final BlockingQueue<Runnable> queue;
+        if (BACKPRESSURE_ENABLED.getValue()) {
+            // Do _not_ use  ThreadPoolExecutor.CallerRunsPolicy, as that can mess up the order in which stanzas are processed!
+            queue = new BlocksOnOfferQueue<>(QUEUE_CAPACITY.getValue());
+        }
+        else {
+            queue = new LinkedBlockingQueue<>(QUEUE_CAPACITY.getValue());
+        }
+
         stanzaWorkerPool = new ThreadPoolExecutor(MIN_POOL_SIZE.getValue(), MAX_POOL_SIZE.getValue(), POOL_KEEP_ALIVE.getValue().getSeconds(), TimeUnit.SECONDS,
-                new LinkedBlockingQueue<>(), // unbounded task queue
-                new NamedThreadFactory( "httpbind-worker-", true, null, Thread.currentThread().getThreadGroup(), null )
+            queue, new NamedThreadFactory("httpbind-worker-", true, null, Thread.currentThread().getThreadGroup(), null)
         );
 
         if (JMXManager.isEnabled()) {
@@ -141,6 +143,27 @@ public void start() {
         .setDynamic(false)
         .build();
 
+    /**
+     * Maximum amount of tasks (roughly: stanzas) that can be queued for processing, before the processing executor
+     * will start to invoke the rejection policy.
+     */
+    public static SystemProperty<Integer> QUEUE_CAPACITY = SystemProperty.Builder.ofType(Integer.class)
+        .setKey("xmpp.httpbind.worker.queue-capacity")
+        .setDynamic(false)
+        .setDefaultValue(Integer.MAX_VALUE)
+        .setMinValue(1)
+        .build();
+
+    /**
+     * Backpressure slows down the provider of tasks when the queue is full, by making the threads invoking
+     * {@link #execute(Runnable)} block until there is room again in the queue.
+     */
+    public static SystemProperty<Boolean> BACKPRESSURE_ENABLED = SystemProperty.Builder.ofType(Boolean.class)
+        .setKey("xmpp.httpbind.worker.backpressure-enabled")
+        .setDynamic(false)
+        .setDefaultValue(false)
+        .build();
+
     /**
      * Interval in which a check is executed that will cleanup unused/inactive BOSH sessions.
      */

xmppserver/src/main/java/org/jivesoftware/openfire/util/ThreadPoolExecutorRejectionPolicy.java

@@ -0,0 +1,29 @@
+package org.jivesoftware.openfire.http;
+
+/**
+ * Enumerates frequently used RejectedExecutionHandler implementations.
+ *
+ * @author Guus der Kinderen, [email protected]
+ */
+public enum ThreadPoolExecutorRejectionPolicy
+{
+    /**
+     * Indicates desired usage of {@link java.util.concurrent.ThreadPoolExecutor.DiscardOldestPolicy}.
+     */
+    DiscardOldestPolicy,
+
+    /**
+     * Indicates desired usage of {@link java.util.concurrent.ThreadPoolExecutor.AbortPolicy}.
+     */
+    AbortPolicy,
+
+    /**
+     * Indicates desired usage of {@link java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy}.
+     */
+    CallerRunsPolicy,
+
+    /**
+     * Indicates desired usage of {@link java.util.concurrent.ThreadPoolExecutor.DiscardPolicy}.
+     */
+    DiscardPolicy
+}

xmppserver/src/main/java/org/jivesoftware/util/BlocksOnOfferQueue.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright (C) 2022 Ignite Realtime Foundation. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.jivesoftware.util;
+
+import javax.annotation.Nonnull;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A LinkedBlockingQueue of which the {@link #offer(Object)} method blocks, instead of immediately returning.
+ *
+ * This class is designed to be used as a queue for ThreadPoolExecutors that wish to slow down the producing threads
+ * when the queue is reaching capacity, and where {@link java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy}
+ * cannot be used (for example, because it would cause rejected tasks to be executed _before_ already queued tasks).
+ *
+ * Note that the lock used to guard access in {@link LinkedBlockingQueue#offer(Object, long, TimeUnit)}, which is used
+ * by this implementation of {@link #offer(Object)}, uses an unfair lock. No strict ordering of execution of produced
+ * tasks can be guaranteed.
+ *
+ * @param <E>
+ * @author Guus der Kinderen, [email protected]
+ */
+public class BlocksOnOfferQueue<E> extends LinkedBlockingQueue<E>
+{
+    public BlocksOnOfferQueue(int capacity) {
+        super(capacity);
+    }
+
+    @Override
+    public boolean offer(@Nonnull E e) {
+        try {
+            return super.offer(e, 999, TimeUnit.DAYS); // 'indefinitely'.
+        } catch (InterruptedException ex) {
+            return false;
+        }
+    }
+}