Issue 3070: Fix bug where checkAllLedgers gets stuck when read throttling is enabled

[massakam]

Motivation

The read throttling feature in Auditor#checkAllLedgers added by #2973, but when it was enabled checkAllLedgers got stuck.

The cause is that, as mentioned in #3070, all the BookKeeperClientWorker-OrderedExecutor threads wait for the semaphore to be released, and there is no thread that can execute EntryExistsCallback#readEntryComplete where the semaphore is released.

Changes

Add a new thread pool to the LedgerChecker class. The BookKeeperClientWorker-OrderedExecutor threads threads delegate subsequent processing (this includes waiting for the semaphore to be released) to threads in this thread pool as soon as they release the semaphore.

This will prevent the BookKeeperClientWorker-OrderedExecutor from getting stuck due to waiting for the semaphore to be released.

Master Issue: #3070

[massakam]

rerun failure checks

[dlg99]

@massakam is it possible to avoid the extra executor?

I think the problem is that checkFragments/verifyLedgerFragment tries to acquire a permit from a callback in bookieClient.readEntry call; but there is already a permit request before that call. I think we can simply add a flag to not request/release extra permit in checkFragments/verifyLedgerFragment call chain in this case.

[massakam]

rerun failure checks

[massakam]

rerun failure checks

[massakam]

rerun failure checks

[massakam]

rerun failure checks

[massakam]

@dlg99 Changed the implementation to achieve throttling using RateLimiter instead of Semaphore. What do you think?

I think this is simpler than the implementation of adding a new flag and requesting/releasing a permit according to its value.

[gaozhangmin]

@dlg99 Changed the implementation to achieve throttling using RateLimiter instead of Semaphore. What do you think?

I think this is simpler than the implementation of adding a new flag and requesting/releasing a permit according to its value.

+1

[massakam]

@dlg99 PTAL

[eolivelli]

LGTM

[dlg99]

@massakam I think that Throttle is not a good approach here.

Requests-in-progress (RIPs) approach is very similar when everything works but easier to tune and self-adjusts to remote system's dynamically changing capabilities while throttle is impossible to tuen for that.
Imagine you configured 20 RIPs and one request takes 100ms normally. Now you have 200 RPS (requests per second).
Remote system (bookie) slowed down and now take 500ms to process a request, now you are down to 40RPS without overloading the bookie, changing config etc.
Bookie speeds up and can process request in 10ms - the rate dynamically goes up to 2000rps.

Let's use throttle now: set it to 200 request per second.
In the normal case everything looks ok and the same as in normal case above.
Now the bookie slows down.
Throttle keeps on letting through 200 RPS even though the bookie can only handle 40, with that requests will pile up and eventually make result in OOM on bookie or client side.
Bookie speeds up and can process request in 10ms - the rate stays at 200rps.

I'd prefer to track down where the semaphore is not released and fix that.

Throttle is simple, does not require release etc. but it has its drawback and was not used there for a reason.

[massakam]

@dlg99 Hmm... you have a point there.

However, I once tried to fix it in the way you suggested first, but I gave it up because the code became complicated and there was a high risk of missing the release of the acquired permit. In this approach, if an additional read is required in EntryExistsCallback#readEntryComplete(), it is necessary not to release the permit immediately, but after the subsequent read is complete.

I think it would be better to move to throttling with a rate limiter rather than to keep the RIPs approach and leaving the possibility of getting stuck.

[massakam]

Another option I have is to revert to the first approach with the extra executor.

[hangc0276]

The PR #2973 has been released in BookKeeper 4.15.0, we'd better mark the old one as deprecated instead of removing it. And set the old value to the new one if the user sets the old one.

[massakam]

@hangc0276 I don't think the value of inFlightReadEntryNumInLedgerChecker should be set to readEntryRateInLedgerChecker, as the two settings have slightly different meanings. Therefore, I decided to restore the old setting. If both readEntryRateInLedgerChecker and inFlightReadEntryNumInLedgerChecker are set, readEntryRateInLedgerChecker takes precedence and inFlightReadEntryNumInLedgerChecker is ignored.

[massakam]

PTAL

[massakam]

ping @dlg99 @hangc0276

[massakam]

@dlg99

This gist https://gist.github.com/dlg99/505849e1010a20c6d439ecd53f500a85 is more of a demo of what's going on than actual fix (after all, it breaks testShouldGetTwoFrgamentsIfTwoBookiesFailedInSameEnsemble - might be a test issue).

