Add batch synchronisation #19

Sytten · 2016-11-03T20:35:28Z

This is a first implementation of a new async storage that sync in batch. Suggestions and critiques are welcome.
This storage was necessary since we just can't make a query to our database (even async) for each query that arrives, that does not scale.

malaporte · 2016-11-05T16:48:14Z

src/main/java/com/coveo/spillway/storage/AsyncBatchLimitUsageStorage.java

+  public AsyncBatchLimitUsageStorage(
+      LimitUsageStorage wrappedLimitUsageStorage,
+      int timeBetweenSynchronisations,
+      int delayBeforeFirstSync) {


Not sure I see a real usefulness for this param - why not just use timeBetweenSynchronization ?

Needed for tests :(

Alright then.

package protected then ?

malaporte · 2016-11-05T16:52:31Z

src/test/java/com/coveo/spillway/functional/SpillwayFunctionalTests.java

@@ -146,6 +148,10 @@ public void asyncPerformance() {
      lastResponse =
          asyncStorage.incrementAndGet(RESOURCE1, LIMIT1, PROPERTY1, EXPIRATION, TIMESTAMP);
    }
+    asyncStorage.shutdownStorage();
+    while (!asyncStorage.isTerminated()) {


We should provide a method that blocks until terminated.

Yes, I just wasn't sure of the implementation. Is a Thread sleep of like 10ms between checks good enough?

Well it's a start, but have a look at awaitTermination on ExecutorService

yes thats true, forgot about that

malaporte · 2016-11-05T16:56:14Z

src/main/java/com/coveo/spillway/storage/utils/CacheSynchronisation.java

+                              .withCost(cost)
+                              .build();
+
+                      Pair<LimitKey, Integer> reponse = storage.addAndGet(request);


Until there is something I got wrong you're sending the events using separate requests here right?

Yup. Why not send them in one go through the other overload?

I don't think its possible since we iterate over the map and we sync each value one by one.

I agree that pipelining the queries would be better, but I don't really know how to do that while iterating over the map. We have to remember that it is less critical since its an async job and we don't have that many entries.

I suggest to go with that first version for now. It's already a good improvement over the previous version. Further optimizations can be done in a separate issue.

malaporte · 2016-11-05T17:55:51Z

src/main/java/com/coveo/spillway/storage/utils/CacheSynchronisation.java

+                      valueEntry.getValue().setTotal(reponse.getValue());
+                    });
+          } catch (IllegalStateException e) {
+            logger.warn("Key was deleted during synchronisation, ignoring");


Can't we catch this particular error in a more precise location?

Also, maybe provide a catch all handler that logs an error?

What do you mean by a more precise location? It makes no sens to catch it inside the inner forEach since if the key is deleted, all values in the inner map will be deleted too.
This exception should not be thrown by Oracle JVM anyway, I included it since the javadoc says that some implement might throw it.

Yes I will include a catch for runtime exception, maybe surrounding the whole function?

I changed that for a catch Exception and log the error.

malaporte · 2016-11-06T14:36:25Z

src/main/java/com/coveo/spillway/storage/utils/CacheSynchronisation.java

+ * @author Emile Fugulin
+ * @since 1.0.0
+ */
+public class CacheSynchronisation extends TimerTask {


Pretty sure that Synchronization should take a Z

malaporte · 2016-11-06T14:37:11Z

src/main/java/com/coveo/spillway/storage/utils/CacheSynchronisation.java

+
+  @Override
+  public void run() {
+    cache.applyOnEach(


I would expect that we'd somehow clear the cache after we flush it's content downstream, no?

Not sure what you mean by that...

I will add a removeExpiredEntry after, good point

malaporte · 2016-11-06T14:38:40Z

src/main/java/com/coveo/spillway/storage/AsyncLimitUsageStorage.java

@@ -78,6 +78,14 @@ public AsyncLimitUsageStorage(LimitUsageStorage wrappedLimitUsageStorage) {
    return wrappedLimitUsageStorage.debugCurrentLimitCounters();
  }

+  public void shutdownStorage() {
+    executorService.shutdown();


No guarantee that we flush the remaining data to the underlying storage?

Not necessary since the keys will expire by themselves

Well yes, but it does mean that those events will never get counted. Probably not a problem but still something that could be better easily no?

Ho I see what you mean. Well the shutdown isn't meant to be called during the life of the async storage. Its something that I would have put in a destructor, but Java doesn't have one...

Why shouldn't it be called? Makes sense to do it (maybe indirectly?) when shutting down the app no?

Yes you have to do it when you shutdown the app, but I don't see your point of counting the events on the remote storage if you are shutting down everything :)

You might be shutting down a single server in a cluster that has 10s of them.

well the shutdown method is only to stop gracefully the sync between the local and the remote storage, it won't actually stop the remote storage

Sytten · 2016-11-07T21:36:27Z

I also replied to some comments marked as outdated because I renamed the class, but I would still like your feedback @malaporte

Sytten · 2016-11-08T13:46:11Z

If you want to take a look too @GuiSim

francistb · 2016-11-08T14:56:31Z

src/main/java/com/coveo/spillway/storage/AsyncBatchLimitUsageStorage.java

+  public AsyncBatchLimitUsageStorage(
+      LimitUsageStorage wrappedLimitUsageStorage,
+      int timeBetweenSynchronisations,
+      int delayBeforeFirstSync) {


package protected then ?

francistb · 2016-11-08T14:58:54Z

src/main/java/com/coveo/spillway/storage/AsyncBatchLimitUsageStorage.java

+  private Timer timer;
+
+  public AsyncBatchLimitUsageStorage(
+      LimitUsageStorage wrappedLimitUsageStorage, int timeBetweenSynchronisations) {


Duration instead of a int ? or we should specify the unit somehow...

francistb · 2016-11-08T14:59:33Z

src/main/java/com/coveo/spillway/storage/AsyncBatchLimitUsageStorage.java

+
+  @Override
+  public Map<LimitKey, Integer> addAndGet(Collection<AddAndGetRequest> requests) {
+    Map<LimitKey, Integer> cachedEntries = cache.addAndGet(requests);


return directly

francistb · 2016-11-08T15:11:59Z

src/main/java/com/coveo/spillway/storage/utils/CacheSynchronization.java

+                      valueEntry.getValue().setTotal(reponse.getValue());
+                    });
+          } catch (IllegalStateException e) {
+            logger.warn("Key was deleted during synchronisation, ignoring");


"synchronization"
add a "." at the end

francistb · 2016-11-08T15:13:24Z

src/main/java/com/coveo/spillway/storage/utils/CacheSynchronization.java

+
+                      Pair<LimitKey, Integer> reponse = storage.addAndGet(request);
+
+                      valueEntry.getValue().addAndGet(-cost);


should we add a decrementAndGet method ?

We could, but under it it will still be an addAndGet

why not a decrementAndGet under also ?

Sytten · 2016-11-08T16:40:42Z

If everybody approve, i will merge and create a V1 published on maven central.

Sytten added 3 commits November 3, 2016 09:58

Layout the implementation

68aece7

Finish basic feature and add tests

f230b16

Change test and add documentation

e665587

malaporte reviewed Nov 6, 2016

View reviewed changes

Rename cache sync task

abe758a

Sytten assigned francistb Nov 8, 2016

Small fixes related to review

65ce47d

francistb reviewed Nov 8, 2016

View reviewed changes

Sytten added 2 commits November 8, 2016 10:49

Other fixes related to review

e4958bb

Use awaitTermination instead of sleep

38692fe

Sytten mentioned this pull request Nov 8, 2016

Publish to Maven Central #16

Closed

Sytten added 2 commits November 8, 2016 11:34

Ignore all exceptions when sync with remote DB

b5a59ad

Fix formatting

0395008

Sytten mentioned this pull request Nov 8, 2016

ASync Batch Storage Optimization #21

Open

Sytten merged commit 21a738a into master Nov 8, 2016

Sytten deleted the add-batch-async branch November 8, 2016 17:25

malaporte unassigned francistb Jun 24, 2019

This pull request was closed.


		Pair<LimitKey, Integer> reponse = storage.addAndGet(request);

		valueEntry.getValue().addAndGet(-cost);

Add batch synchronisation #19

Add batch synchronisation #19

Conversation

Sytten commented Nov 3, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Sytten commented Nov 7, 2016

Sytten commented Nov 8, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Sytten commented Nov 8, 2016