Limit the amount of file processed at a time.

Author: tburch

src/main/java/com/lowtuna/jsonblob/core/BlobCleanupConsumer.java

@@ -0,0 +1,68 @@
+package com.lowtuna.jsonblob.core;
+
+import com.fasterxml.jackson.core.JsonParseException;
+import com.fasterxml.jackson.databind.JsonMappingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Optional;
+import io.dropwizard.util.Duration;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.joda.time.DateTime;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.concurrent.BlockingQueue;
+
+/**
+ * Created by tburch on 8/20/17.
+ */
+@Slf4j
+@RequiredArgsConstructor
+public class BlobCleanupConsumer implements Runnable {
+  private final BlockingQueue<File> filesToProcess;
+  private final Duration blobAccessTtl;
+  private final FileSystemJsonBlobManager fileSystemJsonBlobManager;
+  private final ObjectMapper om;
+
+  @Override
+  public void run() {
+    try {
+      File file = filesToProcess.take();
+      log.debug("Processing {}", file.getAbsolutePath());
+      String blobId = file.getName().split("\\.", 2)[0];
+      File metadataFile = fileSystemJsonBlobManager.getMetaDataFile(file.getParentFile());
+
+      if (file.equals(metadataFile)) {
+        return;
+      }
+
+      BlobMetadataContainer metadataContainer = metadataFile.exists() ? om.readValue(fileSystemJsonBlobManager.readFile(metadataFile), BlobMetadataContainer.class) : new BlobMetadataContainer();
+
+      Optional<DateTime> lastAccessed = fileSystemJsonBlobManager.resolveTimestamp(blobId);
+      if (metadataContainer.getLastAccessedByBlobId().containsKey(blobId)) {
+        lastAccessed = Optional.of(metadataContainer.getLastAccessedByBlobId().get(blobId));
+      }
+
+      if (!lastAccessed.isPresent()) {
+        log.warn("Couldn't get last accessed timestamp for blob {}", blobId);
+        return;
+      }
+
+      log.debug("Blob {} was last accessed {}", blobId, lastAccessed.get());
+
+      if (lastAccessed.get().plusMillis((int) blobAccessTtl.toMilliseconds()).isBefore(DateTime.now())) {
+        if (file.delete()) {
+          log.info("Blob {} is older than {} (last accessed {}), so it's going to be deleted", blobId, blobAccessTtl, lastAccessed.get());
+        }
+      }
+    } catch (InterruptedException e) {
+      e.printStackTrace();
+    } catch (JsonParseException e) {
+      e.printStackTrace();
+    } catch (JsonMappingException e) {
+      e.printStackTrace();
+    } catch (IOException e) {
+      e.printStackTrace();
+    }
+  }
+}

src/main/java/com/lowtuna/jsonblob/core/BlobCleanupProducer.java

@@ -0,0 +1,77 @@
+package com.lowtuna.jsonblob.core;
+
+import com.google.common.base.Stopwatch;
+import io.dropwizard.util.Duration;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.io.DirectoryWalker;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.time.LocalDate;
+import java.util.Collection;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Created by tburch on 8/18/17.
+ */
+@AllArgsConstructor
+@Slf4j
+public class BlobCleanupProducer extends DirectoryWalker<Void> implements Runnable {
+  private final Path dataDirectoryPath;
+  private final Duration blobAccessTtl;
+  private final BlockingQueue<File> filesToProcess;
+
+  @Override
+  protected boolean handleDirectory(File directory, int depth, Collection<Void> results) throws IOException {
+    if (directory.listFiles() != null && directory.listFiles().length == 0) {
+      if (directory.delete()) log.info("{} has no files, so it's being deleted", directory.getAbsolutePath());
+      return false;
+    }
+
+    if (directory.listFiles().length == 1) {
+      if (directory.listFiles()[0].getName().startsWith(FileSystemJsonBlobManager.BLOB_METADATA_FILE_NAME)) {
+        if (directory.delete()) log.info("{} has only a metadata file, so it's being deleted", directory.getAbsolutePath());
+        return false;
+      }
+    }
+
+    boolean process = true;
+    if (isDataDir(directory.getAbsolutePath())) {
+      String[] dateParts = directory.getAbsolutePath().replace(dataDirectoryPath.toFile().getAbsolutePath(), "").split("/", 4);
+      LocalDate localDate = LocalDate.of(Integer.parseInt(dateParts[1]), Integer.parseInt(dateParts[2]), Integer.parseInt(dateParts[3]));
+      process = localDate.isBefore(LocalDate.now().minusDays(blobAccessTtl.toDays()));
+      if (process) {
+        log.info("Processing {} with {} blobs for un-accessed blobs", directory.getAbsolutePath(), directory.listFiles().length - 1);
+        for (File file: directory.listFiles()) {
+          try {
+            filesToProcess.put(file);
+          } catch (InterruptedException e) {
+            log.warn("Interrupted while trying to add file to be processed at {}", file.getAbsolutePath(), e);
+          }
+        }
+        process = false;
+      }
+    }
+
+    return process;
+  }
+
+  private boolean isDataDir(String path) {
+    return path.replace(dataDirectoryPath.toFile().getAbsolutePath(), "").split("/").length == 4;
+  }
+
+  @Override
+  public void run() {
+    Stopwatch stopwatch = new Stopwatch().start();
+    try {
+      walk(dataDirectoryPath.toFile(), null);
+      log.info("Completed cleaning up un-accessed blobs in {}ms", stopwatch.elapsed(TimeUnit.MILLISECONDS));
+    } catch (Exception e) {
+      e.printStackTrace();
+    }
+  }
+
+}

