Batcher implementation with Synchronized `Batcher#flush`(without triggers) #716

rahulKQL · 2019-05-29T18:00:46Z

This is a follow-up PR for #692

What this PR contains

This implementation does not include any triggers/thresholds/flowController for sending received elements for batching(This will be added in followup PRs).
As of this implementation, The User has to explicitly call the Batcher#flush() to send the accumulated request for batching.
Here BatcherImpl expects v2.BatchingDescriptor, UnaryCallable and a request prototype containing repetitive data which would be copied over to each batch along with the bundle of elements.
v2.BatchingCallSettings is an extension from existing BatchingCallSettings.
Have added default wait time for unfinished batch as 250ms.

Follow-ups

Once this change is merged I will raise a separate PR with triggers, thresholds, and flowController for BatcherImpl.

rahulKQL · 2019-05-29T18:02:36Z

@igorbernstein2 Please have a look.

codecov · 2019-05-29T18:05:04Z

Codecov Report

Merging #716 into master will increase coverage by 1.91%.
The diff coverage is 92.64%.

@@             Coverage Diff              @@
##             master     #716      +/-   ##
============================================
+ Coverage     75.74%   77.65%   +1.91%     
- Complexity     1041     1091      +50     
============================================
  Files           196      197       +1     
  Lines          4679     4776      +97     
  Branches        363      371       +8     
============================================
+ Hits           3544     3709     +165     
+ Misses          975      898      -77     
- Partials        160      169       +9

Impacted Files	Coverage Δ	Complexity Δ
...va/com/google/api/gax/batching/v2/BatcherImpl.java	`92.64% <92.64%> (ø)`	`10 <10> (?)`
...ain/java/com/google/api/gax/rpc/ClientContext.java	`81.81% <0%> (-1%)`	`8% <0%> (ø)`
...api/gax/grpc/InstantiatingGrpcChannelProvider.java	`77.97% <0%> (+2.8%)`	`36% <0%> (+9%)`	⬆️
...gle/api/gax/httpjson/HttpJsonTransportChannel.java	`47.05% <0%> (+17.64%)`	`5% <0%> (+2%)`	⬆️
...gle/api/gax/rpc/FixedTransportChannelProvider.java	`100% <0%> (+26.66%)`	`15% <0%> (+6%)`	⬆️
...oogle/api/gax/httpjson/ManagedHttpJsonChannel.java	`54% <0%> (+52%)`	`5% <0%> (+4%)`	⬆️
...httpjson/InstantiatingHttpJsonChannelProvider.java	`70.96% <0%> (+70.96%)`	`19% <0%> (+19%)`	⬆️

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update bb00cb7...05eb03c. Read the comment docs.

igorbernstein2

I add some first pass comments. Will take a deeper look in the next couple of days

gax/src/main/java/com/google/api/gax/batching/v2/BatcherImpl.java

igorbernstein2 · 2019-05-29T18:19:42Z

gax/src/main/java/com/google/api/gax/batching/v2/BatcherImpl.java

+    sendBatch();
+    while (numOfRpcs.get() > 0) {
+      LockSupport.parkNanos(DEFAULT_FINISH_WAIT_NANOS);
+    }


I don't love this. Can we use a semaphore instead?

I have added a semaphore implementation 0 permits. Please let me know if this impl looks good to you
private final AtomicInteger numOfRpcs = new AtomicInteger(0);

Hmm, semaphore doesn't seem quite right for this. Sorry for misleading you on that

Maybe something like this:

AtomicInteger numOutstandingBatches = new AtomicInteger(); Object flushLock = new Object(); void sendBatch() { numOutstandingBatches.increment(); } void onBatchCompletion() { if (numOutstandingBatches.decrementAndGet() == 0) { flushLock.notifyAll(); } } void flush() { sendBatch(); awaitAllOutstandingBatches(); } void awaitAllOutstandingBatches() { while(numOutstandingBatches.get() > 0) { synchronized(lock) { flushLock.wait(); } } }

Thanks for the detailed steps, I have implemented with the same for Batcher#flush(). Please have a look.

gax/src/main/java/com/google/api/gax/batching/v2/BatcherImpl.java

gax/src/main/java/com/google/api/gax/batching/v2/BatchingCallSettings.java

