--- a/src/com/goldencode/p2j/distributed/cache/CommitNotice.java +++ b/src/com/goldencode/p2j/distributed/cache/CommitNotice.java @@ -15,6 +15,12 @@ ** 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). */ /* @@ -105,9 +111,23 @@ * 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. + *

+ * 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: + *

+ * 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 @@ -135,8 +155,9 @@ * 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). *

- * 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 @@ -150,9 +171,12 @@ /** 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; @@ -170,6 +194,18 @@ /** 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 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; @@ -213,13 +249,19 @@ } /** - * 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(); } /** @@ -274,45 +316,47 @@ * 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 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 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 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) @@ -326,6 +370,75 @@ 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 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 members, String why) + { + synchronized (resyncPending) + { + if (members == null) + { + resyncAll = true; + } + else + { + resyncPending.addAll(members); + } + + if (resyncScheduled) + { + return; + } + resyncScheduled = true; + } + + resyncer().schedule(() -> + { + Set 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); } /** @@ -392,6 +505,7 @@ catch (Exception exc) { LOG.log(Level.WARNING, "Commit notice batch not sent: " + exc, exc); + resync(null, "commit notice batch not sent"); } } @@ -418,6 +532,28 @@ } /** + * 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. @@ -460,8 +596,6 @@ dmoIds.add(FastFindCache.combine(dmoMeta.getId(), null)); } } - // a remote request: invalidate here only, never broadcast again - FastFindCache.setIsFromRemoteRequest(); ffCache.invalidate(dmoIds); } @@ -541,7 +675,8 @@ * 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. @@ -576,6 +711,18 @@ 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); } @@ -639,4 +786,200 @@ 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 + { + /** {@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 answered = ConcurrentHashMap.newKeySet(); + + /** The answers. */ + private final List 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 missing(Set sentTo) + { + Set 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 + { + /** 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 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 @@ -10,6 +10,8 @@ ** 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. */ /* @@ -191,6 +193,12 @@ */ 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; @@ -252,6 +260,12 @@ */ 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 PARTICIPANTS = new CopyOnWriteArrayList<>(); @@ -604,6 +618,48 @@ } /** + * 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 @@ -723,13 +779,17 @@ 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() { @@ -754,6 +814,8 @@ } LOG.log(Level.INFO, "Cluster gate: closed (the server shuts down)"); + + awaitCommits(); } /** @@ -810,6 +872,44 @@ } /** + * 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 @@ -28,6 +28,9 @@ ** 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(). */ /* @@ -279,8 +282,10 @@ 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()); @@ -334,8 +339,7 @@ */ private static void missedNotices() { - FastFindCache.invalidateAllPermanent(); - DmoVersioning.incrementRemoteAll(); + com.goldencode.p2j.distributed.cache.CommitNotice.missed("this server was not a member of the cluster"); } /** @@ -546,6 +550,19 @@ } /** + * 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 @@ -8,6 +8,8 @@ ** 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). */ /* @@ -91,6 +93,11 @@ * 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. * + * 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 { @@ -100,12 +107,21 @@ /** 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 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 GONE = new ConcurrentHashMap<>(); + /** The cached view; {@code null} to build a new one at the next check. */ private static volatile View view = null; @@ -154,6 +170,33 @@ } /** + * 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 lockerChecker() + { + Predicate 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). * @@ -193,6 +236,9 @@ 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; } @@ -205,6 +251,27 @@ } /** + * 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 @@ -9,6 +9,8 @@ ** 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()). */ /* @@ -153,7 +155,7 @@ @Override public void memberLeft(MembershipEvent event) { - departed(event.getMemberId(), "left"); + departed(event, "left"); } /** @@ -165,20 +167,24 @@ @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()) @@ -193,6 +199,37 @@ 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 @@ -34,6 +34,8 @@ ** 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()). */ /* @@ -1127,7 +1129,8 @@ */ private static boolean purgeHere(Region region, Object key) { - Predicate live = LiveMembers.checker(); + // a departed member's locks stand until its membership lease expired (it may still be committing) + Predicate live = LiveMembers.lockerChecker(); boolean changed = false; List front = Collections.emptyList(); Object qk = new QueueKey(region.getName(), key); @@ -1219,18 +1222,18 @@ } LockType requested = LockType.fromInt(req.type).toWaitVariant(); - Predicate 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 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"); @@ -1241,8 +1244,7 @@ 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 @@ -67,6 +67,9 @@ ** 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()). */ /* @@ -128,6 +131,8 @@ 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; @@ -958,8 +963,10 @@ } /** - * 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. @@ -971,9 +978,11 @@ 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) @@ -1029,6 +1038,41 @@ } /** + * 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, Object> consumer, LongConsumer recordIds) + { + for (OwnerLockManager manager : MANAGERS.values()) + { + Context local = manager.context.get(false); + Region region = manager.lockTable; + if (local == null || region == null || local.locked.isEmpty()) + { + continue; + } + + for (Map, 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 @@ -2926,10 +2970,12 @@ 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> rc = (ResultCollector>) 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 @@ -8,6 +8,8 @@ ** 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()). */ /* @@ -300,9 +302,9 @@ 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; @@ -324,7 +326,8 @@ } /** - * 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. @@ -341,7 +344,7 @@ { 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 @@ -30,6 +30,9 @@ ** 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()). */ /* @@ -88,9 +91,11 @@ 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; @@ -171,6 +176,9 @@ /** End-of-transaction removals of the current session, not sent yet. */ private static final ThreadLocal> REMOVED = ThreadLocal.withInitial(ArrayList::new); + /** Every index of this server (weakly held, by identity), for {@link #pendingKeys} (guarded by itself). */ + private static final List> ALL = new ArrayList<>(); + /** Reservation regions by database name (lower case). */ private static final Map> REGIONS = new ConcurrentHashMap<>(); @@ -233,6 +241,64 @@ 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 recordIds, BiConsumer, Object> consumer) + { + if (recordIds.isEmpty()) + { + return; + } + + List indexes = new ArrayList<>(); + synchronized (ALL) + { + for (WeakReference r : ALL) + { + OwnerUniqueIndex index = r.get(); + if (index != null) + { + indexes.add(index); + } + } + } + + long epoch = ClusterGate.epoch(); + for (OwnerUniqueIndex index : indexes) + { + Region 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); + } + } + } } /** @@ -422,10 +488,12 @@ Set filter, OwnerUniqueFunction.Request request) { + // the collector notes the owners which answer (their answers renew their membership lease) ResultCollector> rc = (ResultCollector>) FunctionService.onRegion(region) .withFilter(filter) .setArguments(request) + .withCollector(MembershipLease.collector()) .execute(OwnerUniqueFunction.ID); Map answers = new HashMap<>(); @@ -1298,8 +1366,9 @@ } /** - * 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. @@ -1311,9 +1380,11 @@ 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 @@ -61,6 +61,9 @@ ** 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. */ /* @@ -119,8 +122,6 @@ 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.*; @@ -128,8 +129,6 @@ 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.*; @@ -212,13 +211,10 @@ /** 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; @@ -230,11 +226,6 @@ private final AtomicLong generation = new AtomicLong(); /** - * Thread-local flag to mark threads that are handling remote invalidation requests. - */ - private static final ThreadLocal isFromRemoteRequest = ThreadLocal.withInitial(() -> false); - - /** * Constructor for internal use. * * @param isPermanent @@ -243,7 +234,6 @@ private FastFindCache(boolean isPermanent, Database database) { this.isPermanent = isPermanent; - this.distributedMode = ClusterConfig.isEnabled(); this.database = database; this.disabled = isPermanent && !permanentEnabled; @@ -258,8 +248,7 @@ * 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 @@ -307,29 +296,6 @@ } /** - * 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 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. @@ -488,15 +454,6 @@ } /** - * 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 @@ -746,12 +703,6 @@ }); } } - - if (distributedMode && isPermanent) - { - broadcast(new FastFindCacheInvalidationRequest.InvalidateAllIndicesForTable(database.getName(), - dmoUid)); - } } /** @@ -928,13 +879,6 @@ } } } - - if (distributedMode && isPermanent) - { - broadcast(new FastFindCacheInvalidationRequest.InvalidateIndices( - database.getName(), dmoUid, unique, nonUnique)); - - } } /** @@ -957,12 +901,6 @@ { l3.clear(); } - - if (distributedMode && isPermanent) - { - broadcast(new FastFindCacheInvalidationRequest.InvalidateIndex( - database.getName(), dmoUid, idxUid)); - } } } @@ -982,58 +920,9 @@ 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. - *

- * 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. - *

- * - * @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 otherMembers = GeodeConfig.getOtherServerMembers(); - - if (otherMembers.isEmpty()) - { - return; - } - - FunctionService.onMembers(otherMembers) - .setArguments(request) - .execute(FastFindCacheInvalidationFunction.class.getName()); - } - - - /** * The second level key for Fast Find algorithm. It uses this class as cache key for quickly checking * whether a particular record was found recently. */ --- a/src/com/goldencode/p2j/persist/TxWrapper.java +++ b/src/com/goldencode/p2j/persist/TxWrapper.java @@ -42,6 +42,8 @@ ** the record locks are released, which happens also if sending fails ** (WorkArea.commitNotices, given to the SavepointManager). commit() refuses a session ** terminated by the loss of the cluster membership (ClusterGate.checkCommit). +** CA 20261029 commit() runs in the cluster gate's commit fence and waits for a valid membership lease +** (MembershipLease.awaitCommit) before the gate's checks (not for the meta database). */ /* @@ -305,7 +307,11 @@ *

* In cluster mode, a session terminated by the loss of this server's cluster membership is terminated here * instead of committing, and a commit while this server is not a member of the cluster raises the STOP of - * a disconnected database ({@link ClusterGate#checkCommit}). + * a disconnected database ({@link ClusterGate#checkCommit}). The commit first waits until the owners of + * the locks and unique keys it relies on are known to still count this server a member + * ({@link MembershipLease#awaitCommit}: after a pause longer than Geode's member timeout they do not, + * although Geode has not told this server yet), and it runs inside the gate's commit fence + * ({@link ClusterGate#enterCommit}), so FWD's shutdown does not close the cache meanwhile. * * @param transaction * {@code true} if this is a full transaction and {@code false} if this is only a @@ -324,20 +330,34 @@ LOG.log(Level.FINE, message(persistenceCtx.getPersistence(), "master transaction commit")); } - // cluster: a session terminated by the loss of this server's cluster membership never commits (its - // locks and reservations were purged by the other servers), and nothing commits while this server is - // not a member of the cluster - if (!database.isTemporary() && ClusterConfig.isEnabled()) + boolean cluster = !database.isTemporary() && ClusterConfig.isEnabled(); + + // the meta database is not shared through the cluster: no lease, no fence (the gate's checks stay) + boolean fenced = cluster && !database.isMeta(); + boolean interrupted = false; + if (fenced) { - ClusterGate.checkCommit(database); - if (OwnerLockManager.holdsStaleLocks()) - { - ClusterGate.terminateCurrent(null); - } + // FWD's shutdown waits for this commit before it closes the cache (or refuses it once closing) + ClusterGate.enterCommit(database); } try { + if (cluster) + { + // cluster: no commit unless the owners of its locks still count this server a member (it may + // have been paused past the member timeout, its locks freed by the others); a session terminated + // by the loss of this server's cluster membership never commits (its locks and reservations were + // purged by the other servers), and nothing commits while this server is not a member of the + // cluster + interrupted = fenced && MembershipLease.awaitCommit(database); + ClusterGate.checkCommit(database); + if (OwnerLockManager.holdsStaleLocks()) + { + ClusterGate.terminateCurrent(null); + } + } + initialTx = false; autoRollback = false; persistenceCtx.commit(); @@ -353,6 +373,19 @@ // No good way to recover from this. Let it abend. throw new RuntimeException(exc); } + finally + { + if (fenced) + { + ClusterGate.exitCommit(); + } + + if (interrupted) + { + // kept during the lease wait, set again after the database commit + Thread.currentThread().interrupt(); + } + } } /** --- /dev/null +++ b/src/com/goldencode/p2j/distributed/config/MembershipLease.java @@ -0,0 +1,1108 @@ +/* +** Module : MembershipLease.java +** Abstract : The per-owner membership lease: a commit waits until every member which could free a lock or +** unique key reservation the transaction relies on is known to still count this server a member. +** +** Copyright (c) 2026, Golden Code Development Corporation. +** +** -#- -I- --Date-- ---------------------------------------Description---------------------------------------- +** 001 CA 20261029 Initial version (#4369): per-member freshness from the send time of the last answered call +** (monotonic and wall clock); commits wait until the primary and redundant hosts of their +** keys are fresh (awaitCommit; Check routed by key, Ping); background pings of the owners +** recent commits needed; departed members' locks are kept GRACE_MS (LiveMembers). +*/ + +/* +** This program is free software: you can redistribute it and/or modify +** it under the terms of the GNU Affero General Public License as +** published by the Free Software Foundation, either version 3 of the +** License, or (at your option) any later version. +** +** This program is distributed in the hope that it will be useful, +** but WITHOUT ANY WARRANTY; without even the implied warranty of +** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +** GNU Affero General Public License for more details. +** +** You may find a copy of the GNU Affero GPL version 3 at the following +** location: https://www.gnu.org/licenses/agpl-3.0.en.html +** +** Additional terms under GNU Affero GPL version 3 section 7: +** +** Under Section 7 of the GNU Affero GPL version 3, the following additional +** terms apply to the works covered under the License. These additional terms +** are non-permissive additional terms allowed under Section 7 of the GNU +** Affero GPL version 3 and may not be removed by you. +** +** 0. Attribution Requirement. +** +** You must preserve all legal notices or author attributions in the covered +** work or Appropriate Legal Notices displayed by works containing the covered +** work. You may not remove from the covered work any author or developer +** credit already included within the covered work. +** +** 1. No License To Use Trademarks. +** +** This license does not grant any license or rights to use the trademarks +** Golden Code, FWD, any Golden Code or FWD logo, or any other trademarks +** of Golden Code Development Corporation. You are not authorized to use the +** name Golden Code, FWD, or the names of any author or contributor, for +** publicity purposes without written authorization. +** +** 2. No Misrepresentation of Affiliation. +** +** You may not represent yourself as Golden Code Development Corporation or FWD. +** +** You may not represent yourself for publicity purposes as associated with +** Golden Code Development Corporation, FWD, or any author or contributor to +** the covered work, without written authorization. +** +** 3. No Misrepresentation of Source or Origin. +** +** You may not represent the covered work as solely your work. All modified +** versions of the covered work must be marked in a reasonable way to make it +** clear that the modified work is not originating from Golden Code Development +** Corporation or FWD. All modified versions must contain the notices of +** attribution required in this license. +*/ + +package com.goldencode.p2j.distributed.config; + +import java.util.*; +import java.util.concurrent.*; +import java.util.logging.*; + +import org.apache.geode.*; +import org.apache.geode.cache.*; +import org.apache.geode.cache.execute.*; +import org.apache.geode.cache.partition.*; +import org.apache.geode.distributed.*; +import org.apache.geode.internal.cache.PartitionedRegion; +import org.apache.geode.internal.cache.PartitionedRegionHelper; + +import com.goldencode.p2j.distributed.lock.*; +import com.goldencode.p2j.distributed.orm.*; +import com.goldencode.p2j.persist.*; +import com.goldencode.p2j.util.StopConditionException; +import com.goldencode.p2j.util.logging.*; + +/** + * The per-owner membership lease: a commit to a database shared by the cluster is allowed only while every + * member which could free a lock or unique key reservation the transaction relies on is known to still count + * this server as a member, so that it never commits after its locks were given to another server (a pause + * longer than Geode's member timeout: a long GC, a frozen VM, an OS suspend). + *

+ * Who can free a lock. A lock or reservation is an entry of a partitioned region, kept by the member + * hosting the key's primary bucket (the owner); a member hosting a redundant copy becomes the owner if + * the primary leaves. Only the owner drops another member's entries, and only after it saw that member depart + * and {@link #GRACE_MS} more ({@code LiveMembers}). The members which matter for a key are therefore its + * primary and its redundant copies, as this server's partition metadata names them; this server itself + * never drops its own entries. A key whose primary is stale or not known is checked through a call routed by + * key; a member which is not the primary any more answers so ("moved") and the key is checked again. + *

+ * Freshness, per member. An answer of member M to a call this server sent at time {@code t} (a + * lock or unique key call, a {@link Ping}, a {@link Check}, a commit notice) proves that M still had this + * server in its view at {@code t}: Geode drops the messages of a member removed from the view. So M cannot + * drop this server's entries before {@code t + GRACE_MS}. The send time is kept per member + * ({@link #answered}, monotonic and wall clock, in the cluster gate's epoch); M is fresh for + * {@link #LEASE_MS} ms after it. A commit whose required members are all fresh ({@link #awaitCommit}) + * reaches the database at least {@code GRACE_MS - LEASE_MS} ms before any of them could drop an entry it + * relies on, unless this server stalls that long between the check and the database commit (that window + * can only be closed by fencing at the database). The monotonic clock keeps running during a pause; the + * wall clock also counts an OS suspend. + *

+ * Cost. A commit costs a look-up of the owners of its keys and clock reads while they are fresh. A + * stale owner gets one round of calls, sent to all stale members in parallel (a {@link Check} routed by key + * per region, and a {@link Ping} to the stale redundant copies), repeated until all are fresh, at most + * {@link #waitMs} ms in all: the wait is per transaction and does not grow with the number of servers. A + * background thread pings, every {@link #PING_MS} ms, the members which answered this server's owner calls + * (lock and unique key calls) or which commits needed, within the last {@link #INTEREST_MS} ms, and which no + * call answered within {@link #PING_MS} ms, so that a commit seldom waits (a transaction which held its + * locks longer than that may wait for one round). Members which own nothing this server relies on are + * never asked: a slow or paused server delays or refuses only the transactions which hold locks or + * reservations on its buckets (as a primary or a copy). + */ +public final class MembershipLease +{ + /** How long an answer keeps its member fresh, in ms from the time the call was sent. */ + public static final long LEASE_MS = 2000L; + + /** + * How long the owners keep a departed member's locks and unique key reservations, in ms after they learned + * of its departure: the lease and a margin for the time between the lease check and the database commit, + * and for the time the servers install a new view. + */ + public static final long GRACE_MS = LEASE_MS + 3000L; + + /** Logger. */ + private static final CentralLogger LOG = CentralLogger.get(MembershipLease.class.getName()); + + /** Interval (ms) of the background pings; an answer within it makes a ping unnecessary. */ + private static final long PING_MS = 500L; + + /** How long (ms) a member which answered an owner call or was needed by a commit stays pinged. */ + private static final long INTEREST_MS = 60_000L; + + /** The longest wait (ms) of one round of calls for the answers. */ + private static final long CALL_MS = 5000L; + + /** The least bound (ms) of a commit's wait for its owners ({@link #waitMs}). */ + private static final long MIN_WAIT_MS = 30_000L; + + /** Pause (ms) after a round of calls which ended at once without making every owner fresh. */ + private static final long RETRY_PAUSE_MS = 100L; + + /** Geode's default member timeout (ms), when the distributed system does not tell it. */ + private static final long DEFAULT_MEMBER_TIMEOUT_MS = 5000L; + + /** + * The send time of the last answered call of each member: {@code System.nanoTime()}, + * {@code System.currentTimeMillis()} and the cluster gate's epoch. + */ + private static final ConcurrentHashMap ANSWERED = new ConcurrentHashMap<>(); + + /** + * The members which answered owner calls or which commits needed ({@code System.nanoTime()} of the last + * time), for the background pings. + */ + private static final ConcurrentHashMap INTEREST = new ConcurrentHashMap<>(); + + /** The background pinger, once started (guarded by INTEREST). */ + private static Thread pinger = null; + + /** + * The bound (ms) of a commit's wait for its owners: at least {@link #MIN_WAIT_MS}, and long enough for + * Geode to remove an unresponsive owner from the view and another member to take over its buckets (four + * member timeouts and 5 s). + */ + private static volatile long waitMs = MIN_WAIT_MS; + + /** + * Not instantiable. + */ + private MembershipLease() + { + } + + /** + * Note an answer of a member to a call this server sent: the member is fresh from the send time on (see + * the class description). Older send times never replace newer ones. + * + * @param member + * The member which answered. + * @param sentNanos + * {@code System.nanoTime()} when the call was sent. + * @param sentWall + * {@code System.currentTimeMillis()} when the call was sent. + * @param epoch + * The cluster gate's epoch when the call was sent. + */ + public static void answered(DistributedMember member, long sentNanos, long sentWall, long epoch) + { + if (member == null) + { + return; + } + + ANSWERED.merge(member.getId(), new long[] { sentNanos, sentWall, epoch }, (old, now) -> + { + return (now[2] > old[2] || (now[2] == old[2] && now[0] - old[0] > 0)) ? now : old; + }); + } + + /** + * A result collector for the owner calls of the lock and unique key regions: it notes every member which + * answered ({@link #answered}) and keeps it among the members the background pings keep fresh (it owns a + * lock or reservation this server asked for), and collects the answers in a list, as Geode's default + * collector does. Create it right before the call is sent. + * + * @return The collector. + */ + public static ResultCollector> collector() + { + return new Answers(true); + } + + /** + * Wait until every member which could free a lock or unique key reservation the current context relies + * on is fresh (see the class description), before a commit to a database shared by the cluster. Returns at + * once while the cluster is not started yet, or when they are fresh (or there is none). Called on the + * session's thread by {@code TxWrapper}, before the gate's own checks. + * + * @param db + * The database whose transaction commits. + * + * @return {@code true} if the thread was interrupted during the wait: the caller sets the interrupt + * again once the database commit is done (it is not set again before it, as a JDBC driver may + * close an embedded database's files on an interrupt). + * + * @throws StopConditionException + * the STOP of a disconnected database ({@link DatabaseConnectionException}) if the owners could + * not be confirmed in time, or the cluster gate's STOP or termination of the session if the gate + * is not running ({@link ClusterGate#checkCommit}). + */ + public static boolean awaitCommit(Database db) + { + if (ClusterGate.state() == ClusterGate.State.STARTING) + { + return false; + } + + List held = held(); + if (held.isEmpty()) + { + return false; + } + + boolean interrupted = false; + long start = System.nanoTime(); + long bound = waitMs; + int rounds = 0; + + // keys a check found moved: checked by key again, whatever the metadata says, until a check succeeds; + // keys whose current primary answered a check routed by key (send time), whatever the metadata says + Map, Set> moved = new HashMap<>(); + Map, Map> confirmed = new HashMap<>(); + while (true) + { + if (!ClusterGate.isRunning()) + { + if (interrupted) + { + Thread.currentThread().interrupt(); + } + + // the gate terminates a session of an older epoch, or raises its STOP; CLOSED is refused here + ClusterGate.checkCommit(db); + + throw ClusterGate.unavailable(db, null); + } + + Required req = required(held, moved, confirmed); + if (req.isEmpty()) + { + break; + } + + long left = bound - (System.nanoTime() - start) / 1_000_000L; + if (left <= 0) + { + if (interrupted) + { + Thread.currentThread().interrupt(); + } + + String msg = "Database " + db.getName() + " is not available. " + ClusterGate.NOT_READY + ": " + + "the owners of the locks of this transaction (" + req + ") could not confirm the " + + "membership of this server in " + bound + " ms (" + ClusterGate.describe() + ")"; + LOG.log(Level.WARNING, "Membership lease: commit refused: " + msg); + + throw new DatabaseConnectionException(msg, null, db); + } + + rounds++; + long roundStart = System.nanoTime(); + interrupted |= round(req, Math.min(CALL_MS, left), moved, confirmed); + if (!required(held, moved, confirmed).isEmpty() && + (System.nanoTime() - roundStart) / 1_000_000L < RETRY_PAUSE_MS) + { + // ended at once (no member reachable, moved buckets): not a busy loop + try + { + Thread.sleep(RETRY_PAUSE_MS); + } + catch (InterruptedException exc) + { + interrupted = true; + } + } + } + + if (rounds > 0) + { + long waited = (System.nanoTime() - start) / 1_000_000L; + LOG.log(waited >= LEASE_MS ? Level.INFO : Level.FINE, + "Membership lease: a commit waited " + waited + " ms (" + rounds + " round(s)) for its " + + "owners"); + } + + return interrupted; + } + + /** + * Start the background pinger, once. Called by {@link ClusterGate} when the cache was created. + * + * @param c + * The cache, whose distributed system tells Geode's member timeout. + */ + static void start(Cache c) + { + long mt = DEFAULT_MEMBER_TIMEOUT_MS; + try + { + String v = c.getDistributedSystem().getProperties().getProperty("member-timeout"); + if (v != null) + { + mt = Long.parseLong(v.trim()); + } + } + catch (RuntimeException exc) + { + LOG.log(Level.FINE, "Membership lease: member-timeout not known, " + mt + " ms assumed", exc); + } + waitMs = Math.max(MIN_WAIT_MS, 4L * mt + 5000L); + + synchronized (INTEREST) + { + if (pinger != null) + { + return; + } + + pinger = new Thread(MembershipLease::pingLoop, "FWD-Cluster-Lease"); + pinger.setDaemon(true); + pinger.start(); + } + + LOG.log(Level.INFO, "Membership lease (per owner): " + LEASE_MS + " ms; owners pinged every " + + PING_MS + " ms; departed members' locks kept " + GRACE_MS + " ms; a commit waits " + + "at most " + waitMs + " ms for its owners (member-timeout " + mt + " ms)"); + } + + /** + * Whether a member is fresh: an answer to a call sent less than {@link #LEASE_MS} ms ago (by the monotonic + * and by the wall clock; a wall clock time in the future counts as stale), in the current epoch. + * + * @param member + * The member id. + * @param within + * The age (ms) below which the member counts as fresh. + * + * @return {@code true} if it is. + */ + private static boolean isFresh(String member, long within) + { + long[] a = ANSWERED.get(member); + if (a == null || a[2] != ClusterGate.epoch()) + { + return false; + } + + long wallAge = System.currentTimeMillis() - a[1]; + + return System.nanoTime() - a[0] < within * 1_000_000L && wallAge >= 0 && wallAge < within; + } + + /** + * The locks and unique key reservations the current context relies on: its record and table locks in + * every lock region, and the pending unique keys of the records it holds locked (a session changes or + * creates a record only under its lock). + * + * @return The keys with their regions. + */ + private static List held() + { + List held = new ArrayList<>(); + Set ids = new HashSet<>(); + OwnerLockManager.heldKeys((region, key) -> held.add(new Held(region, key)), ids::add); + + // a session holds the lock of every record whose unique keys it has pending (changed or created) + OwnerUniqueIndex.pendingKeys(ids, (region, key) -> held.add(new Held(region, key))); + + return held; + } + + /** + * The members which must answer before the commit: for each key, its primary and the members hosting a + * redundant copy, as this server's partition metadata names them, except this server; and the keys whose + * primary is not known or which a check found moved. Fresh members are left out. The members are noted + * for the background pings. + * + * @param held + * The keys. + * @param moved + * The keys a check found moved, by region: checked by key again. + * @param confirmed + * The keys whose current primary answered a check routed by key, by region, with the send time + * ({@code System.nanoTime()}, {@code System.currentTimeMillis()}): their primary needs nothing + * more within {@link #LEASE_MS} ms. + * + * @return What must answer. + */ + private static Required required(List held, + Map, Set> moved, + Map, Map> confirmed) + { + Required req = new Required(); + moved.forEach((r, k) -> req.byKey.computeIfAbsent(r, x -> new HashSet<>()).addAll(k)); + String self = selfId(); + long now = System.nanoTime(); + Map, Map> buckets = new HashMap<>(); + for (Held h : held) + { + // one look-up per bucket (Geode waits for a primary only while a bucket's promotion is under way) + Object[] hosts = hosts(h, buckets); + DistributedMember primary = (DistributedMember) hosts[0]; + @SuppressWarnings("unchecked") + Set owners = (Set) hosts[1]; + + Map ok = confirmed.get(h.region); + long[] at = (ok == null) ? null : ok.get(h.key); + boolean confirmedNow = at != null && + now - at[0] < LEASE_MS * 1_000_000L && + System.currentTimeMillis() - at[1] >= 0 && + System.currentTimeMillis() - at[1] < LEASE_MS; + if (!confirmedNow) + { + if (primary == null) + { + // not sure: checked by a call routed by key (moved or no primary: answered so, or failed) + req.byKey.computeIfAbsent(h.region, r -> new HashSet<>()).add(h.key); + } + else + { + need(req, h, primary, self, now, true); + } + } + + for (DistributedMember m : owners) + { + if (!m.equals(primary)) + { + need(req, h, m, self, now, false); + } + } + } + + return req; + } + + /** + * The primary and all hosts of the bucket of a key, from this server's partition metadata, once per bucket + * and call ({@code buckets}). Geode waits for the election of a primary only while a bucket which has + * hosts has none (the moment of a promotion after its primary left), so at most once per such bucket. + * + * @param h + * The key. + * @param buckets + * The hosts already looked up, by region and bucket id. + * + * @return The primary ({@code null} if none is known now) and the set of all hosts. + */ + private static Object[] hosts(Held h, Map, Map> buckets) + { + try + { + PartitionedRegion pr = (PartitionedRegion) h.region; + int bucket = PartitionedRegionHelper.getHashKey(pr, h.key); + Map known = buckets.computeIfAbsent(h.region, r -> new HashMap<>()); + Object[] hosts = known.get(bucket); + if (hosts == null) + { + @SuppressWarnings("unchecked") + Region r = (Region) h.region; + Set all = new HashSet<>(PartitionRegionHelper.getAllMembersForKey(r, h.key)); + + // without a host the bucket has no primary to wait for (Geode would wait for its election) + DistributedMember primary = null; + if (!all.isEmpty()) + { + primary = PartitionRegionHelper.getPrimaryMemberForKey(r, h.key); + } + hosts = new Object[] { primary, all }; + known.put(bucket, hosts); + } + + return hosts; + } + catch (RuntimeException exc) + { + // not known: checked by key + return new Object[] { null, Collections.emptySet() }; + } + } + + /** + * Add a member which must answer to the required ones, unless it is this server or fresh. + * + * @param req + * The required ones. + * @param h + * The key. + * @param m + * The member. + * @param self + * This server's member id, or {@code null}. + * @param now + * {@code System.nanoTime()} now. + * @param primary + * {@code true} if the member hosts the key's primary. + */ + private static void need(Required req, Held h, DistributedMember m, String self, long now, boolean primary) + { + String id = m.getId(); + if (id.equals(self)) + { + return; + } + + INTEREST.put(id, now); + if (isFresh(id, LEASE_MS)) + { + return; + } + + if (primary) + { + // routed by key (by this server's metadata): the member asked answers whether it still is the + // primary; a lagging metadata shows as a "moved" answer or a failure, checked again + req.byKey.computeIfAbsent(h.region, r -> new HashSet<>()).add(h.key); + } + else + { + req.members.put(id, m); + } + } + + /** + * One round of calls: a {@link Check} routed by the keys of each region whose primary is stale or not + * known, and a {@link Ping} to the stale members hosting redundant copies, all sent before any is waited + * for; then the answers are awaited, at most {@code timeoutMs} ms in all. Answers are noted by the + * collectors ({@link #answered}); failures and timeouts only leave members stale. The keys of a region + * whose check got a "moved" answer, or no complete answer, stay in {@code moved} (checked by key again); + * those of a region whose check every primary answered are removed from it. + * + * @param req + * What must answer. + * @param timeoutMs + * The longest wait (ms). + * @param moved + * The keys to check by key again, by region (updated). + * @param confirmed + * The keys whose current primary answered a check, by region, with the send times (updated). + * + * @return {@code true} if the thread was interrupted (the interrupt is cleared and kept by the caller). + */ + private static boolean round(Required req, + long timeoutMs, + Map, Set> moved, + Map, Map> confirmed) + { + // a pending interrupt could make the sends fail: cleared first, returned to the caller + boolean interrupted = Thread.interrupted(); + long[] sent = { System.nanoTime(), System.currentTimeMillis() }; + List>> calls = new ArrayList<>(); + Map>, Region> checks = new IdentityHashMap<>(); + for (Map.Entry, Set> e : req.byKey.entrySet()) + { + try + { + ResultCollector> rc = FunctionService.onRegion(e.getKey()) + .withFilter(e.getValue()) + .withCollector(new Answers(false)) + .execute(Check.ID); + calls.add(rc); + checks.put(rc, e.getKey()); + } + catch (RuntimeException exc) + { + LOG.log(Level.FINE, "Membership lease: check of " + e.getKey().getName() + " not sent: " + exc); + } + } + + if (!req.members.isEmpty()) + { + try + { + calls.add(FunctionService.onMembers(new HashSet<>(req.members.values())) + .withCollector(new Answers(false)) + .execute(Ping.ID)); + } + catch (RuntimeException exc) + { + LOG.log(Level.FINE, "Membership lease: ping not sent: " + exc); + } + } + + interrupted |= Thread.interrupted(); + long end = System.nanoTime() + timeoutMs * 1_000_000L; + for (ResultCollector> rc : calls) + { + long ms = Math.max(1L, (end - System.nanoTime()) / 1_000_000L); + Region region = checks.get(rc); + try + { + List answers = rc.getResult(ms, TimeUnit.MILLISECONDS); + if (region != null) + { + if (!answers.isEmpty() && !answers.contains(Boolean.FALSE) && allBoolean(answers)) + { + moved.remove(region); + Map ok = confirmed.computeIfAbsent(region, r -> new HashMap<>()); + req.byKey.get(region).forEach(k -> ok.put(k, sent)); + } + else + { + moved.computeIfAbsent(region, r -> new HashSet<>()).addAll(req.byKey.get(region)); + } + } + } + catch (InterruptedException exc) + { + interrupted = true; + } + catch (RuntimeException exc) + { + // a member did not answer in time, left meanwhile, or this member was removed; the keys of a + // check are checked by key again (their metadata may name a member which no longer is primary) + LOG.log(Level.FINE, "Membership lease: call not answered: " + exc); + if (region != null) + { + moved.computeIfAbsent(region, r -> new HashSet<>()).addAll(req.byKey.get(region)); + } + } + } + + return interrupted | Thread.interrupted(); + } + + /** + * Whether every answer of a check is a {@code Boolean} (no exception sent back). + * + * @param answers + * The answers. + * + * @return {@code true} if they all are. + */ + private static boolean allBoolean(List answers) + { + synchronized (answers) + { + for (Object a : answers) + { + if (!(a instanceof Boolean)) + { + return false; + } + } + } + + return true; + } + + /** + * The id of this member. + * + * @return The id, or {@code null} if the cache is closed. + */ + private static String selfId() + { + try + { + return GeodeConfig.getCache().getDistributedSystem().getDistributedMember().getId(); + } + catch (CancelException exc) + { + return null; + } + } + + /** + * The body of the background pinger: every {@link #PING_MS} ms while the gate is running, ping (in one + * call, without waiting longer than {@link #PING_MS} ms) the members of {@link #INTEREST} which no call + * answered within {@link #PING_MS} ms and which are still in the view; end when the gate is closed. + */ + private static void pingLoop() + { + while (true) + { + try + { + Thread.sleep(PING_MS); + ClusterGate.State s = ClusterGate.state(); + if (s == ClusterGate.State.CLOSED) + { + return; + } + + if (s == ClusterGate.State.RUNNING) + { + ping(); + } + } + catch (InterruptedException exc) + { + return; + } + catch (Throwable thr) + { + LOG.log(Level.WARNING, "Membership lease: ping failed", thr); + } + } + } + + /** + * One background ping (see {@link #pingLoop}). + * + * @throws InterruptedException + * if the pinger is interrupted. + */ + private static void ping() + throws InterruptedException + { + long now = System.nanoTime(); + INTEREST.values().removeIf(t -> now - t > INTEREST_MS * 1_000_000L); + ANSWERED.keySet().removeIf(m -> !INTEREST.containsKey(m) && !isFresh(m, INTEREST_MS)); + if (INTEREST.isEmpty()) + { + return; + } + + Set to = new HashSet<>(); + for (DistributedMember m : GeodeConfig.getOtherServerMembers()) + { + if (INTEREST.containsKey(m.getId()) && !isFresh(m.getId(), PING_MS)) + { + to.add(m); + } + } + + if (to.isEmpty()) + { + return; + } + + try + { + // waited for one interval at most: a stalled member never holds up the pings of the others + FunctionService.onMembers(to) + .withCollector(new Answers(false)) + .execute(Ping.ID) + .getResult(PING_MS, TimeUnit.MILLISECONDS); + } + catch (FunctionException | CancelException exc) + { + // a member did not answer in time, left meanwhile, or this member was removed + LOG.log(Level.FINE, "Membership lease: ping not answered: " + exc); + } + } + + /** A key the current context relies on, with its region. */ + private static final class Held + { + /** The region. */ + final Region region; + + /** The key. */ + final Object key; + + /** + * Constructor. + * + * @param region + * The region. + * @param key + * The key. + */ + Held(Region region, Object key) + { + this.region = region; + this.key = key; + } + } + + /** What must answer before a commit (see {@link MembershipLease#required}). */ + private static final class Required + { + /** Stale members hosting redundant copies, by id. */ + final Map members = new HashMap<>(); + + /** Keys whose primary is stale or not known, by region. */ + final Map, Set> byKey = new HashMap<>(); + + /** + * Whether nothing must answer. + * + * @return {@code true} if nothing must. + */ + boolean isEmpty() + { + return members.isEmpty() && byKey.isEmpty(); + } + + /** + * Describe it, for messages. + * + * @return The stale members and the regions with keys to check. + */ + @Override + public String toString() + { + StringJoiner j = new StringJoiner(", "); + members.keySet().forEach(j::add); + byKey.forEach((r, k) -> j.add(k.size() + " key(s) of " + r.getName())); + + return j.toString(); + } + } + + /** + * The collector of {@link #collector} (and of the pings and checks): notes the members which answered, + * collects the answers. + */ + private static final class Answers + implements ResultCollector> + { + /** {@code System.nanoTime()} when the collector was made (right before the call 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 answers. */ + private final List results = Collections.synchronizedList(new ArrayList<>()); + + /** {@code true} to keep the members which answer among those the background pings keep fresh. */ + private final boolean interest; + + /** + * Constructor. + * + * @param interest + * {@code true} to keep the members which answer among those the background pings keep fresh + * (owner calls). + */ + Answers(boolean interest) + { + this.interest = interest; + } + + /** + * A member answered: noted unless the answer is an exception made here (e.g. for a member which left). + * + * @param member + * The member. + * @param result + * Its answer. + */ + @Override + public void addResult(DistributedMember member, Object result) + { + results.add(result); + if (!(result instanceof Throwable)) + { + answered(member, sentNanos, sentWall, epoch); + if (interest && member != null) + { + INTEREST.put(member.getId(), System.nanoTime()); + } + } + } + + /** + * The answers. + * + * @return The answers. + */ + @Override + public List getResult() + { + return results; + } + + /** + * The answers (Geode's execution waits for them before it calls this). + * + * @param timeout + * Ignored. + * @param unit + * Ignored. + * + * @return The answers. + */ + @Override + public List getResult(long timeout, TimeUnit unit) + { + return results; + } + + /** + * Every member answered: nothing to do. + */ + @Override + public void endResults() + { + } + + /** + * Forget the answers (Geode executes the call again); the members noted stay noted. + */ + @Override + public void clearResults() + { + results.clear(); + } + } + + /** + * The Geode function of a ping: answers on the other servers (only while the calling server is in their + * view of the cluster). + */ + public static class Ping + implements Function + { + /** The function id. */ + static final String ID = Ping.class.getName(); + + /** Serial version. */ + private static final long serialVersionUID = 1L; + + /** + * Answer the call. + * + * @param context + * The function context. + */ + @Override + public void execute(FunctionContext context) + { + context.getResultSender().lastResult(Boolean.TRUE); + } + + /** + * The function id. + * + * @return {@link #ID}. + */ + @Override + public String getId() + { + return ID; + } + + /** + * The caller waits for the answer. + * + * @return {@code true}. + */ + @Override + public boolean hasResult() + { + return true; + } + + /** + * A ping is 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; + } + } + + /** + * The Geode function of a check routed by key: Geode sends it to the member which the caller's partition + * metadata names as the primary of the keys (of a lock or unique key region), which answers (only while + * the calling server is in its view). A member which is not the primary of some key (the caller's metadata + * lags a move) answers so ("moved"), or the call fails if it no longer hosts the bucket: the keys are + * checked again at the next round; a "moved" answer never counts as a check of the new primary. + */ + public static class Check + implements Function + { + /** The function id. */ + static final String ID = Check.class.getName(); + + /** Serial version. */ + private static final long serialVersionUID = 1L; + + /** + * Answer the call: {@code TRUE} if this member is the primary of every key of the call here, else + * {@code FALSE} (moved). + * + * @param context + * The function context (a region function context). + */ + @Override + public void execute(FunctionContext context) + { + boolean primary = true; + if (context instanceof RegionFunctionContext) + { + RegionFunctionContext rfc = (RegionFunctionContext) context; + Region region = rfc.getDataSet(); + DistributedMember self = context.getCache().getDistributedSystem().getDistributedMember(); + for (Object key : rfc.getFilter()) + { + if (!self.equals(PartitionRegionHelper.getPrimaryMemberForKey(region, key))) + { + primary = false; + break; + } + } + } + + context.getResultSender().lastResult(primary); + } + + /** + * The function id. + * + * @return {@link #ID}. + */ + @Override + public String getId() + { + return ID; + } + + /** + * The caller waits for the answer. + * + * @return {@code true}. + */ + @Override + public boolean hasResult() + { + return true; + } + + /** + * A check is not sent again when a member fails (the caller checks again). + * + * @return {@code false}. + */ + @Override + public boolean isHA() + { + return false; + } + + /** + * Run on the primaries of the keys. + * + * @return {@code true}. + */ + @Override + public boolean optimizeForWrite() + { + return true; + } + } +} --- a/src/com/goldencode/p2j/distributed/cache/FastFindCacheInvalidationFunction.java +++ /dev/null @@ -1,180 +0,0 @@ -/* -** Module : FastFindCacheInvalidationFunction.java -** Abstract : Invalidation function for FFCache which will be executed across the cluster. -** -** Copyright (c) 2026, Golden Code Development Corporation. -** -** -#- -I- --Date-- --------------------------------------Description---------------------------------------- -** 001 AP 20260212 Created initial version. -** 002 ES 20261025 Do nothing when the FastFindCache of permanent databases is disabled. -** ES 20261027 Resolve the cache with FastFindCache.getPermanentInstance: DatabaseManager needs a user -** context, which a Geode thread lacks, so every remote invalidation was dropped. -** CA 20261028 Removed the temporary [4369-ffc] instrumentation and the fwd.4369.ffc.recvFix switch, with -** the context-bound lookup it kept. -*/ - -/* -** This program is free software: you can redistribute it and/or modify -** it under the terms of the GNU Affero General Public License as -** published by the Free Software Foundation, either version 3 of the -** License, or (at your option) any later version. -** -** This program is distributed in the hope that it will be useful, -** but WITHOUT ANY WARRANTY; without even the implied warranty of -** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -** GNU Affero General Public License for more details. -** -** You may find a copy of the GNU Affero GPL version 3 at the following -** location: https://www.gnu.org/licenses/agpl-3.0.en.html -** -** Additional terms under GNU Affero GPL version 3 section 7: -** -** Under Section 7 of the GNU Affero GPL version 3, the following additional -** terms apply to the works covered under the License. These additional terms -** are non-permissive additional terms allowed under Section 7 of the GNU -** Affero GPL version 3 and may not be removed by you. -** -** 0. Attribution Requirement. -** -** You must preserve all legal notices or author attributions in the covered -** work or Appropriate Legal Notices displayed by works containing the covered -** work. You may not remove from the covered work any author or developer -** credit already included within the covered work. -** -** 1. No License To Use Trademarks. -** -** This license does not grant any license or rights to use the trademarks -** Golden Code, FWD, any Golden Code or FWD logo, or any other trademarks -** of Golden Code Development Corporation. You are not authorized to use the -** name Golden Code, FWD, or the names of any author or contributor, for -** publicity purposes without written authorization. -** -** 2. No Misrepresentation of Affiliation. -** -** You may not represent yourself as Golden Code Development Corporation or FWD. -** -** You may not represent yourself for publicity purposes as associated with -** Golden Code Development Corporation, FWD, or any author or contributor to -** the covered work, without written authorization. -** -** 3. No Misrepresentation of Source or Origin. -** -** You may not represent the covered work as solely your work. All modified -** versions of the covered work must be marked in a reasonable way to make it -** clear that the modified work is not originating from Golden Code Development -** Corporation or FWD. All modified versions must contain the notices of -** attribution required in this license. -*/ -package com.goldencode.p2j.distributed.cache; - -import com.goldencode.p2j.persist.*; -import org.apache.geode.cache.execute.*; - -/** - * A distributed function responsible for processing FastFind cache invalidation - * requests across a Geode cluster. - *

- * This function listens for various types of invalidation events (e.g., whole tables, - * specific indices, or specific DMO lists) and applies them to the local - * {@link FastFindCache} instance of the named permanent database. - *

- */ -public class FastFindCacheInvalidationFunction -implements Function -{ - /** - * Executes the distributed invalidation request. - *

- * This method first flags the current context to indicate the request originated - * remotely, which prevents cyclic broadcast echoes. It then resolves the target - * cache by database name and uses pattern matching on the request type to delegate - * to the appropriate specific invalidation logic. - *

- * - * @param context - * The function context containing the invalidation request payload. - */ - @Override - public void execute(FunctionContext context) - { - if (!FastFindCache.isPermanentEnabled()) - { - // the cache of the permanent databases is switched off here, there is nothing to invalidate - return; - } - - FastFindCache.setIsFromRemoteRequest(); - - FastFindCacheInvalidationRequest request = context.getArguments(); - FastFindCache ffCache = getFFCache(request.getDatabaseName()); - if (ffCache == null) - { - return; - } - if (request instanceof FastFindCacheInvalidationRequest.InvalidateTable req) - { - ffCache.invalidate(req.getDmoUid(), null); - } - else if (request instanceof FastFindCacheInvalidationRequest.InvalidateIndices req) - { - ffCache.invalidate(req.getDmoUid(), null, - req.getUniqueIndices(), req.getNonUniqueIndices()); - } - else if (request instanceof FastFindCacheInvalidationRequest.InvalidateDmoList req) - { - ffCache.invalidate(req.getDmoIds()); - } - else if (request instanceof FastFindCacheInvalidationRequest.InvalidateIndex req) - { - ffCache.invalidate(req.getDmoUid(), null, req.getIdxId()); - } - else if (request instanceof FastFindCacheInvalidationRequest.InvalidateAllIndicesForTable req) - { - ffCache.invalidate(req.getDmoUid(), null, false); - } - } - - /** - * Retrieves the {@link FastFindCache} of the permanent database with the specified name. A Geode thread - * has no FWD context, so the {@code DatabaseManager} cannot resolve the name here. - * - * @param dbName the name of the database to resolve the cache for. - * @return the corresponding {@code FastFindCache}, or {@code null} if this server has none. - */ - private FastFindCache getFFCache(String dbName) - { - return FastFindCache.getPermanentInstance(dbName); - } - - /** - * Returns the unique identifier for this function. - * - * @return the fully qualified class name. - */ - @Override - public String getId() - { - return getClass().getName(); - } - - /** - * Indicates whether this function returns a result to the caller. - * - * @return {@code false}, as invalidations are processed strictly as fire-and-forget operations. - */ - @Override - public boolean hasResult() - { - return false; // Fire-and-forget - } - - /** - * Indicates whether this function requires highly available (HA) execution. - * - * @return {@code false}, as HA re-execution is not required for these cache invalidations. - */ - @Override - public boolean isHA() { - return false; - } -} --- a/src/com/goldencode/p2j/distributed/cache/FastFindCacheInvalidationRequest.java +++ /dev/null @@ -1,316 +0,0 @@ -/* -** Module : FastFindCacheInvalidationRequest.java -** Abstract : Invalidation request for FFCache which will be distributed across the cluster. -** -** Copyright (c) 2026, Golden Code Development Corporation. -** -** -#- -I- --Date-- --------------------------------------Description---------------------------------------- -** 001 AP 20260212 Created initial version. -*/ - -/* -** This program is free software: you can redistribute it and/or modify -** it under the terms of the GNU Affero General Public License as -** published by the Free Software Foundation, either version 3 of the -** License, or (at your option) any later version. -** -** This program is distributed in the hope that it will be useful, -** but WITHOUT ANY WARRANTY; without even the implied warranty of -** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -** GNU Affero General Public License for more details. -** -** You may find a copy of the GNU Affero GPL version 3 at the following -** location: https://www.gnu.org/licenses/agpl-3.0.en.html -** -** Additional terms under GNU Affero GPL version 3 section 7: -** -** Under Section 7 of the GNU Affero GPL version 3, the following additional -** terms apply to the works covered under the License. These additional terms -** are non-permissive additional terms allowed under Section 7 of the GNU -** Affero GPL version 3 and may not be removed by you. -** -** 0. Attribution Requirement. -** -** You must preserve all legal notices or author attributions in the covered -** work or Appropriate Legal Notices displayed by works containing the covered -** work. You may not remove from the covered work any author or developer -** credit already included within the covered work. -** -** 1. No License To Use Trademarks. -** -** This license does not grant any license or rights to use the trademarks -** Golden Code, FWD, any Golden Code or FWD logo, or any other trademarks -** of Golden Code Development Corporation. You are not authorized to use the -** name Golden Code, FWD, or the names of any author or contributor, for -** publicity purposes without written authorization. -** -** 2. No Misrepresentation of Affiliation. -** -** You may not represent yourself as Golden Code Development Corporation or FWD. -** -** You may not represent yourself for publicity purposes as associated with -** Golden Code Development Corporation, FWD, or any author or contributor to -** the covered work, without written authorization. -** -** 3. No Misrepresentation of Source or Origin. -** -** You may not represent the covered work as solely your work. All modified -** versions of the covered work must be marked in a reasonable way to make it -** clear that the modified work is not originating from Golden Code Development -** Corporation or FWD. All modified versions must contain the notices of -** attribution required in this license. -*/ -package com.goldencode.p2j.distributed.cache; - -import java.io.*; -import java.util.*; - -/** - * Base data transfer object (DTO) representing an operation to invalidate elements of - * a {@code FastFindCache} across a distributed cluster. - *

- * This class is designed to be serialized and passed into distributed tasks or functions. - * Specific sub-types specify the fine-grained scope of the cache eviction. - *

- */ -public class FastFindCacheInvalidationRequest -implements Serializable -{ - /** The target database name associated with the cache being invalidated. */ - private final String databaseName; - - /** - * Constructs a new base invalidation request. - * - * @param databaseName - * The name of the database scope for the invalidation event. - */ - protected FastFindCacheInvalidationRequest(String databaseName) - { - this.databaseName = databaseName; - } - - /** - * Gets the name of the target database. - * - * @return the database name. - */ - public String getDatabaseName() - { - return databaseName; - } - - /** - * Represents a request to invalidate all entries and indices corresponding to a specified table. - */ - public static final class InvalidateTable extends FastFindCacheInvalidationRequest - { - /** The unique identifier of the Data Management Object (DMO) / table. */ - private final int dmoUid; - - /** - * Constructs a request to invalidate a whole table. - * - * @param databaseName - * The name of the database. - * @param dmoUid - * The table's DMO unique identifier. - */ - public InvalidateTable(String databaseName, int dmoUid) - { - super(databaseName); - this.dmoUid = dmoUid; - } - - /** - * Gets the target table's unique identifier. - * - * @return the DMO unique identifier. - */ - public int getDmoUid() - { - return dmoUid; - } - } - - /** - * Represents a request to invalidate specific unique and non-unique indices belonging to a table. - */ - public static final class InvalidateIndices extends FastFindCacheInvalidationRequest - { - /** The unique identifier of the Data Management Object (DMO) / table. */ - private final int dmoUid; - - /** Bitset containing flags for specific unique indices to be evicted. */ - private final BitSet uniqueIndices; - - /** Bitset containing flags for specific non-unique indices to be evicted. */ - private final BitSet nonUniqueIndices; - - /** - * Constructs a request to invalidate targeted indices. - * - * @param databaseName - * The name of the database. - * @param dmoUid - * The table's DMO unique identifier. - * @param uniqueIndices - * The set of unique index identifiers to invalidate. - * @param nonUniqueIndices - * The set of non-unique index identifiers to invalidate. - */ - public InvalidateIndices(String databaseName, int dmoUid, - BitSet uniqueIndices, BitSet nonUniqueIndices) - { - super(databaseName); - this.dmoUid = dmoUid; - this.uniqueIndices = uniqueIndices; - this.nonUniqueIndices = nonUniqueIndices; - } - - /** - * Gets the target table's unique identifier. - * - * @return the DMO unique identifier. - */ - public int getDmoUid() - { - return dmoUid; - } - - /** - * Gets the collection of target unique index flags. - * - * @return a {@link BitSet} indicating unique indices to clear. - */ - public BitSet getUniqueIndices() - { - return uniqueIndices; - } - - /** - * Gets the collection of target non-unique index flags. - * - * @return a {@link BitSet} indicating non-unique indices to clear. - */ - public BitSet getNonUniqueIndices() - { - return nonUniqueIndices; - } - } - - /** - * Represents a request to invalidate specific unique and non-unique indices belonging to a table. - */ - public static final class InvalidateIndex extends FastFindCacheInvalidationRequest - { - /** The unique identifier of the Data Management Object (DMO) / table. */ - private final int dmoUid; - - /** The identifier of the specific index being evicted. */ - private final int idxUid; - - /** - * Constructs a request to invalidate an isolated index configuration. - * - * @param databaseName - * The name of the database. - * @param dmoUid - * The table's DMO unique identifier. - * @param idxUid - * The unique identifier of the targeted index. - */ - public InvalidateIndex(String databaseName, int dmoUid, Integer idxUid) - { - super(databaseName); - this.dmoUid = dmoUid; - this.idxUid = idxUid; - } - - /** - * Gets the target table's unique identifier. - * - * @return the DMO unique identifier. - */ - public int getDmoUid() - { - return dmoUid; - } - - /** - * Gets the specific index identifier. - * - * @return the index identifier. - */ - public int getIdxId() - { - return idxUid; - } - } - - /** - * Represents a request to flush and invalidate all configured index paths for a specified table. - */ - public static final class InvalidateAllIndicesForTable extends FastFindCacheInvalidationRequest - { - /** The unique identifier of the Data Management Object (DMO) / table. */ - private final int dmoUid; - - /** - * Constructs a request to wipe out all index variations linked to the table. - * - * @param databaseName - * The name of the database. - * @param dmoUid - The table's DMO unique identifier. - */ - public InvalidateAllIndicesForTable(String databaseName, int dmoUid) - { - super(databaseName); - this.dmoUid = dmoUid; - } - - /** - * Gets the target table's unique identifier. - * - * @return the DMO unique identifier. - */ - public int getDmoUid() - { - return dmoUid; - } - } - - /** - * Represents a request to invalidate a specific collection of distinct Data Management Objects (DMOs) by identity. - */ - public static final class InvalidateDmoList extends FastFindCacheInvalidationRequest - { - /** The explicit list of individual DMO primary record IDs to evict. */ - private final List dmoIds; - - /** - * Constructs a request to remove specific entries from the cache by their keys. - * - * @param databaseName - he name of the database. - * @param dmoIds - The primary tracking IDs of the records to invalidate. - */ - public InvalidateDmoList(String databaseName, Collection dmoIds) - { - super(databaseName); - this.dmoIds = new ArrayList<>(dmoIds); - } - - /** - * Gets the list of DMO identities scheduled for eviction. - * - * @return a list containing the targeted record identifiers. - */ - public List getDmoIds() - { - return dmoIds; - } - } -}