Bug 37846548 - [36898832->22.06.13] RFA: Proxy leaking processedEvents causing heap stress

[git-p4: depot-paths = "//dev/coherence-ce/release/coherence-ce-v22.06/": change = 115733]

Author: mgamanho

prj/coherence-core-components/src/main/java/com/tangosol/coherence/component/net/Message.java

@@ -1,6 +1,6 @@
 
 /*
- * Copyright (c) 2000, 2024, 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
  * https://oss.oracle.com/licenses/upl.
@@ -87,7 +87,14 @@ public class Message
      * MessageHandler.sendMulti
      */
     private transient int __m_BufferUsageCounter;
-    
+
+    /**
+     * Property fDelivered
+     *
+     * Only true iff releaseOutgoing determined that message has been delivered to all recipients.
+     */
+    private transient volatile boolean m_fDelivered = false;
+
     /**
      * Property DeserializationRequired
      *
@@ -141,7 +148,7 @@ public class Message
      * Messages generated by the Cluster Service (service 0).
      */
     private int __m_MessageType;
-    
+
     /**
      * Property NotifyDelivery
      *
@@ -788,7 +795,7 @@ public com.tangosol.internal.tracing.SpanContext getTracingSpanContext()
     public boolean isDelivered()
         {
         // see releaseOutgoing
-        return getBufferController() == null;
+        return  m_fDelivered;
         }
 
     // Accessor for the property "DeserializationRequired"
@@ -1250,12 +1257,13 @@ public void releaseOutgoing(boolean fSuspect, boolean fOrdered)
                 setBufferUsageCounter(cUsage);
                 }
             }
-        
+
         if ((cUsage & ~BUFFER_COUNT_ORDER_BIT) == 0)
             {
             // all receipts are in
             controller.dispose();
             setBufferController(null);
+            m_fDelivered = true;
 
             if (cUsage == 0) // not marked as unordered
                 {
@@ -1276,6 +1284,7 @@ public void releaseOutgoingComplete()
         if (getBufferUsageCounter() == BUFFER_COUNT_ORDER_BIT) // dirty read; no users left; common case
             {
             setBufferUsageCounter(0);
+            m_fDelivered = true;
             notifyDelivery();
             }
         else // potentially some users left; may be concurrently releasing
@@ -1291,6 +1300,7 @@ public void releaseOutgoingComplete()
 
                 if (cUsage == BUFFER_COUNT_ORDER_BIT) // no users left
                     {
+                    m_fDelivered = true;
                     notifyDelivery();
                     }
                 // else; still users left; they will notify if needed during release

prj/coherence-core-components/src/main/java/com/tangosol/coherence/component/util/daemon/queueProcessor/service/grid/partitionedService/PartitionedCache.java

@@ -5088,12 +5088,21 @@ protected void onInterval()
 
         getRequestCoordinator().onInterval();
 
-        // on some environments the "NotifyDelivery" ack may become lost or be
-        // held up (e.g.: hundreds of ViewCache clients with heavy writes)
-        // the following attempts to get stuck events "unstuck" by re-sending
-        // the oldest and hopefully cause a chain of events to clear the
-        // pending events long array which can become clogged on backup members
-        // due to said clean-up being solely based on oldest ack'ed event.
+        // event sending may miss one or more events due to concurrency on
+        // entry status processing, see ResourceCoordinator.processEvent.
+        // Synthetic events are processed in finalizeInvoke*, which in turn
+        // processes events through ResourceCoordinator.processEvent one by
+        // one. If the event is not "managed" (== synthetic), it is then passed
+        // back to the caller as an OOB event to be sent to the client.
+        // If the event is "managed", it is assumed that the caller is
+        // responsible for sending the event.
+        // There is a narrow race condition where an entry status can be locked
+        // (setting isManaged() to true) while synthetic events are being
+        // processed, for example entries evicted.
+        // In this case the path is that of synthetic events, but the event
+        // is not returned into the OOB set.
+        // This is caught below when we see the oldest event staying too long
+        // in the pending events LongArray structure.
         LongArray laPending = getPendingEvents();
         if (laPending == null) // used in getOldestPendingEventSUID
             {
@@ -5102,7 +5111,6 @@ protected void onInterval()
 
         long ldtNow      = Base.getSafeTimeMillis();
         long lOldestSUID = getOldestPendingEventSUID();
-
         if (ldtNow > getOldestEventResendNextMillis())
             {
             if (lOldestSUID > 0 &&
@@ -8662,7 +8670,7 @@ protected void postEvent(PartitionedCache.MapEvent msgEvent)
             {
             return;
             }
-        
+
         post(msgEvent);
         }
 
@@ -9538,7 +9546,7 @@ protected void publishChanges(com.tangosol.coherence.component.net.RequestContex
 
                 // remember the events-holder
                 oHolder = status.getMapEventHolder();
-        
+
                 // increment the partition version
                 int               nPartition = status.getPartition();
                 PartitionedCache.PartitionControl ctrlPart   = (PartitionedCache.PartitionControl) getPartitionControl(nPartition);
@@ -9926,8 +9934,11 @@ protected void publishChanges(com.tangosol.coherence.component.net.RequestContex
         //
         // The batch context data structure is used for "synchronization" and both
         // conditions are checked by $BatchContext#onJobCompleted
-        
-        ctxBatch.onJobCompleted(job);  // may be null
+
+        synchronized (ctxBatch)
+            {
+            ctxBatch.onJobCompleted(job);  // may be null
+            }
 
         // Now, only after backup messages/events have been queued/sent, unlock the
         // keys that were locked during #processEvent
@@ -17566,7 +17577,7 @@ protected boolean tryBatchCompletion()
 
                     // all jobs have run, and all backups have completed; respond to the client
                     service.publishToClients(getPrimaryResponse(), getEvents());
-            
+
                     return true;
                     }
                 }
@@ -29688,7 +29699,7 @@ public void onDelivery()
         public void onReceived()
             {
             super.onReceived();
-            
+
             ((PartitionedCache) getService()).onMapEvent(this);
             }
         }
@@ -38012,7 +38023,7 @@ protected void checkResourceDeadlock(Object oContender)
             }
 
         /**
-         * Specialized helper method to collect aynchronously observed
+         * Specialized helper method to collect asynchronously observed
         * $EventStatus objects.
         * 
         * @param setStatus    the Set<$EntryStatus> of asynchronously observed
@@ -38697,7 +38708,7 @@ else if (!service.isPrimaryOwner(status.getPartition()))
                // else this is a troubling case; process the com.tangosol.util.MapEvent updating ancillary
                //      data structures
                }
-            
+
             // event could be null if this is a "synthetic" event holder used
             // to force a flush of the backup & client event changes
             if (event != null)
@@ -38895,7 +38906,7 @@ else if (fSynthetic)
             // 3) the status is managed by another thread
             //
             // If the status is unmanaged, we should try to lock the key and "take"
-            // ownership of the EntryStatus (and become reponsible for publishing
+            // ownership of the EntryStatus (and become responsible for publishing
             // backup and client-event changes).  Note: we must not be overly aggressive
             // in attempting to lock due to the possibility of deadlock (see COH-5436).
             //
@@ -38944,7 +38955,10 @@ else if (fSynthetic)
                 }
 
             // return true iff the status has been locked by this method
-            // and therefore needs to be added to the OOB status collection
+            // and therefore needs to be added to the OOB status collection.
+            // Caveat: if a "managed" event has grabbed the same lock it could
+            // lead to a "forgotten" event if the caller is the synthetic event
+            // path. See onInterval
             return !fOOBEvent;
             }