Concurrent file directories index

src/main/java/build/buildfarm/CASTest.java

@@ -65,8 +65,8 @@ public static void main(String[] args) throws Exception {
     CASFileCache fileCache =
         new LocalCASFileCache(
             root,
-            /* maxSizeInBytes=*/ GBtoBytes(500),
-            new DigestUtil(HashFunction.SHA1),
+            /* maxSizeInBytes=*/ GBtoBytes(2 * 1024),
+            new DigestUtil(HashFunction.SHA256),
             /* expireService=*/ newDirectExecutorService(),
             /* accessRecorder=*/ directExecutor());
 

src/main/java/build/buildfarm/cas/CASFileCache.java

@@ -116,7 +116,6 @@ public abstract class CASFileCache implements ContentAddressableStorage {
   private static final Logger logger = Logger.getLogger(CASFileCache.class.getName());
 
   protected static final String DEFAULT_DIRECTORIES_INDEX_NAME = "directories.sqlite";
-  protected static final String DIRECTORIES_INDEX_NAME_MEMORY = ":memory:";
 
   private final Path root;
   private final FileStore fileStore;
@@ -129,7 +128,7 @@ public abstract class CASFileCache implements ContentAddressableStorage {
   private final Executor accessRecorder;
   private final ExecutorService expireService;
 
-  private final Map<Digest, DirectoryEntry> directoryStorage = Maps.newHashMap();
+  private final Map<Digest, DirectoryEntry> directoryStorage = Maps.newConcurrentMap();
   private final DirectoriesIndex directoriesIndex;
   private final String directoriesIndexDbName;
   private final LockMap locks = new LockMap();
@@ -286,24 +285,7 @@ public CASFileCache(
     this.onExpire = onExpire;
     this.delegate = delegate;
     this.directoriesIndexDbName = directoriesIndexDbName;
-
-    String directoriesIndexUrl = "jdbc:sqlite:";
-    if (directoriesIndexDbName.equals(DIRECTORIES_INDEX_NAME_MEMORY)) {
-      directoriesIndexUrl += directoriesIndexDbName;
-    } else {
-      // db is ephemeral for now, no reuse occurs to match it, computation
-      // occurs each time anyway, and expected use of put is noop on collision
-      Path path = getPath(directoriesIndexDbName);
-      try {
-        if (Files.exists(path)) {
-          Files.delete(path);
-        }
-      } catch (IOException e) {
-        throw new RuntimeException(e);
-      }
-      directoriesIndexUrl += path.toString();
-    }
-    this.directoriesIndex = new FileDirectoriesIndex(directoriesIndexUrl, root);
+    this.directoriesIndex = new FileDirectoriesIndex(directoriesIndexDbName, root);
 
     header.before = header.after = header;
   }
@@ -1295,16 +1277,33 @@ public StartupCacheResults start(Consumer<Digest> onPut, ExecutorService removeD
     CacheScanResults cacheScanResults = scanRoot();
     LogCacheScanResults(cacheScanResults);
     deleteInvalidFileContent(cacheScanResults.deleteFiles, removeDirectoryService);
+    Instant phase1Ending = Instant.now();
+    logger.log(
+        Level.INFO,
+        "Phase 1 Time: "
+            + Duration.between(startTime, phase1Ending).getSeconds()
+            + "s");
 
     // Phase 2: Compute
     // recursively construct all directory structures.
     List<Path> invalidDirectories = computeDirectories(cacheScanResults);
     LogComputeDirectoriesResults(invalidDirectories);
     deleteInvalidFileContent(invalidDirectories, removeDirectoryService);
 
-    logger.log(Level.INFO, "Creating Index");
+    Instant beforeIndexing = Instant.now();
+    logger.log(
+        Level.INFO,
+        "Phase 2 Inserting Time: "
+            + Duration.between(phase1Ending, beforeIndexing).getSeconds()
+            + "s");
+
     directoriesIndex.start();
-    logger.log(Level.INFO, "Index Created");
+    Instant afterIndexing = Instant.now();
+    logger.log(
+        Level.INFO,
+        "Phase 2 Building Index Time: "
+            + Duration.between(beforeIndexing, afterIndexing).getSeconds()
+            + "s");
 
     // Calculate Startup time
     Instant endTime = Instant.now();
@@ -1400,7 +1399,7 @@ private void processRootFile(
 
     // ignore our directories index database
     // indexes will be removed and rebuilt for compute
-    if (!basename.equals(directoriesIndexDbName)) {
+    if (!basename.startsWith(directoriesIndexDbName)) {
       FileStatus stat = stat(file, false);
 
       // mark directory for later key compute
@@ -1458,14 +1457,15 @@ private List<Path> computeDirectories(CacheScanResults cacheScanResults)
       throws IOException, InterruptedException {
 
     // create thread pool
-    int nThreads = Runtime.getRuntime().availableProcessors();
+    int nThreads = Runtime.getRuntime().availableProcessors() * 2;
     String threadNameFormat = "compute-cache-pool-%d";
     ExecutorService pool =
         Executors.newFixedThreadPool(
             nThreads, new ThreadFactoryBuilder().setNameFormat(threadNameFormat).build());
 
     ImmutableList.Builder<Path> invalidDirectories = new ImmutableList.Builder<>();
 
+    directoriesIndex.setBatchMode(true);
     for (Path path : cacheScanResults.computeDirs) {
       pool.execute(
           () -> {
@@ -1482,9 +1482,7 @@ private List<Path> computeDirectories(CacheScanResults cacheScanResults)
               if (digest != null && getDirectoryPath(digest).equals(path)) {
                 DirectoryEntry e = new DirectoryEntry(directory, Deadline.after(10, SECONDS));
                 directoriesIndex.put(digest, inputsBuilder.build());
-                synchronized (this) {
                   directoryStorage.put(digest, e);
-                }
               } else {
                 synchronized (invalidDirectories) {
                   invalidDirectories.add(path);
@@ -1497,6 +1495,7 @@ private List<Path> computeDirectories(CacheScanResults cacheScanResults)
     }
 
     joinThreads(pool, "Populating Directories...", 1, MINUTES);
+    directoriesIndex.setBatchMode(false);
 
     return invalidDirectories.build();
   }
@@ -2226,9 +2225,7 @@ private ListenableFuture<Path> putDirectorySynchronized(
                       ? Directory.getDefaultInstance()
                       : directoriesByDigest.get(digest),
                   Deadline.after(10, SECONDS));
-          synchronized (this) {
-            directoryStorage.put(digest, e);
-          }
+          directoryStorage.put(digest, e);
           return path;
         },
         service);

src/main/java/build/buildfarm/cas/DirectoriesIndex.java

@@ -34,4 +34,6 @@ interface DirectoriesIndex {
   void remove(Digest directory);
 
   void start();
+
+  void setBatchMode(boolean batchMode) throws InterruptedException;
 }