commit-guard-fastfind-r16812.patch

Constantin Asofiei, 10/29/2026 05:06 PM

     **                  the changes as other servers' (incrementRemoteIfTracked) and resolves the FastFindCache
     **                  by database type and name (databaseType; the meta database has the primary's name).
     **                  The per-database publish() is replaced by Pending; serialVersionUID 2.
     **     CA  20261029 Removed the off mode (isEnabled()). A server which may have missed a notice clears its
     **                  fast-find caches and counts a remote change of every cached record (missed()): a receiver
     **                  which failed to apply it; the servers which did not confirm it in time, or all others
     **                  when the send failed, through the Missed function (resync()); the survivors of a server
     **                  which departed (LockMembershipHandler). The receiver no longer marks Geode's thread.
     **                  An answer to a notice renews the answering server's membership lease (Confirmations).
     */
     /*
-...
      * there has locked a changed record and read it again from the database. It must not make that current copy
      * stale, or the session's update would not be saved: the receiver counts it apart from the local changes,
      * and a session does not treat as stale a copy read under a lock its context still holds (see
      * {@code Session.isStale}). A late notice therefore only delays NO-LOCK reads of other sessions there; a
      * notice which is lost (not delivered, or not applied: the receiver answers -1) leaves their cached copies
      * and FastFindCache entries of those records stale until they are evicted or change again. This
      * {@code Session.isStale}). A late notice therefore only delays NO-LOCK reads of other sessions there.
      * <p>
      * A server which may have missed a notice clears all its FastFindCache results and counts a remote change
      * of every record it caches ({@link #missed}), so nothing it cached before is used without reading it again:
      * <ul>
      *   <li>a receiver which could not apply a notice, at once (it answers -1);</li>
      *   <li>the servers which did not confirm a notice in time (the sender's wait timed out, they left during
      *       it, or the wait failed), or every other server when the send itself failed: the sender tells them
      *       ({@link Missed}, without waiting; the servers to tell are coalesced while a call is pending); a
      *       server which is unresponsive gets it when it answers again, or Geode removes it from the view and
      *       its re-initialisation clears; a server which answers but cannot run it logs that;</li>
      *   <li>every server, when another FWD server departs (it may have committed without sending, or while
      *       sending, its notices), once the departed server can no longer commit ({@code LockMembershipHandler},
      *       after {@code MembershipLease.GRACE_MS});</li>
      *   <li>a server which lost its membership, in its re-initialisation ({@code GeodeConfig}).</li>
      * </ul>
      * In the batch and async modes a notice which is sent but lost without an error is not noticed. This
      * needs a lock manager which tells since which value of the JVM-wide sequence (no clock) a lock is held
      * ({@code LockManager.heldSinceSeq}): the owner-side lock manager, the cluster default; with another one a
      * late notice can still drop a lock holder's update. Prerequisite of the cluster mode: the database runs at
-...
      * to each other server, waited for (sync), or at most one per {@link #BATCH_MS} (batch); the message holds
      * the changed records' ids. This replaces the FastFindCache's own broadcasts (several per transaction).
      * <p>
      * The mode is {@link #MODE}: {@code sync} (the default), {@code batch}, {@code async} (one call per
      * transaction, not waited for) or {@code off}.
      * The mode is {@link #MODE}: {@code sync} (the default), {@code batch} or {@code async} (one call per
      * transaction, not waited for). The commit notice is the only invalidation of the FastFindCache across the
      * servers.
      */
     public final class CommitNotice
     implements Serializable
-...
        /** How long a sender waits for the other servers to apply its notice, in ms (sync mode). */
        private static final long TIMEOUT_MS = 10000L;
        /** The mode: sync (the default), batch, async or off. */
        /** The mode: sync (the default), batch or async. */
        private static final String MODE = "sync";
        /** The delay (ms) of a call of {@link #resync}, so that the servers named meanwhile are told together. */
        private static final long RESYNC_DELAY_MS = 1000L;
        /** Batch mode: the least ms between two sends of a server. */
        private static final long BATCH_MS = 5L;
-...
        /** Batch mode: the sender thread, created on first use (guarded by {@link #batchPending}). */
        private static ScheduledExecutorService flusher = null;
        /** The thread of {@link #resync}, created on first use (guarded by {@link #batchPending}). */
        private static ScheduledExecutorService resyncer = null;
        /** The servers {@link #resync} tells in its next call. */
        private static final Set<DistributedMember> resyncPending = new HashSet<>();
        /** {@link #resync}'s next call tells every other server (guarded by {@link #resyncPending}). */
        private static boolean resyncAll = false;
        /** A call of {@link #resync} is scheduled and has not taken the pending servers yet (guarded likewise). */
        private static boolean resyncScheduled = false;
        /** The cluster key of the {@link DmoVersioning} of the database and tenant. */
        private final String versioningKey;
-...
+       }
        /**
         * Check whether the servers send commit notices.
         * This server may have missed commit notices of the other servers (see the class description): clear
         * every result of its FastFindCaches and count a remote change of every record it tracks, so that every
         * cached copy is read again before it is used.
+        *
         * @return   {@code true} unless the mode is {@code off}.
         * @param    why
         *           Why, for the log.
         */
        public static boolean isEnabled()
        public static void missed(String why)
+       {
           return !"off".equals(MODE);
           LOG.warning("Commit notices may have been missed (" + why + "): the fast-find caches are cleared and " +
                       "the cached records are read again");
           FastFindCache.invalidateAllPermanent();
           DmoVersioning.incrementRemoteAll();
+       }
        /**
-...
         * as the mode says: in sync mode (the default) in one call, and wait once (at most {@link #TIMEOUT_MS})
         * until every other server has applied them; a server which does not answer in time or an error is
         * logged and the commit goes on (Geode's wait is not interruptible; its interrupt case is defensive).
         * Batch mode queues them, async mode sends them without waiting, off sends nothing.
         * The servers which did not confirm the notices, or all other servers if the send failed, are told that
         * they missed them ({@link #resync}). Batch mode queues them, async mode sends them without waiting (a
         * failed send is told to all other servers too).
+        *
         * @param    notices
         *           The notices; not empty.
         */
        private static void send(CommitNotice[] notices)