I have a question about this gist. Why is releasePermit() added on line 79 of LedgerChecker.java?
https://gist.github.com/dlg99/505849e1010a20c6d439ecd53f500a85#file-pr3214-diff-L79

EntryExistsCallback#readEntryComplete also calls releasePermit(), so it seems that one extra permit is released.

bookkeeper/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerChecker.java

Line 321 in 7004d99

releasePermit();

And if releasePermit() on line 75 of LedgerChecker.java is removed, TestLedgerChecker times out.

[massakam]

rerun failure checks

[massakam]

rerun failure checks

[massakam]

rerun failure checks

[dlg99]

@massakam

Why is releasePermit() added on line 79 of LedgerChecker.java?

It is in the callback.
Permits requested before bookieClient.readEntry called, and must be released in the callback passed there.

                final long entryToRead = curEntryId;

                final EntryExistsCallback eecb = new EntryExistsCallback(lh.getLedgerMetadata().getWriteQuorumSize(),
                                              new GenericCallback<Boolean>() {
                                                  @Override
                                                  public void operationComplete(int rc, Boolean result) {
                                                      releasePermit();
                                                      if (result) {
                                                          fragments.add(lastLedgerFragment);
                                                      }
                                                      checkFragments(fragments, cb,
                                                          percentageOfLedgerFragmentToBeVerified);
                                                  }
                                              });

                try {
                    DistributionSchedule.WriteSet writeSet = lh.getDistributionSchedule().getWriteSet(entryToRead);
                    acquirePermits(writeSet.size());
                    for (int i = 0; i < writeSet.size(); i++) {
                            BookieId addr = curEnsemble.get(writeSet.get(i));
                            bookieClient.readEntry(addr, lh.getId(), entryToRead,
                                    eecb, null, BookieProtocol.FLAG_NONE);
                    }
                    writeSet.recycle();
                } catch (InterruptedException e) {
                    LOG.error("InterruptedException when checking entry : {}", entryToRead, e);
                }

[massakam]

@dlg99 That callback method is called from EntryExistsCallback#readEntryComplete. releasePermit() is executed numReads times in EntryExistsCallback#readEntryComplete, so releasePermit() will be executed numReads + 1 times in total.

    private class EntryExistsCallback implements ReadEntryCallback {
        AtomicBoolean entryMayExist = new AtomicBoolean(false);
        final AtomicInteger numReads;
        final GenericCallback<Boolean> cb;

        EntryExistsCallback(int numReads,
                            GenericCallback<Boolean> cb) {
            this.numReads = new AtomicInteger(numReads);
            this.cb = cb;
        }

        @Override
        public void readEntryComplete(int rc, long ledgerId, long entryId,
                                      ByteBuf buffer, Object ctx) {
            releasePermit(); // Executed `numReads` times
            if (BKException.Code.NoSuchEntryException != rc && BKException.Code.NoSuchLedgerExistsException != rc
                    && BKException.Code.NoSuchLedgerExistsOnMetadataServerException != rc) {
                entryMayExist.set(true);
            }

            if (numReads.decrementAndGet() == 0) {
                cb.operationComplete(rc, entryMayExist.get()); // Call `GenericCallback#operationComplete` and execute `releasePermit()` one more time
            }
        }
    }

In fact, when I logged the value of semaphore.availablePermits() after the test was completed, it was inFlightReadEntryNum + 1. But it should return to inFlightReadEntryNum.

[dlg99]

@massakam you are absolutely right, that was an extra release.
So, we should not call checkFragments() in the callback that runs on the bookie's ordered executor (acquire blocks a chance to run of the bookie callback that releases permit).
https://gist.github.com/dlg99/3e1524f0804c4688ccb31e3300b89c4e basically your initial idea with the executor, needs other tests to close LedgerChecker.
This one: https://gist.github.com/dlg99/28f07f3bcd3f71000b1faba6b906fb09 is kind of the same but uses common pool executor (via whenCompleteAsync)

[massakam]

rerun failure checks

[massakam]

rerun failure checks

[massakam]

rerun failure checks

[dlg99]

Thank you!

[massakam]

This one: https://gist.github.com/dlg99/28f07f3bcd3f71000b1faba6b906fb09 is kind of the same but uses common pool executor (via whenCompleteAsync)

Adopted this approach and modified the code.

[massakam]

@hangc0276 @rdhabalia @eolivelli @StevenLuMT PTAL. And if there is no problem, please merge this.

[eolivelli]

@massakam
Merged, thanks !