igorbernstein2 · 2019-05-29T18:33:57Z

gax/src/main/java/com/google/api/gax/batching/v2/BatcherImpl.java

+   * This class represent one logical Batch. It accumulates all the elements and it's corresponding
+   * future element results for one batch.
+   */
+  class Batch {


My intention was not to declare type params for inner class... but Have updated it to private static class.

Please add a comment to make that clear

gax/src/main/java/com/google/api/gax/batching/v2/BatcherImpl.java

gax/src/main/java/com/google/api/gax/batching/v2/BatchingCallSettings.java

gax/src/main/java/com/google/api/gax/batching/v2/BatchingDescriptor.java

gax/src/main/java/com/google/api/gax/batching/v2/BatchingCallSettings.java

gax/src/test/java/com/google/api/gax/batching/v2/BatchingCallSettingsTest.java

gax/src/test/java/com/google/api/gax/rpc/testing/FakeBatchableApi.java

gax/src/test/java/com/google/api/gax/batching/v2/BatchingCallSettingsTest.java

…gers) - This implementation does not include any triggers for sending received elements for batching.(This will be added in followup PRs). - User has to explicitly call the `Batcher#flush()` to send the accumulated request for batching. - Here `BatcherImpl` expects `v2.BatchingDescriptor`, `v2.BatchingCallSetting`, UnaryCallable and a request prototype containing repetitive data which would be copied over to each batch along with the elements. - `v2.BatchingCallSettings` is an extension from existing `BatchingCallSettings`.

