3232import java .util .concurrent .atomic .AtomicLong ;
3333
3434/** Maps _uid value to its version information. */
35- class LiveVersionMap implements ReferenceManager .RefreshListener , Accountable {
35+ final class LiveVersionMap implements ReferenceManager .RefreshListener , Accountable {
3636
3737 /**
3838 * Resets the internal map and adjusts it's capacity as if there were no indexing operations.
@@ -46,29 +46,110 @@ void adjustMapSizeUnderLock() {
4646 maps = new Maps ();
4747 }
4848
49- private static class Maps {
49+ private static final class VersionLookup {
50+
51+ private static final VersionLookup EMPTY = new VersionLookup (Collections .emptyMap ());
52+ private final Map <BytesRef ,VersionValue > map ;
53+
54+ // each version map has a notion of safe / unsafe which allows us to apply certain optimization in the auto-generated ID usecase
55+ // where we know that documents can't have any duplicates so we can skip the version map entirely. This reduces
56+ // the memory pressure significantly for this use-case where we often get a massive amount of small document (metrics).
57+ // if the version map is in safeAccess mode we track all version in the version map. yet if a document comes in that needs
58+ // safe access but we are not in this mode we force a refresh and make the map as safe access required. All subsequent ops will
59+ // respect that and fill the version map. The nice part here is that we are only really requiring this for a single ID and since
60+ // we hold the ID lock in the engine while we do all this it's safe to do it globally unlocked.
61+ // NOTE: these values can both be non-volatile since it's ok to read a stale value per doc ID. We serialize changes in the engine
62+ // that will prevent concurrent updates to the same document ID and therefore we can rely on the happens-before guanratee of the
63+ // map reference itself.
64+ private boolean unsafe ;
65+
66+ private VersionLookup (Map <BytesRef , VersionValue > map ) {
67+ this .map = map ;
68+ }
69+
70+ VersionValue get (BytesRef key ) {
71+ return map .get (key );
72+ }
73+
74+ VersionValue put (BytesRef key , VersionValue value ) {
75+ return map .put (key , value );
76+ }
77+
78+ boolean isEmpty () {
79+ return map .isEmpty ();
80+ }
81+
82+
83+ int size () {
84+ return map .size ();
85+ }
86+
87+ boolean isUnsafe () {
88+ return unsafe ;
89+ }
90+
91+ void markAsUnsafe () {
92+ unsafe = true ;
93+ }
94+ }
95+
96+ private static final class Maps {
5097
5198 // All writes (adds and deletes) go into here:
52- final Map < BytesRef , VersionValue > current ;
99+ final VersionLookup current ;
53100
54101 // Used while refresh is running, and to hold adds/deletes until refresh finishes. We read from both current and old on lookup:
55- final Map <BytesRef ,VersionValue > old ;
102+ final VersionLookup old ;
103+
104+ // this is not volatile since we don't need to maintain a happens before relation ship across doc IDs so it's enough to
105+ // have the volatile read of the Maps reference to make it visible even across threads.
106+ boolean needsSafeAccess ;
107+ final boolean previousMapsNeededSafeAccess ;
56108
57- Maps (Map <BytesRef ,VersionValue > current , Map <BytesRef ,VersionValue > old ) {
58- this .current = current ;
59- this .old = old ;
109+ Maps (VersionLookup current , VersionLookup old , boolean previousMapsNeededSafeAccess ) {
110+ this .current = current ;
111+ this .old = old ;
112+ this .previousMapsNeededSafeAccess = previousMapsNeededSafeAccess ;
60113 }
61114
62115 Maps () {
63- this (ConcurrentCollections .<BytesRef ,VersionValue >newConcurrentMapWithAggressiveConcurrency (),
64- Collections .emptyMap ());
116+ this (new VersionLookup (ConcurrentCollections .newConcurrentMapWithAggressiveConcurrency ()), VersionLookup .EMPTY , false );
117+ }
118+
119+ boolean isSafeAccessMode () {
120+ return needsSafeAccess || previousMapsNeededSafeAccess ;
121+ }
122+
123+ boolean shouldInheritSafeAccess () {
124+ final boolean mapHasNotSeenAnyOperations = current .isEmpty () && current .isUnsafe () == false ;
125+ return needsSafeAccess
126+ // we haven't seen any ops and map before needed it so we maintain it
127+ || (mapHasNotSeenAnyOperations && previousMapsNeededSafeAccess );
128+ }
129+
130+ /**
131+ * Builds a new map for the refresh transition this should be called in beforeRefresh()
132+ */
133+ Maps buildTransitionMap () {
134+ return new Maps (new VersionLookup (ConcurrentCollections .newConcurrentMapWithAggressiveConcurrency (current .size ())),
135+ current , shouldInheritSafeAccess ());
136+ }
137+
138+ /**
139+ * builds a new map that invalidates the old map but maintains the current. This should be called in afterRefresh()
140+ */
141+ Maps invalidateOldMap () {
142+ return new Maps (current , VersionLookup .EMPTY , previousMapsNeededSafeAccess );
65143 }
66144 }
67145
68146 // All deletes also go here, and delete "tombstones" are retained after refresh:
69147 private final Map <BytesRef ,DeleteVersionValue > tombstones = ConcurrentCollections .newConcurrentMapWithAggressiveConcurrency ();
70148
71149 private volatile Maps maps = new Maps ();
150+ // we maintain a second map that only receives the updates that we skip on the actual map (unsafe ops)
151+ // this map is only maintained if assertions are enabled
152+ private volatile Maps unsafeKeysMap = new Maps ();
72153
73154 /** Bytes consumed for each BytesRef UID:
74155 * In this base value, we account for the {@link BytesRef} object itself as
@@ -113,8 +194,8 @@ public void beforeRefresh() throws IOException {
113194 // map. While reopen is running, any lookup will first
114195 // try this new map, then fallback to old, then to the
115196 // current searcher:
116- maps = new Maps ( ConcurrentCollections . newConcurrentMapWithAggressiveConcurrency ( maps .current . size ()), maps . current );
117-
197+ maps = maps .buildTransitionMap ( );
198+ assert ( unsafeKeysMap = unsafeKeysMap . buildTransitionMap ()) != null ;
118199 // This is not 100% correct, since concurrent indexing ops can change these counters in between our execution of the previous
119200 // line and this one, but that should be minor, and the error won't accumulate over time:
120201 ramBytesUsedCurrent .set (0 );
@@ -128,13 +209,18 @@ public void afterRefresh(boolean didRefresh) throws IOException {
128209 // case. This is because we assign new maps (in beforeRefresh) slightly before Lucene actually flushes any segments for the
129210 // reopen, and so any concurrent indexing requests can still sneak in a few additions to that current map that are in fact reflected
130211 // in the previous reader. We don't touch tombstones here: they expire on their own index.gc_deletes timeframe:
131- maps = new Maps (maps .current , Collections .emptyMap ());
212+
213+ maps = maps .invalidateOldMap ();
214+ assert (unsafeKeysMap = unsafeKeysMap .invalidateOldMap ()) != null ;
215+
132216 }
133217
134218 /** Returns the live version (add or delete) for this uid. */
135219 VersionValue getUnderLock (final BytesRef uid ) {
136- Maps currentMaps = maps ;
220+ return getUnderLock (uid , maps );
221+ }
137222
223+ private VersionValue getUnderLock (final BytesRef uid , Maps currentMaps ) {
138224 // First try to get the "live" value:
139225 VersionValue value = currentMaps .current .get (uid );
140226 if (value != null ) {
@@ -149,11 +235,52 @@ VersionValue getUnderLock(final BytesRef uid) {
149235 return tombstones .get (uid );
150236 }
151237
238+ VersionValue getVersionForAssert (final BytesRef uid ) {
239+ VersionValue value = getUnderLock (uid , maps );
240+ if (value == null ) {
241+ value = getUnderLock (uid , unsafeKeysMap );
242+ }
243+ return value ;
244+ }
245+
246+ boolean isUnsafe () {
247+ return maps .current .isUnsafe () || maps .old .isUnsafe ();
248+ }
249+
250+ void enforceSafeAccess () {
251+ maps .needsSafeAccess = true ;
252+ }
253+
254+ boolean isSafeAccessRequired () {
255+ return maps .isSafeAccessMode ();
256+ }
257+
258+ /** Adds this uid/version to the pending adds map iff the map needs safe access. */
259+ void maybePutUnderLock (BytesRef uid , VersionValue version ) {
260+ Maps maps = this .maps ;
261+ if (maps .isSafeAccessMode ()) {
262+ putUnderLock (uid , version , maps );
263+ } else {
264+ maps .current .markAsUnsafe ();
265+ assert putAssertionMap (uid , version );
266+ }
267+ }
268+
269+ private boolean putAssertionMap (BytesRef uid , VersionValue version ) {
270+ putUnderLock (uid , version , unsafeKeysMap );
271+ return true ;
272+ }
273+
152274 /** Adds this uid/version to the pending adds map. */
153275 void putUnderLock (BytesRef uid , VersionValue version ) {
276+ Maps maps = this .maps ;
277+ putUnderLock (uid , version , maps );
278+ }
279+
280+ /** Adds this uid/version to the pending adds map. */
281+ private void putUnderLock (BytesRef uid , VersionValue version , Maps maps ) {
154282 assert uid .bytes .length == uid .length : "Oversized _uid! UID length: " + uid .length + ", bytes length: " + uid .bytes .length ;
155283 long uidRAMBytesUsed = BASE_BYTES_PER_BYTESREF + uid .bytes .length ;
156-
157284 final VersionValue prev = maps .current .put (uid , version );
158285 if (prev != null ) {
159286 // Deduct RAM for the version we just replaced:
@@ -264,5 +391,5 @@ public Collection<Accountable> getChildResources() {
264391
265392 /** Returns the current internal versions as a point in time snapshot*/
266393 Map <BytesRef , VersionValue > getAllCurrent () {
267- return maps .current ;
394+ return maps .current . map ;
268395 }}
0 commit comments