src/main/java/com/lowtuna/jsonblob/core/BlobDataDirectoryCleaner.java

@@ -31,6 +31,8 @@
 import java.util.Map;
 import java.util.TimeZone;
 import java.util.UUID;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -45,8 +47,9 @@
  */
 @Slf4j
 public class FileSystemJsonBlobManager implements JsonBlobManager, Runnable, Managed {
-  public static final String BLOB_METADATA_FILE_NAME = "blobMetadata";
+  static final String BLOB_METADATA_FILE_NAME = "blobMetadata";
 
+  private static final int CONSUMER_COUNT = 7;
   private static final DateTimeFormatter DIRECTORY_FORMAT = DateTimeFormat.forPattern("yyyy/MM/dd");
 
   @GuardedBy("lastAccessedLock")
@@ -308,8 +311,12 @@ public void start() throws Exception {
     scheduledExecutorService.scheduleWithFixedDelay(this, 1, 1, TimeUnit.MINUTES);
 
     log.info("Scheduling blob cleanup job");
-    BlobDataDirectoryCleaner dataDirectoryCleaner = new BlobDataDirectoryCleaner(blobDataDirectory.toPath(), blobAccessTtl, this,objectMapper);
+    BlockingQueue<File> filesToProcess = new ArrayBlockingQueue<>(1024);
+    for (int i = 0; i < CONSUMER_COUNT; i++) {
+      cleanupScheduledExecutorService.scheduleAtFixedRate(new BlobCleanupConsumer(filesToProcess, blobAccessTtl, this, objectMapper), 0, 250, TimeUnit.MILLISECONDS);
+    }
 
+    BlobCleanupProducer dataDirectoryCleaner = new BlobCleanupProducer(blobDataDirectory.toPath(), blobAccessTtl, filesToProcess);
     cleanupScheduledExecutorService.scheduleWithFixedDelay(dataDirectoryCleaner, 0, 1, TimeUnit.DAYS);
   }
 

src/main/java/com/lowtuna/jsonblob/core/FileSystemJsonBlobManager.java

@@ -20,7 +20,7 @@
  * Created by tburch on 8/16/17.
  */
 @Log
-public class BlobCleanupJobTest {
+public class TryBlobCleanupJob {
 
   private static final File TEMP;
   static {
@@ -36,7 +36,7 @@ public class BlobCleanupJobTest {
 
   @Before
   public void initBlobManage() {
-    this.blobManager = new FileSystemJsonBlobManager(TEMP, Executors.newSingleThreadScheduledExecutor(), Executors.newSingleThreadScheduledExecutor(), new ObjectMapper(), blobTtl, true);
+    this.blobManager = new FileSystemJsonBlobManager(TEMP, Executors.newSingleThreadScheduledExecutor(), Executors.newScheduledThreadPool(10), new ObjectMapper(), blobTtl, true);
   }
 
   @Test