Skip to content

Commit 61dab6a

Browse files
jdconradjdconrad
committed
Remove the middle list.
1 parent 8d6f055 commit 61dab6a

File tree

2 files changed

+34
-95
lines changed

2 files changed

+34
-95
lines changed

‎x-pack/plugin/blob-cache/src/main/java/org/elasticsearch/blobcache/shared/SharedBlobCacheService.java‎

Lines changed: 15 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -627,7 +627,7 @@ public void maybeFetchRange(
627627
final Executor fetchExecutor,
628628
final ActionListener<Boolean> listener
629629
) {
630-
if (freeRegionCount() < 1 && maybeEvictLeastUsed() == false) {
630+
if (freeRegionCount() < 1 && maybeEvict() == false) {
631631
// no free page available and no old enough unused region to be evicted
632632
logger.info("No free regions, skipping loading region [{}]", region);
633633
listener.onResponse(false);
@@ -1307,15 +1307,7 @@ void touch() {
13071307
}
13081308

13091309
private final ConcurrentHashMap<RegionKey<KeyType>, LRUCacheEntry> keyMapping = new ConcurrentHashMap<>();
1310-
private final int maxSize;
13111310
private final LRUCacheList front = new LRUCacheList();
1312-
private final LRUCacheList middle = new LRUCacheList();
1313-
1314-
LRUCache() {
1315-
// we split the front and middle lists in half, but require a minimum of 2
1316-
// for front to avoid strange head/tail semantics
1317-
maxSize = Math.min(2, numRegions / 2);
1318-
}
13191311

13201312
@Override
13211313
public void close() {
@@ -1417,30 +1409,12 @@ private void assignToSlot(LRUCacheEntry entry, SharedBytes.IO freeSlot) {
14171409
keyMapping.remove(entry.chunk.regionKey, entry);
14181410
throwAlreadyClosed("evicted during free region allocation");
14191411
}
1420-
pushEntryToMiddle(entry);
1412+
pushEntryToFront(entry);
14211413
// assign io only when chunk is ready for use. Under lock to avoid concurrent tryEvict.
14221414
entry.chunk.io = freeSlot;
14231415
}
14241416
}
14251417

1426-
private void pushEntryToMiddle(final LRUCacheEntry entry) {
1427-
assert Thread.holdsLock(SharedBlobCacheService.this);
1428-
assert invariant(entry, false);
1429-
assert entry.prev == null;
1430-
assert entry.next == null;
1431-
if (middle.head == null) {
1432-
middle.head = entry;
1433-
middle.tail = entry;
1434-
} else {
1435-
entry.next = middle.head;
1436-
middle.head.prev = entry;
1437-
middle.head = entry;
1438-
}
1439-
entry.list = middle;
1440-
++middle.size;
1441-
assert invariant(entry, true);
1442-
}
1443-
14441418
private void pushEntryToFront(final LRUCacheEntry entry) {
14451419
assert Thread.holdsLock(SharedBlobCacheService.this);
14461420
assert invariant(entry, false);
@@ -1457,29 +1431,22 @@ private void pushEntryToFront(final LRUCacheEntry entry) {
14571431
entry.list = front;
14581432
++front.size;
14591433
assert invariant(entry, true);
1460-
if (front.size > maxSize) {
1461-
LRUCacheEntry move = front.tail;
1462-
unlink(move);
1463-
pushEntryToMiddle(move);
1464-
}
14651434
}
14661435

14671436
private synchronized boolean invariant(final LRUCacheEntry e, boolean present) {
14681437
boolean found = false;
1469-
for (LRUCacheList list : List.of(front, middle)) {
1470-
for (LRUCacheEntry entry = list.head; entry != null; entry = entry.next) {
1471-
assert entry.next == null || entry.next.prev == entry;
1472-
assert entry.prev == null || entry.prev.next == entry;
1473-
if (entry == e) {
1474-
found = true;
1475-
}
1438+
for (LRUCacheEntry entry = front.head; entry != null; entry = entry.next) {
1439+
assert entry.next == null || entry.next.prev == entry;
1440+
assert entry.prev == null || entry.prev.next == entry;
1441+
if (entry == e) {
1442+
found = true;
14761443
}
1477-
for (LRUCacheEntry entry = list.tail; entry != null; entry = entry.prev) {
1478-
assert entry.next == null || entry.next.prev == entry;
1479-
assert entry.prev == null || entry.prev.next == entry;
1480-
if (entry == e) {
1481-
found = true;
1482-
}
1444+
}
1445+
for (LRUCacheEntry entry = front.tail; entry != null; entry = entry.prev) {
1446+
assert entry.next == null || entry.next.prev == entry;
1447+
assert entry.prev == null || entry.prev.next == entry;
1448+
if (entry == e) {
1449+
found = true;
14831450
}
14841451
}
14851452
assert found == present;
@@ -1533,7 +1500,7 @@ private void unlink(final LRUCacheEntry entry) {
15331500
*/
15341501
private SharedBytes.IO maybeEvictAndTake() {
15351502
assert Thread.holdsLock(SharedBlobCacheService.this);
1536-
LRUCacheEntry entry = middle.tail;
1503+
LRUCacheEntry entry = front.tail;
15371504
if (entry != null) {
15381505
boolean evicted = entry.chunk.tryEvictNoDecRef();
15391506
if (evicted) {
@@ -1568,7 +1535,7 @@ private SharedBytes.IO maybeEvictAndTake() {
15681535
*/
15691536
public boolean maybeEvictLeastRecent() {
15701537
synchronized (SharedBlobCacheService.this) {
1571-
LRUCacheEntry entry = middle.tail;
1538+
LRUCacheEntry entry = front.tail;
15721539
if (entry != null) {
15731540
boolean evicted = entry.chunk.tryEvict();
15741541
if (evicted && entry.chunk.io != null) {

‎x-pack/plugin/blob-cache/src/test/java/org/elasticsearch/blobcache/shared/SharedBlobCacheServiceUsingLRUCacheTests.java‎

Lines changed: 19 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -344,20 +344,12 @@ public void testFetchFullCacheEntry() throws Exception {
344344
.build();
345345

346346
AtomicInteger bulkTaskCount = new AtomicInteger(0);
347-
ThreadPool threadPool = new TestThreadPool("test") {
347+
final var threadPool = new TestThreadPool("test");
348+
final var bulkExecutor = new StoppableExecutorServiceWrapper(threadPool.generic()) {
348349
@Override
349-
public ExecutorService executor(String name) {
350-
ExecutorService generic = super.executor(Names.GENERIC);
351-
if (Objects.equals(name, "bulk")) {
352-
return new StoppableExecutorServiceWrapper(generic) {
353-
@Override
354-
public void execute(Runnable command) {
355-
super.execute(command);
356-
bulkTaskCount.incrementAndGet();
357-
}
358-
};
359-
}
360-
return generic;
350+
public void execute(Runnable command) {
351+
super.execute(command);
352+
bulkTaskCount.incrementAndGet();
361353
}
362354
};
363355

@@ -368,7 +360,6 @@ public void execute(Runnable command) {
368360
settings,
369361
threadPool,
370362
ThreadPool.Names.GENERIC,
371-
"bulk",
372363
BlobCacheMetrics.NOOP
373364
)
374365
) {
@@ -381,7 +372,7 @@ public void execute(Runnable command) {
381372
cacheService.maybeFetchFullEntry(cacheKey, size, (channel, channelPos, relativePos, length, progressUpdater) -> {
382373
bytesRead.addAndGet(-length);
383374
progressUpdater.accept(length);
384-
}, future);
375+
}, bulkExecutor, future);
385376

386377
future.get(10, TimeUnit.SECONDS);
387378
assertEquals(0L, bytesRead.get());
@@ -394,7 +385,7 @@ public void execute(Runnable command) {
394385
assertEquals(2, cacheService.freeRegionCount());
395386
var configured = cacheService.maybeFetchFullEntry(cacheKey, size(500), (ch, chPos, relPos, len, update) -> {
396387
throw new AssertionError("Should never reach here");
397-
}, ActionListener.noop());
388+
}, bulkExecutor, ActionListener.noop());
398389
assertFalse(configured);
399390
assertEquals(2, cacheService.freeRegionCount());
400391
}
@@ -411,16 +402,8 @@ public void testFetchFullCacheEntryConcurrently() throws Exception {
411402
.put("path.home", createTempDir())
412403
.build();
413404

414-
ThreadPool threadPool = new TestThreadPool("test") {
415-
@Override
416-
public ExecutorService executor(String name) {
417-
ExecutorService generic = super.executor(Names.GENERIC);
418-
if (Objects.equals(name, "bulk")) {
419-
return new StoppableExecutorServiceWrapper(generic);
420-
}
421-
return generic;
422-
}
423-
};
405+
final var threadPool = new TestThreadPool("test");
406+
final var bulkExecutor = new StoppableExecutorServiceWrapper(threadPool.generic());
424407

425408
try (
426409
NodeEnvironment environment = new NodeEnvironment(settings, TestEnvironment.newEnvironment(settings));
@@ -429,7 +412,6 @@ public ExecutorService executor(String name) {
429412
settings,
430413
threadPool,
431414
ThreadPool.Names.GENERIC,
432-
"bulk",
433415
BlobCacheMetrics.NOOP
434416
)
435417
) {
@@ -446,6 +428,7 @@ public ExecutorService executor(String name) {
446428
cacheKey,
447429
size,
448430
(channel, channelPos, relativePos, length, progressUpdater) -> progressUpdater.accept(length),
431+
bulkExecutor,
449432
f
450433
)
451434
);
@@ -677,7 +660,6 @@ public void testMaybeEvictRecentUsed() throws Exception {
677660
settings,
678661
taskQueue.getThreadPool(),
679662
ThreadPool.Names.GENERIC,
680-
"bulk",
681663
BlobCacheMetrics.NOOP
682664
)
683665
) {
@@ -716,21 +698,13 @@ public void testMaybeFetchRegion() throws Exception {
716698
.put("path.home", createTempDir())
717699
.build();
718700

719-
AtomicInteger bulkTaskCount = new AtomicInteger(0);
720-
ThreadPool threadPool = new TestThreadPool("test") {
701+
final var bulkTaskCount = new AtomicInteger(0);
702+
final var threadPool = new TestThreadPool("test");
703+
final var bulkExecutor = new StoppableExecutorServiceWrapper(threadPool.generic()) {
721704
@Override
722-
public ExecutorService executor(String name) {
723-
ExecutorService generic = super.executor(Names.GENERIC);
724-
if (Objects.equals(name, "bulk")) {
725-
return new StoppableExecutorServiceWrapper(generic) {
726-
@Override
727-
public void execute(Runnable command) {
728-
super.execute(command);
729-
bulkTaskCount.incrementAndGet();
730-
}
731-
};
732-
}
733-
return generic;
705+
public void execute(Runnable command) {
706+
super.execute(command);
707+
bulkTaskCount.incrementAndGet();
734708
}
735709
};
736710
try (
@@ -740,7 +714,6 @@ public void execute(Runnable command) {
740714
settings,
741715
threadPool,
742716
ThreadPool.Names.GENERIC,
743-
"bulk",
744717
BlobCacheMetrics.NOOP
745718
)
746719
) {
@@ -754,7 +727,7 @@ public void execute(Runnable command) {
754727
cacheService.maybeFetchRegion(cacheKey, 0, blobLength, (channel, channelPos, relativePos, length, progressUpdater) -> {
755728
bytesRead.addAndGet(length);
756729
progressUpdater.accept(length);
757-
}, future);
730+
}, bulkExecutor, future);
758731

759732
var fetched = future.get(10, TimeUnit.SECONDS);
760733
assertThat("Region has been fetched", fetched, is(true));
@@ -782,6 +755,7 @@ public void execute(Runnable command) {
782755
bytesRead.addAndGet(length);
783756
progressUpdater.accept(length);
784757
},
758+
bulkExecutor,
785759
listener
786760
);
787761
}
@@ -802,7 +776,7 @@ public void execute(Runnable command) {
802776
cacheService.maybeFetchRegion(cacheKey, 0, blobLength, (channel, channelPos, relativePos, length, progressUpdater) -> {
803777
bytesRead.addAndGet(length);
804778
progressUpdater.accept(length);
805-
}, future);
779+
}, bulkExecutor, future);
806780

807781
var fetched = future.get(10, TimeUnit.SECONDS);
808782
assertThat("Region has been fetched", fetched, is(true));
@@ -831,7 +805,6 @@ public void testPopulate() throws Exception {
831805
settings,
832806
taskQueue.getThreadPool(),
833807
ThreadPool.Names.GENERIC,
834-
ThreadPool.Names.GENERIC,
835808
BlobCacheMetrics.NOOP
836809
)
837810
) {
@@ -924,7 +897,6 @@ public void testUseFullRegionSize() throws IOException {
924897
settings,
925898
taskQueue.getThreadPool(),
926899
ThreadPool.Names.GENERIC,
927-
ThreadPool.Names.GENERIC,
928900
BlobCacheMetrics.NOOP
929901
) {
930902
@Override

0 commit comments

Comments
 (0)