external-software/github_OpenIdentityPlatform_OpenDJ.git

			@@ -19,9 +19,11 @@

			import java.util.ArrayList;
			import java.util.Collection;
			import java.util.HashSet;
			import java.util.List;
			import java.util.Map.Entry;
			import java.util.NoSuchElementException;
			import java.util.Set;
			import java.util.TreeMap;
			import java.util.concurrent.Callable;

			@@ -56,9 +58,33 @@
			*/
			static final int MAX_KEPT_CHANGES = 10000;
			static final long MAX_KEPT_BYTES = 32L * 1024 * 1024;
			/**
			* How many of the last local changes the domain keeps the messages of at most, and how much of the
			* heap those take at most, as {@link #footprint(LDAPUpdateMsg)} reckons it: a recovery publishes the
			* changes its replication server misses from them, and rebuilds the ones before them from the
			* historical information - see {@link LastLocalChanges}. Under the load of the replication convergence
			* run, a few hundred changes a second, a replication server more than two minutes behind is covered.
			* <p>
			* Each domain written to keeps that much, whether a recovery runs or not, so its bytes are a sixteenth
			* of the heap at most; multiple written domains can keep that share each. A server given a small heap
			* covers a shorter lag. A recovery round holds the messages it publishes besides: while it runs, as much
			* again can stay on the heap.
			*/
			static final int MAX_RECENT_CHANGES = 100000;
			static final long MAX_RECENT_BYTES = Math.min(64L * 1024 * 1024, Runtime.getRuntime().maxMemory() / 16);
			/**
			* What a message kept takes besides twice its {@link LDAPUpdateMsg#size()}, which counts its encoded
			* changes once, and its encoding holds them again: its DN, entry UUID and CSN, the objects of its
			* changes, and its place in the map. Measured on the messages of the load of the convergence run,
			* kept as they are here: 880 bytes of heap for a modify whose size() is 168, 1300 for an add whose
			* size() is 370.
			*/
			static final int MESSAGE_OVERHEAD = 560;

			/** A map used to store the pending changes. */
			private final TreeMap<CSN, PendingChange> pendingChanges = new TreeMap<>();
			/** Local operations pending at a reload, which may have written to the replaced data. */
			private final Set<CSN> pendingAtDataReload = new HashSet<>();

			/**
			* The {@link CSNGenerator} to use to create new unique CSNs
			@@ -81,6 +107,10 @@
			private Recovery recovery;
			/** How many local changes a recovery keeps at most: fewer in the tests which make it run over. */
			private int maxKeptChanges = MAX_KEPT_CHANGES;
			/** The messages of the last local changes, which a recovery publishes as they were done. */
			private final LastLocalChanges lastLocalChanges = new LastLocalChanges();
			/** How many of the last local changes the domain keeps the messages of at most: fewer in tests. */
			private int maxRecentChanges = MAX_RECENT_CHANGES;

			/**
			* Creates a new PendingChanges using the provided CSNGenerator.
			@@ -108,6 +138,7 @@
			synchronized void remove(CSN csn)
			{
			pendingChanges.remove(csn);
			pendingAtDataReload.remove(csn);
			}

			/**
			@@ -193,8 +224,8 @@
			* {@code null} if none was pushed or the broker refused it. The announcement that a
			* replica goes offline is the one message whose delivery the caller must know about:
			* it is stored nowhere, so nothing publishes it again, while a change the broker
			* refuses is republished from the historical information of its entry on the next
			* session.
			* refuses is republished on the next session, from the message kept of it or, once that
			* is dropped, from the historical information of its entry.
			*/
			synchronized CSN pushCommittedChanges()
			{
			@@ -217,6 +248,16 @@
			&& op != null
			&& !op.isSynchronizationOperation())
			{
			// Kept before the domain state takes it, published or not: a recovery publishes it as it was done.
			if (pendingAtDataReload.contains(firstChange.getCSN()))
			{
			// There is a gap in the messages kept after the reload; earlier ones cannot cover it.
			lastLocalChanges.forget();
			}
			else
			{
			lastLocalChanges.add(firstChange.getCSN(), (LDAPUpdateMsg) msg);
			}
			if (!recoveringOldChanges)
			{
			domain.publish(msg);
			@@ -268,6 +309,7 @@

			// false warning: firstEntry will not be null if firstChange is not null
			pendingChanges.remove(firstEntry.getKey());
			pendingAtDataReload.remove(firstChange.getCSN());

			// peek the oldest change
			firstEntry = pendingChanges.firstEntry();
			@@ -536,6 +578,185 @@
			}
			}

			/**
			* Returns the messages the domain kept of its local changes which come after the provided CSN, up
			* to the provided one, in order, or tells where the messages it kept begin if they do not go back
			* that far.
			*
			* @param after
			* where the recovery has got to
			* @param upTo
			* where the changes to publish end, or {@code null} for the last local change
			* @return the messages, or where the ones kept begin
			*/
			synchronized RecentChanges recentChangesAfter(CSN after, CSN upTo)
			{
			return lastLocalChanges.after(after, upTo != null ? upTo : domain.getLastLocalChange());
			}

			/**
			* Forgets the messages kept of the last local changes: the domain read its state from data which
			* may lack their changes - it was restored or imported since - and a recovery which published them
			* would have the other replicas do changes this one does not hold. Local operations still pending
			* may have written to the replaced data even if they have not yet committed, so their messages are
			* not kept when they are pushed. Later messages are kept, and a recovery rebuilds the changes before
			* them from the historical information.
			*/
			synchronized void forgetRecentChanges()
			{
			lastLocalChanges.forget();
			for (PendingChange change : pendingChanges.values())
			{
			if (change.getOp() != null && !change.getOp().isSynchronizationOperation())
			{
			pendingAtDataReload.add(change.getCSN());
			}
			}
			}

			/**
			* Sets how many of the last local changes the domain keeps the messages of at most: with none, a
			* recovery rebuilds every change from the historical information.
			*
			* @param maxRecentChanges
			* how many of the last local changes the domain keeps the messages of at most
			*/
			@VisibleForTesting
			synchronized void setMaxRecentChanges(int maxRecentChanges)
			{
			this.maxRecentChanges = maxRecentChanges;
			lastLocalChanges.trim();
			}

			/**
			* The messages of the last local changes, in order, which a recovery publishes as they were done.
			* <p>
			* A recovery publishes the changes its replication server misses of this replica, and rebuilt from
			* the historical information of the entries, those are not the changes: it holds nothing of a change
			* whose values a later one replaced, nor of the add of an entry deleted since. A replication server
			* which took the later changes takes none of the older ones afterwards - from the peer which got them
			* from this replica either, whose changes it takes in order, as a log ordered by CSN - so a directory
			* server which failed over to a replication server behind it, or which the load balancing moved to
			* one, left gaps in its changelog for good (issue #1246). The message of every local change pushed -
			* published, refused by the broker, or left to a recovery - is kept here instead, for as many changes
			* as {@link #maxRecentChanges} and {@link #MAX_RECENT_BYTES} allow, the oldest dropped first.
			* <p>
			* Every local change after {@link #keptAfter} has its message kept: the CSN right before the first one
			* kept, and the newest one dropped once some were. A recovery from further back rebuilds the changes up
			* to there from the historical information. It is not the last local change the domain state holds:
			* the state read back once the data was restored, or once the domain was enabled again, can be older
			* than changes done since the domain started - a delete leaves no historical information to bring it
			* up to - and taking the messages kept for the changes after it skipped those changes.
			*/
			private final class LastLocalChanges
			{
			private final TreeMap<CSN, LDAPUpdateMsg> messages = new TreeMap<>();
			private long bytes;
			/**
			* Every local change after this one has its message kept; {@code null} if none was kept since the
			* domain started or forgot them.
			*/
			private CSN keptAfter;

			private void add(CSN csn, LDAPUpdateMsg msg)
			{
			if (maxRecentChanges <= 0)
			{
			return;
			}
			if (keptAfter == null)
			{
			keptAfter = justBefore(csn);
			}
			// A change is pushed again when the publish of the push before threw: it is counted once.
			final LDAPUpdateMsg replaced = messages.put(csn, msg);
			bytes += footprint(msg) - (replaced != null ? footprint(replaced) : 0);
			trim();
			}

			private void trim()
			{
			if (maxRecentChanges <= 0)
			{
			forget();
			return;
			}
			while (!messages.isEmpty() && (messages.size() > maxRecentChanges \|\| bytes > MAX_RECENT_BYTES))
			{
			final Entry<CSN, LDAPUpdateMsg> oldest = messages.pollFirstEntry();
			bytes -= footprint(oldest.getValue());
			keptAfter = oldest.getKey();
			}
			}

			private void forget()
			{
			messages.clear();
			bytes = 0;
			keptAfter = null;
			}

			private RecentChanges after(CSN after, CSN upTo)
			{
			if (keptAfter == null)
			{
			return new RecentChanges(null, upTo, null);
			}
			if (after.isOlderThan(keptAfter))
			{
			return new RecentChanges(null, upTo, keptAfter);
			}
			final List<LDAPUpdateMsg> found = upTo != null && after.isOlderThan(upTo)
			? new ArrayList<>(messages.subMap(after, false, upTo, true).values())
			: new ArrayList<LDAPUpdateMsg>();
			return new RecentChanges(found, upTo, keptAfter);
			}
			}

			/**
			* Returns the CSN right before the provided one among the CSNs of its server: no CSN of that server
			* comes in between. For a sequence number of 0, it is the greatest one of the millisecond before, as
			* {@code Integer.MAX_VALUE} rather than -1: a search window of a recovery may start there, and the
			* index of the historical information reads -1 as the greatest sequence number of its millisecond.
			*/
			private static CSN justBefore(CSN csn)
			{
			return csn.getSeqnum() > 0
			? new CSN(csn.getTime(), csn.getSeqnum() - 1, csn.getServerId())
			: new CSN(csn.getTime() - 1, Integer.MAX_VALUE, csn.getServerId());
			}

			/** Returns how much of the heap a message kept takes, as reckoned for {@link #MAX_RECENT_BYTES}. */
			static long footprint(LDAPUpdateMsg msg)
			{
			return 2L * msg.size() + MESSAGE_OVERHEAD;
			}

			/** What the domain kept of the messages of its local changes from a CSN on, up to another one. */
			static final class RecentChanges
			{
			/**
			* The messages of the local changes after the CSN asked from, up to {@link #upTo}, in order;
			* {@code null} if the messages kept do not go back that far, or the domain keeps none.
			*/
			final List<LDAPUpdateMsg> messages;
			/** Where the changes asked for end: the CSN asked for, or the last local change. */
			final CSN upTo;
			/**
			* Every local change after this one has its message kept: where the changes to rebuild from the
			* historical information end when {@link #messages} is {@code null}. {@code null} if no message is
			* kept, and the changes are all to rebuild.
			*/
			final CSN keptAfter;

			private RecentChanges(List<LDAPUpdateMsg> messages, CSN upTo, CSN keptAfter)
			{
			this.messages = messages;
			this.upTo = upTo;
			this.keptAfter = keptAfter;
			}
			}

			/** A recovery which keeps the local changes pushed while it runs. */
			static final class KeptChanges
			{