feat: dynamic flow control for batcher part 2 #1310

mutianf · 2021-02-22T14:53:16Z

Implementation of go/veneer-dynamic-flow-control part 2.
Adding a DynamicFlowControlSetting and new aPIs in FlowController so the flow control limits in the FlowController can be adjusted.

codecov · 2021-02-22T14:55:34Z

Codecov Report

Merging #1310 (39fb9fe) into master (7f7aa25) will increase coverage by 0.06%.
The diff coverage is 86.74%.

@@             Coverage Diff              @@
##             master    #1310      +/-   ##
============================================
+ Coverage     80.27%   80.34%   +0.06%     
- Complexity     1304     1331      +27     
============================================
  Files           209      210       +1     
  Lines          5547     5682     +135     
  Branches        479      519      +40     
============================================
+ Hits           4453     4565     +112     
- Misses          905      912       +7     
- Partials        189      205      +16

Impacted Files	Coverage Δ	Complexity Δ
.../google/api/gax/batching/NonBlockingSemaphore.java	`76.31% <74.07%> (-4.94%)`	`11.00 <6.00> (+6.00)`	⬇️
...e/api/gax/batching/DynamicFlowControlSettings.java	`83.72% <83.72%> (ø)`	`2.00 <2.00> (?)`
...com/google/api/gax/batching/BlockingSemaphore.java	`81.81% <85.18%> (+0.86%)`	`12.00 <6.00> (+6.00)`
...va/com/google/api/gax/batching/FlowController.java	`88.49% <93.84%> (+2.13%)`	`34.00 <22.00> (+14.00)`
.../java/com/google/api/gax/batching/BatcherImpl.java	`94.94% <100.00%> (-1.13%)`	`22.00 <0.00> (-1.00)`

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 7f7aa25...39fb9fe. Read the comment docs.

mutianf · 2021-02-22T22:47:16Z

@igorbernstein2 The second part is ready for review. This part adds the DynamicFlowControlSettings and makes flow control limits in the FlowController adjustable. Thanks! :)

elharo

There are some common style issues here that I haven't noted on every occurrence, but please fix globally.

gax/src/main/java/com/google/api/gax/batching/DynamicFlowControlSettings.java

gax/src/main/java/com/google/api/gax/batching/NonBlockingSemaphore.java

gax/src/test/java/com/google/api/gax/batching/DynamicFlowControlSettingsTest.java

gax/src/test/java/com/google/api/gax/batching/Semaphore64Test.java

gax/src/main/java/com/google/api/gax/batching/BlockingSemaphore.java

gax/src/main/java/com/google/api/gax/batching/DynamicFlowControlSettings.java

gax/src/main/java/com/google/api/gax/batching/FlowController.java

igorbernstein2 · 2021-03-01T18:27:22Z

gax/src/main/java/com/google/api/gax/batching/FlowController.java