…chingCallSettings.java`

rahulKQL · 2019-05-31T15:41:04Z

@igorbernstein2
Please have another round of review, I have addressed the feedback comments.

Also, I decided to remove BatchingCallSettings.java from this PR. I believe till this PR, we are not expecting any special configuration for UnaryCallable so a generic unaryCallable should suffice to perform batching RPCs.
Would add BatchingCallSettings along with Default Batching Settings and flow controller in followup PRs.

gax/src/main/java/com/google/api/gax/batching/v2/BatcherImpl.java

igorbernstein2 · 2019-06-04T15:41:37Z

gax/src/main/java/com/google/api/gax/batching/v2/BatcherImpl.java

+    sendBatch();
+    while (numOfRpcs.get() > 0) {
+      LockSupport.parkNanos(DEFAULT_FINISH_WAIT_NANOS);
+    }


Hmm, semaphore doesn't seem quite right for this. Sorry for misleading you on that

Maybe something like this:

AtomicInteger numOutstandingBatches = new AtomicInteger(); Object flushLock = new Object(); void sendBatch() { numOutstandingBatches.increment(); } void onBatchCompletion() { if (numOutstandingBatches.decrementAndGet() == 0) { flushLock.notifyAll(); } } void flush() { sendBatch(); awaitAllOutstandingBatches(); } void awaitAllOutstandingBatches() { while(numOutstandingBatches.get() > 0) { synchronized(lock) { flushLock.wait(); } } }

igorbernstein2

LGTM, except for the timeout.
I'll assume that you will deal with synchronization when implementing timed flush

@vam-google can you take a look as well?

igorbernstein2 · 2019-06-06T15:35:35Z

gax/src/main/java/com/google/api/gax/batching/v2/BatcherImpl.java

+  private void awaitAllOutstandingBatches() throws InterruptedException {
+    while (numOfOutstandingBatches.get() > 0) {
+      synchronized (flushLock) {
+        flushLock.wait(DEFAULT_WAIT_TIME_MS);


I'm not sure I understand why we need a timeout

I added this as a defense, Also inspired by existing CBT-OperationalAccountant

What is it defending against?

I added this for the case of RPC taking too long, which would enable this thread to keep on checking the numOfOutStandingBatch.

I now realized that this would not add any meaning. The only update we are waiting here would be from onBatchComplete's notifyAll(), which will execute on all cases.
Thanks a lot for pointing it out. I will remove this timeout.

gax/src/test/java/com/google/api/gax/batching/v2/BatcherImplTest.java

gax/src/main/java/com/google/api/gax/batching/v2/BatcherImpl.java

gax/src/test/java/com/google/api/gax/rpc/testing/FakeBatchableApi.java

gax/src/main/java/com/google/api/gax/batching/v2/BatcherImpl.java

vam-google · 2019-06-07T20:06:52Z

gax/src/main/java/com/google/api/gax/batching/v2/BatcherImpl.java

+  private boolean isClosed = false;
+
+  private BatcherImpl(Builder<ElementT, ElementResultT, RequestT, ResponseT> builder) {
+    this.prototype = checkNotNull(builder.prototype, "RequestPrototype cannot be null.");


Please do not put period (.) at the end of messages like this to stay consistent with the rest of the codebase (applies to other places as well, like line 113).

Also the message probably should not use the class names to indicate what is null (either use field names i.e prototype instead of RequestPrototype, or regular langugae to describe what is null, otherwise it looks like the "class" itself is null).

vam-google · 2019-06-07T20:10:23Z

gax/src/main/java/com/google/api/gax/batching/v2/BatcherImpl.java

+  }
+
+  /** Builder for a BatcherImpl. */
+  public static class Builder<ElementT, ElementResultT, RequestT, ResponseT> {


can this be @Autovalue? I'm not a big fan of @Autovalue but we already use it extensively in gax-java, so lets reuse it here if possible to reduce boilerplate code typing.

I don't consider BatcherImpl is a value object. So i'm not sure how AutoValue can help here

I don't think it has to be a value object, it can be anything, which can be constructed with a builder.
Now when I look at it, do we even need a builder here? I'm just trying to figure out if we can reduce boilerplate code here (either by generating builder with the autovalue annotation or not having it at all).

I dont think a builder is necessary here. Given the choice between AutoValue and removing the builder I would prefer removing the builder

I have removed the Builder from the current change and annotated this class with @AutoValue.

Just a side note, We may need to come back on this when introducing flow controllers. As after implementing triggers, we would be expecting 6 arguments(i.e. existing ones + BatchingSettings, FlowController, and an Executor).

TLDR version: please remove AutoValue for the time being.

Longer version:
I've given this more thought, and I really think that AutoValue is really the wrong tool for this. In the best practices it states

Don't use AutoValue to implement value semantics unless you really want value semantics. In particular, you should never care about the difference between two equal instances.

This is definitely not the case for BatcherImpl since a Batcher has mutable state and we do care which instance we call flush on. Also, using AutoValue here implies that equality will use value semantics and be determined by the prototype, callable & batchingDescriptor:

BatcherImpl.Builder builder = ...; Batcher b1 = builder.build(); Batcher b2 = builder.build(); b2.add(...) b1.equals(b2) == true;

Which makes no sense. I think the only meaningful equality here is referential and goes against the best practices of AutoValue.

Since you already removed the Builder until a later PR, please remove the use of AutoValue here since it buys you very little. Once that PR comes up I'll chat with vam about the best path forward.

gax/src/main/java/com/google/api/gax/batching/v2/BatcherImpl.java

vam-google · 2019-06-07T20:25:23Z

gax/src/main/java/com/google/api/gax/batching/v2/BatcherImpl.java

+
+  private final AtomicInteger numOfOutstandingBatches = new AtomicInteger(0);
+  private final Object flushLock = new Object();
+  private boolean isClosed = false;


Please make it volatile

This is only used by the single caller thread, so it doesn't need to be volatile

Ok, but it is practically free to make it such, ans feels safer (a little bit of defensive programming never hurts in concurrent programming).

I generally prefer to avoid defensive programming because you end up hiding the intention of the code. But I agree with you that volatile is free here. So if you feel strongly, then I won't object

I would still add volatile

@rahulKQL can you add the volatile and I will merge this PR

I am sorry, I haven't marked this ongoing conversation as closed.
I have already added a volatile for isClosed flag.

rahulKQL · 2019-06-10T16:38:18Z

@vam-google, @igorbernstein2, I have tried to address your feedback, Please have a fresh look at this change.

@autovalue

Now BatcherImpl is created with @autovalue. Also made `isClosed` field as volatile.

vam-google

LGTM

igorbernstein2 · 2019-06-13T17:21:33Z

@vam-google Thanks for the review!

googlebot added the cla: yes This human has signed the Contributor License Agreement. label May 29, 2019

igorbernstein2 reviewed May 29, 2019

View reviewed changes

igorbernstein2 reviewed May 30, 2019

View reviewed changes

rahulKQL added 2 commits May 31, 2019 12:29

Addressing feedback comments for BatcherImpl.java and removing `Bat…

