#5614

Committed 11 Jul 2026 09:20PM UTC coverage: 99.988% (-0.01%) from 100.0%

Build # #5614

Build Type

push

github

Committed by

ben-manes

Commit Message

Probe quietly in the async VarExpiration putIfAbsent

BoundedVarExpiration.putIfAbsentAsync probed for a present entry with a
non-quiet cache.getIfPresent, which runs the read path through the
cache's configured AsyncExpiry and so applies the user's expireAfterRead
to the existing entry. A present-entry putIfAbsent then rewrote the
entry's expiration, violating the "no effect on the duration if the
entry was present" contract. The synchronous path routes the same probe
through FixedExpireAfterWrite, whose read leg is identity, so it never
recomputes the duration.

Probe with getIfPresentQuietly instead; the retries were already quiet,
so the loop-carried priorFuture collapses into the loop. This does not
apply to AsMapView.putIfAbsent, whose synchronous counterpart passes the
user Expiry and so correctly consults expireAfterRead on both paths.

Coverage Stats

4124 of 4134 branches covered (99.76%)

1 of 1 new or added line in 1 file covered. (100.0%)

1 existing line in 1 file now uncovered.

8385 of 8386 relevant lines covered (99.99%)

1.0 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.78

/caffeine/src/main/java/com/github/benmanes/caffeine/cache/StripedBuffer.java

/*
 * Copyright 2015 Ben Manes. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
/*
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain, as explained at
 * http://creativecommons.org/publicdomain/zero/1.0/
 */
package com.github.benmanes.caffeine.cache;

import static com.github.benmanes.caffeine.cache.Caffeine.ceilingPowerOfTwo;
import static java.lang.invoke.ConstantBootstraps.fieldVarHandle;

import java.lang.invoke.MethodHandles;
import java.lang.invoke.VarHandle;
import java.util.Arrays;
import java.util.function.Consumer;

import org.jspecify.annotations.Nullable;

import com.google.errorprone.annotations.Var;

/**
 * A base class providing the mechanics for supporting dynamic striping of bounded buffers. This
 * implementation is an adaption of the numeric 64-bit <i>java.util.concurrent.atomic.Striped64</i>
 * class, which is used by atomic counters. The approach was modified to lazily grow an array of
 * buffers in order to minimize memory usage for caches that are not heavily contended on.
 *
 * @author dl@cs.oswego.edu (Doug Lea)
 * @author ben.manes@gmail.com (Ben Manes)
 */
abstract class StripedBuffer<E> implements Buffer<E> {
  /*
   * This class maintains a lazily-initialized table of atomically updated buffers. The table size
   * is a power of two. Indexing uses masked per-thread hash codes. Nearly all declarations in this
   * class are package-private, accessed directly by subclasses.
   *
   * Table entries are of class Buffer and should be padded to reduce cache contention. Padding is
   * overkill for most atomics because they are usually irregularly scattered in memory and thus
   * don't interfere much with each other. But atomic objects residing in arrays will tend to be
   * placed adjacent to each other, and so will most often share cache lines (with a huge negative
   * performance impact) without this precaution.
   *
   * In part because Buffers are relatively large, we avoid creating them until they are needed.
   * When there is no contention, all updates are made to a single buffer. Upon contention (a failed
   * CAS inserting into the buffer), the table is expanded to size 2. The table size is doubled upon
   * further contention until reaching the nearest power of two greater than or equal to the number
   * of CPUS. Table slots remain empty (null) until they are needed.
   *
   * A single spinlock ("tableBusy") is used for initializing and resizing the table, as well as
   * populating slots with new Buffers. There is no need for a blocking lock; when the lock is not
   * available, threads try other slots. During these retries, there is increased contention and
   * reduced locality, which is still better than alternatives.
   *
   * Contention and/or table collisions are indicated by failed CASes when performing an update
   * operation. Upon a collision, if the table size is less than the capacity, it is doubled in size
   * unless some other thread holds the lock. If a hashed slot is empty, and lock is available, a
   * new Buffer is created. Otherwise, if the slot exists, a CAS is tried. The thread id serves as
   * the base for per-thread hash codes. Retries proceed by "incremental hashing", using the top
   * half of the seed to increment the bottom half which is used as a probe to try to find a free
   * slot.
   *
   * The table size is capped because, when there are more threads than CPUs, supposing that each
   * thread were bound to a CPU, there would exist a perfect hash function mapping threads to slots
   * that eliminates collisions. When we reach capacity, we search for this mapping by varying the
   * hash codes of colliding threads. Because search is random, and collisions only become known via
   * CAS failures, convergence can be slow, and because threads are typically not bound to CPUs
   * forever, may not occur at all. However, despite these limitations, observed contention rates
   * are typically low in these cases.
   *
   * It is possible for a Buffer to become unused when threads that once hashed to it terminate, as
   * well as in the case where doubling the table causes no thread to hash to it under expanded
   * mask. We do not try to detect or remove buffers, under the assumption that for long-running
   * instances, observed contention levels will recur, so the buffers will eventually be needed
   * again; and for short-lived ones, it does not matter.
   */