+       {
           if ("off".equals(MODE))
+          {
              return;
+          }
           if ("batch".equals(MODE))
+          {
              enqueue(notices);
              return;
+          }
           Set<DistributedMember> others = null;
           Confirmations confirmed = new Confirmations();
           try
+          {
              // inside the try: on a server whose cache is closed (expelled) this fails, and the commit goes on
              Set<DistributedMember> others = GeodeConfig.getOtherServerMembers();
              others = GeodeConfig.getOtherServerMembers();
              if (others.isEmpty())
+             {
                 return;
+             }
              ResultCollector<?, ?> rc = FunctionService.onMembers(others)
                                                        .setArguments(notices)
                                                        .execute(Receiver.ID);
              if ("sync".equals(MODE))
+             {
                 // one wait for all the other servers, which apply the notices in parallel
                 Object results = rc.getResult(TIMEOUT_MS, TimeUnit.MILLISECONDS);
                 if (results instanceof List && ((List<?>) results).contains(-1))
+                {
                    LOG.warning("Commit notice of " + describe(notices) + " not applied by every server (see " +
                                "their log)");
+                }
              ResultCollector<Object, Object> rc = FunctionService.onMembers(others)
                                                                  .setArguments(notices)
                                                                  .withCollector(confirmed)
                                                                  .execute(Receiver.ID);
              if (!"sync".equals(MODE))
+             {
                 return;
+             }
              // one wait for all the other servers, which apply the notices in parallel
              rc.getResult(TIMEOUT_MS, TimeUnit.MILLISECONDS);
              if (confirmed.failed())
+             {
                 LOG.warning("Commit notice of " + describe(notices) + " not applied by every server (see " +
                             "their log)");
+             }
+          }
           catch (InterruptedException exc)
-...
              LOG.warning("Commit notice of " + describe(notices) + " not confirmed by every server: " + exc);
              LOG.log(Level.FINE, "Commit notice not confirmed", exc);
+          }
           // the servers which did not answer may have missed the notices (one which answered -1 cleared itself);
           // in async mode, only a failed send is known
           Set<DistributedMember> missing = (others == null) ? null : confirmed.missing(others);
           if (missing == null || !missing.isEmpty())
+          {
              resync(missing, "commit notice of " + describe(notices) + " not confirmed");
+          }
+       }
        /**
         * Tell other servers that they may have missed commit notices: they clear what they cached
         * ({@link #missed}). Sent on a daemon thread without waiting for an answer; the servers named while a call
         * is pending are told in the next call, once each (so a series of timed-out notices makes a server clear
         * once, not once per notice): a call starts {@link #RESYNC_DELAY_MS} ms after the first server was named,
         * so that the notices timing out during one stall are told together. If this server is not a member of
         * the cluster any more the call fails, which is logged (its departure makes the others clear).
+        *
         * @param    members
         *           The servers, or {@code null} for every other server.
         * @param    why
         *           Why, for their log.
         */
        private static void resync(Set<DistributedMember> members, String why)
+       {
           synchronized (resyncPending)
+          {
              if (members == null)
+             {
                 resyncAll = true;
+             }
              else
+             {
                 resyncPending.addAll(members);
+             }
              if (resyncScheduled)
+             {
                 return;
+             }
              resyncScheduled = true;
+          }
           resyncer().schedule(() ->
+          {
              Set<DistributedMember> to;
              synchronized (resyncPending)
+             {
                 to = resyncAll ? GeodeConfig.getOtherServerMembers() : new HashSet<>(resyncPending);
                 resyncPending.clear();
                 resyncAll = false;
                 resyncScheduled = false;
+             }
              try
+             {
                 if (!to.isEmpty())
+                {
                    FunctionService.onMembers(to).setArguments(why).execute(Missed.ID);
                    LOG.warning("Commit notices: told " + to.size() + " server(s) that they may have missed " +
                                "notices (" + why + ")");
+                }
+             }
              catch (RuntimeException exc)
+             {
                 LOG.warning("Commit notices: could not tell the other servers that they may have missed " +
                             "notices (" + why + "): " + exc);
+             }
           }, RESYNC_DELAY_MS, TimeUnit.MILLISECONDS);
+       }
        /**
-...
           catch (Exception exc)
+          {
              LOG.log(Level.WARNING, "Commit notice batch not sent: " + exc, exc);
              resync(null, "commit notice batch not sent");
+          }
+       }
-...
+       }
        /**
         * The daemon thread which tells other servers that they may have missed notices ({@link #resync}).
+        *
         * @return   The executor, created on first use.
         */
        private static ScheduledExecutorService resyncer()
+       {
           synchronized (batchPending)
+          {
              if (resyncer == null)
+             {
                 resyncer = Executors.newSingleThreadScheduledExecutor(r ->
+                {
                    Thread t = new Thread(r, "FWD commit notice resync");
                    t.setDaemon(true);
                    return t;
                 });
+             }
              return resyncer;
+          }
+       }
        /**
         * Apply the notice to the record versions of this server.
+        *
         * @return   The number of records whose version this server tracked.
-...
                    dmoIds.add(FastFindCache.combine(dmoMeta.getId(), null));
+                }
+             }
              // a remote request: invalidate here only, never broadcast again
              FastFindCache.setIsFromRemoteRequest();
              ffCache.invalidate(dmoIds);
+          }
-...
            * Apply the notices (one per database and tenant) and answer the number of tracked records. The record
            * versions of every notice are applied first, then the FastFindCache invalidations, each notice on its
            * own, so a failure leaves the others applied. Always answers: an exception sent back to the sender
            * would end its wait for every server, not only this one; a failure is logged and answered with -1.
            * would end its wait for every server, not only this one; a failure is logged, this server clears what
            * it cached ({@link CommitNotice#missed}), and it answers -1.
+           *
            * @param    context
            *           The function context, whose argument is the notices.
-...
                    notApplied(notice, exc);
+                }
+             }
              if (failed)
+             {
                 try
+                {
                    missed("a commit notice could not be applied on this server");
+                }
                 catch (RuntimeException | LinkageError | AssertionError exc)
+                {
                    // always answer (see above)
                    LOG.warning("Fast-find caches not cleared after a failed commit notice: " + exc);
+                }
+             }
              context.getResultSender().lastResult(failed ? -1 : tracked);
+          }
-...
              LOG.log(Level.FINE, "Commit notice not applied", exc);
+          }
+       }
        /**
         * The collector of the answers to a notice in sync mode (Geode waits for them, see
         * {@link ResultCollector#getResult(long, TimeUnit)}): it records which servers answered, so that the
         * others can be told that they may have missed the notice.
         */
        private static final class Confirmations
        implements ResultCollector<Object, Object>
