- If the thread pool is full, use a separate thread to send the view (https://issues.redhat.com/browse/JGRP-2880)

- Use a separate thread to send LEAVE request in Leaver

Author: belaban

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

@@ -560,7 +560,11 @@ public void castViewChangeAndSendJoinRsps(View new_view, Digest digest, Collecti
                 down_prot.down(view_change_msg);
                 sendJoinResponses(jr, joiners);
             };
-            thread_pool.execute(r);
+            boolean rc=thread_pool.execute(r);
+            if(!rc) { // https://issues.redhat.com/browse/JGRP-2880
+                Thread th=getThreadFactory().newThread(r);
+                th.start();
+            }
         }
 
         try {

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

@@ -116,6 +116,19 @@ public void reset() {
     protected void sendLeaveRequest(Address coord, Address leaving_mbr) {
         Message msg=new EmptyMessage(coord).setFlag(Message.Flag.OOB).setFlag(Message.TransientFlag.DONT_BLOCK)
           .putHeader(gms.getId(), new GMS.GmsHeader(GMS.GmsHeader.LEAVE_REQ, leaving_mbr));
-        gms.getDownProtocol().down(msg);
+        if(gms.thread_pool == null)
+            gms.getDownProtocol().down(msg);
+        else {
+            // If the mcast protocol can block, we need to send a view asynchronously. The views will still
+            // be delivered in order, see https://issues.redhat.com/browse/JGRP-2875 for details
+            Runnable r=() -> {
+                gms.getDownProtocol().down(msg);
+            };
+            boolean rc=gms.thread_pool.execute(r);
+            if(!rc) { // https://issues.redhat.com/browse/JGRP-2880
+                Thread th=gms.getThreadFactory().newThread(r);
+                th.start();
+            }
+        }
     }
 }

src/org/jgroups/util/ThreadPool.java

@@ -75,6 +75,14 @@ public class ThreadPool implements Lifecycle {
     @ManagedAttribute(description="The number of messages dropped because the thread pool was full",type= SCALAR)
     protected final LongAdder     num_rejected_msgs=new LongAdder();
 
+    protected final RejectedExecutionHandler rejected_handler=(r, pool) -> {
+        num_rejected_msgs.increment();
+        //https://issues.redhat.com/browse/JGRP-2802
+        String thread_dump=thread_dumps_enabled? String.format(". Threads:\n%s", Util.dumpThreads()) : "";
+        thread_pool_full_log.log(warn, "thread-pool-full", thread_pool_full_suppress_time,
+                                 address, max_threads, getThreadPoolSize(), thread_dump);
+    };
+
     public ThreadPool() {
     }
 
@@ -231,17 +239,17 @@ public void doExecute(Runnable task) {
 
     public Executor pool() {return thread_pool;}
 
-    public boolean execute(Runnable task) {
+    public boolean execute(Runnable task, RejectedExecutionHandler h) {
         try {
             thread_pool.execute(task);
             return true;
         }
         catch(RejectedExecutionException ex) {
-            num_rejected_msgs.increment();
-            //https://issues.redhat.com/browse/JGRP-2802
-            String thread_dump=thread_dumps_enabled? String.format(". Threads:\n%s", Util.dumpThreads()) : "";
-            thread_pool_full_log.log(warn, "thread-pool-full", thread_pool_full_suppress_time,
-                                     address, max_threads, getThreadPoolSize(), thread_dump);
+            rejected_handler.rejectedExecution(task, null); // we want the default behavior (incr rej. msgs)
+            if(h != null) {
+                ThreadPoolExecutor tmp=thread_pool instanceof ThreadPoolExecutor? (ThreadPoolExecutor)thread_pool : null;
+                h.rejectedExecution(task, tmp);
+            }
             return false;
         }
         catch(Throwable t) {
@@ -251,6 +259,10 @@ public boolean execute(Runnable task) {
         }
     }
 
+    public boolean execute(Runnable task) {
+        return execute(task, null);
+    }
+
     public String toString() {
         return thread_pool != null? thread_pool.toString() : "n/a";
     }

tests/junit-functional/org/jgroups/tests/ReliableUnicastBlockTest.java

@@ -155,7 +155,8 @@ public void testConnectionCloseThenReopen() throws Exception {
         GMS gms_b=b.stack().findProtocol(GMS.class);
         gms_b.installView(mv);
 
-        Util.waitUntil(2000, 100, () -> rb.size() == 10);
+        Util.waitUntil(2000, 100, () -> rb.size() == 10,
+                       () -> String.format("B has %d messages: %s", rb.size(), rb.list()));
         assert rb.list().equals(EXPECTED);
         System.out.printf("-- rb: %s\n", print(rb));
     }