diff --git a/dotCMS/src/main/java/com/dotcms/browser/BrowserAPIImpl.java b/dotCMS/src/main/java/com/dotcms/browser/BrowserAPIImpl.java index cec5b06df16..2a71d0f19aa 100644 --- a/dotCMS/src/main/java/com/dotcms/browser/BrowserAPIImpl.java +++ b/dotCMS/src/main/java/com/dotcms/browser/BrowserAPIImpl.java @@ -93,7 +93,9 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.function.Function; +import java.util.function.Function; import java.util.stream.Collectors; +import java.util.stream.Gatherers; import java.util.stream.Stream; import static com.dotcms.content.elasticsearch.business.ESMappingAPIImpl.INCLUDE_DOTRAW_METADATA_FIELDS; @@ -1030,6 +1032,43 @@ private int calculateContentletChunkSize(int totalInodes) { return calculatedSize; } + /** + * How many chunks may be in flight at once. Bounded by what the database can serve + * concurrently, not by the CPU: the loaders block on a socket, so the ceiling that matters is + * the connection pool. + */ + private static final int MAX_CONCURRENT_CHUNKS = + Config.getIntProperty("browser.contentlet.load.max.concurrent.chunks", 8); + + /** + * Splits {@code items} into chunks of {@code chunkSize}, hands each chunk to {@code loader} + * with at most {@code concurrency} chunks in flight, and flattens the results. + * + *

Two gatherers, one per concern: {@code windowFixed} cuts (emitting the short final chunk + * itself) and {@code mapConcurrent} runs the loaders on virtual threads while preserving + * input order, which is what a {@code parallelStream()} does not promise.

+ * + *

Concurrency is bounded by the connection pool, not by the CPU. The loader here goes + * to PostgreSQL, so more in-flight chunks than the pool has connections only moves the queue + * from this method into Hikari. That is why the limit is a parameter and not + * {@code availableProcessors()}.

+ * + *

What this does not give back: {@code mapConcurrent} has no timeout of any kind. + * The shape it replaces carried {@code orTimeout(90s)} per chunk and a 180s ceiling on the + * whole set. Neither survives here. Re-adding them means either putting a timeout inside the + * loader or reaching for {@code StructuredTaskScope}, which is still preview in Java 25 — so + * this is a deliberate, documented loss, not an oversight.

+ */ + static List inParallelChunks(final List items, final int chunkSize, + final int concurrency, final Function, List> loader) { + + return items.stream() + .gather(Gatherers.windowFixed(chunkSize)) + .gather(Gatherers.mapConcurrent(concurrency, loader::apply)) + .flatMap(List::stream) + .toList(); + } + /** * Loads contentlets in a single request (for small sets). */ @@ -1048,81 +1087,37 @@ private List loadContentletsSingle(List inodes, long startTi /** * Loads contentlets in parallel chunks for better performance and reliability. + * + *

See {@link #inParallelChunks(List, int, int, Function)} for what the two gatherers replace + * and for the per-chunk and global timeouts that this shape no longer carries.

*/ - private List loadContentletsParallel(Set inodes, int chunkSize, long startTime) { - final List inodesList = new ArrayList<>(inodes); - final int totalInodes = inodesList.size(); - - // Create chunks for parallel processing - final List> chunks = createChunks(inodesList, chunkSize); - final int chunkCount = chunks.size(); - Logger.debug(this, String.format("Loading contentlets in parallel: %d inodes in %d chunks (chunk size: %d)", - totalInodes, chunkCount, chunkSize)); + private List loadContentletsParallel(final Set inodes, final int chunkSize, + final long startTime) { - // Process chunks in parallel using CompletableFuture - final DotSubmitter submitter = DotConcurrentFactory.getInstance().getSubmitter(); - final CompletableFuture>[] futures = new CompletableFuture[chunkCount]; + final List inodesList = new ArrayList<>(inodes); final ContentletAPI contentletAPI = APILocator.getContentletAPI(); - for (int i = 0; i < chunkCount; i++) { - final List chunk = chunks.get(i); - final int chunkIndex = i + 1; - - futures[i] = CompletableFuture - .supplyAsync(() -> { - final long chunkStartTime = System.currentTimeMillis(); - Logger.debug(BrowserAPIImpl.this, String.format("Loading contentlet chunk %d/%d: %d inodes", - chunkIndex, chunkCount, chunk.size())); + final List allContentlets = inParallelChunks(inodesList, chunkSize, + MAX_CONCURRENT_CHUNKS, + chunk -> { + final long chunkStart = System.currentTimeMillis(); try { - final List chunkContentlets = contentletAPI.findContentlets(chunk); - final long chunkDuration = System.currentTimeMillis() - chunkStartTime; - Logger.debug(BrowserAPIImpl.this, String.format( - "Contentlet chunk %d/%d completed: %d inodes → %d contentlets in %d ms", - chunkIndex, chunkCount, chunk.size(), chunkContentlets.size(), chunkDuration)); - return chunkContentlets; - } catch (Exception e) { - Logger.error(BrowserAPIImpl.this, String.format("Contentlet chunk %d failed: %s", - chunkIndex, e.getMessage()), e); - return new ArrayList(); + final List loaded = contentletAPI.findContentlets(chunk); + Logger.debug(this, String.format( + "Contentlet chunk completed: %d inodes -> %d contentlets in %d ms", + chunk.size(), loaded.size(), System.currentTimeMillis() - chunkStart)); + return loaded; + } catch (final Exception e) { + // Same contract as before: one bad chunk must not fail the whole load. + Logger.error(this, String.format("Contentlet chunk of %d inodes failed: %s", + chunk.size(), e.getMessage()), e); + return List.of(); } - }, submitter) - .orTimeout(90, TimeUnit.SECONDS) // Longer timeout for DB operations - .exceptionally(throwable -> { - Logger.error(BrowserAPIImpl.this, String.format("Contentlet chunk %d timed out or failed: %s", - chunkIndex, throwable.getMessage()), throwable); - return new ArrayList<>(); }); - } - // Collect results from all chunks - final List allContentlets = new ArrayList<>(); - try { - CompletableFuture allFutures = CompletableFuture.allOf(futures); - allFutures.get(180, TimeUnit.SECONDS); // Global timeout for all chunks - - for (CompletableFuture> future : futures) { - try { - List chunkContentlets = future.get(); - allContentlets.addAll(chunkContentlets); - } catch (Exception e) { - Logger.warn(this, "Failed to get result from contentlet chunk future: " + e.getMessage()); - Thread.currentThread().interrupt(); - } - } - - final long totalDuration = System.currentTimeMillis() - startTime; - Logger.debug(this, String.format( - "Parallel contentlet loading completed: %d inodes in %d chunks → %d contentlets in %d ms", - totalInodes, chunkCount, allContentlets.size(), totalDuration)); - - } catch (InterruptedException e) { - Logger.error(this, "Parallel contentlet loading interrupted: " + e.getMessage(), e); - Thread.currentThread().interrupt(); - } catch (ExecutionException e) { - Logger.error(this, "Parallel contentlet loading execution error: " + e.getMessage(), e); - } catch (TimeoutException e) { - Logger.error(this, "Parallel contentlet loading timed out: " + e.getMessage(), e); - } + Logger.debug(this, String.format( + "Parallel contentlet loading completed: %d inodes -> %d contentlets in %d ms", + inodesList.size(), allContentlets.size(), System.currentTimeMillis() - startTime)); return allContentlets; } @@ -1137,51 +1132,25 @@ private List loadContentletsParallel(Set inodes, int chunkSi private List> hydrateContentletsInParallel(final List contentlets, final BrowserQuery browserQuery, final Role[] roles) { - final List> resultList = new ArrayList<>(); - final int totalContentlets = contentlets.size(); - final int chunkSize = Math.max(1, Math.min(10, totalContentlets / 4)); - final List> chunks = createChunks(contentlets, chunkSize); - - final List>>> futures = chunks.stream() - .map(chunk -> CompletableFuture.supplyAsync(() -> { - final List> chunkResults = new ArrayList<>(chunk.size()); - for (final Contentlet contentlet : chunk) { - try { - final Map contentMap = hydrate(browserQuery, contentlet, roles); - chunkResults.add(contentMap); - } catch (DotDataException | DotSecurityException e) { - Logger.error(this, "Error hydrating contentlet " + contentlet.getInode() + ": " + e.getMessage(), e); - throw new DotRuntimeException("Failed to hydrate contentlet: " + contentlet.getInode(), e); - } - } - return chunkResults; - }, DotConcurrentFactory.getInstance().getSubmitter())) - .collect(Collectors.toList()); - // Collect results maintaining order - for (final CompletableFuture>> future : futures) { - try { - resultList.addAll(future.get(30, TimeUnit.SECONDS)); - } catch (InterruptedException | ExecutionException | TimeoutException e) { - Logger.error(this, "Error in parallel hydration: " + e.getMessage(), e); - if (e instanceof InterruptedException) { - Thread.currentThread().interrupt(); - } - throw new DotRuntimeException("Failed to hydrate contentlets in parallel", e); - } - } - return resultList; - } - - /** - * Creates chunks from a list with the specified chunk size. - * - * @param list The list to split into chunks - * @param chunkSize The size of each chunk - * @return List of chunks, each containing at most chunkSize elements - */ - private List> createChunks(List list, int chunkSize) { - return Lists.partition(list, chunkSize); + final int chunkSize = Math.max(1, Math.min(10, contentlets.size() / 4)); + + // The comment this replaces read "Collect results maintaining order", and the ordering was + // a property of iterating the futures in creation order. mapConcurrent promises it. + return inParallelChunks(contentlets, chunkSize, MAX_CONCURRENT_CHUNKS, + chunk -> chunk.stream() + .map(contentlet -> { + try { + return hydrate(browserQuery, contentlet, roles); + } catch (final DotDataException | DotSecurityException e) { + // Same contract as before: a failed hydration fails the request. + Logger.error(this, "Error hydrating contentlet " + + contentlet.getInode() + ": " + e.getMessage(), e); + throw new DotRuntimeException( + "Failed to hydrate contentlet: " + contentlet.getInode(), e); + } + }) + .collect(Collectors.toList())); } /** diff --git a/dotCMS/src/test/java/com/dotcms/browser/BrowserAPIImplParallelChunksTest.java b/dotCMS/src/test/java/com/dotcms/browser/BrowserAPIImplParallelChunksTest.java new file mode 100644 index 00000000000..a1bcce0e6c5 --- /dev/null +++ b/dotCMS/src/test/java/com/dotcms/browser/BrowserAPIImplParallelChunksTest.java @@ -0,0 +1,150 @@ +package com.dotcms.browser; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.IntStream; +import org.junit.Test; + +/** + * Covers {@link BrowserAPIImpl#inParallelChunks(List, int, int, java.util.function.Function)} — the + * two-gatherer pipeline that replaced a hand-built {@code CompletableFuture} array in + * {@code loadContentletsParallel} and a future list in {@code hydrateContentletsInParallel}. + * + *

The old shape produced ordered results as a side effect of iterating the futures in creation + * order, and one of the two sites said so in a comment. These tests make that a checked property + * rather than a remark, and pin the concurrency bound, which is the parameter that matters when the + * loader is holding a database connection.

+ */ +public class BrowserAPIImplParallelChunksTest { + + /** + * Method to test: {@link BrowserAPIImpl#inParallelChunks(List, int, int, java.util.function.Function)} + * Given scenario: a chunk size that does not divide the input evenly. + * Expected result: every chunk is seen, the last one short, and nothing is dropped. + */ + @Test + public void test_inParallelChunks_chunksCoverEveryElement_lastOneShort() { + + final List chunkSizes = Collections.synchronizedList(new ArrayList<>()); + + final List out = BrowserAPIImpl.inParallelChunks( + IntStream.rangeClosed(1, 25).boxed().toList(), 10, 4, + chunk -> { + chunkSizes.add(chunk.size()); + return chunk; + }); + + assertEquals(25, out.size()); + assertEquals("three chunks: 10, 10 and the short tail", 3, chunkSizes.size()); + assertEquals(5, (int) chunkSizes.stream().sorted().findFirst().orElseThrow()); + } + + /** + * Method to test: {@link BrowserAPIImpl#inParallelChunks(List, int, int, java.util.function.Function)} + * Given scenario: loaders that finish in a deliberately reversed order — the first chunk sleeps + * longest, so a shape that returned results as they completed would come back scrambled. + * Expected result: input order regardless. This is the property the old code got from iterating + * the futures in creation order, and the one {@code parallelStream()} does not promise. + */ + @Test + public void test_inParallelChunks_preservesInputOrder_whateverTheCompletionOrder() { + + final AtomicInteger seen = new AtomicInteger(); + + final List out = BrowserAPIImpl.inParallelChunks( + IntStream.rangeClosed(1, 20).boxed().toList(), 5, 4, + chunk -> { + // The earlier the chunk, the longer it takes to come back. + final int order = seen.getAndIncrement(); + try { + Thread.sleep(60L - (order * 15L)); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + } + return chunk; + }); + + assertEquals(IntStream.rangeClosed(1, 20).boxed().toList(), out); + } + + /** + * Method to test: {@link BrowserAPIImpl#inParallelChunks(List, int, int, java.util.function.Function)} + * Given scenario: more chunks than the permitted concurrency. + * Expected result: never more than that many loaders in flight at once. With a loader that + * holds a database connection, this bound is the whole point of the parameter. + */ + @Test + public void test_inParallelChunks_neverExceedsTheConcurrencyBound() { + + final int limit = 3; + final AtomicInteger inFlight = new AtomicInteger(); + final AtomicInteger peak = new AtomicInteger(); + + BrowserAPIImpl.inParallelChunks( + IntStream.rangeClosed(1, 60).boxed().toList(), 2, limit, + chunk -> { + peak.accumulateAndGet(inFlight.incrementAndGet(), Math::max); + try { + Thread.sleep(20); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + inFlight.decrementAndGet(); + } + return chunk; + }); + + assertTrue("peak " + peak.get() + " must not exceed " + limit, peak.get() <= limit); + assertTrue("and the work must actually overlap", peak.get() > 1); + } + + /** + * Method to test: {@link BrowserAPIImpl#inParallelChunks(List, int, int, java.util.function.Function)} + * Given scenario: any loader at all. + * Expected result: it runs on a virtual thread, not on a platform pool thread. The loaders here + * block on a socket, which is the case virtual threads are for. + */ + @Test + public void test_inParallelChunks_runsOnVirtualThreads() { + + final Set virtual = ConcurrentHashMap.newKeySet(); + + BrowserAPIImpl.inParallelChunks( + IntStream.rangeClosed(1, 12).boxed().toList(), 3, 4, + chunk -> { + virtual.add(Thread.currentThread().isVirtual()); + return chunk; + }); + + assertEquals("every loader ran on a virtual thread", Set.of(Boolean.TRUE), virtual); + } + + /** + * Method to test: {@link BrowserAPIImpl#inParallelChunks(List, int, int, java.util.function.Function)} + * Given scenario: an empty input. + * Expected result: no loader is invoked and the result is empty — no empty chunk reaches the + * database. + */ + @Test + public void test_inParallelChunks_empty_neverCallsTheLoader() { + + final AtomicInteger calls = new AtomicInteger(); + + final List out = BrowserAPIImpl.inParallelChunks( + List.of(), 10, 4, + chunk -> { + calls.incrementAndGet(); + return chunk; + }); + + assertTrue(out.isEmpty()); + assertEquals(0, calls.get()); + } +}