+       {
           /** {@code System.nanoTime()} when the collector was made (right before the notice was sent). */
           private final long sentNanos = System.nanoTime();
           /** {@code System.currentTimeMillis()} likewise. */
           private final long sentWall = System.currentTimeMillis();
           /** The cluster gate's epoch likewise. */
           private final long epoch = ClusterGate.epoch();
           /** The servers which answered. */
           private final Set<DistributedMember> answered = ConcurrentHashMap.newKeySet();
           /** The answers. */
           private final List<Object> results = Collections.synchronizedList(new ArrayList<>());
           /** {@code true} if a server answered that it could not apply the notice (-1). */
           private volatile boolean failed = false;
           /**
            * A server answered (an exception, e.g. for a server which left, is no confirmation).
+           *
            * @param    member
            *           The server.
            * @param    result
            *           Its answer.
            */
           @Override
           public void addResult(DistributedMember member, Object result)
+          {
              results.add(result);
              if (result instanceof Throwable)
+             {
                 return;
+             }
              if (Integer.valueOf(-1).equals(result))
+             {
                 failed = true;
+             }
              answered.add(member);
              // an answer also shows that the server still counts this one a member (its membership lease)
              MembershipLease.answered(member, sentNanos, sentWall, epoch);
+          }
           /**
            * The answers so far.
+           *
            * @return   The answers.
            */
           @Override
           public Object getResult()
+          {
              return results;
+          }
           /**
            * The answers so far (Geode's execution waits for them before it calls this).
+           *
            * @param    timeout
            *           Ignored.
            * @param    unit
            *           Ignored.
+           *
            * @return   The answers.
            */
           @Override
           public Object getResult(long timeout, TimeUnit unit)
+          {
              return results;
+          }
           /**
            * Every server answered: nothing to do.
            */
           @Override
           public void endResults()
+          {
+          }
           /**
            * Forget the answers (Geode calls this before it executes again; notices are not re-executed).
            */
           @Override
           public void clearResults()
+          {
              results.clear();
              answered.clear();
              failed = false;
+          }
           /**
            * Whether a server answered that it could not apply the notice.
+           *
            * @return   {@code true} if one did (it cleared what it cached).
            */
           boolean failed()
+          {
              return failed;
+          }
           /**
            * The servers which did not answer.
+           *
            * @param    sentTo
            *           The servers the notice was sent to.
+           *
            * @return   Those of them which did not answer (a new set).
            */
           Set<DistributedMember> missing(Set<DistributedMember> sentTo)
+          {
              Set<DistributedMember> missing = new HashSet<>(sentTo);
              missing.removeAll(answered);
              return missing;
+          }
+       }
        /**
         * The Geode function which tells a server that it may have missed commit notices
         * ({@link CommitNotice#resync}): it clears what it cached ({@link CommitNotice#missed}).
         */
        public static class Missed
        implements Function<String>
+       {
           /** The function id. */
           static final String ID = Missed.class.getName();
           /** Serial version. */
           private static final long serialVersionUID = 1L;
           /**
            * Clear what this server cached.
+           *
            * @param    context
            *           The function context, whose argument tells why.
            */
           @Override
           public void execute(FunctionContext<String> context)
+          {
              missed("told by another server: " + context.getArguments());
+          }
           /**
            * The function id.
+           *
            * @return   {@link #ID}.
            */
           @Override
           public String getId()
+          {
              return ID;
+          }
           /**
            * The sender does not wait for an answer.
+           *
            * @return   {@code false}.
            */
           @Override
           public boolean hasResult()
+          {
              return false;
+          }
           /**
            * Not sent again to another member when a member fails.
+           *
            * @return   {@code false}.
            */
           @Override
           public boolean isHA()
+          {
              return false;
+          }
           /**
            * Not a data-dependent function.
+           *
            * @return   {@code false}.
            */
           @Override
           public boolean optimizeForWrite()
+          {
              return false;
+          }
+       }
+    }
     -- a/src/com/goldencode/p2j/distributed/config/ClusterGate.java
     ++ b/src/com/goldencode/p2j/distributed/config/ClusterGate.java
-...
     **                  every session is terminated, logins and appserver requests are refused until the
     **                  in-process re-initialisation after Geode's reconnect, through participants, is done;
     **                  the appserver agents are started again after it).
     **     CA  20261029 Commit fence (enterCommit()/exitCommit()): closing() refuses new commits and waits for
     **                  those in flight before the cache closes; started() starts the MembershipLease pinger.
     */
     /*
-...
         */
        private static final long MSA_UNWIND_MS = 30_000L;
        /** The longest wait (ms) of {@link #closing} for the commits in flight ({@link #enterCommit}). */
        private static final long FENCE_WAIT_MS = 10_000L;
        /** Guards {@link #commitsInFlight} and {@link #fenceClosed}; {@link #closing} waits on it. */
        private static final Object FENCE = new Object();
        /** Session ids above this one are local ({@link #localSessionId}); the cluster's sequence stays below. */
        private static final int LOCAL_SESSION_ID_FLOOR = 2_000_000_000;
-...
         */
        private static volatile boolean rejoinFailed = false;
        /** The number of commits between {@link #enterCommit} and {@link #exitCommit} (guarded by FENCE). */
        private static int commitsInFlight = 0;
        /** {@code true} once {@link #closing} started: no commit enters any more (guarded by FENCE). */
        private static boolean fenceClosed = false;
        /** The re-initialisation participants, kept sorted by their order. */
        private static final List<Registered> PARTICIPANTS = new CopyOnWriteArrayList<>();
-...
+       }
        /**
         * Enter the commit fence before the commit of a database transaction ({@code TxWrapper}), before every
         * other check of the commit; {@link #exitCommit} must follow once the database commit is done or failed.
         * While a commit is inside the fence, FWD's shutdown does not close the cache ({@link #closing} waits for
         * it), so its locks are still held by this member when its change reaches the database; once the shutdown
         * started, no commit enters.
+        *
         * @param   db
         *          The database whose transaction commits.
+        *
         * @throws  DatabaseConnectionException
         *          if FWD's shutdown closes the cache (the database transaction is rolled back).
         */
        public static void enterCommit(Database db)