-      if (!outstandingByteCount.acquire(permitsToDraw)) {
+      long permitsToDraw, permitsOwed;
+      boolean acquired;
+      synchronized (this) {


Please avoid locking on this, its better to use a private lock Object

gax/src/main/java/com/google/api/gax/batching/FlowController.java

gax/src/main/java/com/google/api/gax/batching/NonBlockingSemaphore.java

vam-google

The concurrency part looks a bit scary to me (it seems like it can unintentionally an indirectly break pretty much everything else in gax).

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

gax/src/main/java/com/google/api/gax/batching/BlockingSemaphore.java

vam-google · 2021-03-04T22:35:38Z

gax/src/main/java/com/google/api/gax/batching/BlockingSemaphore.java

-        wait();
-      } catch (InterruptedException e) {
-        interrupted = true;
+    while (!interrupted) {


(Partially this comment applies even to the original implementation, pushed in part 1, which I LGTM'ed but I missed it during reiew of part 1)

It looks like the only way the thread, which entered this loop, can exit is is by receiving interrupt signal. It looks very dangerous not only for correctness of this new feature but for correctness of the rest of gax in an indirect way.

In general, pretty much all the code in gax-java is executed in the context of the thread executor pool, which has 4 threads by default (or number of current CPUs if > 4). My biggest concern is that once code enters this block, it means one of those 4 threads is completely owned by this block and there are only two possible ways out:

this code will never end and will keep attempting to wait forever;

this thread gets interrupted, then the execution reaches line 91 and Thread.currentTrhead.interrupt() is called.

Then it is not clear what will happen with this thread after interrupt() is called, as it only sets the interrupt flag to true, and does not guarantee that the thread does not get "reinterrupted" (interrupt flag cleared).

Also, if the thread actually gets terminated, it means it will have to be re-created in the executor pool, basically defeating the purpose of using thread pool (compared to simply spawning new thread for every shingle task). In other words this code block messes with executors pool in a very brutal way:

4 parallel entries in this block and the executor pool is dead (assuming 4 is the number of configured threads)

if the execution flow assumes frequent entering/leaving this code block, it would result into frequent thread_interrupt->thread_kill->thread_create, which is a heavy operation and is not how threads arfe supposed to be used in the context of a thread pool (threads must be shared by tasks, and tasks should never dictate the lifecycle of a thread it is executed in).

Basically this code block does not "respect" the other potential tasks in the pool, assuming that the thread it executes in now belongs to it and will never be released back to the pool.

@igorbernstein2 PTAL.

Reverted back to the original logic, which will wait for the permits and interrupt. And it has always been implemented this way https://github.com/googleapis/gax-java/blame/master/gax/src/main/java/com/google/api/gax/batching/BlockingSemaphore.java#L59-L70

4 parallel entries in this block and the executor pool is dead

This class is used in FlowController, so wherever it's called, it'll want to block the thread from doing anything for throttling?

if the execution flow assumes frequent entering/leaving this code block, it would result into frequent thread_interrupt->thread_kill->thread_create, which is a heavy operation and is not how threads are supposed to be used in the context of a thread pool

Sorry again for the confusing changes from using 2 locks (which I've reverted back). But in both implementations, the expected way of leaving this code block is when there are more permits to give to the caller. And interrupt should rarely happen.

gax/src/main/java/com/google/api/gax/batching/BlockingSemaphore.java

vam-google

(forgot to select "request changes" in previous review, please check the comments from above for context)

vam-google

Overall looks good, but the concurrency/synchronization part is still concerning, unfortunately.

gax/src/main/java/com/google/api/gax/batching/FlowController.java

gax/src/main/java/com/google/api/gax/batching/NonBlockingSemaphore.java

vam-google · 2021-03-11T23:15:55Z

gax/src/main/java/com/google/api/gax/batching/FlowController.java

+  @Nullable private final Long maxRequestBytesLimit;
+  @Nullable private final Long minElementCountLimit;
+  @Nullable private final Long minRequestBytesLimit;
+  @Nullable private Long currentElementCountLimit;


Since these are used in multithreaded environment, it would be a good practice to declare them volatile.

Also, given that these values are used only in simple context, it might be possible to resort to using non-blocking approach (and convert this to AtomicLong)

~~currentElementCountLimit is modified and accessed in synchronized blocks (will add synchronize to the getters), so is it still necessary to make it volatile? Or is it just to make the code clear?~~
Moved the limit to the semaphore class

gax/src/main/java/com/google/api/gax/batching/BlockingSemaphore.java

igorbernstein2

This looks good to me...just a couple of nits on naming

gax/src/main/java/com/google/api/gax/batching/BlockingSemaphore.java

gax/src/main/java/com/google/api/gax/batching/DynamicFlowControlSettings.java

gax/src/main/java/com/google/api/gax/batching/NonBlockingSemaphore.java

igorbernstein2

lgtm!

vam-google

Some more comments about multithreading part, but LGTM =)

vam-google · 2021-03-19T20:09:11Z

gax/src/main/java/com/google/api/gax/batching/NonBlockingSemaphore.java

+  @Override
+  public void increasePermitLimit(long permits) {
+    checkNotNegative(permits);
+    availablePermits.addAndGet(permits);


This looks suspicious. Even though both availablePermits and limit are atomic, their joint update here is not atomic (if they were under the same lock, it would be). Is it ok here?

Very good point and I struggled with it a bit but I think it's ok:

If multiple threads are calling increasePermitsLimit, this is ok because updates to the variables are atomic and we just need to make sure the numbers add up.

If it happens concurrently with reduceLimit (which shouldn't really happen in reality), we need to make sure limit don't go < 0, and the checks in reducePermit should handle that.

If increasePermitLimit happens concurrently with acquire, limit is not accessed / updated in acquire, and availablePermits is atomic, so it should be fine.

If increasePermitLimit happens concurrently with acquirePartial, limit is accessed to check if there's enough permits. If limit is updated after if (old < Math.min(limit.get(), permits)), it means either availablePermits was also updated midway so if (availablePermits.compareAndSet(old, old - permits)) won't succeed and it'll loop one more time, or availablePermits is also updated and it passed / failed the check.

vam-google · 2021-03-19T20:10:06Z

gax/src/main/java/com/google/api/gax/batching/NonBlockingSemaphore.java

+      long oldLimit = limit.get();
+      Preconditions.checkState(oldLimit - reduction > 0, "permit limit underflow");
+      if (limit.compareAndSet(oldLimit, oldLimit - reduction)) {
+        availablePermits.addAndGet(-reduction);


Should we call aquirePartial() here instead to update availablePermits() in a non-blocking algorithm fashion, instead of updating it directly?

🤔 I think it's pretty safe the just update it, calling aquirePartial() seems to work too but I feel it's a bit harder to read?

vam-google · 2021-03-19T20:13:06Z

gax/src/main/java/com/google/api/gax/batching/BlockingSemaphore.java

+    checkNotNegative(reduction);
+    Preconditions.checkState(limit - reduction > 0, "permit limit underflow");
+    availablePermits -= reduction;
+    limit -= reduction;


I see increasePermitLimit calls notifyAll(), but reduce does not. Looks suspicious, as it is a reverse operation to increase (so I would assume it should follow the same principles).

It should be ok here, because the wait() in other threads are waiting for more permits to become available, while this method removes permits. So even if we call notifyAll() here, the threads are likely to go back to sleep again :). It's similar to how we don't call notifyAll() in reserve().

igorbernstein2 · 2021-03-22T14:42:26Z

@vam-google I'm going to merge this, if you still have concerns after @mutianf responses, we can open follow up PRs.

google-cla bot added the cla: yes This human has signed the Contributor License Agreement. label Feb 22, 2021

mutianf marked this pull request as ready for review February 22, 2021 22:45

mutianf requested review from a team as code owners February 22, 2021 22:45

elharo reviewed Feb 23, 2021

View reviewed changes

mutianf force-pushed the dynamic_flow_control_p2 branch from 8566969 to 53d8d7e Compare February 25, 2021 23:46

elharo approved these changes Feb 26, 2021

View reviewed changes

igorbernstein2 reviewed Mar 1, 2021

View reviewed changes

mutianf force-pushed the dynamic_flow_control_p2 branch from 868e6aa to 9064e26 Compare March 4, 2021 16:26

vam-google reviewed Mar 4, 2021

View reviewed changes

vam-google suggested changes Mar 4, 2021

View reviewed changes

vam-google suggested changes Mar 11, 2021

View reviewed changes

vam-google reviewed Mar 11, 2021

View reviewed changes

elharo reviewed Mar 12, 2021

View reviewed changes

gax/src/main/java/com/google/api/gax/batching/BlockingSemaphore.java Show resolved Hide resolved

mutianf force-pushed the dynamic_flow_control_p2 branch from 677e470 to b7b525d Compare March 15, 2021 19:12

mutianf added 13 commits March 16, 2021 16:49

feat: dynamic flow control for batcher part 2

8cf73e0

add test on construtor for code coverage

1906ea2

Update comments based on the review

0f62981

missed some error messages in test

c4d3337

Fix possible deadlock

692099a

fix comment

722d775

update comments

2401993

update naming

80f70c9

removing setThreshold because there's no use case for it

517ceb7

no need to use 2 locks on semaphore

d374225

fix comment

cfe12b9

fix test name

eac770d

make nonblocking semaphore not blocking

c1a0543

move limit to semaphore class

96ef367

mutianf force-pushed the dynamic_flow_control_p2 branch from d83fdc1 to 96ef367 Compare March 16, 2021 16:51

add an addPermits method

59f29e9

igorbernstein2 suggested changes Mar 17, 2021

View reviewed changes

fixing naming, don't throw on overflow and add some tests

a4067ce

igorbernstein2 approved these changes Mar 18, 2021

View reviewed changes

vam-google approved these changes Mar 19, 2021

View reviewed changes

Merge branch 'master' into dynamic_flow_control_p2

39fb9fe

igorbernstein2 merged commit 20f6ecf into googleapis:master Mar 22, 2021

mutianf deleted the dynamic_flow_control_p2 branch March 22, 2021 14:56

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: dynamic flow control for batcher part 2 #1310

feat: dynamic flow control for batcher part 2 #1310

mutianf commented Feb 22, 2021

codecov bot commented Feb 22, 2021 •

edited

mutianf commented Feb 22, 2021

elharo left a comment

igorbernstein2 Mar 1, 2021

vam-google left a comment

vam-google Mar 4, 2021 •

edited

vam-google Mar 4, 2021

mutianf Mar 8, 2021 •

edited

vam-google left a comment •

edited

vam-google left a comment

vam-google Mar 11, 2021

mutianf Mar 12, 2021 •

edited

igorbernstein2 left a comment

igorbernstein2 left a comment

vam-google left a comment

vam-google Mar 19, 2021

mutianf Mar 19, 2021

vam-google Mar 19, 2021

mutianf Mar 19, 2021

vam-google Mar 19, 2021

mutianf Mar 19, 2021

igorbernstein2 commented Mar 22, 2021

feat: dynamic flow control for batcher part 2 #1310

feat: dynamic flow control for batcher part 2 #1310

Conversation

mutianf commented Feb 22, 2021

codecov bot commented Feb 22, 2021 • edited

Codecov Report

mutianf commented Feb 22, 2021

elharo left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vam-google left a comment

Choose a reason for hiding this comment

vam-google Mar 4, 2021 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mutianf Mar 8, 2021 • edited

Choose a reason for hiding this comment

vam-google left a comment • edited

Choose a reason for hiding this comment

vam-google left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mutianf Mar 12, 2021 • edited

Choose a reason for hiding this comment

igorbernstein2 left a comment

Choose a reason for hiding this comment

igorbernstein2 left a comment

Choose a reason for hiding this comment

vam-google 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 commented Mar 22, 2021

codecov bot commented Feb 22, 2021 •

edited

vam-google Mar 4, 2021 •

edited

mutianf Mar 8, 2021 •

edited

vam-google left a comment •

edited

mutianf Mar 12, 2021 •

edited