Parallelize resource verification.

Ad-hoc testing with ~320MB of resources reduces verification time from
~2400ms to ~800ms on MBP with four-core i7.

src/main/java/com/threerings/getdown/data/Application.java

@@ -20,6 +20,7 @@ import java.nio.channels.OverlappingFileLockException;
 import java.security.*;
 import java.security.cert.Certificate;
 import java.util.*;
+import java.util.concurrent.*;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -1282,64 +1283,119 @@ public class Application
      */
     public void verifyResources (
         ProgressObserver obs, int[] alreadyValid, Set<Resource> unpacked,
-        List<Resource> toInstall, List<Resource> toDownload)
+        Set<Resource> toInstall, Set<Resource> toDownload)
         throws InterruptedException
     {
-        List<Resource> rsrcs = getAllActiveResources();
+        // resources are verified on backgroudn threads supplied by the thread pool, and progress
+        // is reported by posting runnable actions to the actions queue which is processed by the
+        // main (UI) thread
+        Executor exec = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
+        final BlockingQueue<Runnable> actions = new LinkedBlockingQueue<Runnable>();
+        final int[] completed = new int[1];
+
+        long start = System.currentTimeMillis();
 
         // obtain the sizes of the resources to validate
+        List<Resource> rsrcs = getAllActiveResources();
         long[] sizes = new long[rsrcs.size()];
+        long totalSize = 0;
         for (int ii = 0; ii < sizes.length; ii++) {
-            sizes[ii] = rsrcs.get(ii).getLocal().length();
+            totalSize += sizes[ii] = rsrcs.get(ii).getLocal().length();
         }
+        final ProgressObserver fobs = obs;
+        // as long as we forward aggregated progress updates to the UI thread, having multiple
+        // threads update a progress aggregator is "mostly" thread-safe
+        final ProgressAggregator pagg = new ProgressAggregator(new ProgressObserver() {
+            public void progress (final int percent) {
+                actions.add(new Runnable() {
+                    public void run () {
+                        fobs.progress(percent);
+                    }
+                });
+            }
+        }, sizes);
+
+        final boolean noUnpack = SysProps.noUnpack();
+        final int[] fAlreadyValid = alreadyValid;
+        final Set<Resource> toInstallAsync = new ConcurrentSkipListSet<Resource>(toInstall);
+        final Set<Resource> toDownloadAsync = new ConcurrentSkipListSet<Resource>();
+        final Set<Resource> unpackedAsync = new ConcurrentSkipListSet<Resource>();
 
-        ProgressAggregator pagg = new ProgressAggregator(obs, sizes);
-        boolean noUnpack = SysProps.noUnpack();
         for (int ii = 0; ii < sizes.length; ii++) {
-            Resource rsrc = rsrcs.get(ii);
+            final Resource rsrc = rsrcs.get(ii);
             if (Thread.interrupted()) {
                 throw new InterruptedException("m.applet_stopped");
             }
-
+            final int index = ii;
-            ProgressObserver robs = pagg.startElement(ii);
+            exec.execute(new Runnable() {
-            if (rsrc.isMarkedValid()) {
+                public void run () {
-                if (alreadyValid != null) {
+                    verifyResource(rsrc, pagg.startElement(index), fAlreadyValid, noUnpack,
-                    alreadyValid[0]++;
+                                   unpackedAsync, toInstallAsync, toDownloadAsync);
+                    actions.add(new Runnable() {
+                        public void run () {
+                            completed[0] += 1;
+                        }
+                    });
                 }
-                robs.progress(100);
+            });
-                continue;
-            }
-
-            try {
-                if (_digest.validateResource(rsrc, robs)) {
-                    // if the resource has a _new file, add it to to-install list
-                    if (!toInstall.contains(rsrc) && rsrc.getLocalNew().exists()) {
-                        toInstall.add(rsrc);
-                        continue;
-                    }
-                    // unpack this resource if appropriate
-                    if (noUnpack || !rsrc.shouldUnpack()) {
-                        // finally note that this resource is kosher
-                        rsrc.markAsValid();
-                        continue;
-                    }
-                    if (rsrc.unpack()) {
-                        unpacked.add(rsrc);
-                        rsrc.markAsValid();
-                        continue;
-                    }
-                    log.info("Failure unpacking resource", "rsrc", rsrc);
-                }
-
-            } catch (Exception e) {
-                log.info("Failure validating resource. Requesting redownload...",
-                    "rsrc", rsrc, "error", e);
-
-            } finally {
-                robs.progress(100);
-            }
-            toDownload.add(rsrc);
         }
+
+        while (completed[0] < rsrcs.size()) {
+            // we should be getting progress completion updates WAY more often than one every
+            // minute, so if things freeze up for 60 seconds, abandon ship
+            Runnable action = actions.poll(60, TimeUnit.SECONDS);
+            action.run();
+        }
+
+        toInstall.addAll(toInstallAsync);
+        toDownload.addAll(toDownloadAsync);
+        unpacked.addAll(unpackedAsync);
+
+        long complete = System.currentTimeMillis();
+        log.info("Verified resources", "count", rsrcs.size(), "size", (totalSize/1024) + "k",
+                 "duration", (complete-start) + "ms");
+    }
+
+    private void verifyResource (Resource rsrc, ProgressObserver obs, int[] alreadyValid,
+                                 boolean noUnpack, Set<Resource> unpacked,
+                                 Set<Resource> toInstall, Set<Resource> toDownload) {
+        if (rsrc.isMarkedValid()) {
+            if (alreadyValid != null) {
+                alreadyValid[0]++;
+            }
+            obs.progress(100);
+            return;
+        }
+
+        try {
+            if (_digest.validateResource(rsrc, obs)) {
+                // if the resource has a _new file, add it to to-install list
+                if (!toInstall.contains(rsrc) && rsrc.getLocalNew().exists()) {
+                    toInstall.add(rsrc);
+                    return;
+                }
+                // unpack this resource if appropriate
+                if (noUnpack || !rsrc.shouldUnpack()) {
+                    // finally note that this resource is kosher
+                    rsrc.markAsValid();
+                    return;
+                }
+                if (rsrc.unpack()) {
+                    unpacked.add(rsrc);
+                    rsrc.markAsValid();
+                    return;
+                }
+                log.info("Failure unpacking resource", "rsrc", rsrc);
+            }
+
+        } catch (Exception e) {
+            log.info("Failure validating resource. Requesting redownload...",
+                     "rsrc", rsrc, "error", e);
+
+        } finally {
+            obs.progress(100);
+        }
+        toDownload.add(rsrc);
     }
 
     /**

src/main/java/com/threerings/getdown/launcher/Getdown.java

@@ -38,17 +38,7 @@ import java.net.URLConnection;
 
 import java.security.cert.Certificate;
 
-import java.util.ArrayList;
+import java.util.*;
-import java.util.Arrays;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Locale;
-import java.util.Map;
-import java.util.ResourceBundle;
-import java.util.Set;
-import java.util.Timer;
-import java.util.TimerTask;
 
 import ca.beq.util.win32.registry.RegistryKey;
 import ca.beq.util.win32.registry.RegistryValue;
@@ -142,7 +132,7 @@ public abstract class Getdown extends Thread
     public void install () throws IOException, InterruptedException
     {
         if (_readyToInstall) {
-            log.info("Installing downloaded resources:");
+            log.info("Installing " + _toInstallResources.size() + " downloaded resources:");
             for (Resource resource : _toInstallResources) {
                 resource.install();
                 if (Thread.interrupted()) {
@@ -442,7 +432,7 @@ public abstract class Getdown extends Thread
             // we'll keep track of all the resources we unpack
             Set<Resource> unpacked = new HashSet<Resource>();
 
-            _toInstallResources = new ArrayList<Resource>();
+            _toInstallResources = new HashSet<Resource>();
             _readyToInstall = false;
 
             //setStep(Step.START);
@@ -476,12 +466,10 @@ public abstract class Getdown extends Thread
                 // now verify our resources...
                 setStep(Step.VERIFY_RESOURCES);
                 setStatusAsync("m.validating", -1, -1L, false);
-                List<Resource> toDownload = new ArrayList<Resource>();
+                Set<Resource> toDownload = new HashSet<Resource>();
                 _app.verifyResources(_progobs, alreadyValid, unpacked,
                                      _toInstallResources, toDownload);
                 if (toDownload.size() == 0) {
-                    log.info("Resources verified.");
-
                     // if we were downloaded in full from another service (say, Steam), we may
                     // not have unpacked all of our resources yet
                     if (Boolean.getBoolean("check_unpacked")) {
@@ -758,7 +746,7 @@ public abstract class Getdown extends Thread
     /**
      * Called if the application is determined to require resource downloads.
      */
-    protected void download (List<Resource> resources)
+    protected void download (Collection<Resource> resources)
         throws IOException, InterruptedException
     {
         // create our user interface
@@ -1282,7 +1270,7 @@ public abstract class Getdown extends Thread
     protected boolean _launchInSilent;
     protected long _startup;
 
-    protected List<Resource> _toInstallResources;
+    protected Set<Resource> _toInstallResources;
     protected boolean _readyToInstall;
 
     protected boolean _enableTracking = true;