  static final VarHandle TABLE_BUSY = fieldVarHandle(MethodHandles.lookup(),
      "tableBusy", VarHandle.class, StripedBuffer.class, int.class);

  /** Number of CPUS. */
  static final int NCPU = Runtime.getRuntime().availableProcessors();

  /** The bound on the table size. */
  static final int MAXIMUM_TABLE_SIZE = 4 * ceilingPowerOfTwo(NCPU);

  /** The maximum number of attempts when trying to expand the table. */
  static final int ATTEMPTS = 3;

  /** Table of buffers. When non-null, size is a power of 2. */
  volatile Buffer<E> @Nullable[] table;

  /** Spinlock (locked via CAS) used when resizing and/or creating Buffers. */
  volatile int tableBusy;

  /** CASes the tableBusy field from 0 to 1 to acquire lock. */
  final boolean casTableBusy() {
    return TABLE_BUSY.compareAndSet(this, 0, 1);
  }

  /**
   * Creates a new buffer instance after resizing to accommodate a producer.
   *
   * @param e the producer's element
   * @return a newly created buffer populated with a single element
   */
  protected abstract Buffer<E> create(E e);

  @Override
  @SuppressWarnings("Varifier")
  public int offer(E e) {
    @SuppressWarnings("deprecation")
    long z = mix64(Thread.currentThread().getId());
    int increment = ((int) (z >>> 32)) | 1;
    int h = (int) z;

    int mask;
    int result;
    Buffer<E> buffer;
    @Var boolean uncontended = true;
    @Nullable Buffer<E>[] buffers = table;
    if ((buffers == null)
        || ((mask = buffers.length - 1) < 0)
        || ((buffer = buffers[h & mask]) == null)
        || !(uncontended = ((result = buffer.offer(e)) != Buffer.FAILED))) {
      return expandOrRetry(e, h, increment, uncontended);
    }
    return result;
  }

  /**
   * Handles cases of updates involving initialization, resizing, creating new Buffers, and/or
   * contention. See above for explanation. This method suffers the usual non-modularity problems of
   * optimistic retry code, relying on rechecked sets of reads.
   *
   * @param e the element to add
   * @param h the thread's hash
   * @param increment the amount to increment by when rehashing
   * @param wasUncontended false if CAS failed before this call
   * @return {@code Buffer.SUCCESS}, {@code Buffer.FAILED}, or {@code Buffer.FULL}
   */
  final int expandOrRetry(E e, @Var int h, int increment, @Var boolean wasUncontended) {
    @Var int result = Buffer.FAILED;
    @Var boolean collide = false; // True if last slot nonempty
    for (int attempt = 0; attempt < ATTEMPTS; attempt++) {
      @Nullable Buffer<E>[] buffers;
      Buffer<E> buffer;
      int n;
      if (((buffers = table) != null) && ((n = buffers.length) > 0)) {
        if ((buffer = buffers[(n - 1) & h]) == null) {
          if ((tableBusy == 0) && casTableBusy()) { // Try to attach new Buffer
            @Var boolean created = false;
            try { // Recheck under lock
              @Nullable Buffer<E>[] rs;
              int mask;
              int j;
              if (((rs = table) != null) && ((mask = rs.length) > 0)
                  && (rs[j = (mask - 1) & h] == null)) {
                rs[j] = create(e);
                created = true;
              }
            } finally {
              tableBusy = 0;
            }
            if (created) {
              result = Buffer.SUCCESS;
              break;
            }
            continue; // Slot is now non-empty
          }
          collide = false;
        } else if (!wasUncontended) { // CAS already known to fail
          wasUncontended = true;      // Continue after rehash
        } else if ((result = buffer.offer(e)) != Buffer.FAILED) {
          break;
        } else if ((n >= MAXIMUM_TABLE_SIZE) || (table != buffers)) {
          collide = false; // At max size or stale
        } else if (!collide) {
          collide = true;
        } else if ((tableBusy == 0) && casTableBusy()) {
          try {
            if (table == buffers) { // Expand table unless stale
              table = Arrays.copyOf(buffers, n << 1);
            }
          } finally {
            tableBusy = 0;
          }
          collide = false;
          continue; // Retry with expanded table
        }
        h += increment;
      } else if ((tableBusy == 0) && (table == buffers) && casTableBusy()) {
        @Var boolean init = false;
        try { // Initialize table
          if (table == buffers) {
            @SuppressWarnings({"rawtypes", "unchecked"})
            Buffer<E>[] rs = new Buffer[1];
            rs[0] = create(e);
            table = rs;
            init = true;
          }
        } finally {
          tableBusy = 0;
        }
        if (init) {
          result = Buffer.SUCCESS;
          break;
        }
      }
    }
    return result;
  }

