external-software/github_OpenIdentityPlatform_OpenDJ.git

			@@ -74,10 +74,10 @@
			@GuardedBy("dependentChangesLock")
			private final SortedSet<PendingChange> dependentChanges = new TreeSet<>();
			/**
			* {@code activeAndDependentChanges} also contains changes discovered to be dependent
			* on currently in progress changes.
			* Uncommitted changes, registered before enqueue. A queued change whose hand-off
			* fails must remain a predecessor: otherwise its DELETE can run before its ADD retry.
			*/
			private final ConcurrentSkipListSet<PendingChange> activeAndDependentChanges = new ConcurrentSkipListSet<>();
			private final ConcurrentSkipListSet<PendingChange> replayDependencies;

			/**
			* The change each replay thread is replaying right now, read by the give-back on the way
			@@ -106,7 +106,7 @@
			* races nothing for it. {@link #clear()} is the one writer of every entry, and it holds
			* both locks.
			*/
			private final ConcurrentMap<Thread, CSN> changeBeingReplayed = new ConcurrentHashMap<>();
			private final ConcurrentMap<Thread, CSN> changeBeingReplayed;

			private final ReentrantReadWriteLock pendingChangesLock = new ReentrantReadWriteLock(true);
			private final ReentrantReadWriteLock.ReadLock pendingChangesReadLock = pendingChangesLock.readLock();
			@@ -115,6 +115,16 @@

			/** The ServerState that will be updated when LDAPUpdateMsg are fully replayed. */
			private final ServerState state;
			private final int localServerId;
			/** Own-origin replay progress; local publication must not cover restore catch-up. */
			@GuardedBy("pendingChangesLock")
			private CSN localReplayCSN;

			/** Whether a delivery needs replay or how its duplicate should be completed. */
			enum Admission
			{
			ACCEPTED, QUEUED_DUPLICATE, DUPLICATE
			}

			/**
			* How many of the changes listed here have a replay failure recorded against them.
			@@ -137,7 +147,43 @@
			*/
			public RemotePendingChanges(ServerState state)
			{
			this(state, -1);
			}

			RemotePendingChanges(ServerState state, int localServerId)
			{
			this(state, localServerId, new ConcurrentHashMap<Thread, CSN>());
			}

			RemotePendingChanges(ServerState state, int localServerId,
			ConcurrentMap<Thread, CSN> changeBeingReplayed)
			{
			this(state, localServerId, changeBeingReplayed, new ConcurrentSkipListSet<PendingChange>());
			}

			RemotePendingChanges(ServerState state, int localServerId,
			ConcurrentMap<Thread, CSN> changeBeingReplayed,
			ConcurrentSkipListSet<PendingChange> replayDependencies)
			{
			this.replayDependencies = replayDependencies;
			this.state = state;
			this.localServerId = localServerId;
			this.changeBeingReplayed = changeBeingReplayed;
			resetReplayState();
			}

			/** Reloads the own-origin watermark only when the backend state is loaded. */
			void resetReplayState()
			{
			pendingChangesWriteLock.lock();
			try
			{
			localReplayCSN = state.getCSN(localServerId);
			}
			finally
			{
			pendingChangesWriteLock.unlock();
			}
			}

			/**
			@@ -169,7 +215,23 @@
			*/
			public int changesInProgressSize()
			{
			return activeAndDependentChanges.size();
			pendingChangesReadLock.lock();
			try
			{
			int count = 0;
			for (PendingChange change : replayDependencies)
			{
			if (change.hasStartedReplay())
			{
			count++;
			}
			}
			return count;
			}
			finally
			{
			pendingChangesReadLock.unlock();
			}
			}

			/**
			@@ -194,47 +256,82 @@
			* Add a new LDAPUpdateMsg that was received from the replication server
			* to the pendingList.
			* <p>
			* A change which is already listed and which a replay thread owns - it is being
			* replayed, it waits for the change it depends on, or it has been replayed and waits
			* for the changes before it - is left alone: that copy knows what this one does not,
			* and replaying it again is exactly what the duplicate check is there to prevent
			* (OPENDJ-1115).
			* Changes covered by the applied ServerState are rejected even after they have left
			* the pending list. For this replica's own origin, only the state loaded from the backend
			* and subsequent replays count: a newer local publication does not prove that an older
			* change sent back after restore is in the data. A listed change is also left alone while its delivery is queued,
			* owned by a replay thread, or committed and waiting for earlier changes. Replacing a
			* queued delivery would move it behind newer changes: a DELETE could then be replayed
			* before its ADD, leaving the entry present once the ADD is replayed (issue #1269).
			* <p>
			* A change which is listed but which no replay thread owns is taken over by this
			* delivery: it is the one to replay, and the copy which may still wait in the replay
			* queue is dropped by {@link #markInProgress(LDAPUpdateMsg)}. That is a change whose
			* replay failed - it is deliberately left out of the ServerState, so the replication
			* server sends it again over the session which was restarted (issue #889) - and it is
			* also, harmlessly, a change which was listed a moment ago by a delivery no replay
			* thread has picked up yet: the two deliveries carry the same change, and the last one
			* listed is the one replayed.
			* Only a delivery which was released after failed or abandoned replay, or cancelled
			* before being enqueued, may be replaced. It is deliberately left out of the
			* ServerState so the replication server sends it again over the restarted session
			* (issue #889). Neither an owner nor a queued delivery remains to finish it.
			* <p>
			* The failures the change went through are kept, whichever delivery replays it: they
			* are what has this replica eventually give up on a change it can not apply.
			*
			* @param update The LDAPUpdateMsg that was received from the replication
			* server and that will be added to the pending list.
			* @return {@code false} if the update was already registered in the pending
			* changes and is owned by a replay thread.
			* @return {@code false} if the change is already applied or has a delivery which is
			* still queued, owned, or committed
			*/
			public boolean putRemoteUpdate(LDAPUpdateMsg update)
			{
			return putRemoteUpdate(new UpdateToReplay(update, null)) == Admission.ACCEPTED;
			}

			Admission putRemoteUpdate(UpdateToReplay delivery)
			{
			final LDAPUpdateMsg update = delivery.getUpdateMessage();
			pendingChangesWriteLock.lock();
			try
			{
			final CSN csn = update.getCSN();
			if (csn.getServerId() == localServerId
			? localReplayCSN != null && localReplayCSN.compareTo(csn) >= 0 : state.cover(csn))
			{
			return Admission.DUPLICATE;
			}
			final PendingChange listed = pendingChanges.get(csn);
			if (listed == null)
			{
			pendingChanges.put(csn, new PendingChange(csn, null, update));
			return true;
			final PendingChange change = new PendingChange(csn, null, update);
			change.setDelivery(delivery);
			pendingChanges.put(csn, change);
			try
			{
			replayDependencies.add(change);
			}
			catch (RuntimeException \| Error e)
			{
			// No other admission sees this partial registration while the write lock is held.
			// Request recovery before rollback: removing a skip-list node can also fail.
			delivery.cancel();
			pendingChanges.remove(csn);
			replayDependencies.remove(change);
			throw e;
			}
			return Admission.ACCEPTED;
			}
			if (listed.isCommitted() \|\| listed.isOwned())
			{
			return false;
			return Admission.DUPLICATE;
			}
			listed.setMsg(update);
			return true;
			if (listed.isQueuedForReplay())
			{
			// Keep the queue position, but complete for the latest delivery's assured mode.
			// This changes in-memory ack metadata only; the received message's cached wire
			// encoding is stale and must not be reused to publish this message.
			final LDAPUpdateMsg accepted = listed.getLDAPUpdateMsg();
			accepted.setAssured(update.isAssured());
			accepted.setAssuredMode(update.getAssuredMode());
			accepted.setSafeDataLevel(update.getSafeDataLevel());
			return Admission.QUEUED_DUPLICATE;
			}
			listed.setDelivery(delivery);
			return Admission.ACCEPTED;
			}
			finally
			{
			@@ -278,7 +375,7 @@
			curChange.setCommitted(true);
			curChange.setOwner(null);
			changeBeingReplayed.remove(Thread.currentThread(), csn);
			activeAndDependentChanges.remove(curChange);
			replayDependencies.remove(curChange);

			final Iterator<PendingChange> it = pendingChanges.values().iterator();
			while (it.hasNext())
			@@ -291,6 +388,10 @@
			if (pendingChange.getMsg().contributesToDomainState())
			{
			state.update(pendingChange.getCSN());
			if (pendingChange.getCSN().getServerId() == localServerId)
			{
			localReplayCSN = pendingChange.getCSN();
			}
			}
			if (pendingChange.getReplayFailures() > 0)
			{
			@@ -504,7 +605,7 @@
			{
			pendingChanges.clear();
			dependentChanges.clear();
			activeAndDependentChanges.clear();
			replayDependencies.clear();
			changeBeingReplayed.clear();
			failingChanges = 0;
			}
			@@ -520,12 +621,8 @@
			*
			* @param msg
			* the message whose change is being replayed
			* @return {@code false} if this message is not the delivery which is listed as
			* pending, which happens when the session was restarted after a failed replay
			* while this message was still waiting in the replay queue: the replication
			* server delivered the change again and that delivery took over from this one,
			* or the domain was disabled and forgot the change, so this copy must not be
			* replayed.
			* @return {@code false} if the change is already committed, the domain has forgotten
			* it, the delivery was cancelled, or this is a stale delivery which a retry has replaced
			*/
			public boolean markInProgress(LDAPUpdateMsg msg)
			{
			@@ -533,18 +630,19 @@
			try
			{
			final PendingChange change = pendingChanges.get(msg.getCSN());
			if (change == null \|\| change.isCommitted() \|\| change.getLDAPUpdateMsg() != msg)
			if (change == null \|\| change.isCommitted() \|\| change.isDeliveryCancelled()
			\|\| change.getLDAPUpdateMsg() != msg)
			{
			return false;
			}
			// This delivery has left the queue. If an allocation below fails before it gets
			// an owner, a later delivery must be able to take it over.
			change.markDequeued();
			/*
			* Listed as being replayed before it is owned, and not the other way round: the
			* caller enters the replay - where the give-back on the way out lives - once this
			* returns, so nothing which allocates must run between the owner being stamped and
			* that. A change listed here without an owner is the state a failed replay leaves
			* behind, and the dependency checks which read this set do not read the owner.
			* The dependency was registered before enqueue, so a failed claim still holds back
			* its successors. Allocate the thread index before stamping the owner: the caller
			* reaches replay's give-back only once this method returns.
			*/
			activeAndDependentChanges.add(change);
			changeBeingReplayed.put(Thread.currentThread(), change.getCSN());
			change.setOwner(Thread.currentThread());
			return true;
			@@ -836,7 +934,7 @@

			boolean hasDependencies = false;
			final DN targetDN = op.getEntryDN();
			for (PendingChange pendingChange : activeAndDependentChanges)
			for (PendingChange pendingChange : replayDependencies)
			{
			if (pendingChange.getCSN().isNewerThanOrEqualTo(csn))
			{
			@@ -910,7 +1008,7 @@

			boolean hasDependencies = false;
			final DN targetDN = change.getLDAPUpdateMsg().getDN();
			for (PendingChange pendingChange : activeAndDependentChanges)
			for (PendingChange pendingChange : replayDependencies)
			{
			if (pendingChange.getCSN().isNewerThanOrEqualTo(csn))
			{
			@@ -977,7 +1075,7 @@

			boolean hasDependencies = false;
			final DN targetDN = change.getLDAPUpdateMsg().getDN();
			for (PendingChange pendingChange : activeAndDependentChanges)
			for (PendingChange pendingChange : replayDependencies)
			{
			if (pendingChange.getCSN().isNewerThanOrEqualTo(csn))
			{
			@@ -1052,7 +1150,7 @@

			boolean hasDependencies = false;
			final DN targetDN = op.getEntryDN();
			for (PendingChange pendingChange : activeAndDependentChanges)
			for (PendingChange pendingChange : replayDependencies)
			{
			if (pendingChange.getCSN().isNewerThanOrEqualTo(csn))
			{