Skip to content

Commit 7e2cd4c

Browse files
LinShunKangLinShunKang
committed
Perf: Reduce the memory footprint of AtomicIntHashCounter
1 parent bc28abb commit 7e2cd4c

File tree

2 files changed

+59
-66
lines changed

2 files changed

+59
-66
lines changed

‎MyPerf4J-Base/src/main/java/cn/myperf4j/base/util/UnsafeUtils.java

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -78,6 +78,14 @@ public static Object getAndSetObject(Object obj, long offset, Object newVal) {
7878
return oldVal;
7979
}
8080

81+
public static long getAndAddLong(Object obj, long offset, long delta) {
82+
long oldVal;
83+
do {
84+
oldVal = UNSAFE.getLongVolatile(obj, offset);
85+
} while (!UNSAFE.compareAndSwapLong(obj, offset, oldVal, oldVal + delta));
86+
return oldVal;
87+
}
88+
8189
private UnsafeUtils() {
8290
//empty
8391
}

‎MyPerf4J-Base/src/main/java/cn/myperf4j/base/util/concurrent/AtomicIntHashCounter.java

Lines changed: 51 additions & 66 deletions
Original file line numberDiff line numberDiff line change
@@ -6,15 +6,9 @@
66

77
import java.io.Serializable;
88
import java.util.Arrays;
9-
import java.util.concurrent.atomic.AtomicInteger;
10-
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
11-
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
129

1310
import static cn.myperf4j.base.util.UnsafeUtils.fieldOffset;
14-
import static java.lang.Integer.MAX_VALUE;
1511
import static java.lang.Integer.MIN_VALUE;
16-
import static java.util.concurrent.atomic.AtomicLongFieldUpdater.newUpdater;
17-
import static java.util.concurrent.atomic.AtomicReferenceFieldUpdater.newUpdater;
1812