  @Override
  public void drainTo(Consumer<E> consumer) {
    @Nullable Buffer<E>[] buffers = table;
    if (buffers == null) {
      return;
    }
    for (Buffer<E> buffer : buffers) {
      if (buffer != null) {
        buffer.drainTo(consumer);
      }
    }
  }

  @Override
  public long reads() {
    @Nullable Buffer<E>[] buffers = table;
    if (buffers == null) {
      return 0;
    }
    @Var long reads = 0;
    for (Buffer<E> buffer : buffers) {
      if (buffer != null) {
        reads += buffer.reads();
      }
    }
    return reads;
  }

  @Override
  public long writes() {
    @Nullable Buffer<E>[] buffers = table;
    if (buffers == null) {
      return 0;
    }
    @Var long writes = 0;
    for (Buffer<E> buffer : buffers) {
      if (buffer != null) {
        writes += buffer.writes();
      }
    }
    return writes;
  }

  /** Computes Stafford variant 13 of 64-bit mix function. */
  static long mix64(@Var long z) {
    z = (z ^ (z >>> 30)) * 0xbf58476d1ce4e5b9L;
    z = (z ^ (z >>> 27)) * 0x94d049bb133111ebL;
    return z ^ (z >>> 31);
  }
}

1	/*
2	* Copyright 2015 Ben Manes. All Rights Reserved.
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16	/*
17	* Written by Doug Lea with assistance from members of JCP JSR-166
18	* Expert Group and released to the public domain, as explained at
19	* http://creativecommons.org/publicdomain/zero/1.0/
20	*/
21	package com.github.benmanes.caffeine.cache;
22
23	import static com.github.benmanes.caffeine.cache.Caffeine.ceilingPowerOfTwo;
24	import static java.lang.invoke.ConstantBootstraps.fieldVarHandle;
25
26	import java.lang.invoke.MethodHandles;
27	import java.lang.invoke.VarHandle;
28	import java.util.Arrays;
29	import java.util.function.Consumer;
30
31	import org.jspecify.annotations.Nullable;
32
33	import com.google.errorprone.annotations.Var;
34
35	/**
36	* A base class providing the mechanics for supporting dynamic striping of bounded buffers. This
37	* implementation is an adaption of the numeric 64-bit <i>java.util.concurrent.atomic.Striped64</i>
38	* class, which is used by atomic counters. The approach was modified to lazily grow an array of
39	* buffers in order to minimize memory usage for caches that are not heavily contended on.
40	*
41	* @author dl@cs.oswego.edu (Doug Lea)
42	* @author ben.manes@gmail.com (Ben Manes)
43	*/
44	abstract class StripedBuffer<E> implements Buffer<E> {	1✔
45	/*
46	* This class maintains a lazily-initialized table of atomically updated buffers. The table size
47	* is a power of two. Indexing uses masked per-thread hash codes. Nearly all declarations in this
48	* class are package-private, accessed directly by subclasses.
49	*
50	* Table entries are of class Buffer and should be padded to reduce cache contention. Padding is
51	* overkill for most atomics because they are usually irregularly scattered in memory and thus
52	* don't interfere much with each other. But atomic objects residing in arrays will tend to be
53	* placed adjacent to each other, and so will most often share cache lines (with a huge negative
54	* performance impact) without this precaution.
55	*
56	* In part because Buffers are relatively large, we avoid creating them until they are needed.
57	* When there is no contention, all updates are made to a single buffer. Upon contention (a failed
58	* CAS inserting into the buffer), the table is expanded to size 2. The table size is doubled upon
59	* further contention until reaching the nearest power of two greater than or equal to the number
60	* of CPUS. Table slots remain empty (null) until they are needed.
61	*
62	* A single spinlock ("tableBusy") is used for initializing and resizing the table, as well as
63	* populating slots with new Buffers. There is no need for a blocking lock; when the lock is not
64	* available, threads try other slots. During these retries, there is increased contention and
65	* reduced locality, which is still better than alternatives.
66	*
67	* Contention and/or table collisions are indicated by failed CASes when performing an update
68	* operation. Upon a collision, if the table size is less than the capacity, it is doubled in size
69	* unless some other thread holds the lock. If a hashed slot is empty, and lock is available, a
70	* new Buffer is created. Otherwise, if the slot exists, a CAS is tried. The thread id serves as
71	* the base for per-thread hash codes. Retries proceed by "incremental hashing", using the top
72	* half of the seed to increment the bottom half which is used as a probe to try to find a free
73	* slot.
74	*
75	* The table size is capped because, when there are more threads than CPUs, supposing that each
76	* thread were bound to a CPU, there would exist a perfect hash function mapping threads to slots
77	* that eliminates collisions. When we reach capacity, we search for this mapping by varying the
78	* hash codes of colliding threads. Because search is random, and collisions only become known via
79	* CAS failures, convergence can be slow, and because threads are typically not bound to CPUs
80	* forever, may not occur at all. However, despite these limitations, observed contention rates
81	* are typically low in these cases.
82	*
83	* It is possible for a Buffer to become unused when threads that once hashed to it terminate, as
84	* well as in the case where doubling the table causes no thread to hash to it under expanded
85	* mask. We do not try to detect or remove buffers, under the assumption that for long-running
86	* instances, observed contention levels will recur, so the buffers will eventually be needed
87	* again; and for short-lived ones, it does not matter.
88	*/
89
90	static final VarHandle TABLE_BUSY = fieldVarHandle(MethodHandles.lookup(),	1✔
91	"tableBusy", VarHandle.class, StripedBuffer.class, int.class);
92
93	/** Number of CPUS. */
94	static final int NCPU = Runtime.getRuntime().availableProcessors();	1✔
95
96	/** The bound on the table size. */
97	static final int MAXIMUM_TABLE_SIZE = 4 * ceilingPowerOfTwo(NCPU);	1✔
98
99	/** The maximum number of attempts when trying to expand the table. */
100	static final int ATTEMPTS = 3;
101
102	/** Table of buffers. When non-null, size is a power of 2. */
103	volatile Buffer<E> @Nullable[] table;
104
105	/** Spinlock (locked via CAS) used when resizing and/or creating Buffers. */
106	volatile int tableBusy;
107
108	/** CASes the tableBusy field from 0 to 1 to acquire lock. */
109	final boolean casTableBusy() {
110	return TABLE_BUSY.compareAndSet(this, 0, 1);	1✔
111	}
112
113	/**
114	* Creates a new buffer instance after resizing to accommodate a producer.
115	*
116	* @param e the producer's element
117	* @return a newly created buffer populated with a single element
118	*/
119	protected abstract Buffer<E> create(E e);
120
121	@Override
122	@SuppressWarnings("Varifier")
123	public int offer(E e) {
124	@SuppressWarnings("deprecation")
125	long z = mix64(Thread.currentThread().getId());	1✔
126	int increment = ((int) (z >>> 32)) \| 1;	1✔
127	int h = (int) z;	1✔
128
129	int mask;
130	int result;
131	Buffer<E> buffer;
132	@Var boolean uncontended = true;	1✔
133	@Nullable Buffer<E>[] buffers = table;	1✔
134	if ((buffers == null)	1✔
135	\|\| ((mask = buffers.length - 1) < 0)
136	\|\| ((buffer = buffers[h & mask]) == null)
137	\|\| !(uncontended = ((result = buffer.offer(e)) != Buffer.FAILED))) {	1✔
138	return expandOrRetry(e, h, increment, uncontended);	1✔
139	}
140	return result;	1✔
141	}
142
143	/**
144	* Handles cases of updates involving initialization, resizing, creating new Buffers, and/or
145	* contention. See above for explanation. This method suffers the usual non-modularity problems of
146	* optimistic retry code, relying on rechecked sets of reads.
147	*
148	* @param e the element to add
149	* @param h the thread's hash
150	* @param increment the amount to increment by when rehashing
151	* @param wasUncontended false if CAS failed before this call
152	* @return {@code Buffer.SUCCESS}, {@code Buffer.FAILED}, or {@code Buffer.FULL}
153	*/
154	final int expandOrRetry(E e, @Var int h, int increment, @Var boolean wasUncontended) {
155	@Var int result = Buffer.FAILED;	1✔
156	@Var boolean collide = false; // True if last slot nonempty	1✔
157	for (int attempt = 0; attempt < ATTEMPTS; attempt++) {	1✔
158	@Nullable Buffer<E>[] buffers;
159	Buffer<E> buffer;
160	int n;
161	if (((buffers = table) != null) && ((n = buffers.length) > 0)) {	1✔
162	if ((buffer = buffers[(n - 1) & h]) == null) {	1✔
163	if ((tableBusy == 0) && casTableBusy()) { // Try to attach new Buffer	1!
164	@Var boolean created = false;	1✔
165	try { // Recheck under lock
166	@Nullable Buffer<E>[] rs;
167	int mask;
168	int j;
169	if (((rs = table) != null) && ((mask = rs.length) > 0)	1!
170	&& (rs[j = (mask - 1) & h] == null)) {
171	rs[j] = create(e);	1✔
172	created = true;	1✔
173	}
174	} finally {
175	tableBusy = 0;	1✔
176	}
177	if (created) {	1!
178	result = Buffer.SUCCESS;	1✔
179	break;	1✔
180	}
181	continue; // Slot is now non-empty
182	}
UNCOV 183	collide = false;	×
184	} else if (!wasUncontended) { // CAS already known to fail	1✔
185	wasUncontended = true; // Continue after rehash	1✔
186	} else if ((result = buffer.offer(e)) != Buffer.FAILED) {	1✔
187	break;	1✔
188	} else if ((n >= MAXIMUM_TABLE_SIZE) \|\| (table != buffers)) {	1!
189	collide = false; // At max size or stale	1✔
190	} else if (!collide) {	1✔
191	collide = true;	1✔
192	} else if ((tableBusy == 0) && casTableBusy()) {	1!
193	try {
194	if (table == buffers) { // Expand table unless stale	1!
195	table = Arrays.copyOf(buffers, n << 1);	1✔
196	}
197	} finally {
198	tableBusy = 0;	1✔
199	}
200	collide = false;	1✔
201	continue; // Retry with expanded table	1✔
202	}
203	h += increment;	1✔
204	} else if ((tableBusy == 0) && (table == buffers) && casTableBusy()) {	1✔
205	@Var boolean init = false;	1✔
206	try { // Initialize table
207	if (table == buffers) {	1✔
208	@SuppressWarnings({"rawtypes", "unchecked"})
209	Buffer<E>[] rs = new Buffer[1];	1✔
210	rs[0] = create(e);	1✔
211	table = rs;	1✔
212	init = true;	1✔
213	}
214	} finally {
215	tableBusy = 0;	1✔
216	}
217	if (init) {	1✔
218	result = Buffer.SUCCESS;	1✔
219	break;	1✔
220	}
221	}
222	}
223	return result;	1✔
224	}
225
226	@Override
227	public void drainTo(Consumer<E> consumer) {
228	@Nullable Buffer<E>[] buffers = table;	1✔
229	if (buffers == null) {	1✔
230	return;	1✔
231	}
232	for (Buffer<E> buffer : buffers) {	1✔
233	if (buffer != null) {	1✔
234	buffer.drainTo(consumer);	1✔
235	}
236	}
237	}	1✔
238
239	@Override
240	public long reads() {
241	@Nullable Buffer<E>[] buffers = table;	1✔
242	if (buffers == null) {	1✔
243	return 0;	1✔
244	}
245	@Var long reads = 0;	1✔
246	for (Buffer<E> buffer : buffers) {	1✔
247	if (buffer != null) {	1✔
248	reads += buffer.reads();	1✔
249	}
250	}
251	return reads;	1✔
252	}
253
254	@Override
255	public long writes() {
256	@Nullable Buffer<E>[] buffers = table;	1✔
257	if (buffers == null) {	1✔
258	return 0;	1✔
259	}
260	@Var long writes = 0;	1✔
261	for (Buffer<E> buffer : buffers) {	1✔
262	if (buffer != null) {	1✔
263	writes += buffer.writes();	1✔
264	}
265	}
266	return writes;	1✔
267	}
268
269	/** Computes Stafford variant 13 of 64-bit mix function. */
270	static long mix64(@Var long z) {
271	z = (z ^ (z >>> 30)) * 0xbf58476d1ce4e5b9L;	1✔
272	z = (z ^ (z >>> 27)) * 0x94d049bb133111ebL;	1✔
273	return z ^ (z >>> 31);	1✔
274	}
275	}

ben-manes / caffeine / #5614

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous