[ManagedLedger] Pin executor and scheduled executor threads for ManagedLedgerImpl #11387
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -93,7 +93,7 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) { | |
|
|
||
| if (!entries.isEmpty()) { | ||
| // There were already some entries that were read before, we can return them | ||
| cursor.ledger.getPinnedExecutor().execute(safeRun(() -> { | ||
|
There was a problem hiding this comment. I think we should be careful in not serializing every cursor into the managed ledger pinned thread, as it could become a bottleneck where there are many cursors on a topic. There was a problem hiding this comment. Yes that's true. The reason to use the pinned executor is to adhere to Java Memory Model rules of correct synchronization. There's a generic problem in OpReadEntry since it's sharing an array that is mutated by multiple threads. JLS 17.4 explains that "Incorrectly Synchronized Programs May Exhibit Surprising Behavior". I would assume that "entries" would have to be copied to a new list before sharing if we want to use multiple threads. Is that right? There was a problem hiding this comment. I think the entry reading happens one by one, if we got the read entries failed here, this means we will not get a chance to add more elements to the list right(all the previous read operations are done)? |
||
| callback.readEntriesComplete(entries, ctx); | ||
| recycle(); | ||
| })); | ||
|
|
@@ -141,8 +141,8 @@ void checkReadCompletion() { | |
| cursor.ledger.startReadOperationOnLedger(nextReadPosition, OpReadEntry.this); | ||
| } | ||
|
|
||
|
There was a problem hiding this comment. This is a behaviour change There was a problem hiding this comment.
testing, testing, testing. we need more Fallout tests. :) |
||
| // Schedule next read | ||
| cursor.ledger.getPinnedExecutor().execute(safeRun(() -> { | ||
|
There was a problem hiding this comment. Other than the consideration that different cursors shouldn't be pinned on a single thread, the reason for jumping to a different thread here is to avoid a stack overflow. When the read is being served from the ML cache, it's coming back from same thread. There are some conditions in which we ask for next read. eg. If you ask to read 100 entries and we only got 20 entries from current ledger, we'll schedule a read for the remaining 80 on next ledger. In some cases there could be abnormal distributions, like 1 entry per ledger and it would be chaining all the reads and callback within the same stack. Therefore, the "jump to a random thread" was introduced to break that chain. There was a problem hiding this comment. Wouldn't the usage of the pinned executor achieve the same result? It prevents the stack from going deeper and deeper. The only reason that comes into my mind is the case where there's a completable future that gets triggered as part of the call flow and it is being waited to complete in the same thread as where the result should be executed in. That would never complete and would dead lock. Would that be the case here to use a different executor? There was a problem hiding this comment. It looks like the stack should be There was a problem hiding this comment.
@lhotari Uhm, I think that some executors are short-circuiting the queue if they detect that you're trying to add a task from the same executor thread. That is the case for Netty IO thread, though I just check that it shouldn't happen on the |
||
| readPosition = cursor.ledger.startReadOperationOnLedger(nextReadPosition, OpReadEntry.this); | ||
| cursor.ledger.asyncReadEntries(OpReadEntry.this); | ||
| })); | ||
|
|
@@ -152,7 +152,7 @@ void checkReadCompletion() { | |
| cursor.readOperationCompleted(); | ||
|
|
||
| } finally { | ||
| cursor.ledger.getPinnedExecutor().execute(safeRun(() -> { | ||
| callback.readEntriesComplete(entries, ctx); | ||
| recycle(); | ||
| })); | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
If we are using 2 threads, one with the regular executor (which is more efficient) and the other for the
pinnedScheduledExecutor, wouldn't that mean that we still have more than 1 thread accessing some of the objects?Would it make sense to use the generic
scheduledExecutor(just for deferring purposes) and then jump back into the samepinnedExecutor?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
That's true.
Perhaps a more optimal solution would be to have the capability for scheduling tasks on the pinned scheduler. I don't know why this solution isn't available in the underlying Bookkeeper libraries that are used. The benefit of that is that there isn't an additional thread switch when the scheduled task triggers.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@merlimat Do you mean scheduledExecutor.schedule(pinnedExecutor.execute() ...) ?
Seems to be a feasible way right now :)
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@lhotari The scheduled executor is less efficient compared to the normal executor because it has to maintain the delayed tasks. For that it's preferable not to use it directly in the critical data path, but only when we want to defer actions or for background tasks.