perf: Parallelize bundled app install at startup (#24109)

david-mackessy · web-flow · commit 3a3c42c52ed8 · 2026-06-09T16:23:56.000+02:00
* Parallelize file upload for bundled apps

* install largest apps first

* Add number to each new thread for clarity
diff --git a/dhis-2/dhis-api/src/main/java/org/hisp/dhis/appmanager/AppStorageService.java b/dhis-2/dhis-api/src/main/java/org/hisp/dhis/appmanager/AppStorageService.java
@@ -67,13 +67,19 @@ public interface AppStorageService {
    * @param file the zip file containing the app
    * @param appCache The app cache
    * @param bundledAppInfo bundled app info, can be null
+   * @param removeExisting whether to scan storage for and remove other already-installed apps that
+   *     share the new app's key (i.e. older versions in a different folder). Pass {@code false} to
+   *     skip this scan when it is known there can be no duplicates — e.g. installing bundled apps
+   *     into an empty store on first startup — since the scan is an expensive per-app rescan of all
+   *     installed apps.
    * @return The status of the installation
    */
   @Nonnull
   App installApp(
       @Nonnull File file,
       @Nonnull Cache<App> appCache,
-      @CheckForNull BundledAppInfo bundledAppInfo);
+      @CheckForNull BundledAppInfo bundledAppInfo,
+      boolean removeExisting);
 
   /**
    * Deletes the app from storage.
diff --git a/dhis-2/dhis-services/dhis-service-core/src/main/java/org/hisp/dhis/appmanager/BlobStoreAppStorageService.java b/dhis-2/dhis-services/dhis-service-core/src/main/java/org/hisp/dhis/appmanager/BlobStoreAppStorageService.java
@@ -32,12 +32,15 @@
 import static org.hisp.dhis.util.ZipFileUtils.getFilePath;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.UncheckedIOException;
 import java.net.URI;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Enumeration;
 import java.util.HashMap;
@@ -46,6 +49,12 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 import java.util.function.BiConsumer;
 import java.util.zip.ZipEntry;
 import java.util.zip.ZipFile;
@@ -85,11 +94,43 @@ public class BlobStoreAppStorageService implements AppStorageService {
   private static final String BUNDLED_APP_INFO_FILENAME = "bundled-app-info.json";
   public static final String MANIFEST_WEBAPP_FILENAME = "manifest.webapp";
 
+  /**
+   * Max concurrent file uploads to blob storage across <em>all</em> apps installing at once. Apps
+   * may install in parallel (see {@code DefaultAppManager}); this is a single shared pool so the
+   * total number of in-flight uploads (and buffered file payloads) stays bounded regardless of how
+   * many apps install concurrently.
+   *
+   * <p>Note this value also bounds transient memory: the S3 backend buffers each non-resetable zip
+   * entry stream fully in memory, so worst case is this many file payloads on the heap at once.
+   * Benchmarks against AWS S3 showed no install-time gain from 64 over 32, so 32 is used to halve
+   * the worst-case memory footprint.
+   */
+  private static final int APP_UPLOAD_PARALLELISM = 32;
+
   private final BlobStoreService blobStore;
   private final LocationManager locationManager;
   private final ObjectMapper jsonMapper;
   private final FileResourceContentStore fileResourceContentStore;
 
+  /**
+   * Shared, lazily-populated pool for concurrent file uploads. Core threads time out when idle so
+   * the pool holds no threads outside of (re)install activity, e.g. after startup.
+   */
+  private final ExecutorService uploadExecutor = newUploadExecutor();
+
+  private static ExecutorService newUploadExecutor() {
+    ThreadPoolExecutor executor =
+        new ThreadPoolExecutor(
+            APP_UPLOAD_PARALLELISM,
+            APP_UPLOAD_PARALLELISM,
+            60L,
+            TimeUnit.SECONDS,
+            new LinkedBlockingQueue<>(),
+            new ThreadFactoryBuilder().setNameFormat("app-upload-%d").setDaemon(true).build());
+    executor.allowCoreThreadTimeOut(true);
+    return executor;
+  }
+
   @Override
   @Nonnull
   public Map<String, Pair<App, BundledAppInfo>> discoverInstalledApps() {
@@ -224,7 +265,8 @@ private void validateAppAdditionalNamespacesAreWellDefined(App app) {
   public App installApp(
       @Nonnull File file,
       @Nonnull Cache<App> appCache,
-      @CheckForNull BundledAppInfo bundledAppInfo) {
+      @CheckForNull BundledAppInfo bundledAppInfo,
+      boolean removeExisting) {
     App app;
     AppFolderName folder;
     String topLevelFolder;
@@ -259,7 +301,9 @@ public App installApp(
       ZipFileUtils.validateZip(file, folder.path(), topLevelFolder);
       unzipFile(file, folder, topLevelFolder);
 
-      removeOtherAppsWithSameKey(app);
+      if (removeExisting) {
+        removeOtherAppsWithSameKey(app);
+      }
 
       app.setAppState(AppStatus.OK);
       logInstallSuccess(app, folder.path());
@@ -286,6 +330,8 @@ public App installApp(
   private void unzipFile(File file, AppFolderName folder, String topLevelFolder)
       throws IOException, ZipSlipException {
     try (ZipFile zipFile = new ZipFile(file)) {
+      List<ZipEntry> fileEntries = new ArrayList<>();
+      List<String> filePaths = new ArrayList<>();
       Enumeration<? extends ZipEntry> entries = zipFile.entries();
       while (entries.hasMoreElements()) {
         ZipEntry zipEntry = entries.nextElement();
@@ -298,11 +344,48 @@ private void unzipFile(File file, AppFolderName folder, String topLevelFolder)
           blobStore.createDirectory(BlobKeyPrefix.of(filePath));
           continue;
         }
-        try (InputStream zipInputStream = zipFile.getInputStream(zipEntry)) {
-          blobStore.putBlob(
-              BlobKey.of(filePath), zipInputStream, zipEntry.getSize(), null, null, null);
-        }
+        fileEntries.add(zipEntry);
+        filePaths.add(filePath);
+      }
+      uploadEntriesInParallel(zipFile, fileEntries, filePaths);
+    }
+  }
+
+  /**
+   * Uploads the given zip file entries to blob storage concurrently via the shared {@link
+   * #uploadExecutor}. App zips contain many small files and each upload is dominated by per-request
+   * latency (network round-trip on object stores), so uploading them in parallel rather than
+   * one-at-a-time substantially reduces install time. The pool is shared across concurrently
+   * installing apps so total in-flight uploads stay bounded; this call only waits on its own files.
+   */
+  private void uploadEntriesInParallel(
+      ZipFile zipFile, List<ZipEntry> fileEntries, List<String> filePaths) throws IOException {
+    if (fileEntries.isEmpty()) return;
+
+    try {
+      CompletableFuture<?>[] futures = new CompletableFuture<?>[fileEntries.size()];
+      for (int i = 0; i < fileEntries.size(); i++) {
+        ZipEntry zipEntry = fileEntries.get(i);
+        String filePath = filePaths.get(i);
+        futures[i] =
+            CompletableFuture.runAsync(
+                () -> uploadZipEntry(zipFile, zipEntry, filePath), uploadExecutor);
       }
+      CompletableFuture.allOf(futures).join();
+    } catch (CompletionException e) {
+      Throwable cause = e.getCause() != null ? e.getCause() : e;
+      if (cause instanceof UncheckedIOException uioe) throw uioe.getCause();
+      if (cause instanceof IOException ioe) throw ioe;
+      if (cause instanceof RuntimeException re) throw re;
+      throw new IOException("Failed to upload app files", cause);
+    }
+  }
+
+  private void uploadZipEntry(ZipFile zipFile, ZipEntry zipEntry, String filePath) {
+    try (InputStream zipInputStream = zipFile.getInputStream(zipEntry)) {
+      blobStore.putBlob(BlobKey.of(filePath), zipInputStream, zipEntry.getSize(), null, null, null);
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
     }
   }
 
diff --git a/dhis-2/dhis-services/dhis-service-core/src/main/java/org/hisp/dhis/appmanager/DefaultAppManager.java b/dhis-2/dhis-services/dhis-service-core/src/main/java/org/hisp/dhis/appmanager/DefaultAppManager.java
@@ -36,6 +36,7 @@
 import static org.hisp.dhis.datastore.DatastoreNamespaceProtection.ProtectionType.RESTRICTED;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import java.io.*;
 import java.io.InputStream;
 import java.net.URISyntaxException;
@@ -47,12 +48,16 @@
 import java.nio.file.StandardCopyOption;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Comparator;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.function.Predicate;
 import java.util.stream.Stream;
 import javax.annotation.CheckForNull;
@@ -99,6 +104,13 @@ public class DefaultAppManager implements AppManager {
   public static final String INVALID_FILTER_MSG = "Invalid filter: ";
   private static final Set<String> EXCLUSION_APPS = Set.of("Line Listing");
 
+  /**
+   * Max number of bundled apps installed concurrently on startup. File uploads across all of these
+   * share a single bounded pool in {@link BlobStoreAppStorageService}, so this only governs how
+   * many app zips are read/decompressed at once, not total upload concurrency.
+   */
+  private static final int APP_INSTALL_PARALLELISM = 16;
+
   @Autowired private UserService userService;
   @Autowired private ObjectMapper jsonMapper;
   @Autowired private LocaleManager localeManager;
@@ -154,49 +166,123 @@ public void reloadApps() {
     Map<String, Pair<App, BundledAppInfo>> installedApps =
         blobStoreAppStorageService.discoverInstalledApps();
 
-    installBundledApps(installedApps);
+    // When the store starts empty (typical first startup) there can be no pre-existing apps to
+    // replace, so the per-app duplicate-removal scan is skipped to avoid an expensive O(n^2)
+    // rescan of storage during bundled app installation.
+    boolean storeStartedEmpty = installedApps.isEmpty();
+    installBundledApps(installedApps, storeStartedEmpty);
     // Invalidate the previous app cache
     appCache.invalidateAll();
     // Cache all discovered apps
     installedApps.values().forEach(app -> cacheApp(app.getLeft()));
     log.info("Loaded {} apps.", installedApps.size());
   }
 
+  /**
+   * A bundled app that needs to be (re)installed, together with its zip resource and the zip's
+   * compressed size in bytes (computed once; used to order installs largest-first).
+   */
+  private record BundledAppInstall(
+      String appKey, BundledAppInfo bundledAppInfo, Resource zipFileResource, long zipSize) {}
+
   /**
    * Installs bundled apps, by looking in the classpath for app .zip files. If the bundled app is
    * already installed, it can be overwritten with a newer one if the Etag is different.
    *
+   * <p>Apps are installed concurrently: each app is large enough that its files already upload in
+   * parallel, but the apps themselves used to install one-after-another, so the many small apps
+   * waited behind the few large ones. Installing apps in parallel lets the small apps upload in the
+   * shadow of the large ones (each app reads its own zip, so decompression also parallelizes). The
+   * shared upload pool in {@link BlobStoreAppStorageService} keeps total in-flight uploads bounded.
+   *
    * @param installedApps the Map with all existing apps, we overwrite existing apps in this Map
    *     with new ones.
+   * @param storeStartedEmpty whether the app store was empty before this install run; when {@code
+   *     true} the per-app duplicate-removal scan is skipped as there can be nothing to replace.
    */
-  private void installBundledApps(@Nonnull Map<String, Pair<App, BundledAppInfo>> installedApps) {
+  private void installBundledApps(
+      @Nonnull Map<String, Pair<App, BundledAppInfo>> installedApps, boolean storeStartedEmpty) {
+    boolean removeExisting = !storeStartedEmpty;
+
+    // First collect which bundled apps actually need (re)installing, without touching storage. An
+    // app is (re)installed if it isn't installed yet, or if its installed Etag differs from the
+    // bundled one. Apps that are already up to date keep their existing map entry untouched.
+    List<BundledAppInstall> appsToInstall = new ArrayList<>();
     bundledAppManager.installBundledApps(
         (app, bundledAppInfo, zipFileResource) -> {
-          String appKey = app.getKey();
-          installedApps.computeIfAbsent(
-              appKey,
-              x ->
-                  Pair.of(
-                      installBundledAppResource(zipFileResource, bundledAppInfo), bundledAppInfo));
-
-          // If the bundled app is already installed and the Etag is different, overwrite the
-          // existing.
-          if (installedApps.containsKey(appKey)) {
-            BundledAppInfo installedAppInfo = installedApps.get(appKey).getRight();
-            if (installedAppInfo != null
-                && installedAppInfo.getEtag() != null
-                && !installedAppInfo.getEtag().equals(bundledAppInfo.getEtag())) {
-              installedApps.put(
-                  appKey,
-                  Pair.of(
-                      installBundledAppResource(zipFileResource, bundledAppInfo), bundledAppInfo));
-
+          Pair<App, BundledAppInfo> existing = installedApps.get(app.getKey());
+          if (needsInstall(existing, bundledAppInfo)) {
+            if (existing != null) {
               log.info(
-                  "A bundled app with a different Etag was installed and replaced the existing one. App name: '{}'",
-                  installedAppInfo.getName());
+                  "Bundled app '{}' has a different Etag than the installed one and will replace it",
+                  bundledAppInfo.getName());
             }
+            appsToInstall.add(
+                new BundledAppInstall(
+                    app.getKey(), bundledAppInfo, zipFileResource, zipSize(zipFileResource)));
           }
         });
+
+    if (appsToInstall.isEmpty()) return;
+
+    // Install the largest apps first. A few apps (e.g. Maintenance) are far bigger than the rest,
+    // if they start last they end up running alone at the tail with no smaller apps left to overlap
+    // them. Starting them first lets the many small apps install in their shadow, so total time
+    // approaches the largest app's runtime rather than largest + the rest.
+    appsToInstall.sort(Comparator.comparingLong(BundledAppInstall::zipSize).reversed());
+
+    // Install in parallel, then merge results into the map on this thread (single-writer).
+    ExecutorService appExecutor =
+        Executors.newFixedThreadPool(
+            Math.min(APP_INSTALL_PARALLELISM, appsToInstall.size()),
+            new ThreadFactoryBuilder().setNameFormat("app-install-%d").setDaemon(true).build());
+    try {
+      List<CompletableFuture<Map.Entry<String, Pair<App, BundledAppInfo>>>> futures =
+          new ArrayList<>(appsToInstall.size());
+      for (BundledAppInstall item : appsToInstall) {
+        futures.add(
+            CompletableFuture.supplyAsync(
+                () ->
+                    Map.entry(
+                        item.appKey(),
+                        Pair.of(
+                            installBundledAppResource(
+                                item.zipFileResource(), item.bundledAppInfo(), removeExisting),
+                            item.bundledAppInfo())),
+                appExecutor));
+      }
+      CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new)).join();
+      for (CompletableFuture<Map.Entry<String, Pair<App, BundledAppInfo>>> future : futures) {
+        Map.Entry<String, Pair<App, BundledAppInfo>> result = future.join();
+        installedApps.put(result.getKey(), result.getValue());
+      }
+    } finally {
+      appExecutor.shutdown();
+    }
+  }
+
+  /**
+   * Whether a bundled app needs to be (re)installed: {@code true} when it is not yet installed, or
+   * when its installed bundled-app Etag differs from the one being offered.
+   */
+  private static boolean needsInstall(
+      @CheckForNull Pair<App, BundledAppInfo> existing, @Nonnull BundledAppInfo bundledAppInfo) {
+    if (existing == null) {
+      return true;
+    }
+    BundledAppInfo installedAppInfo = existing.getRight();
+    return installedAppInfo != null
+        && installedAppInfo.getEtag() != null
+        && !installedAppInfo.getEtag().equals(bundledAppInfo.getEtag());
+  }
+
+  /** Compressed size of the app's zip in bytes, or {@code 0} when it cannot be determined. */
+  private static long zipSize(Resource zipFileResource) {
+    try {
+      return zipFileResource.contentLength();
+    } catch (IOException e) {
+      return 0L;
+    }
   }
 
   private void cacheApp(@Nonnull App app) {
@@ -434,12 +520,14 @@ public App getApp(@Nonnull String key, @Nonnull String contextPath) {
   @Override
   @Nonnull
   public App installApp(@Nonnull File file) {
-    return installAppZipFile(file, null);
+    // User-initiated installs may be upgrading an existing app, so always remove duplicates.
+    return installAppZipFile(file, null, true);
   }
 
   @Nonnull
-  private App installAppZipFile(@Nonnull File file, @CheckForNull BundledAppInfo bundledAppInfo) {
-    App app = blobStoreAppStorageService.installApp(file, appCache, bundledAppInfo);
+  private App installAppZipFile(
+      @Nonnull File file, @CheckForNull BundledAppInfo bundledAppInfo, boolean removeExisting) {
+    App app = blobStoreAppStorageService.installApp(file, appCache, bundledAppInfo, removeExisting);
     log.debug(
         String.format(
             "Installed App with AppHub ID %s (status: %s)", app.getAppHubId(), app.getAppState()));
@@ -449,12 +537,12 @@ private App installAppZipFile(@Nonnull File file, @CheckForNull BundledAppInfo b
 
   @Nonnull
   public App installBundledAppResource(
-      @Nonnull Resource resource, @Nonnull BundledAppInfo bundledAppInfo) {
+      @Nonnull Resource resource, @Nonnull BundledAppInfo bundledAppInfo, boolean removeExisting) {
     try {
       Path tempFile = Files.createTempFile("tmp-bundled-app-", CodeGenerator.generateUid());
       Files.copy(resource.getInputStream(), tempFile, StandardCopyOption.REPLACE_EXISTING);
       try {
-        return installAppZipFile(tempFile.toFile(), bundledAppInfo);
+        return installAppZipFile(tempFile.toFile(), bundledAppInfo, removeExisting);
       } finally {
         Files.deleteIfExists(tempFile);
       }