1913
/**
2014
* Created by LinShunkang on 2022/07/17
@@ -33,14 +27,16 @@ public class AtomicIntHashCounter implements IntHashCounter {
3327

3428
private static final long IHC_OFFSET = fieldOffset(AtomicIntHashCounter.class, "ihc");
3529

36-
private static final long VAL_0_OFFSET = fieldOffset(AtomicIntHashCounter.class, "val0");
30+
private static final long SIZE_OFFSET = fieldOffset(AtomicIntHashCounter.class, "size");
3731

38-
private static final int MAX_CAPACITY = MAX_VALUE >> 1;
32+
private static final long VAL_0_OFFSET = fieldOffset(AtomicIntHashCounter.class, "val0");
3933

4034
private static final int MIN_LOG_SIZE = 4;
4135

4236
private static final int MAX_LOG_SIZE = 29;
4337

38+
private static final int MAX_CAPACITY = 1 << MAX_LOG_SIZE;
39+
4440
private static final int MIN_SIZE = 1 << MIN_LOG_SIZE; // Must be power of 2
4541

4642
private static final int RE_PROBE_LIMIT = 10;
@@ -59,7 +55,11 @@ public class AtomicIntHashCounter implements IntHashCounter {
5955

6056
private IHC ihc;
6157

62-
private int val0; // Value for Key: NO_KEY
58+
// Size in active K,V pairs
59+
private int size;
60+
61+
// Value for Key: NO_KEY
62+
private int val0;
6363

6464
public AtomicIntHashCounter() {
6565
this(MIN_SIZE);
@@ -72,7 +72,8 @@ public AtomicIntHashCounter(int initialCapacity) {
7272
throw new IllegalArgumentException("Max initialCapacity need low than " + MAX_CAPACITY);
7373
}
7474

75-
this.ihc = new IHC(this, new AtomicInteger(), log2Size(initialCapacity));
75+
this.ihc = new IHC(this, log2Size(initialCapacity));
76+
this.size = 0;
7677
this.val0 = 0;
7778
}
7879

@@ -100,6 +101,10 @@ private void helpCopy() {
100101
topIhc.helpCopy(false);
101102
}
102103

104+
private void incrementSize() {
105+
UnsafeUtils.getAndAddInt(this, SIZE_OFFSET, 1);
106+
}
107+
103108
@Override
104109
public int get(int key) {
105110
if (key == NO_KEY) {
@@ -143,11 +148,12 @@ public int addAndGet(int key, int delta) {
143148

144149
@Override
145150
public int size() {
146-
return (val0 == 0 ? 0 : 1) + ihc.size();
151+
return (val0 == 0 ? 0 : 1) + size;
147152
}
148153

149154
@Override
150155
public void reset() {
156+
this.size = 0;
151157
this.val0 = 0;
152158
finishCopy().reset();
153159
}
@@ -197,27 +203,26 @@ private static final class IHC implements Serializable {
197203

198204
private static final long serialVersionUID = -156965978265678243L;
199205

200-
private static final AtomicReferenceFieldUpdater<IHC, IHC> NEXT_IHC_UPDATER =
201-
newUpdater(IHC.class, IHC.class, "nextIhc");
206+
private static final long SLOTS_OFFSET = fieldOffset(IHC.class, "slots");
207+
208+
private static final long NEXT_IHC_OFFSET = fieldOffset(IHC.class, "nextIhc");
202209

203-
private static final AtomicLongFieldUpdater<IHC> RESIZE_THREADS_UPDATER =
204-
newUpdater(IHC.class, "resizeThreads");
210+
private static final long RESIZE_THREADS_OFFSET = fieldOffset(IHC.class, "resizeThreads");
205211

206-
private static final AtomicLongFieldUpdater<IHC> COPY_DONE_UPDATER = newUpdater(IHC.class, "copyDone");
212+
private static final long COPY_DONE_OFFSET = fieldOffset(IHC.class, "copyDone");
207213

208-
private static final AtomicLongFieldUpdater<IHC> COPY_IDX_UPDATER = newUpdater(IHC.class, "copyIdx");
214+
private static final long COPY_IDX_OFFSET = fieldOffset(IHC.class, "copyIdx");
209215

210216
// Back-pointer to top-level structure
211217
private final AtomicIntHashCounter aihc;
212218

213-
// Size in active K,V pairs
214-
private final AtomicInteger size;
215-
216-
private final AtomicInteger slots;
219+
// Count of used slots, to tell when table is full of dead unusable slots
220+
private volatile int slots;
217221

218222
private volatile IHC nextIhc;
219223

220-
private volatile long resizeThreads; // Count of threads attempting an initial resize
224+
// Count of threads attempting an initial resize
225+
private volatile long resizeThreads;
221226

222227
private volatile long copyDone;
223228

@@ -229,31 +234,25 @@ private static final class IHC implements Serializable {
229234

230235
private final int reProbeLimit;
231236

232-
IHC(AtomicIntHashCounter aihc, AtomicInteger size, int logSize) {
237+
IHC(AtomicIntHashCounter aihc, int logSize) {
233238
this.aihc = aihc;
234-
this.size = size;
235-
this.slots = new AtomicInteger(0);
239+
this.slots = 0;
236240
this.kvs = new int[(1 << logSize) << 1];
237241
this.len = len(this.kvs);
238242
this.reProbeLimit = reProbeLimit(this.len);
239243
}
240244

241245
public void reset() {
242-
UNSAFE.setMemory(kvs, byteOffset(0), ((long) kvs.length) * I_SCALE, (byte) 0);
246+
this.slots = 0;
243247
this.nextIhc = null;
244-
this.size.set(0);
245-
this.slots.set(0);
246248
this.resizeThreads = 0L;
247249
this.copyDone = 0L;
248250
this.copyIdx = 0L;
249-
}
250-
251-
public int size() {
252-
return size.get();
251+
UNSAFE.setMemory(kvs, byteOffset(0), ((long) kvs.length) * I_SCALE, (byte) 0);
253252
}
254253

255254
private boolean casNextIhc(IHC newIhc) {
256-
return NEXT_IHC_UPDATER.compareAndSet(this, null, newIhc);
255+
return UNSAFE.compareAndSwapObject(this, NEXT_IHC_OFFSET, null, newIhc);
257256
}
258257

259258
private boolean casKv(int idx, long oldKv, long newKv) {
@@ -294,12 +293,7 @@ private int get(final int key) {
294293
return nextIhc.get(key); // Retry in the new table
295294
}
296295

297-
if (k == TOMB_PRIME) {
298-
aihc.helpCopy();
299-
return nextIhc.get(key); // Retry in the new table
300-
}
301-
302-
if (++reProbeTimes >= reProbeLimit) {
296+
if (++reProbeTimes >= reProbeLimit || k == TOMB_PRIME) {
303297
if (nextIhc != null) { // Table copy in progress?
304298
aihc.helpCopy();
305299
return nextIhc.get(key); // Retry in the new table
@@ -323,10 +317,20 @@ private int addDelta(final int key, final int delta, final boolean fromTableCopy
323317
k = key(kv);
324318
v = value(kv);
325319
if (k == NO_KEY) { //No key!
320+
assert v == 0;
326321
if (casKv(idx, kv, kv(key, delta))) {
327-
slots.addAndGet(1);
328322
if (!fromTableCopy) {
329-
size.addAndGet(1);
323+
aihc.incrementSize();
324+
}
325+
326+
// See if we want to move to a new table (to avoid high average re-probe counts).
327+
// We only check on the initial set of a Value from zero to not-zero
328+
final int slots = UnsafeUtils.getAndAddInt(this, SLOTS_OFFSET, 1);
329+
if (slots >= (len >> 1) + (len >> 2)) { // Table is full? slots > len * 3/4
330+
resize(); // Force the new table copy to start
331+
if (!fromTableCopy) {
332+
aihc.helpCopy(); // help along an existing copy
333+
}
330334
}
331335
return v;
332336
}
@@ -347,16 +351,6 @@ private int addDelta(final int key, final int delta, final boolean fromTableCopy
347351
return nextIhc.addDelta(key, delta, v != TOMB_PRIME);
348352
}
349353

350-
// See if we want to move to a new table (to avoid high average re-probe counts).
351-
// We only check on the initial set of a Value from zero to not-zero
352-
if (v == 0 && slots.get() >= len >> 1) {
353-
final IHC newIhc = resize(); // Force the new table copy to start
354-
if (!fromTableCopy) {
355-
aihc.helpCopy(); // help along an existing copy
356-
}
357-
return newIhc.addDelta(key, delta, fromTableCopy);
358-
}
359-
360354
// Try to increase the value with delta
361355
if (casValue(idx, v, v + delta)) {
362356
return v;
@@ -390,16 +384,13 @@ private IHC resize() {
390384
// Prevent integer overflow - limit of 2^31 elements in a Java array.
391385
// So here, 2^30 is the largest number of elements in the hash table
392386
if (log2 > MAX_LOG_SIZE) {
393-
throw new RuntimeException("Table is full, size=" + size + ", newSize=" + newSize + ", log2=" + log2);
387+
throw new RuntimeException("Table is full, size=" + aihc.size + ", newSize=" + newSize);
394388
}
395389

396390
// Now limit the number of threads actually allocating memory to a
397391
// handful - lest we have 750 threads all trying to allocate a giant
398392
// resized array.
399-
long r = resizeThreads;
400-
while (!RESIZE_THREADS_UPDATER.compareAndSet(this, r, r + 1)) {
401-
r = resizeThreads;
402-
}
393+
final long r = UnsafeUtils.getAndAddLong(this, RESIZE_THREADS_OFFSET, 1);
403394

404395
// Size calculation: 2 words (K+V) per table entry, plus a handful. We
405396
// guess at 64-bit pointers; 32-bit pointers screws up the size calc by
@@ -428,7 +419,7 @@ private IHC resize() {
428419
}
429420

430421
// New IHC - actually allocate the big arrays
431-
newIhc = new IHC(aihc, this.size, log2);
422+
newIhc = new IHC(aihc, log2);
432423

433424
// Another check after the slow allocation
434425
if (nextIhc != null) { // See if resize is already in progress
@@ -461,11 +452,7 @@ private void helpCopy(final boolean copyAll) {
461452
// algorithm) or do the copy work ourselves. Tiny tables with huge
462453
// thread counts trying to copy the table often 'panic'.
463454
if (panicStart == -1) { // No panic?
464-
copyIdx = (int) this.copyIdx;
465-
while (!COPY_IDX_UPDATER.compareAndSet(this, copyIdx, copyIdx + MIN_COPY_WORK)) {
466-
copyIdx = (int) this.copyIdx; // Re-read
467-
}
468-
455+
copyIdx = (int) UnsafeUtils.getAndAddLong(this, COPY_IDX_OFFSET, MIN_COPY_WORK);
469456
if (!(copyIdx < (oldLen << 1))) { // Panic!
470457
panicStart = copyIdx; // Record where we started to panic-copy
471458
}
@@ -525,10 +512,8 @@ private void copyCheckAndPromote(int workDone) {
525512
long copyDone = this.copyDone;
526513
assert (copyDone + workDone) <= oldLen;
527514
if (workDone > 0) {
528-
while (!COPY_DONE_UPDATER.compareAndSet(this, copyDone, copyDone + workDone)) {
529-
copyDone = this.copyDone; // Reload, retry
530-
assert (copyDone + workDone) <= oldLen;
531-
}
515+
copyDone = UnsafeUtils.getAndAddLong(this, COPY_DONE_OFFSET, workDone);
516+
assert (copyDone + workDone) <= oldLen;
532517
}
533518

534519
// Check for copy being ALL done, and promote. Note that we might have

0 commit comments

Comments
 (0)