Merge disk space aware take 3 #127828
Merge disk space aware take 3 #127828
Conversation
| private void dropBudgetOverflow() { | ||
| synchronized (mutex) { | ||
| for (E element : budgetOverflow) { | ||
| budgetOverflowConsumer.accept(element); | ||
| } | ||
| budgetOverflow.clear(); | ||
| } | ||
| } |
There was a problem hiding this comment.
As soon as more budget becomes available, re-enqueue all the merge tasks that were over-budget.
It's complicated to re-enqueue only "within" available budget.
There was a problem hiding this comment.
That sounds reasonable to me though we'll need a combined view of the two queues (the budgetoverflow and the real queue) for autoscaling to trigger heap based autoscaling.
| this.executorService = threadPool.executor(ThreadPool.Names.MERGE); | ||
| this.maxConcurrentMerges = threadPool.info(ThreadPool.Names.MERGE).getMax(); | ||
| // the intent here is to throttle down whenever we submit a task and no other task is running | ||
| this.concurrentMergesFloorLimitForThrottling = 2; | ||
| this.concurrentMergesCeilLimitForThrottling = maxConcurrentMerges * 2; | ||
| assert concurrentMergesFloorLimitForThrottling <= concurrentMergesCeilLimitForThrottling; | ||
| this.budgetTracker = new BudgetTracker<>(MergeTask::estimatedRemainingMergeSize, Long.MAX_VALUE, queuedMergeTasks::add); |
There was a problem hiding this comment.
the last argument, i.e. queuedMergeTasks::add, is invoked for the over-budget merge tasks (as soon as more budget becomes available).
| // it's then the duty of the said merge scheduler to re-enqueue the backlogged merge task when it can be run | ||
| } | ||
| } catch (BudgetTracker.BudgetOverflowException e) { | ||
| // continue to poll for new merge tasks, over-budget merge task will be reenqueued |
There was a problem hiding this comment.
over-budget merge tasks have been "backlogged" until some more budget will become available, at which point all will be re-enqueued.
| synchronized (mutex) { | ||
| if (elementBudget > availableBudget) { | ||
| budgetOverflow.add(element); | ||
| throw new BudgetOverflowException(); |
There was a problem hiding this comment.
I think I'd prefer a return value, perhaps just null?
| private void dropBudgetOverflow() { | ||
| synchronized (mutex) { | ||
| for (E element : budgetOverflow) { | ||
| budgetOverflowConsumer.accept(element); | ||
| } | ||
| budgetOverflow.clear(); | ||
| } | ||
| } |
There was a problem hiding this comment.
That sounds reasonable to me though we'll need a combined view of the two queues (the budgetoverflow and the real queue) for autoscaling to trigger heap based autoscaling.
| } | ||
|
|
||
| void updateBudget(long availableBudget) { | ||
| assert availableBudget > 0; |
There was a problem hiding this comment.
Is == 0 not legal too?
| static class BudgetTracker<E> { | ||
| private final ToLongFunction<? super E> budgetFunction; | ||
| private long availableBudget; | ||
| private final IdentityHashMap<E, Long> commitedBudgetPerElement; |
There was a problem hiding this comment.
nit:
| private final IdentityHashMap<E, Long> commitedBudgetPerElement; | |
| private final IdentityHashMap<E, Long> committedBudgetPerElement; |
| private long availableBudget; | ||
| private final IdentityHashMap<E, Long> commitedBudgetPerElement; | ||
| private final List<E> budgetOverflow; | ||
| private final Consumer<? super E> budgetOverflowConsumer; |
There was a problem hiding this comment.
I am not sure I understand "overflow" in this name. Is it the requeueConsumer? Or maybe the moreBudgetAvailableConsumer?
No description provided.