Added xmit_interval=0 to NAKACK2, disabling retransmission (https://issues.redhat.com/browse/JGRP-2675)

Added rejection_policy field to ThreadPool

Author: belaban

conf/tcp-low-latency.xml

@@ -18,7 +18,7 @@
          bind_port="${jgroups.bind_port:7800}"
 
          buffered_input_stream_size="8192"
-         buffered_output_stream_size="0"
+         buffered_output_stream_size="8192"
 
          thread_pool.enabled="false"
 

doc/design/NAKACK2_NoRetransmission.txt

@@ -0,0 +1,86 @@
+
+NAKACK2 without retransmission over TCP
+=======================================
+
+Author:    Bela Ban
+Date:      March 2023
+JIRA:      https://issues.redhat.com/browse/JGRP-2675
+Unit test: NAKACK2_NoRetransmissionTest
+
+
+Goal
+----
+When run over TCP (and TCP_NIO2, see below) as transport, NAKACK2 can disable retransmission, as this is already
+performed by TCP.
+
+Contrary to [1], which completely removes NAKACK2, this design uses the retransmit table *for the sole purpose of
+establishing ordering*, but not for retransmission.
+
+
+Implementation
+--------------
+* Seqnos are still added to messages sent down the stack
+* On the receiver, messages (or message batches) are added to the retransmit table and delivered in order of seqnos
+* Digests received by STABLE purge messages (from the retransmit table) that have been see by everyone
+* xmit_interval <= 0 disables retransmission
+  * No retransmit task is running
+  * stable() will not trigger a retransmission
+  * The last-seqno-sender task is also disabled
+
+
+become_server_queue
+-------------------
+* A joiner queues messages received before JChannel.connect() returns
+* When getting a BECOME_SERVER event, the queued messages are sent up the stack
+* The become_server_queue is bounded (default: 50), so the following can happen:
+  * C joins a cluster {A,B}. B's highest seqno is 100, sent as a digest with the view {A,B,C} to C
+  * Before C can become a server, B sends 70 messages
+  * C queues 50 of those (101-150), but drops 151-170 as the queue is bounded
+    ** Correction: BoundedList drops the oldest N messages when the capacity has been exceeded, so 101-120
+                   However, this doesn't change anything on a concetual level
+  * C now creates the retransmit window for B at 50 (next seqno to be expected: 51)
+  * When C receives the BECOME_SERVER, it'll drain the 50 messages from the queue, so B's window is at 150
+    --> Since there is not retransmission, B's messages 151-170 will not get retransmitted, so further messages sent
+        by B will be added to B's window at C, but will not get delivered as there is a gap
+    --> Eventually, C's window for B might cause an OOME, as it only grows and never shrinks
+==> Solution: when xmit_interval is <= 0, make become_server_queue *unbounded*. The number of messages received in
+    that queue should be relatively small, as the time window between sending a join request and becoming server
+    is short
+==> xmit_interval should be able to be changed dynamically
+    * Start / stop the retransmission task
+    * Make become_server_queue bounded / unbounded
+
+
+Thread pool rejection policy
+----------------------------
+* The transport's thread pool must not drop messages, as this would create a gap in the retransmit table. This is
+  required, as there is no retransmission at the NAKACK2 level.
+
+
+Summary
+-------
+This design requires that a message received over TCP *cannot be dropped before it reaches NAKACK2*. So tests such as
+LastMessageDroppedTest, which inserts DISCARD just below NAKACK2 to drop selected messages, won't pass with
+retransmission disabled. As TCP/IP considers the message delivered, if it is nevertheless dropped later on, it will
+never be retransmitted (not by TCP/IP and not by NAKACK2).
+
+
+
+
+
+
+
+
+
+
+
+
+TCP_NIO2
+--------
+* When TCP_NIO2 is used as transport, when sending a message, the gathering-write buffer 'Buffers' cannot drop the
+  message, e.g. because the max capacity has been exceeded. Investigate making Buffers unbounded.
+
+
+Links
+-----
+[1] https://issues.redhat.com/browse/JGRP-2566

src/org/jgroups/protocols/pbcast/NAKACK2.java

@@ -10,10 +10,7 @@
 import org.jgroups.util.*;
 
 import java.util.*;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
@@ -60,20 +57,18 @@ public class NAKACK2 extends Protocol implements DiagnosticsHandler.ProbeHandler
     @Property(description="Use a multicast to request retransmission of missing messages")
     protected boolean use_mcast_xmit_req;
 
-
     /**
      * Ask a random member for retransmission of a missing message. If set to
      * true, discard_delivered_msgs will be set to false
      */
     @Property(description="Ask a random member for retransmission of a missing message. Default is false")
     protected boolean xmit_from_random_member;
 
-
     /**
      * Messages that have been received in order are sent up the stack (= delivered to the application).
-     * Delivered messages are removed from the retransmission buffer, so they can get GC'ed by the JVM. When this
-     * property is true, everyone (except the sender of a message) removes the message from their retransission
-     * buffers as soon as it has been delivered to the application
+     * Delivered messages are removed from the retransmit table, so they can get GC'ed by the JVM. When this
+     * property is true, everyone (except the sender of a message) removes the message from their retransmit
+     * table as soon as it has been delivered to the application
      */
     @Property(description="Should messages delivered to application be discarded")
     protected boolean discard_delivered_msgs=true;
@@ -110,7 +105,7 @@ public class NAKACK2 extends Protocol implements DiagnosticsHandler.ProbeHandler
     @Property(description="Size of the queue to hold messages received after creating the channel, but before being " +
       "connected (is_server=false). After becoming the server, the messages in the queue are fed into up() and the " +
       "queue is cleared. The motivation is to avoid retransmissions (see https://issues.redhat.com/browse/JGRP-1509 " +
-      "for details). 0 disables the queue.")
+      "for details).")
     protected int     become_server_queue_size=50;
 
     @Property(description="Time during which identical warnings about messages from a non member will be suppressed. " +
@@ -257,7 +252,7 @@ public void setResendLastSeqno(boolean flag) {
     /** Keeps a bounded list of the last N digest sets */
     protected final BoundedList<String> digest_history=new BoundedList<>(10);
 
-    protected BoundedList<Message>      become_server_queue;
+    protected Queue<Message>            become_server_queue;
 
      /** Log to suppress identical warnings for messages from non-members */
     protected SuppressLog<Address>      suppress_log_non_member;
@@ -338,7 +333,7 @@ public <T extends Protocol> T setLevel(String level) {
 
     @ManagedAttribute(description="Actual size of the become_server_queue",type=AttributeType.SCALAR)
     public int getBecomeServerQueueSizeActual() {
-        return become_server_queue != null? become_server_queue.size() : -1;
+        return become_server_queue.size();
     }
 
     /** Returns the receive window for sender; only used for testing. Do not use ! */
@@ -499,8 +494,23 @@ public void init() throws Exception {
             }
         }
 
-        if(become_server_queue_size > 0)
-            become_server_queue=new BoundedList<>(become_server_queue_size);
+        if(xmit_interval <= 0) {
+            // https://issues.redhat.com/browse/JGRP-2675
+            become_server_queue=new ConcurrentLinkedQueue<>();
+            RejectedExecutionHandler handler=transport.getThreadPool().getRejectedExecutionHandler();
+            if(!(handler instanceof ThreadPoolExecutor.CallerRunsPolicy)) {
+                log.warn("%s: xmit_interval of %d requires a CallerRunsPolicy in the thread pool; replacing %s",
+                         local_addr, xmit_interval, handler.getClass().getSimpleName());
+                transport.getThreadPool().setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
+            }
+        }
+        else {
+            if(become_server_queue_size <= 0) {
+                log.warn("%s: %s.become_server_queue_size is <= 0; setting it to 10", local_addr, NAKACK2.class.getSimpleName());
+                become_server_queue_size=10;
+            }
+            become_server_queue=new ArrayBlockingQueue<>(become_server_queue_size);
+        }
 
         if(suppress_time_non_member_warnings > 0)
             suppress_log_non_member=new SuppressLog<>(log, "MsgDroppedNak", "SuppressMsg");
@@ -539,8 +549,7 @@ public void start() throws Exception {
     public void stop() {
         running=false;
         is_server=false;
-        if(become_server_queue != null)
-            become_server_queue.clear();
+        become_server_queue.clear();
         stopRetransmitTask();
         xmit_task_map.clear();
         stable_xmit_map.clear();
@@ -785,12 +794,10 @@ public String[] supportedKeys() {
     /* --------------------------------- Private Methods --------------------------------------- */
 
     protected void queueMessage(Message msg, long seqno) {
-        if(become_server_queue != null) {
-            become_server_queue.add(msg);
-            log.trace("%s: message %s#%d was added to queue (not yet server)", local_addr, msg.getSrc(), seqno);
-        }
+        if(become_server_queue.offer(msg)) // discards item if queue is full
+            log.trace("%s: message %s#%d was queued (not yet server)", local_addr, msg.getSrc(), seqno);
         else
-            log.trace("%s: message %s#%d was discarded (not yet server)", local_addr, msg.getSrc(), seqno);
+            log.trace("%s: message %s#%d was discarded (not yet server, queue full)", local_addr, msg.getSrc(), seqno);
     }
 
     protected void unknownMember(Address sender, Object message) {
@@ -1028,19 +1035,14 @@ protected void deliverBatch(MessageBatch batch) {
      * {@link GMS#installView(org.jgroups.View,org.jgroups.util.Digest)} method (called when a view is installed).
      */
     protected void flushBecomeServerQueue() {
-        if(become_server_queue != null && !become_server_queue.isEmpty()) {
+        if(!become_server_queue.isEmpty()) {
             log.trace("%s: flushing become_server_queue (%d elements)", local_addr, become_server_queue.size());
-
             TP transport=getTransport();
-            for(final Message msg: become_server_queue) {
-                transport.getThreadPool().execute(() -> {
-                    try {
-                        up(msg);
-                    }
-                    finally {
-                        become_server_queue.remove(msg);
-                    }
-                });
+            for(;;) {
+                final Message msg=become_server_queue.poll();
+                if(msg == null)
+                    break;
+                transport.getThreadPool().execute(() -> up(msg));
             }
         }
     }
@@ -1436,7 +1438,7 @@ protected void stable(Digest digest) {
             // check whether the last seqno received for a sender P in the stability digest is > last seqno
             // received for P in my digest. if yes, request retransmission (see "Last Message Dropped" topic in DESIGN)
             Table<Message> buf=xmit_table.get(member);
-            if(buf != null) {
+            if(buf != null && xmit_interval > 0) {
                 long my_hr=buf.getHighestReceived();
                 Long prev_hr=stable_xmit_map.get(member);
                 if(prev_hr != null && prev_hr > my_hr) {
@@ -1501,7 +1503,7 @@ protected static long sizeOfAllMessages(Table<Message> buf, boolean include_head
     }
 
     protected void startRetransmitTask() {
-        if(xmit_task == null || xmit_task.isDone())
+        if(xmit_interval > 0 && (xmit_task == null || xmit_task.isDone()))
             xmit_task=timer.scheduleWithFixedDelay(new RetransmitTask(), 0, xmit_interval, TimeUnit.MILLISECONDS, sends_can_block);
     }
 

src/org/jgroups/util/ThreadPool.java

@@ -42,6 +42,10 @@ public class ThreadPool implements Lifecycle {
     @Property(description="Timeout (ms) to remove idle threads from the pool", type=AttributeType.TIME)
     protected long                keep_alive_time=30000;
 
+    @Property(description="The rejection policy to be used in the thread pool (abort, discard, run, custom etc. " +
+      "See Util.parseRejectionPolicy() for details")
+    protected String              rejection_policy="abort";
+
     @Property(description="The number of times a thread pool needs to be full before a thread dump is logged")
     protected int                 thread_dumps_threshold=1;
 
@@ -111,6 +115,23 @@ public ThreadPool setKeepAliveTime(long time) {
         return this;
     }
 
+    public ThreadPool setRejectionPolicy(String policy) {
+        RejectedExecutionHandler p=Util.parseRejectionPolicy(policy);
+        this.rejection_policy=policy;
+        if(thread_pool instanceof ThreadPoolExecutor)
+            ((ThreadPoolExecutor)thread_pool).setRejectedExecutionHandler(p);
+        return this;
+    }
+
+    public RejectedExecutionHandler getRejectedExecutionHandler() {
+        Executor t=thread_pool;
+        return t instanceof ThreadPoolExecutor? ((ThreadPoolExecutor)t).getRejectedExecutionHandler() : null;
+    }
+
+    public void setRejectedExecutionHandler(RejectedExecutionHandler handler) {
+        if(thread_pool instanceof ThreadPoolExecutor)
+            ((ThreadPoolExecutor)thread_pool).setRejectedExecutionHandler(handler);
+    }
 
     public int getThreadDumpsThreshold() {
         return thread_dumps_threshold;
@@ -160,7 +181,7 @@ public int getLargestSize() {
     public void init() throws Exception {
         if(enabled) {
             thread_pool=ThreadCreator.createThreadPool(min_threads, max_threads, keep_alive_time,
-                  "abort", new SynchronousQueue<>(), tp.getThreadFactory(), tp.useVirtualThreads(), tp.getLog());
+                  rejection_policy, new SynchronousQueue<>(), tp.getThreadFactory(), tp.useVirtualThreads(), tp.getLog());
         }
         else // otherwise use the caller's thread to unmarshal the byte buffer into a message
             thread_pool=new DirectExecutor();

tests/byteman/org/jgroups/tests/byteman/BecomeServerTest.java

@@ -67,7 +67,7 @@ public void receive(Message msg) {
     }
 
 
-    protected void sendMessage(JChannel ch, String message) {
+    protected static void sendMessage(JChannel ch, String message) {
         try {
             ch.send(new ObjectMessage(null, message).setFlag(Message.Flag.OOB));
         }