+       {
           synchronized (FENCE)
+          {
              if (!fenceClosed)
+             {
                 commitsInFlight++;
                 return;
+             }
+          }
           throw (DatabaseConnectionException) unavailable(db, null);
+       }
        /**
         * Leave the commit fence entered by {@link #enterCommit}.
         */
        public static void exitCommit()
+       {
           synchronized (FENCE)
+          {
              commitsInFlight--;
              if (commitsInFlight <= 0)
+             {
                 FENCE.notifyAll();
+             }
+          }
+       }
        /**
         * Build the STOP of the gate.
+        *
         * @param   db
-...
           t.start();
           LOG.log(Level.INFO, "Cluster gate: running (watchdog every " + WATCH_MS + " ms)");
           MembershipLease.start(c);
+       }
        /**
         * FWD closes the cache (server shutdown): the gate is {@link State#CLOSED} from now on, so code needing
         * the cluster gets STOP; the epoch is kept until the cache is closed ({@link #closed}), so that the
         * releases of the sessions still reach the owners meanwhile. Nobody is interrupted. Called by
         * {@link GeodeConfig}'s shutdown hook before it closes the cache.
         * releases of the sessions still reach the owners meanwhile. Nobody is interrupted. No commit enters the
         * commit fence any more, and the commits in it are waited for (at most {@link #FENCE_WAIT_MS} ms), so that
         * none reaches its database after the cache closed and the other members purged this member's locks.
         * Called by {@link GeodeConfig}'s shutdown hook before it closes the cache.
         */
        static void closing()
+       {
-...
+          }
           LOG.log(Level.INFO, "Cluster gate: closed (the server shuts down)");
           awaitCommits();
+       }
        /**
-...
+       }
        /**
         * Close the commit fence and wait (at most {@link #FENCE_WAIT_MS} ms) until no commit is in it; an
         * interrupt ends the wait and is kept.
         */
        private static void awaitCommits()
+       {
           boolean interrupted = false;
           int left;
           synchronized (FENCE)
+          {
              fenceClosed = true;
              long end = System.nanoTime() + FENCE_WAIT_MS * 1_000_000L;
              long ms;
              while (commitsInFlight > 0 && !interrupted && (ms = (end - System.nanoTime()) / 1_000_000L) > 0)
+             {
                 try
+                {
                    FENCE.wait(ms);
+                }
                 catch (InterruptedException exc)
+                {
                    interrupted = true;
+                }
+             }
              left = commitsInFlight;
+          }
           if (interrupted)
+          {
              Thread.currentThread().interrupt();
+          }
           if (left > 0)
+          {
              LOG.log(Level.WARNING, "Cluster gate: " + left + " commit(s) still in flight when the cache closes");
+          }
+       }
        /**
         * Whether a cache is closed or closing.
+        *
         * @param   c
     -- a/src/com/goldencode/p2j/distributed/config/GeodeConfig.java
     ++ b/src/com/goldencode/p2j/distributed/config/GeodeConfig.java
-...
     **                  re-initialisation participants: the reconnected cache is adopted and FWD's functions
     **                  registered again (registerFunctions()), ManagerClasses re-checked, the rebalance
     **                  coordinator started on it, the fast-find caches cleared and the record versions bumped.
     **     CA  20261029 Registered MembershipLease.Ping/Check and CommitNotice.Missed; removed the registration of
     **                  FastFindCacheInvalidationFunction; missedNotices() goes through CommitNotice.missed();
     **                  added isServerMember().
     */
     /*
-...
        private static void registerFunctions()
+       {
           FunctionService.registerFunction(new GetP2JJarVersionFunction());
           FunctionService.registerFunction(new FastFindCacheInvalidationFunction());
           FunctionService.registerFunction(new com.goldencode.p2j.distributed.cache.CommitNotice.Receiver());
           FunctionService.registerFunction(new com.goldencode.p2j.distributed.cache.CommitNotice.Missed());
           FunctionService.registerFunction(new MembershipLease.Ping());
           FunctionService.registerFunction(new MembershipLease.Check());
           FunctionService.registerFunction(new TenantRegistryReloadFunction());
           FunctionService.registerFunction(new com.goldencode.p2j.distributed.lock.OwnerLockFunction());
           FunctionService.registerFunction(new com.goldencode.p2j.distributed.lock.TableLeases.RevokeFunction());
-...
         */
        private static void missedNotices()
+       {
           FastFindCache.invalidateAllPermanent();
           DmoVersioning.incrementRemoteAll();
           com.goldencode.p2j.distributed.cache.CommitNotice.missed("this server was not a member of the cluster");
+       }
        /**
-...
+       }
        /**
         * Whether a member of the distributed system is an FWD server (not, e.g., a locator).
+        *
         * @param   member
         *          The member.
+        *
         * @return  {@code true} if it belongs to the FWD server group.
         */
        public static boolean isServerMember(DistributedMember member)
+       {
           return member != null && member.getGroups().contains(FWD_SERVER_GROUP);
+       }
        /**
         * Constructs the locator string in the format "host[port]".
+        *
         * @return  A string formatted for Geode's "locators" property.
     -- a/src/com/goldencode/p2j/distributed/lock/LiveMembers.java
     ++ b/src/com/goldencode/p2j/distributed/lock/LiveMembers.java
-...
     ** 001 ES  20261026 Initial version (#4369: liveness checks without a member set built per operation).
     **     CA  20261028 The view is built from GeodeConfig's cache, not CacheFactory.getAnyInstance(), which waits
     **                  for Geode's reconnect and then answers the reconnected cache.
     **     CA  20261029 holdsLocks()/lockerChecker(): a departed member's locks and reservations still stand for
     **                  MembershipLease.GRACE_MS after this member learned it is gone (GONE).
     */
     /*
-...
      *       not list it: a member missing from the cached view is checked against the current view before the
      *       answer, so a live member is never purged or refused because of a stale cache.</li>
      * </ul>
      * The locks and unique key reservations of a member which is gone still stand for
      * {@link MembershipLease#GRACE_MS} ms after this member learned of it ({@link #holdsLocks},
      * {@link #lockerChecker}): a member paused past the member timeout may still be committing under them until
      * its membership lease expired. A departed member is never granted anything new ({@link #isLive},
      * {@link #checker}, {@link #hasLeft} have no grace).
      */
     public final class LiveMembers
