Skip to content

Commit 830e38e

Browse files
lvcalvca
committed
perf: used custom maps in hot paths
Fixed issue #3965
1 parent 0aa8995 commit 830e38e

8 files changed

Lines changed: 858 additions & 90 deletions

File tree

‎engine/src/main/java/com/arcadedb/database/TransactionContext.java‎

Lines changed: 12 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@
3939
import com.arcadedb.log.LogManager;
4040
import com.arcadedb.schema.LocalSchema;
4141
import com.arcadedb.utility.IntHashSet;
42+
import com.arcadedb.utility.IntIntHashMap;
4243
import com.arcadedb.utility.RidHashSet;
4344

4445
import java.io.*;
@@ -61,7 +62,9 @@
6162
public class TransactionContext implements Transaction {
6263
private final DatabaseInternal database;
6364
private final Map<Integer, Integer> newPageCounters = new ConcurrentHashMap<>();
64-
private final Map<Integer, AtomicInteger> bucketRecordDelta = new HashMap<>();
65+
// Per-tx record-count delta per bucket. Single-threaded (HashMap was used, not ConcurrentHashMap),
66+
// so AtomicInteger was only a mutable-cell trick. IntIntHashMap.add(key, delta) covers it directly.
67+
private final IntIntHashMap bucketRecordDelta = new IntIntHashMap();
6568
private final Map<RID, Record> immutableRecordsCache = new HashMap<>(1024);
6669
private final Map<RID, Record> modifiedRecordsCache = new HashMap<>(1024);
6770
private final TransactionIndexContext indexChanges;
@@ -501,31 +504,22 @@ private Object getRequester() {
501504

502505
public Map<Integer, Integer> getBucketRecordDelta() {
503506
final Map<Integer, Integer> map = new HashMap<>(bucketRecordDelta.size());
504-
for (Map.Entry<Integer, AtomicInteger> entry : bucketRecordDelta.entrySet())
505-
map.put(entry.getKey(), entry.getValue().get());
507+
bucketRecordDelta.forEach((k, v) -> map.put(k, v));
506508
return map;
507509
}
508510

509511
/**
510512
* Returns the delta of records considering the pending changes in transaction.
511513
*/
512514
public long getBucketRecordDelta(final int bucketId) {
513-
final AtomicInteger delta = bucketRecordDelta.get(bucketId);
514-
if (delta != null)
515-
return delta.get();
516-
return 0;
515+
return bucketRecordDelta.get(bucketId, 0);
517516
}
518517

519518
/**
520519
* Updates the record counter for buckets. At transaction commit, the delta is updated into the schema.
521520
*/
522521
public void updateBucketRecordDelta(final int bucketId, final int delta) {
523-
AtomicInteger counter = bucketRecordDelta.get(bucketId);
524-
if (counter == null) {
525-
counter = new AtomicInteger(delta);
526-
bucketRecordDelta.put(bucketId, counter);
527-
} else
528-
counter.addAndGet(delta);
522+
bucketRecordDelta.add(bucketId, delta);
529523
}
530524

531525
/**
@@ -536,7 +530,7 @@ public void commitFromReplica(final WALFile.WALTransaction buffer,
536530
final Map<Integer, Integer> bucketRecordDelta) throws TransactionException {
537531

538532
for (Map.Entry<Integer, Integer> entry : bucketRecordDelta.entrySet())
539-
this.bucketRecordDelta.put(entry.getKey(), new AtomicInteger(entry.getValue()));
533+
this.bucketRecordDelta.put(entry.getKey(), entry.getValue());
540534

541535
final int totalImpactedPages = buffer.pages.length;
542536
if (totalImpactedPages == 0 && keysTx.isEmpty()) {
@@ -753,13 +747,13 @@ public void commit2ndPhase(final TransactionPhase1 changes) {
753747
((PaginatedComponent) database.getSchema().getFileById(entry.getKey())).updatePageCount(entry.getValue());
754748

755749
// UPDATE RECORD COUNT
756-
for (Map.Entry<Integer, AtomicInteger> entry : bucketRecordDelta.entrySet()) {
750+
bucketRecordDelta.forEach((bucketId, delta) -> {
757751
// THE BUCKET/FILE COULD HAVE BEEN REMOVED IN THE CURRENT TRANSACTION
758-
final LocalBucket bucket = (LocalBucket) database.getSchema().getFileByIdIfExists(entry.getKey());
752+
final LocalBucket bucket = (LocalBucket) database.getSchema().getFileByIdIfExists(bucketId);
759753
if (bucket != null && bucket.getCachedRecordCount() > -1)
760754
// UPDATE THE CACHE COUNTER ONLY IF ALREADY COMPUTED
761-
bucket.setCachedRecordCount(bucket.getCachedRecordCount() + entry.getValue().get());
762-
}
755+
bucket.setCachedRecordCount(bucket.getCachedRecordCount() + delta);
756+
});
763757

764758
for (final Record r : modifiedRecordsCache.values())
765759
((RecordInternal) r).unsetDirty();

‎engine/src/main/java/com/arcadedb/engine/LocalBucket.java‎

Lines changed: 74 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@
3939
import com.arcadedb.serializer.json.JSONArray;
4040
import com.arcadedb.serializer.json.JSONObject;
4141
import com.arcadedb.utility.FileUtils;
42+
import com.arcadedb.utility.IntIntHashMap;
4243

4344
import java.io.*;
4445
import java.util.*;
@@ -88,7 +89,11 @@ public class LocalBucket extends PaginatedComponent implements Bucket {
8889
protected final int contentHeaderSize;
8990
private final int maxRecordsInPage = DEF_MAX_RECORDS_IN_PAGE;
9091
private final AtomicLong cachedRecordCount = new AtomicLong(-1);
91-
private final TreeMap<Integer, Integer> freeSpaceInPages = new TreeMap<>();
92+
// pageId → free-space-bytes. TreeMap ordering is unused (verified by grep), so a primitive
93+
// open-addressing map saves memory and avoids Integer boxing on every read/write/remove on
94+
// the page-allocation hot path. Bounded by MAX_PAGES_GATHER_STATS (100). Single-threaded
95+
// access is enforced by `synchronized (freeSpaceInPages)` blocks at every callsite.
96+
private final IntIntHashMap freeSpaceInPages = new IntIntHashMap();
9297
private final REUSE_SPACE_MODE reuseSpaceMode;
9398
private long timeOfLastStats = 0L;
9499
private long changesFromLastStats = 0L;
@@ -1875,34 +1880,55 @@ private PageAnalysis findAvailableSpace(final int currentPageId, final int space
18751880
private PageAnalysis findAvailableSpaceFromStatistics(final int currentPageId, final int spaceNeeded,
18761881
final boolean multiPageRecord)
18771882
throws IOException {
1883+
// Snapshot keys/values into local arrays so we can safely mutate freeSpaceInPages
1884+
// (put/remove) inside the loop body. The snapshot is bounded by MAX_PAGES_GATHER_STATS (100).
1885+
//
1886+
// The snapshot is sorted by pageId to preserve the deterministic "fill lowest pageId first"
1887+
// allocation behavior that the previous TreeMap-backed implementation provided implicitly via
1888+
// its ordered iteration. Tests like RandomDeleteTest depend on this so that re-inserts after
1889+
// bulk deletes land at the same RIDs as the original inserts.
1890+
final int snapSize = freeSpaceInPages.size();
1891+
final int[] snapPageIds = new int[snapSize];
1892+
final int[] snapPageStats = new int[snapSize];
1893+
final int[] cursor = { 0 };
1894+
freeSpaceInPages.forEach((k, v) -> {
1895+
snapPageIds[cursor[0]] = k;
1896+
snapPageStats[cursor[0]] = v;
1897+
cursor[0]++;
1898+
});
1899+
sortByPageIdAscending(snapPageIds, snapPageStats, snapSize);
1900+
18781901
PageAnalysis bestPageAnalysis = null;
1879-
List<Integer> pagesToRemove = null;
1880-
for (Map.Entry<Integer, Integer> entry : freeSpaceInPages.entrySet()) {
1881-
final int pageId = entry.getKey();
1902+
int[] pagesToRemove = null;
1903+
int pagesToRemoveCount = 0;
1904+
for (int s = 0; s < snapSize; s++) {
1905+
final int pageId = snapPageIds[s];
18821906
if (pageId == currentPageId)
18831907
// ALREADY EVALUATED
18841908
continue;
18851909

1886-
final Integer pageStats = entry.getValue();
1910+
final int pageStats = snapPageStats[s];
18871911

18881912
if (pageStats >= spaceNeeded) {
18891913
// CHECK IF THE SPACE AVAILABLE IS REAL
18901914
final PageAnalysis pageAnalysis = getAvailableSpaceInPage(pageId, spaceNeeded, multiPageRecord);
18911915

18921916
if (pageAnalysis.totalRecordsInPage >= maxRecordsInPage) {
1893-
pagesToRemove = pagesToRemove == null ? new ArrayList<>() : pagesToRemove;
1894-
pagesToRemove.add(pageId);
1917+
if (pagesToRemove == null)
1918+
pagesToRemove = new int[snapSize];
1919+
pagesToRemove[pagesToRemoveCount++] = pageId;
18951920
continue;
18961921
}
18971922

18981923
if (!pageAnalysis.createNewPage) {
18991924
if (pageAnalysis.spaceAvailableInCurrentPage != pageStats) {
19001925
// LOW COST UPDATE OF STATISTICS
19011926
if (pageAnalysis.spaceAvailableInCurrentPage < MINIMUM_SPACE_LEFT_IN_PAGE) {
1902-
pagesToRemove = pagesToRemove == null ? new ArrayList<>() : pagesToRemove;
1903-
pagesToRemove.add(pageId);
1927+
if (pagesToRemove == null)
1928+
pagesToRemove = new int[snapSize];
1929+
pagesToRemove[pagesToRemoveCount++] = pageId;
19041930
} else
1905-
entry.setValue(pageAnalysis.spaceAvailableInCurrentPage);
1931+
freeSpaceInPages.put(pageId, pageAnalysis.spaceAvailableInCurrentPage);
19061932
}
19071933

19081934
if (multiPageRecord && pageAnalysis.page.pageId.getPageNumber() == 0 && pageAnalysis.availablePositionIndex == 0)
@@ -1916,8 +1942,8 @@ private PageAnalysis findAvailableSpaceFromStatistics(final int currentPageId, f
19161942
}
19171943

19181944
if (pagesToRemove != null)
1919-
for (Integer pageId : pagesToRemove)
1920-
freeSpaceInPages.remove(pageId);
1945+
for (int i = 0; i < pagesToRemoveCount; i++)
1946+
freeSpaceInPages.remove(pagesToRemove[i], -1);
19211947

19221948
return bestPageAnalysis;
19231949
}
@@ -1974,35 +2000,36 @@ private void updatePageStatistics(final int pageId, final int availableSpace, fi
19742000

19752001
synchronized (freeSpaceInPages) {
19762002
if (availableSpace + delta == 0)
1977-
freeSpaceInPages.remove(pageId);
2003+
freeSpaceInPages.remove(pageId, -1);
19782004
else {
19792005
final int usableSpaceInPage = getPageSize() - contentHeaderSize;
19802006

1981-
final Integer freeSpaceInPage = freeSpaceInPages.get(pageId);
1982-
final int prevSpace = availableSpace == 0 && freeSpaceInPage != null ? freeSpaceInPage : availableSpace;
2007+
final boolean hasEntry = freeSpaceInPages.containsKey(pageId);
2008+
final int existingFreeSpace = hasEntry ? freeSpaceInPages.get(pageId, 0) : 0;
2009+
final int prevSpace = availableSpace == 0 && hasEntry ? existingFreeSpace : availableSpace;
19832010
final int newSpace = prevSpace + delta;
19842011

1985-
if (freeSpaceInPage != null) {
2012+
if (hasEntry) {
19862013
if (newSpace <= MINIMUM_SPACE_LEFT_IN_PAGE || (freeSpaceInPages.size() >= MAX_PAGES_GATHER_STATS
19872014
&& newSpace * 100 / usableSpaceInPage < GATHER_STATS_MIN_SPACE_PERC))
1988-
freeSpaceInPages.remove(pageId);
2015+
freeSpaceInPages.remove(pageId, -1);
19892016
else
19902017
freeSpaceInPages.put(pageId, newSpace);
19912018
} else if (newSpace * 100 / usableSpaceInPage >= GATHER_STATS_MIN_SPACE_PERC) {
19922019
if (freeSpaceInPages.size() >= MAX_PAGES_GATHER_STATS) {
19932020
// REMOVE THE SMALLEST PAGE
1994-
int lowestPageId = -1;
1995-
int lowestPageSpace = -1;
2021+
final int[] lowestPageId = { -1 };
2022+
final int[] lowestPageSpace = { -1 };
19962023

1997-
for (Map.Entry<Integer, Integer> entry : freeSpaceInPages.entrySet()) {
1998-
if (lowestPageId < 0 || entry.getValue() < lowestPageSpace) {
1999-
lowestPageId = entry.getKey();
2000-
lowestPageSpace = entry.getValue();
2024+
freeSpaceInPages.forEach((k, v) -> {
2025+
if (lowestPageId[0] < 0 || v < lowestPageSpace[0]) {
2026+
lowestPageId[0] = k;
2027+
lowestPageSpace[0] = v;
20012028
}
2002-
}
2029+
});
20032030

2004-
if (lowestPageId > -1 && lowestPageSpace < newSpace) {
2005-
freeSpaceInPages.remove(lowestPageId);
2031+
if (lowestPageId[0] > -1 && lowestPageSpace[0] < newSpace) {
2032+
freeSpaceInPages.remove(lowestPageId[0], -1);
20062033
freeSpaceInPages.put(pageId, newSpace);
20072034
}
20082035
} else
@@ -2022,6 +2049,26 @@ public void setPageStatistics(final JSONArray pages) {
20222049
}
20232050
}
20242051

2052+
/**
2053+
* Insertion sort on two parallel arrays, ordering by ascending {@code pageIds[i]}. Bounded to
2054+
* MAX_PAGES_GATHER_STATS (100) entries so insertion sort is the right choice (low constant,
2055+
* great cache behavior, in-place).
2056+
*/
2057+
private static void sortByPageIdAscending(final int[] pageIds, final int[] pageStats, final int n) {
2058+
for (int i = 1; i < n; i++) {
2059+
final int kPage = pageIds[i];
2060+
final int kStats = pageStats[i];
2061+
int j = i - 1;
2062+
while (j >= 0 && pageIds[j] > kPage) {
2063+
pageIds[j + 1] = pageIds[j];
2064+
pageStats[j + 1] = pageStats[j];
2065+
j--;
2066+
}
2067+
pageIds[j + 1] = kPage;
2068+
pageStats[j + 1] = kStats;
2069+
}
2070+
}
2071+
20252072
public JSONObject getStatistics() {
20262073
final JSONObject json = new JSONObject();
20272074

@@ -2031,10 +2078,7 @@ public JSONObject getStatistics() {
20312078

20322079
final JSONArray pages = new JSONArray();
20332080
synchronized (freeSpaceInPages) {
2034-
for (Map.Entry<Integer, Integer> entry : freeSpaceInPages.entrySet())
2035-
pages.put(//
2036-
new JSONObject().put("id", entry.getKey()).put("free", entry.getValue())//
2037-
);
2081+
freeSpaceInPages.forEach((k, v) -> pages.put(new JSONObject().put("id", k).put("free", v)));
20382082
}
20392083
json.put("pages", pages);
20402084

‎engine/src/main/java/com/arcadedb/graph/GraphBatch.java‎

Lines changed: 35 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@
3535
import com.arcadedb.serializer.BinarySerializer;
3636
import com.arcadedb.serializer.BinaryTypes;
3737
import com.arcadedb.utility.LongHashSet;
38+
import com.arcadedb.utility.LongObjectHashMap;
3839

3940
import java.util.ArrayList;
4041
import java.util.Arrays;
@@ -152,13 +153,34 @@ public class GraphBatch implements AutoCloseable {
152153
private final Map<Long, RID> inChunkRIDCache = new ConcurrentHashMap<>();
153154

154155
// --- Deferred vertex head chunk updates: persisted in one batch pass at close() ---
155-
// vertexKey → latest segment RID that needs to be set on the vertex record
156-
private final Map<Long, RID> deferredOutHead = new ConcurrentHashMap<>();
157-
private final Map<Long, RID> deferredInHead = new ConcurrentHashMap<>();
156+
// vertexKey → latest segment RID that needs to be set on the vertex record.
157+
//
158+
// Lifecycle (matters for the choice of non-concurrent collection):
159+
// 1. Sequential put() calls happen from connectOutgoingEdgesSorted /
160+
// connectIncomingEdgesSequential (single-threaded paths).
161+
// 2. The PARALLEL paths (connectOutEdgesRangeLocal, connectIncomingEdgesRangeLocal)
162+
// do NOT touch these fields directly - they write to per-flush local
163+
// ConcurrentHashMap parameters (parallelDeferredOutHead / parallelDeferredInHead)
164+
// which are merged into these fields via putAll() AFTER async.waitCompletion(),
165+
// establishing a happens-before barrier.
166+
// 3. batchUpdateVertexHeadChunks() reads these single-threaded at flush end.
167+
// LongObjectHashMap (zero-boxing, ~5x less memory than ConcurrentHashMap<Long, RID>)
168+
// is safe because no concurrent writes ever hit these fields.
169+
private final LongObjectHashMap<RID> deferredOutHead = new LongObjectHashMap<>();
170+
private final LongObjectHashMap<RID> deferredInHead = new LongObjectHashMap<>();
158171

159172
// --- Known-new vertices: created by createVertices(), guaranteed no existing segments ---
160-
// Allows skipping vertex record loads when creating first segment
161-
private final Set<Long> knownNewVertexKeys = ConcurrentHashMap.newKeySet();
173+
// Allows skipping vertex record loads when creating first segment.
174+
//
175+
// Lifecycle (matters for the choice of non-concurrent collection):
176+
// 1. add() is only called from createVertices(), single-threaded, terminated by database.commit()
177+
// which establishes happens-before with subsequent reads.
178+
// 2. contains() is called during edge-connect phases that may run with parallelFlush=true,
179+
// but at that point the set is fully populated and never mutated (read-only).
180+
// 3. clear() runs at end of flush, single-threaded after another database.commit().
181+
// Plain LongHashSet (zero-boxing, ~7x less memory than ConcurrentHashMap.newKeySet()) is safe
182+
// because the only concurrent access is read-only after a publishing barrier.
183+
private final LongHashSet knownNewVertexKeys = new LongHashSet();
162184

163185
// --- Statistics ---
164186
private long totalVerticesCreated;
@@ -760,10 +782,8 @@ private void batchUpdateVertexHeadChunks() {
760782
// open-addressing) instead of HashSet<Long> avoids ~70 bytes/entry of overhead
761783
// and Long boxing on every addAll, which matters during 100K+ entry bulk loads.
762784
final LongHashSet allKeys = new LongHashSet(deferredOutHead.size() + deferredInHead.size());
763-
for (final Long k : deferredOutHead.keySet())
764-
allKeys.add(k);
765-
for (final Long k : deferredInHead.keySet())
766-
allKeys.add(k);
785+
deferredOutHead.forEach((k, v) -> allKeys.add(k));
786+
deferredInHead.forEach((k, v) -> allKeys.add(k));
767787

768788
// Sort by vertex key for page locality
769789
final long[] sortedKeys = allKeys.toArray();
@@ -1064,8 +1084,9 @@ private void connectIncomingEdgesParallel(final int[] inSortIndex, final int max
10641084
async.setTransactionUseWAL(savedAsyncWAL);
10651085
async.setTransactionSync(savedAsyncSync);
10661086

1067-
// Merge deferred head pointers back
1068-
deferredInHead.putAll(parallelDeferredInHead);
1087+
// Merge deferred head pointers back into the (non-concurrent) shared maps.
1088+
// putAll is unavailable on LongObjectHashMap by design; iterate and put.
1089+
parallelDeferredInHead.forEach(deferredInHead::put);
10691090
inChunkRIDCache.putAll(parallelInChunkCache);
10701091

10711092
final Throwable t = error.get();
@@ -1732,8 +1753,9 @@ private void connectOutgoingEdgesParallel(final RID[] edgeRIDs, final int maxBuc
17321753
async.setTransactionUseWAL(savedAsyncWAL);
17331754
async.setTransactionSync(savedAsyncSync);
17341755

1735-
// Merge deferred head pointers back
1736-
deferredOutHead.putAll(parallelDeferredOutHead);
1756+
// Merge deferred head pointers back into the (non-concurrent) shared maps.
1757+
// putAll is unavailable on LongObjectHashMap by design; iterate and put.
1758+
parallelDeferredOutHead.forEach(deferredOutHead::put);
17371759
outChunkRIDCache.putAll(parallelOutChunkCache);
17381760

17391761
final Throwable t = error.get();

0 commit comments

Comments
 (0)