7de2c99

…chingCallSettings.java`

rahulKQL force-pushed the basicImpl branch from ed6c84a to 7de2c99 Compare May 31, 2019 15:12

Refactor batch variables names and junit util method

97b931d

igorbernstein2 reviewed Jun 4, 2019

View reviewed changes

yoshi-automation added the 🚨 This issue needs some love. label Jun 5, 2019

Updated with while await for batches in flush()

bc98e36

igorbernstein2 approved these changes Jun 6, 2019

View reviewed changes

yoshi-automation added the 🚨 This issue needs some love. label Jun 6, 2019

romlogic reviewed Jun 6, 2019

View reviewed changes

rahulKQL added 2 commits June 7, 2019 14:25

Address language related feedback

6383a04

Removing timeout time from Batcher#flush()

df02692

vam-google reviewed Jun 7, 2019

View reviewed changes

Addressing more feedback comments

27bb276

rahulKQL added 2 commits June 11, 2019 16:43

Updated BatcherImpl.Builder to AutoValue

3439c02

Now BatcherImpl is created with @autovalue. Also made `isClosed` field as volatile.

Removing @AutoValue from BatcherImpl class

05eb03c

vam-google approved these changes Jun 13, 2019

View reviewed changes

igorbernstein2 merged commit f4554ed into googleapis:master Jun 13, 2019

rahulKQL mentioned this pull request Jun 14, 2019

Implementing ElementCount and ElementBytes Triggers #734

Merged

chingor13 mentioned this pull request Jun 18, 2019

Release gax-java v1.46.0 #736

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Batcher implementation with Synchronized `Batcher#flush`(without triggers) #716

Batcher implementation with Synchronized `Batcher#flush`(without triggers) #716

rahulKQL commented May 29, 2019

rahulKQL commented May 29, 2019

codecov bot commented May 29, 2019 •

edited

Loading

igorbernstein2 left a comment

igorbernstein2 May 29, 2019

rahulKQL May 31, 2019

igorbernstein2 Jun 4, 2019

rahulKQL Jun 6, 2019

igorbernstein2 May 29, 2019

rahulKQL May 31, 2019

igorbernstein2 May 31, 2019

rahulKQL commented May 31, 2019

igorbernstein2 Jun 4, 2019

igorbernstein2 left a comment

igorbernstein2 Jun 6, 2019

rahulKQL Jun 7, 2019

igorbernstein2 Jun 7, 2019

rahulKQL Jun 7, 2019

vam-google Jun 7, 2019

vam-google Jun 7, 2019

igorbernstein2 Jun 7, 2019 •

edited

Loading

vam-google Jun 10, 2019

igorbernstein2 Jun 10, 2019

rahulKQL Jun 11, 2019

igorbernstein2 Jun 11, 2019

vam-google Jun 7, 2019

igorbernstein2 Jun 7, 2019

vam-google Jun 10, 2019

igorbernstein2 Jun 10, 2019

vam-google Jun 13, 2019

igorbernstein2 Jun 13, 2019

rahulKQL Jun 13, 2019

rahulKQL commented Jun 10, 2019

vam-google left a comment

igorbernstein2 commented Jun 13, 2019

Batcher implementation with Synchronized Batcher#flush(without triggers) #716

Batcher implementation with Synchronized Batcher#flush(without triggers) #716

Conversation

rahulKQL commented May 29, 2019

What this PR contains

Follow-ups

rahulKQL commented May 29, 2019

codecov bot commented May 29, 2019 • edited Loading

Codecov Report

igorbernstein2 left a 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

rahulKQL commented May 31, 2019

Choose a reason for hiding this comment

igorbernstein2 left a 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

igorbernstein2 Jun 7, 2019 • edited Loading

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

rahulKQL commented Jun 10, 2019

vam-google left a comment

Choose a reason for hiding this comment

igorbernstein2 commented Jun 13, 2019

Batcher implementation with Synchronized `Batcher#flush`(without triggers) #716

Batcher implementation with Synchronized `Batcher#flush`(without triggers) #716

codecov bot commented May 29, 2019 •

edited

Loading

igorbernstein2 Jun 7, 2019 •

edited

Loading