+    {
-...
        /** Time (ms) a member whose departure was reported is remembered. */
        private static final long DEPARTED_KEEP_MS = 3_600_000L;
        /** Size of {@link #GONE} above which its old entries are dropped at the next grace check. */
        private static final int GONE_PRUNE_SIZE = 256;
        /** The view answering while the cache is closing: every member live, so nothing is purged. */
        private static final View CLOSING = new View(null, null, Collections.emptySet(), 0L);
        /** Members whose departure the membership listener reported, with the time of the report (ms). */
        private static final ConcurrentHashMap<String, Long> DEPARTED = new ConcurrentHashMap<>();
        /**
         * Members known to be gone (departure reported, or missing from a current view), with the time this
         * member learned it ({@code System.nanoTime()}); for the grace of their locks ({@link #holdsLocks}).
         */
        private static final ConcurrentHashMap<String, Long> GONE = new ConcurrentHashMap<>();
        /** The cached view; {@code null} to build a new one at the next check. */
        private static volatile View view = null;
-...
+       }
        /**
         * Whether the locks and unique key reservations a member holds still stand: it is live ({@link #isLive}),
         * or it is gone since less than {@link MembershipLease#GRACE_MS} ms (see the class description).
+        *
         * @param   member
         *          Member id.
+        *
         * @return  {@code true} if they do.
         */
        public static boolean holdsLocks(String member)
+       {
           return isLive(member) || inGrace(member);
+       }
        /**
         * A check for several lockers in one decision, as {@link #checker} (one view at most) but with the grace
         * of {@link #holdsLocks}: whether a locker's locks still stand.
+        *
         * @return  The check; for one decision only (not thread safe).
         */
        public static Predicate<String> lockerChecker()
+       {
           Predicate<String> live = checker();
           return member -> live.test(member) || inGrace(member);
+       }
        /**
         * Whether the membership listener reported the departure of a member (a definite answer: member ids are
         * not reused).
+        *
-...
           long now = System.currentTimeMillis();
           DEPARTED.put(member, now);
           DEPARTED.values().removeIf(t -> now - t > DEPARTED_KEEP_MS);
           long nanos = System.nanoTime();
           GONE.put(member, nanos);
           GONE.values().removeIf(t -> nanos - t > DEPARTED_KEEP_MS * 1_000_000L);
           view = null;
+       }
-...
+       }
        /**
         * Whether a member which is not live is gone since less than {@link MembershipLease#GRACE_MS} ms (from
         * the time this member learned it; the first time is noted here if the departure was not reported).
+        *
         * @param   member
         *          Member id, of a member which is not live.
+        *
         * @return  {@code true} if it is.
         */
        private static boolean inGrace(String member)
+       {
           long now = System.nanoTime();
           long since = GONE.computeIfAbsent(member, m -> now);
           if (GONE.size() > GONE_PRUNE_SIZE)
+          {
              GONE.values().removeIf(t -> now - t > DEPARTED_KEEP_MS * 1_000_000L);
+          }
           return now - since < MembershipLease.GRACE_MS * 1_000_000L;
+       }
        /**
         * The cached view, or a new one if there is none, it is older than {@link #TTL_MS}, or its distributed
         * system is not connected any more (e.g. reconnected with a new one).
+        *
     -- a/src/com/goldencode/p2j/distributed/lock/LockMembershipHandler.java
     ++ b/src/com/goldencode/p2j/distributed/lock/LockMembershipHandler.java
-...
     **     ES  20261026 No purge while this member's cache is closing; a failed purge never reaches the uncaught
     **                  exception handler; membership events update LiveMembers.
     **     CA  20261029 Added reregister(): the listener is registered again on Geode's reconnected cache.
     **     CA  20261029 When an FWD server departs, its commit notices may be lost: once its membership lease
     **                  expired, this server clears what it cached (CommitNotice.missed()).
     */
     /*
-...
        @Override
        public void memberLeft(MembershipEvent event)
+       {
           departed(event.getMemberId(), "left");
           departed(event, "left");
+       }
        /**
-...
        @Override
        public void memberCrashed(MembershipEvent event)
+       {
           departed(event.getMemberId(), "crashed");
           departed(event, "crashed");
+       }
        /**
         * Purge the locks and unique key reservations of a departed member, on separate threads (each makes a
         * second pass later).
         * Purge the locks and unique key reservations of a departed member, on separate threads (each waits for
         * the grace of the member's membership lease first, and makes a second pass later). When the member is an
         * FWD server, its commit notices of the last moments may be lost (it committed and did not send them, or
         * was sending them): once it can no longer commit (the same grace), this server clears what it cached
         * ({@code CommitNotice.missed}).
+        *
         * @param   memberId
         *          Id of the departed member.
         * @param   event
         *          The membership event, naming the departed member.
         * @param   how
         *          Description of the departure.
         */
        private void departed(String memberId, String how)
        private void departed(MembershipEvent event, String how)
+       {
           String memberId = event.getMemberId();
           LiveMembers.departed(memberId);
           if (OwnerLockManager.cacheClosing())
-...
           start(() -> com.goldencode.p2j.distributed.orm.OwnerUniqueIndex.purgeDepartedMember(memberId),
                 "Owner-Unique-Purge",
                 memberId);
           if (GeodeConfig.isServerMember(event.getDistributedMember()))
+          {
              start(() -> missedNotices(memberId, how), "Commit-Notice-Missed", memberId);
+          }
+       }
        /**
         * Clear what this server cached once the grace of a departed server's membership lease ended (see
         * {@link #departed}); nothing if this member's cache is closing meanwhile (its own re-initialisation
         * clears).
+        *
         * @param   memberId
         *          Id of the departed server.
         * @param   how
         *          Description of the departure.
         */
        private static void missedNotices(String memberId, String how)
+       {
           try
+          {
              Thread.sleep(MembershipLease.GRACE_MS);
+          }
           catch (InterruptedException exc)
+          {
              return;
+          }
           if (!OwnerLockManager.cacheClosing())
+          {
              com.goldencode.p2j.distributed.cache.CommitNotice.missed("server " + memberId + " " + how);
+          }
+       }
        /**
     -- a/src/com/goldencode/p2j/distributed/lock/OwnerLockFunction.java
     ++ b/src/com/goldencode/p2j/distributed/lock/OwnerLockFunction.java
-...
     **                  request or an older one (a late give-up never takes a newer lock); a CANCEL makes the
     **                  older requests stale (LockQueue.giveUp); a call of a request older than the session's
     **                  lock is not applied (olderThanLock()); give-ups with an id skip the call-number test.
     **     CA  20261029 The locks of a departed member are removed only after the grace of its membership lease
     **                  (LiveMembers.lockerChecker() in purgeHere() and decide()).
     */
     /*
-...
         */
        private static boolean purgeHere(Region<Object, OwnerLockEntry> region, Object key)
