Bug 38360810 - [38298007->15.1.1.0.0] make RefreshableAddressProvider use an executor for the refresh thread

(merge 15.1.1-0 -> ce/15.1.1-0 118491)

[git-p4: depot-paths = "//dev/coherence-ce/release/coherence-ce-v15.1.1.0/": change = 118492]

Author: thegridman

prj/coherence-core-21/src/main/java/com/tangosol/internal/util/VirtualThreads.java

@@ -13,6 +13,10 @@
 
 import com.tangosol.net.security.SecurityHelper;
 
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+
 import java.util.function.Function;
 
 /**
@@ -94,6 +98,21 @@ public static boolean isEnabled(String sServiceName)
                : Config.getBoolean(PROPERTY_SERVICE_ENABLED.apply(sServiceName), isEnabled());
         }
 
+
+    /**
+     * Returns either a new virtual thread per-task executor on Java 21 or higher
+     * or a single threaded executor if lower than Java 21.
+     *
+     * @param factory  the {@link ThreadFactory} to use if not on Java 21
+     *
+     * @return either a new virtual thread per-task executor on Java 21 or higher
+     *         or a single threaded executor if lower than Java 21.
+     */
+    public static Executor newMaybeVirtualThreadExecutor(ThreadFactory factory)
+        {
+        return Executors.newVirtualThreadPerTaskExecutor();
+        }
+
     // ---- constants -------------------------------------------------------
 
     /**

prj/coherence-core/src/main/java/com/tangosol/internal/util/VirtualThreads.java

@@ -9,6 +9,10 @@
 
 import com.tangosol.util.Base;
 
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+
 /**
  * Helper class for virtual threads functionality.
  * <p>
@@ -68,4 +72,18 @@ public static boolean isEnabled(String serviceName)
         {
         return false;
         }
+
+    /**
+     * Returns either a new virtual thread per-task executor on Java 21 or higher
+     * or a single threaded executor if lower than Java 21.
+     *
+     * @param factory  the {@link ThreadFactory} to use if not on Java 21
+     *
+     * @return either a new virtual thread per-task executor on Java 21 or higher
+     *         or a single threaded executor if lower than Java 21.
+     */
+    public static Executor newMaybeVirtualThreadExecutor(ThreadFactory factory)
+        {
+        return Executors.newSingleThreadExecutor(factory);
+        }
     }

prj/coherence-core/src/main/java/com/tangosol/net/RefreshableAddressProvider.java

@@ -1,24 +1,29 @@
 /*
- * Copyright (c) 2000, 2020, Oracle and/or its affiliates.
+ * Copyright (c) 2000, 2025, Oracle and/or its affiliates.
  *
  * Licensed under the Universal Permissive License v 1.0 as shown at
- * http://oss.oracle.com/licenses/upl.
+ * https://oss.oracle.com/licenses/upl.
  */
 
 package com.tangosol.net;
 
 
 import com.tangosol.coherence.config.Config;
 
+import com.tangosol.internal.util.VirtualThreads;
+
 import com.tangosol.util.Base;
-import com.tangosol.util.Daemon;
 
 import java.net.InetSocketAddress;
 
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
 
+import java.util.concurrent.Executor;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
 
 /**
 * A RefreshableAddressProvider is an AddressProvider implementation 
@@ -57,7 +62,9 @@ public RefreshableAddressProvider(AddressProvider ap, long lRefresh)
         // populate the initial address list cache
         refreshAddressList();
         // initialize the refresh thread
-        f_daemonRefresh = new RefreshThread(lRefresh);
+        String sName = RefreshableAddressProvider.this.getClass().getName() + ": RefreshThread";
+        f_refreshTask   = new RefreshTask(lRefresh);
+        f_daemonRefresh = VirtualThreads.newMaybeVirtualThreadExecutor(r -> new Thread(r, sName));
         }
 
     /**
@@ -178,54 +185,74 @@ protected void refreshAddressList()
     */
     protected void ensureRefreshThread()
         {
-        if (!f_daemonRefresh.isRunning())
+        if (!f_refreshTask.isRunning())
             {
-            f_daemonRefresh.start();
+            f_daemonRefresh.execute(f_refreshTask);
             }
         }
 
 
     // ----- inner class: RefreshThread -------------------------------------
 
-    protected class RefreshThread
-            extends Daemon
+    protected class RefreshTask
+            implements Runnable
         {
         // ----- constructors -----------------------------------------------
 
         /**
-        * Construct a new RefreshThread with the specified refresh interval.
+        * Construct a new RefreshTask with the specified refresh interval.
         *
         * @param lRefresh  the refresh interval
         */
-        protected RefreshThread(long lRefresh)
+        protected RefreshTask(long lRefresh)
             {
-            super(RefreshableAddressProvider.this.getClass().getName() +
-                  ": RefreshThread");
             f_lRefresh = lRefresh;
             }
 
-        /**
-        * {@inheritDoc}
-        */
+        public boolean isRunning()
+            {
+            return m_fRunning;
+            }
+
+        @Override
         public void run()
             {
-            long lRefresh = f_lRefresh;
-            while (!isStopping())
+            if (m_fRunning)
                 {
-                try
-                    {
-                    refreshAddressList();
-                    stop();
-                    }
-                catch (Throwable t)
+                return;
+                }
+
+            f_lock.lock();
+            try
+                {
+                if (!m_fRunning)
                     {
-                    err("An exception occurred while refreshing an address list: " +
-                        "\n" + getStackTrace(t) +
-                        "\nReducing the refresh rate.");
-                    Base.sleep(lRefresh);
-                    lRefresh = 2 * lRefresh;
+                    m_fRunning = true;
+                    long lRefresh = f_lRefresh;
+                    while (m_fRunning)
+                        {
+                        try
+                            {
+                            refreshAddressList();
+                            m_fRunning = false;
+                            }
+                        catch (Throwable t)
+                            {
+                            err("An exception occurred while refreshing an address list: " +
+                                    "\n" + t.getMessage() +
+                                    "\n" + getStackTrace(t) +
+                                    "\nReducing the refresh rate.");
+                            Base.sleep(lRefresh);
+                            lRefresh = 2 * lRefresh;
+                            }
+                        }
                     }
                 }
+            finally
+                {
+                m_fRunning = false;
+                f_lock.unlock();
+                }
             }
 
         // ----- data members -----------------------------------------------
@@ -234,6 +261,16 @@ public void run()
         * The interval with which to attempt to refresh the address list.
         */
         protected final long f_lRefresh;
+
+        /**
+         * The flag to indicate whether the task is running.
+         */
+        private volatile boolean m_fRunning = false;
+
+        /**
+         * A lock to control running only a single task.
+         */
+        private final Lock f_lock = new ReentrantLock();
         }
 
 
@@ -357,7 +394,15 @@ protected void refreshIterator()
     /**
     * The refresh daemon.
     */
-    protected final Daemon f_daemonRefresh;
+    protected final RefreshTask f_refreshTask;
+
+    /**
+    * The refresh executor.
+    * <p>
+    * This will be a single threaded executor on Java 17 or a
+    * virtual thread per-task on Java 21 or higher.
+    */
+    protected final Executor f_daemonRefresh;
 
     /**
     * An Iterator over the cached set of addresses.

prj/coherence-core/src/main/java/com/tangosol/util/Daemon.java

@@ -750,7 +750,10 @@ protected void configureWorker(DaemonWorker worker)
         ThreadGroup curThreadGroup = ensureThreadGroup();
         synchronized (curThreadGroup) // ensures that the thread group is not destroyed concurrently
             {
-            ensureThreadGroup();
+            if (curThreadGroup.isDestroyed())
+                {
+                ensureThreadGroup();
+                }
             threadWorker = makeThread(m_threadGroup, worker, null);
             }
 
@@ -781,9 +784,11 @@ protected void configureWorker(DaemonWorker worker)
     protected ThreadGroup ensureThreadGroup()
         {
         ThreadGroup threadGroup = m_threadGroup;
-        if (threadGroup == null)
+        if (threadGroup == null || threadGroup.isDestroyed())
             {
             threadGroup = m_threadGroup = new ThreadGroup(getConfiguredName());
+            // Make it a daemon so that it is destroyed automatically.
+            threadGroup.setDaemon(true);
             }
         return threadGroup;
         }