cache: concurrency friendly

Author: MykolaGolubyev

webtau-cache/pom.xml

@@ -63,6 +63,14 @@
             <artifactId>groovy-nio</artifactId>
             <scope>test</scope>
         </dependency>
+
+        <dependency>
+            <groupId>org.testingisdocumenting.webtau</groupId>
+            <artifactId>webtau-core</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+            <type>test-jar</type>
+        </dependency>
     </dependencies>
 
     <build>

webtau-cache/src/main/java/org/testingisdocumenting/webtau/cache/FileBasedCache.java

@@ -1,4 +1,5 @@
 /*
+ * Copyright 2021 webtau maintainers
  * Copyright 2019 TWO SIGMA OPEN SOURCE, LLC
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -24,17 +25,29 @@
 import java.nio.file.Path;
 import java.util.LinkedHashMap;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Supplier;
 
 public class FileBasedCache {
     private static final String VALUE_KEY = "value";
     private static final String EXPIRATION_TIME_KEY = "expirationTime";
 
+    private final AtomicBoolean isCacheLoaded;
+
     private final Supplier<Path> cachePathSupplier;
-    private Map<String, Object> loadedCache;
+    private final AtomicLong lastPutTime;
+
+    private final Map<String, Object> loadedCache;
 
     public FileBasedCache(Supplier<Path> cachePathSupplier) {
         this.cachePathSupplier = cachePathSupplier;
+        this.isCacheLoaded = new AtomicBoolean(false);
+        this.lastPutTime = new AtomicLong(Time.currentTimeMillis());
+        this.loadedCache = new ConcurrentHashMap<>();
+
+        registerFlushCacheOnExit();
     }
 
     @SuppressWarnings("unchecked")
@@ -47,22 +60,28 @@ public <E> E get(String key) {
 
         Number expirationTime = (Number) valueWithMeta.get(EXPIRATION_TIME_KEY);
         if (Time.currentTimeMillis() >= expirationTime.longValue()) {
+            loadedCache.remove(key);
+            flushCacheToDiskIfRequired();
             return null;
         }
 
-        return (E) valueWithMeta.get(VALUE_KEY);
+        E result = (E) valueWithMeta.get(VALUE_KEY);
+        flushCacheToDiskIfRequired();
+
+        return result;
     }
 
     public void put(String key, Object value, long expirationTime) {
         loadCacheIfRequired();
         Map<String, Object> valueWithMeta = createValueWithMeta(value, expirationTime);
         loadedCache.put(key, valueWithMeta);
 
-        FileUtils.writeTextContent(cachePathSupplier.get(), JsonUtils.serializePrettyPrint(loadedCache));
+        flushCacheToDiskIfRequired();
     }
 
     public void put(String key, Object value) {
         put(key, value, Long.MAX_VALUE);
+        flushCacheToDiskIfRequired();
     }
 
     private Map<String, Object> createValueWithMeta(Object value, long expirationTime) {
@@ -75,16 +94,36 @@ private Map<String, Object> createValueWithMeta(Object value, long expirationTim
 
     @SuppressWarnings("unchecked")
     private void loadCacheIfRequired() {
-        if (loadedCache != null) {
+        if (isCacheLoaded.get()) {
             return;
         }
 
         Path cachePath = cachePathSupplier.get();
         if (!Files.exists(cachePath)) {
-            loadedCache = new LinkedHashMap<>();
+            isCacheLoaded.set(true);
             return;
         }
 
-        loadedCache = (Map<String, Object>) JsonUtils.deserialize(FileUtils.fileTextContent(cachePath));
+        loadedCache.putAll((Map<String, ?>) JsonUtils.deserialize(FileUtils.fileTextContent(cachePath)));
+        isCacheLoaded.set(true);
+    }
+
+    private void flushCacheToDiskIfRequired() {
+        long lastTime = lastPutTime.get();
+        long current = Time.currentTimeMillis();
+
+        if (current - lastTime > 10_000) {
+            flushCacheToDisk();
+        }
+
+        lastPutTime.set(current);
+    }
+
+    private synchronized void flushCacheToDisk() {
+        FileUtils.writeTextContent(cachePathSupplier.get(), JsonUtils.serializePrettyPrint(loadedCache));
+    }
+
+    private void registerFlushCacheOnExit() {
+        Runtime.getRuntime().addShutdownHook(new Thread(this::flushCacheToDisk));
     }
 }

webtau-cache/src/test/groovy/org/testingisdocumenting/webtau/cache/FileBasedCacheTest.groovy

@@ -16,15 +16,24 @@
 
 package org.testingisdocumenting.webtau.cache
 
-import org.testingisdocumenting.webtau.utils.JsonUtils
+import org.junit.After
 import org.junit.Assert
 import org.junit.Test
+import org.testingisdocumenting.webtau.time.DummyTimeProvider
+import org.testingisdocumenting.webtau.time.Time
+import org.testingisdocumenting.webtau.utils.JsonUtils
 
 import java.nio.file.Files
 import java.nio.file.Path
 import java.nio.file.Paths
+import java.util.concurrent.Executors
 
 class FileBasedCacheTest {
+    @After
+    void cleanUp() {
+        Time.setTimeProvider(null)
+    }
+
     @Test
     void "should load cached values from a provided file"() {
         def cache = [
@@ -51,8 +60,11 @@ class FileBasedCacheTest {
     }
 
     @Test
-    void "should persist values in the file as pretty print json"() {
+    void "should persist values in the file as pretty print json if time passed since last put"() {
         def cacheFile = createTempCacheFile([:])
+
+        Time.timeProvider = new DummyTimeProvider([0, 11_000])
+
         def fileBasedCache = new FileBasedCache({ -> cacheFile })
         fileBasedCache.put('accessToken', 'abc', 400)
 
@@ -64,10 +76,24 @@ class FileBasedCacheTest {
                 '}'), cacheFile.text)
     }
 
+    @Test
+    void "should not persist values in the file right away"() {
+        def cacheFile = createTempCacheFile([:])
+
+        Time.timeProvider = new DummyTimeProvider([0, 100])
+
+        def fileBasedCache = new FileBasedCache({ -> cacheFile })
+        fileBasedCache.put('accessToken', 'abc', 400)
+
+        Assert.assertEquals("{ }", cacheFile.text)
+    }
+
     @Test
     void "should create non expiring values if no expiration time is provided"() {
         def cacheFile = createTempCacheFile([:])
+
         def fileBasedCache = new FileBasedCache({ -> cacheFile })
+        Time.timeProvider = new DummyTimeProvider([0, 11_000])
         fileBasedCache.put('accessToken', 'abc')
 
         Assert.assertEquals(String.format('{%n' +
@@ -78,6 +104,31 @@ class FileBasedCacheTest {
                 '}'), cacheFile.text)
     }
 
+    @Test
+    void "should handle put from multiple threads"() {
+        def cacheFile = createTempCacheFile([:])
+        Time.timeProvider = new DummyTimeProvider(0)
+        def fileBasedCache = new FileBasedCache({ -> cacheFile })
+
+        def executor = Executors.newFixedThreadPool(300)
+
+        def futures = []
+        300.times {idx ->
+            fileBasedCache.put("value" + idx, 0)
+
+            futures << executor.submit {
+                150.times {
+                    fileBasedCache.put("value" + idx, fileBasedCache.get("value" + idx) + 1)
+                }
+            }
+        }
+
+        futures*.get()
+        assert fileBasedCache.get("value0") == 150
+        assert fileBasedCache.get("value1") == 150
+        assert fileBasedCache.get("value9") == 150
+    }
+
     @Test
     void "should handle non existing file"() {
         def cachePath = Paths.get('nonExisting.json')

webtau-core/src/main/java/org/testingisdocumenting/webtau/time/Time.java

@@ -1,4 +1,5 @@
 /*
+ * Copyright 2021 webtau maintainers
  * Copyright 2019 TWO SIGMA OPEN SOURCE, LLC
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -20,7 +21,7 @@
 
 public class Time {
     private static final TimeProvider systemTimeProvider = new SystemTimeProvider();
-    private static AtomicReference<TimeProvider> timeProvider = new AtomicReference<>(systemTimeProvider);
+    private static final AtomicReference<TimeProvider> timeProvider = new AtomicReference<>(systemTimeProvider);
 
     public static long currentTimeMillis() {
         return Time.timeProvider.get().currentTimeMillis();

webtau-core/src/test/groovy/org/testingisdocumenting/webtau/reporter/ConsoleStepReporterTest.groovy

@@ -42,7 +42,7 @@ class ConsoleStepReporterTest implements ConsoleOutput {
         ConsoleOutputs.add(this)
         ConsoleOutputs.add(ansiConsoleOutput)
 
-        Time.setTimeProvider(new DummyTimeProvider([0, 0, 0, 0, 0, 0]))
+        Time.setTimeProvider(new DummyTimeProvider(0))
     }
 
     @After

webtau-core/src/test/groovy/org/testingisdocumenting/webtau/time/DummyTimeProvider.groovy

@@ -1,4 +1,5 @@
 /*
+ * Copyright 2021 webtau maintainers
  * Copyright 2019 TWO SIGMA OPEN SOURCE, LLC
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -16,18 +17,25 @@
 
 package org.testingisdocumenting.webtau.time
 
-import org.testingisdocumenting.webtau.time.TimeProvider
-
 class DummyTimeProvider implements TimeProvider {
-    private List<Integer> timeSnapshots = []
+    private List<Long> timeSnapshots = []
+    private Long constantTime = null
     private int currentSnapshotIdx = 0
 
     DummyTimeProvider(List<Integer> timeSnapshots) {
         this.timeSnapshots.addAll(timeSnapshots)
     }
 
+    DummyTimeProvider(Long constantTime) {
+        this.constantTime = constantTime
+    }
+
     @Override
     long currentTimeMillis() {
+        if (constantTime != null) {
+            return constantTime
+        }
+
         if (currentSnapshotIdx >= timeSnapshots.size()) {
             throw new RuntimeException("$currentSnapshotIdx idx is out of the provided time snapshots $timeSnapshots")
         }