+       {
           Predicate<String> live = LiveMembers.checker();
           // a departed member's locks stand until its membership lease expired (it may still be committing)
           Predicate<String> live = LiveMembers.lockerChecker();
           boolean changed = false;
           List<SessionToken> front = Collections.emptyList();
           Object qk = new QueueKey(region.getName(), key);
-...
+          }
           LockType requested = LockType.fromInt(req.type).toWaitVariant();
           Predicate<String> live = null;
           OwnerLockEntry base = before;
           boolean departed = false;
           LockPolicy.Outcome outcome = decide(base, token, requested);
           if (outcome == LockPolicy.Outcome.WAIT)
+          {
              // remove locks left by departed members so they never block anyone
              live = LiveMembers.checker();
              if (base.hasLockerOutside(live))
              // remove locks left by departed members so they never block anyone, once their membership lease
              // expired (until then they may still be committing under them)
              Predicate<String> lockers = LiveMembers.lockerChecker();
              if (base.hasLockerOutside(lockers))
+             {
                 base = base.purgeMembers(live);
                 base = base.purgeMembers(lockers);
                 outcome = decide(base, token, requested);
                 LOG.log(Level.WARNING, "[" + req.database + "] lock on " + req.key +
                                        " was held by a departed member; removed");
-...
           if (outcome == LockPolicy.Outcome.GRANT && req.update)
+          {
              // a request arriving after its member left must not leave a lock behind
              boolean requesterLive = LiveMembers.isLocal(req.member) ||
                                      (live != null ? live : LiveMembers.checker()).test(req.member);
              boolean requesterLive = LiveMembers.isLocal(req.member) || LiveMembers.isLive(req.member);
              if (requesterLive)
+             {
                 after = base.grant(token, req.member, requested, req.req);
     -- a/src/com/goldencode/p2j/distributed/lock/OwnerLockManager.java
     ++ b/src/com/goldencode/p2j/distributed/lock/OwnerLockManager.java
-...
     **                  holding lost locks; lost locks are released locally; added holdsStaleLocks(); the
     **                  re-initialisation participant rebinds the lock regions of the databases still in use
     **                  (reinitialiseAll(), rebind(); not those of deleted tenant databases).
     **     CA  20261029 purgeDepartedMember() waits for the grace of the departed member's membership lease first;
     **                  heldKeys() names the current context's locks for the commit's per-owner lease; owner calls
     **                  note the member which answered (MembershipLease.collector()).
     */
     /*
-...
     import java.util.*;
     import java.util.concurrent.*;
     import java.util.concurrent.atomic.*;
     import java.util.function.BiConsumer;
     import java.util.function.LongConsumer;
     import java.util.logging.*;
     import org.apache.geode.CancelException;
-...
+       }
        /**
         * Remove the locks of departed members from the local primary entries of every lock region; repeated
         * after a few seconds to catch primaries moved here meanwhile. Called by {@link LockMembershipHandler}.
         * Remove the locks of departed members from the local primary entries of every lock region, once the
         * grace of the departed member's membership lease ended ({@link MembershipLease#GRACE_MS}: it may still be
         * committing under them until then); repeated after a few seconds to catch primaries moved here meanwhile.
         * Called by {@link LockMembershipHandler}.
+        *
         * @param   departed
         *          Id of the departed member, for the log.
-...
              return;
+          }
           purgeLocalPrimaries(departed);
           try
+          {
              // the departed member may still be committing under its locks until its membership lease expired
              Thread.sleep(MembershipLease.GRACE_MS);
              purgeLocalPrimaries(departed);
              Thread.sleep(PURGE_AGAIN_MS);
+          }
           catch (InterruptedException exc)
-...
+       }
        /**
         * Name the record and table locks the current context holds, in every lock region of this server, for
         * the commit's per-owner membership lease ({@code MembershipLease.awaitCommit}).
+        *
         * @param   consumer
         *          Takes the lock region and the key of each lock.
         * @param   recordIds
         *          Takes the record id of each record lock (not of table locks).
         */
        public static void heldKeys(BiConsumer<Region<?, ?>, Object> consumer, LongConsumer recordIds)
+       {
           for (OwnerLockManager<?> manager : MANAGERS.values())
+          {
              Context<?> local = manager.context.get(false);
              Region<OwnerLockKey, OwnerLockEntry> region = manager.lockTable;
              if (local == null || region == null || local.locked.isEmpty())
+             {
                 continue;
+             }
              for (Map<? extends RecordIdentifier<?>, LockType> locks : local.locked.values())
+             {
                 for (RecordIdentifier<?> ident : locks.keySet())
+                {
                    consumer.accept(region, new OwnerLockKey(ident));
                    Long id = ident.getRecordID();
                    if (id != null)
+                   {
                       recordIds.accept(id);
+                   }
+                }
+             }
+          }
+       }
        /**
         * Re-initialisation participant ({@link ClusterGate#ORDER_LOCKS}): on Geode's reconnected cache, register
         * the membership listener again and bind every lock manager of {@link #MANAGERS} to its lock region again
         * (the regions of the closed cache are dropped from {@link #REGIONS}; their leases are replaced by
-...
              boolean fence = request.fence || attempt > 1;
              OwnerLockFunction.Request sent = request.sent(CALL_SEQ.incrementAndGet(), fence);
              long start = System.nanoTime();
              // the collector notes the owner which answers (its answer renews its membership lease)
              ResultCollector<Object, List<Object>> rc = (ResultCollector<Object, List<Object>>)
                 FunctionService.onRegion(region)
                                .withFilter(Collections.singleton(request.key))
                                .setArguments(sent)
                                .withCollector(MembershipLease.collector())
                                .execute(OwnerLockFunction.ID);
              try
+             {
     -- a/src/com/goldencode/p2j/distributed/orm/OwnerUniqueFunction.java
     ++ b/src/com/goldencode/p2j/distributed/orm/OwnerUniqueFunction.java
-...
     ** 001 ES  20261009 Initial version (#4369 owner-side unique tracker).
     **     ES  20261026 Liveness checks use LiveMembers (cached view), checked in purge() under the key's lock.
     **     ES  20261026 Removed an unused import.
     **     CA  20261029 A departed member's reservation is taken over or purged only after the grace of its
     **                  membership lease (LiveMembers.holdsLocks()).
     */
     /*
-...
              return new Answer(key, Outcome.MINE, null);
+          }
           if (!LiveMembers.isLive(stored.getMember()))
           if (!LiveMembers.holdsLocks(stored.getMember()))
+          {
              // a departed member's reservation never blocks anyone
              // a departed member's reservation never blocks anyone once its membership lease expired
              if (!region.replace(key, stored, holder))
+             {
                 return null;
-...
+       }
        /**
         * Remove the reservation of a key if it belongs to a member which left the cluster.
         * Remove the reservation of a key if it belongs to a member which left the cluster (after the grace of its
         * membership lease, {@link LiveMembers#holdsLocks}).
+        *
         * @param   region
         *          The reservation region.
-...
+          {
              UniqueReservation stored = region.get(key);
              return stored != null && !LiveMembers.isLive(stored.getMember()) && region.remove(key, stored);
              return stored != null && !LiveMembers.holdsLocks(stored.getMember()) && region.remove(key, stored);
+          }
           finally
+          {
     -- a/src/com/goldencode/p2j/distributed/orm/OwnerUniqueIndex.java
     ++ b/src/com/goldencode/p2j/distributed/orm/OwnerUniqueIndex.java
-...
     **                  again in a new epoch of the gate (re-initialisation participant); pending keys and keys
     **                  carry the epoch they were made in: a reservation of an older epoch is never released or
     **                  restored, and its state never replaces the state of the new epoch.
     **     CA  20261029 purgeDepartedMember() waits for the grace of the departed member's membership lease first;
     **                  pendingKeys() names the session's pending keys for the commit's per-owner lease; owner
     **                  calls note the member which answered (MembershipLease.collector()).
     */
     /*
-...
     package com.goldencode.p2j.distributed.orm;
     import java.lang.ref.*;
     import java.util.*;
     import java.util.concurrent.*;
     import java.util.concurrent.locks.*;
     import java.util.function.BiConsumer;
     import java.util.logging.*;
     import org.apache.geode.CancelException;
-...
        /** End-of-transaction removals of the current session, not sent yet. */
        private static final ThreadLocal<List<Release>> REMOVED = ThreadLocal.withInitial(ArrayList::new);
        /** Every index of this server (weakly held, by identity), for {@link #pendingKeys} (guarded by itself). */
        private static final List<WeakReference<OwnerUniqueIndex>> ALL = new ArrayList<>();
        /** Reservation regions by database name (lower case). */
        private static final Map<String, Region<ReservedKey, UniqueReservation>> REGIONS =
           new ConcurrentHashMap<>();
-...
           super(fieldMap, NO_LOCK, props);
           this.uniqueId = uniqueId;
           this.dbName = dbName;
           synchronized (ALL)
+          {
              ALL.removeIf(r -> r.get() == null);
              ALL.add(new WeakReference<>(this));
+          }
+       }
        /**
         * Name the unique keys pending in open transactions, of the current epoch,
         * in every reservation region, of the given records, for the commit's per-owner membership lease
         * ({@code MembershipLease.awaitCommit}): the records the current session holds locked (a session changes
         * or creates a record only under its lock), so the keys of other sessions' records are left out (a
         * record id of another table may add a key, which only adds a member to wait for).
+        *
         * @param   recordIds
         *          The record ids.
         * @param   consumer
         *          Takes the reservation region and the reserved key.
         */
        public static void pendingKeys(Set<Long> recordIds, BiConsumer<Region<?, ?>, Object> consumer)
+       {
           if (recordIds.isEmpty())
+          {
              return;
+          }
           List<OwnerUniqueIndex> indexes = new ArrayList<>();
           synchronized (ALL)
+          {
              for (WeakReference<OwnerUniqueIndex> r : ALL)
+             {
                 OwnerUniqueIndex index = r.get();
                 if (index != null)
+                {
                    indexes.add(index);
+                }
+             }
+          }
           long epoch = ClusterGate.epoch();
           for (OwnerUniqueIndex index : indexes)
+          {
              Region<ReservedKey, UniqueReservation> r = index.region;
              if (r == null || index.boundEpoch != epoch || index.pending.isEmpty())
+             {
                 continue;
+             }
              for (Long id : recordIds)
+             {
                 Pending p = index.pending.get(id);
                 if (p != null && p.epoch == epoch && !p.reserved.hasNulls())
+                {
                    consumer.accept(r, p.reserved);
+                }
+             }
+          }
+       }
        /**
-...
                                                              Set<ReservedKey>                       filter,
                                                              OwnerUniqueFunction.Request            request)
+       {
           // the collector notes the owners which answer (their answers renew their membership lease)
           ResultCollector<Object, List<Object>> rc = (ResultCollector<Object, List<Object>>)
              FunctionService.onRegion(region)
                             .withFilter(filter)
                             .setArguments(request)
                             .withCollector(MembershipLease.collector())
                             .execute(OwnerUniqueFunction.ID);
           Map<ReservedKey, OwnerUniqueFunction.Answer> answers = new HashMap<>();
-...
+       }
        /**
         * Remove a departed member's reservations from the keys this member owns, now and again a little
         * later (key ownership may still be moving).
         * Remove a departed member's reservations from the keys this member owns, once the grace of its
         * membership lease ended ({@link MembershipLease#GRACE_MS}), and again a little later (key ownership may
         * still be moving).
+        *
         * @param   departed
         *          Id of the departed member.
-...
              return;
+          }
           purgeLocalPrimaries(departed);
           try
+          {
              // the departed member may still be committing under its locks until its membership lease expired
              Thread.sleep(MembershipLease.GRACE_MS);
              purgeLocalPrimaries(departed);
              Thread.sleep(PURGE_AGAIN_MS);
+          }
           catch (InterruptedException exc)
     -- a/src/com/goldencode/p2j/persist/FastFindCache.java
     ++ b/src/com/goldencode/p2j/persist/FastFindCache.java
-...
     **     CA  20261029 Added getPermanentInstance(Type, String): the lookup by database type and name (the meta
     **                  database has the primary's name), used by the cluster commit notice; added
     **                  invalidateAllPermanent() (notices missed while this server was not a cluster member).
     **     CA  20261029 Removed the per-change invalidation broadcast to the other servers (broadcast(), the
     **                  name-only getPermanentInstance(String), setIsFromRemoteRequest()): the cluster commit
     **                  notice is the only invalidation across the servers.
     */
     /*
-...
     package com.goldencode.p2j.persist;
     import com.goldencode.cache.*;
     import com.goldencode.p2j.distributed.cache.*;
     import com.goldencode.p2j.distributed.config.*;
     import com.goldencode.cache.Cache.Entry;
     import com.goldencode.p2j.persist.orm.*;
     import com.goldencode.p2j.security.*;
-...
     import com.goldencode.p2j.util.logging.*;
     import com.goldencode.util.ResizableHashMap;
     import org.apache.geode.cache.execute.*;
     import org.apache.geode.distributed.*;
     import java.lang.ref.WeakReference;
     import java.util.*;
-...
        /** Differentiate between permanent and temporary tables and caches. */
        private final boolean isPermanent;
        /** Flag indicating distributed mode */
        private final boolean distributedMode;
        /**
         * Flag indicating this instance is switched off: it is for a permanent database and the directory
         * node {@code persistence/fast-find-cache} is {@code false}. A disabled instance never caches, never
         * invalidates and never broadcasts.
         * invalidates.
         */
        private final boolean disabled;
-...
        private final AtomicLong generation = new AtomicLong();
        /**
         * Thread-local flag to mark threads that are handling remote invalidation requests.
         */
        private static final ThreadLocal<Boolean> isFromRemoteRequest = ThreadLocal.withInitial(() -> false);
        /**
         * Constructor for internal use.
+        *
         * @param   isPermanent
-...
        private FastFindCache(boolean isPermanent, Database database)
+       {
           this.isPermanent = isPermanent;
           this.distributedMode = ClusterConfig.isEnabled();
           this.database  = database;
           this.disabled  = isPermanent && !permanentEnabled;
-...
         * for it. For the permanent databases, the cache can be switched off with the directory boolean node
         * {@code persistence/fast-find-cache} (default {@code true}, read once at class load). The instance
         * returned then is inert: {@code get} always misses, {@code put} and all {@code invalidate} variants
         * do nothing (and nothing is broadcast to other servers), and no state is accumulated. The temp-table
         * instances are not affected by this setting.
         * do nothing, and no state is accumulated. The temp-table instances are not affected by this setting.
+        *
         * @param   temporary
         *          Use {@code true} to get the instance object for a temp-table database. They are context-local
-...
+       }
        /**
         * Get the existing instance of a permanent database by the database's name, without a user context
         * (for the cluster invalidation functions, which run on Geode threads).
+        *
         * @param   dbName
         *          The name of the database, as {@link Database#getName()} returns it.
+        *
         * @return  The instance, or {@code null} if this server has none (it then caches nothing for that
         *          database).
         */
        public static FastFindCache getPermanentInstance(String dbName)
+       {
           for (Map.Entry<Database, FastFindCache> e : permanentInstances.entrySet())
+          {
              if (e.getKey().getName().equals(dbName))
+             {
                 return e.getValue();
+             }
+          }
           return null;
+       }
        /**
         * Get the existing instance of a permanent database by the database's type and name, without a user
         * context (for the cluster commit notice, which runs on Geode threads). The type tells apart a primary
         * database and the meta database, which have the same name.
-...
+       }
        /**
         * Marks the current thread as a remote-request handler. Once set, {@code broadcast()} becomes
         * a no-op for this thread, preventing infinite re-distribution of invalidation messages.
         */
        public static void setIsFromRemoteRequest()
+       {
           isFromRemoteRequest.set(true);
+       }
        /**
         * Query the cache for a value.
+        *
         * @param   k
-...
                             });
+             }
+          }
           if (distributedMode && isPermanent)
+          {
              broadcast(new FastFindCacheInvalidationRequest.InvalidateAllIndicesForTable(database.getName(),
                                                                                          dmoUid));
+          }
+       }
        /**
-...
+                }
+             }
+          }
           if (distributedMode && isPermanent)
+          {
              broadcast(new FastFindCacheInvalidationRequest.InvalidateIndices(
                 database.getName(), dmoUid, unique, nonUnique));
+          }
+       }
        /**
-...
+             {
                 l3.clear();
+             }
              if (distributedMode && isPermanent)
+             {
                 broadcast(new FastFindCacheInvalidationRequest.InvalidateIndex(
                    database.getName(), dmoUid, idxUid));
+             }
+          }
+       }
-...
                 map.forEach((idxId, cache) -> cache.clear());
+             }
+          }
           if (distributedMode && isPermanent)
+          {
              broadcast(new FastFindCacheInvalidationRequest.InvalidateDmoList(
                 database.getName(), dmoList));
+          }
+       }
        /**
         * Broadcasts a cache invalidation request to all other FWD server members in the distributed
         * system. Non-server members (e.g. Geode locators) are excluded, since they can neither
         * deserialize nor execute FWD functions.
         * <p>
         * To prevent cyclic broadcasting (echoes) across the cluster, this method immediately
         * returns if the current execution context was triggered by a remote request.
         * If there are other active FWD servers in the Geode cluster, it uses the
         * {@code FunctionService} to execute the {@code FastFindCacheInvalidationFunction}
         * on those peers, passing the invalidation payload.
         * </p>
+        *
         * @param request
         *        The payload containing the specific cache invalidation details to broadcast.
         */
        private void broadcast(FastFindCacheInvalidationRequest request)
+       {
           if (isFromRemoteRequest.get())
+          {
              return;
+          }
           if (com.goldencode.p2j.distributed.cache.CommitNotice.isEnabled())
+          {
              // the other servers learn of committed changes from the commit notice; these invalidations are
              // mostly sent before the commit (useless there: the change is not visible yet) and name the table
              // by this JVM's DMO uid, which is another table, or none, on another server
              return;
+          }
           Set<DistributedMember> otherMembers = GeodeConfig.getOtherServerMembers();
           if (otherMembers.isEmpty())
+          {
              return;
+          }
           FunctionService.onMembers(otherMembers)
                          .setArguments(request)
                          .execute(FastFindCacheInvalidationFunction.class.getName());
+       }

... This diff was truncated because it exceeds the maximum size that can be displayed.

Project

General

Profile

FWD » Core Development » Database

commit-guard-fastfind-r16812.patch