• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

ben-manes / caffeine / #3896

pending completion
#3896

push

github-actions

ben-manes
upgrade jamm library (memory meter)

7542 of 7616 relevant lines covered (99.03%)

0.99 hits per line

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

97.7
/caffeine/src/main/java/com/github/benmanes/caffeine/cache/StripedBuffer.java
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

25
import java.lang.invoke.MethodHandles;
26
import java.lang.invoke.VarHandle;
27
import java.util.Arrays;
28
import java.util.function.Consumer;
29

30
import org.checkerframework.checker.nullness.qual.Nullable;
31

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

87
  static final VarHandle TABLE_BUSY;
88

89
  /** Number of CPUS. */
90
  static final int NCPU = Runtime.getRuntime().availableProcessors();
1✔
91

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

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

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

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

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

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

117
  @Override
118
  public int offer(E e) {
119
    @SuppressWarnings("deprecation")
120
    long z = mix64(Thread.currentThread().getId());
1✔
121
    int increment = ((int) (z >>> 32)) | 1;
1✔
122
    int h = (int) z;
1✔
123

124
    int mask;
125
    int result;
126
    Buffer<E> buffer;
127
    boolean uncontended = true;
1✔
128
    Buffer<E>[] buffers = table;
1✔
129
    if ((buffers == null)
1✔
130
        || ((mask = buffers.length - 1) < 0)
131
        || ((buffer = buffers[h & mask]) == null)
132
        || !(uncontended = ((result = buffer.offer(e)) != Buffer.FAILED))) {
1✔
133
      return expandOrRetry(e, h, increment, uncontended);
1✔
134
    }
135
    return result;
1✔
136
  }
137

138
  /**
139
   * Handles cases of updates involving initialization, resizing, creating new Buffers, and/or
140
   * contention. See above for explanation. This method suffers the usual non-modularity problems of
141
   * optimistic retry code, relying on rechecked sets of reads.
142
   *
143
   * @param e the element to add
144
   * @param h the thread's hash
145
   * @param increment the amount to increment by when rehashing
146
   * @param wasUncontended false if CAS failed before this call
147
   * @return {@code Buffer.SUCCESS}, {@code Buffer.FAILED}, or {@code Buffer.FULL}
148
   */
149
  final int expandOrRetry(E e, int h, int increment, boolean wasUncontended) {
150
    int result = Buffer.FAILED;
1✔
151
    boolean collide = false; // True if last slot nonempty
1✔
152
    for (int attempt = 0; attempt < ATTEMPTS; attempt++) {
1✔
153
      Buffer<E>[] buffers;
154
      Buffer<E> buffer;
155
      int n;
156
      if (((buffers = table) != null) && ((n = buffers.length) > 0)) {
1✔
157
        if ((buffer = buffers[(n - 1) & h]) == null) {
1✔
158
          if ((tableBusy == 0) && casTableBusy()) { // Try to attach new Buffer
1✔
159
            boolean created = false;
1✔
160
            try { // Recheck under lock
161
              Buffer<E>[] rs;
162
              int mask, j;
163
              if (((rs = table) != null) && ((mask = rs.length) > 0)
1✔
164
                  && (rs[j = (mask - 1) & h] == null)) {
165
                rs[j] = create(e);
1✔
166
                created = true;
1✔
167
              }
168
            } finally {
169
              tableBusy = 0;
1✔
170
            }
171
            if (created) {
1✔
172
              result = Buffer.SUCCESS;
1✔
173
              break;
1✔
174
            }
175
            continue; // Slot is now non-empty
176
          }
177
          collide = false;
1✔
178
        } else if (!wasUncontended) { // CAS already known to fail
1✔
179
          wasUncontended = true;      // Continue after rehash
1✔
180
        } else if ((result = buffer.offer(e)) != Buffer.FAILED) {
1✔
181
          break;
1✔
182
        } else if ((n >= MAXIMUM_TABLE_SIZE) || (table != buffers)) {
1✔
183
          collide = false; // At max size or stale
1✔
184
        } else if (!collide) {
1✔
185
          collide = true;
1✔
186
        } else if ((tableBusy == 0) && casTableBusy()) {
1✔
187
          try {
188
            if (table == buffers) { // Expand table unless stale
1✔
189
              table = Arrays.copyOf(buffers, n << 1);
1✔
190
            }
191
          } finally {
192
            tableBusy = 0;
1✔
193
          }
194
          collide = false;
1✔
195
          continue; // Retry with expanded table
1✔
196
        }
197
        h += increment;
1✔
198
      } else if ((tableBusy == 0) && (table == buffers) && casTableBusy()) {
1✔
199
        boolean init = false;
1✔
200
        try { // Initialize table
201
          if (table == buffers) {
1✔
202
            @SuppressWarnings({"rawtypes", "unchecked"})
203
            Buffer<E>[] rs = new Buffer[1];
1✔
204
            rs[0] = create(e);
1✔
205
            table = rs;
1✔
206
            init = true;
1✔
207
          }
208
        } finally {
209
          tableBusy = 0;
1✔
210
        }
211
        if (init) {
1✔
212
          result = Buffer.SUCCESS;
1✔
213
          break;
1✔
214
        }
215
      }
216
    }
217
    return result;
1✔
218
  }
219

220
  @Override
221
  public void drainTo(Consumer<E> consumer) {
222
    Buffer<E>[] buffers = table;
1✔
223
    if (buffers == null) {
1✔
224
      return;
1✔
225
    }
226
    for (Buffer<E> buffer : buffers) {
1✔
227
      if (buffer != null) {
1✔
228
        buffer.drainTo(consumer);
1✔
229
      }
230
    }
231
  }
1✔
232

233
  @Override
234
  public long reads() {
235
    Buffer<E>[] buffers = table;
1✔
236
    if (buffers == null) {
1✔
237
      return 0;
1✔
238
    }
239
    long reads = 0;
1✔
240
    for (Buffer<E> buffer : buffers) {
1✔
241
      if (buffer != null) {
1✔
242
        reads += buffer.reads();
1✔
243
      }
244
    }
245
    return reads;
1✔
246
  }
247

248
  @Override
249
  public long writes() {
250
    Buffer<E>[] buffers = table;
1✔
251
    if (buffers == null) {
1✔
252
      return 0;
1✔
253
    }
254
    long writes = 0;
1✔
255
    for (Buffer<E> buffer : buffers) {
1✔
256
      if (buffer != null) {
1✔
257
        writes += buffer.writes();
1✔
258
      }
259
    }
260
    return writes;
1✔
261
  }
262

263
  /** Computes Stafford variant 13 of 64-bit mix function. */
264
  static long mix64(long z) {
265
    z = (z ^ (z >>> 30)) * 0xbf58476d1ce4e5b9L;
1✔
266
    z = (z ^ (z >>> 27)) * 0x94d049bb133111ebL;
1✔
267
    return z ^ (z >>> 31);
1✔
268
  }
269

270
  static {
271
    try {
272
      TABLE_BUSY = MethodHandles.lookup()
1✔
273
          .findVarHandle(StripedBuffer.class, "tableBusy", int.class);
1✔
274
    } catch (ReflectiveOperationException e) {
×
275
      throw new ExceptionInInitializerError(e);
×
276
    }
1✔
277
  }
1✔
278
}
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc