fix: Wake the selector for every queued immediate task (14.1.2.0 cl 122127 --> 14.1.2.0 CE)

RunnableSelectionService used queue occupancy as a proxy for a durable selector wakeup. A producer could enqueue after the selector consumed the prior wakeup and entered a timed select, delaying TMB progression for approximately 2.5 seconds.

Publish a wakeup for every immediate task; Selector coalesces redundant notifications. Add a regression test for the required enqueue-to-wakeup invariant.

[git-p4: depot-paths = "//dev/coherence-ce/release/coherence-ce-v14.1.2.0/": change = 122152]

Author: fryp

prj/coherence-core/src/main/java/com/oracle/coherence/common/internal/net/RunnableSelectionService.java

@@ -1,8 +1,8 @@
 /*
- * Copyright (c) 2000, 2022, Oracle and/or its affiliates.
+ * Copyright (c) 2000, 2026, 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.oracle.coherence.common.internal.net;
 
@@ -81,12 +81,15 @@ public void invoke(final SelectableChannel chan, final Runnable runnable, long c
 
         if (cMillis == 0)
             {
-            boolean fEmpty = f_tasks.isEmpty();
             f_tasks.add(runnable);
-            if (fEmpty)
-                {
-                wakeup();
-                }
+
+            // A selector wakeup may be consumed by a selection that is already
+            // returning. The queue can also appear non-empty while its consumer
+            // marker is being drained, so queue non-emptiness is not proof that
+            // a durable wakeup is pending for this task. Selector.wakeup()
+            // coalesces redundant notifications, so publish one for every newly
+            // queued immediate task.
+            wakeup();
             }
         else
             {

prj/test/unit/coherence-core-tests/src/test/java/com/oracle/coherence/common/internal/net/RunnableSelectionServiceTest.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright (c) 2026, Oracle and/or its affiliates.
+ *
+ * Licensed under the Universal Permissive License v 1.0 as shown at
+ * https://oss.oracle.com/licenses/upl.
+ */
+package com.oracle.coherence.common.internal.net;
+
+import org.junit.Test;
+
+import java.io.IOException;
+
+import java.nio.channels.Pipe;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+/**
+ * Unit tests for {@link RunnableSelectionService}.
+ *
+ * @author Aleks Seovic  2026.08.13
+ * @since 26.07
+ */
+public class RunnableSelectionServiceTest
+    {
+    @Test
+    public void shouldPublishWakeupForEveryImmediateTask()
+            throws IOException
+        {
+        Pipe                      pipe    = Pipe.open();
+        CountingSelectionService service = new CountingSelectionService();
+
+        try
+            {
+            pipe.source().configureBlocking(false);
+
+            service.invoke(pipe.source(), () -> {}, 0);
+            service.invoke(pipe.source(), () -> {}, 0);
+
+            assertThat(service.m_cWakeups, is(2));
+            }
+        finally
+            {
+            service.shutdown();
+            pipe.source().close();
+            pipe.sink().close();
+            }
+        }
+
+    /**
+     * Selection service that counts published wakeups.
+     */
+    private static class CountingSelectionService
+            extends RunnableSelectionService
+        {
+        @Override
+        protected void wakeup()
+            {
+            ++m_cWakeups;
+            }
+
+        private int m_cWakeups;
+        }
+    }