| 553 | } |
| 554 | |
| 555 | Future<Void> shardMerger(DataDistributionTracker* self, |
| 556 | KeyRange const& keys, |
| 557 | Reference<AsyncVar<Optional<ShardMetrics>>> shardSize) { |
| 558 | int64_t maxShardSize = self->maxShardSize->get().get(); |
| 559 | |
| 560 | auto prevIter = self->shards.rangeContaining(keys.begin); |
| 561 | auto nextIter = self->shards.rangeContaining(keys.begin); |
| 562 | |
| 563 | CODE_PROBE(true, "shard to be merged"); |
| 564 | ASSERT(keys.begin > allKeys.begin); |
| 565 | |
| 566 | // This will merge shards both before and after "this" shard in keyspace. |
| 567 | int shardsMerged = 1; |
| 568 | bool forwardComplete = false; |
| 569 | KeyRangeRef merged; |
| 570 | StorageMetrics endingStats = shardSize->get().get().metrics; |
| 571 | int shardCount = shardSize->get().get().shardCount; |
| 572 | double lastLowBandwidthStartTime = shardSize->get().get().lastLowBandwidthStartTime; |
| 573 | if (FLOW_KNOBS->DELAY_JITTER_OFFSET * SERVER_KNOBS->DD_MERGE_COALESCE_DELAY > |
| 574 | SERVER_KNOBS->DD_LOW_BANDWIDTH_DELAY && |
| 575 | now() - lastLowBandwidthStartTime < SERVER_KNOBS->DD_LOW_BANDWIDTH_DELAY) { |
| 576 | TraceEvent(g_network->isSimulated() ? SevError : SevWarnAlways, "ShardMergeTooSoon", self->distributorId) |
| 577 | .detail("Keys", keys) |
| 578 | .detail("LastLowBandwidthStartTime", lastLowBandwidthStartTime); |
| 579 | } |
| 580 | |
| 581 | int64_t systemBytes = keys.begin >= systemKeys.begin ? shardSize->get().get().metrics.bytes : 0; |
| 582 | |
| 583 | loop { |
| 584 | Optional<ShardMetrics> newMetrics; |
| 585 | if (!forwardComplete) { |
| 586 | if (nextIter->range().end == allKeys.end) { |
| 587 | forwardComplete = true; |
| 588 | continue; |
| 589 | } |
| 590 | ++nextIter; |
| 591 | newMetrics = nextIter->value().stats->get(); |
| 592 | |
| 593 | // If going forward, give up when the next shard's stats are not yet present, or if the |
| 594 | // the shard is already over the merge bounds. |
| 595 | if (!newMetrics.present() || shardCount + newMetrics.get().shardCount >= CLIENT_KNOBS->SHARD_COUNT_LIMIT || |
| 596 | (endingStats.bytes + newMetrics.get().metrics.bytes > maxShardSize)) { |
| 597 | --nextIter; |
| 598 | forwardComplete = true; |
| 599 | continue; |
| 600 | } |
| 601 | } else { |
| 602 | --prevIter; |
| 603 | newMetrics = prevIter->value().stats->get(); |
| 604 | |
| 605 | // If going backward, stop when the stats are not present or if the shard is already over the merge |
| 606 | // bounds. If this check triggers right away (if we have not merged anything) then return a trigger |
| 607 | // on the previous shard changing "size". |
| 608 | if (!newMetrics.present() || shardCount + newMetrics.get().shardCount >= CLIENT_KNOBS->SHARD_COUNT_LIMIT || |
| 609 | (endingStats.bytes + newMetrics.get().metrics.bytes > maxShardSize)) { |
| 610 | if (shardsMerged == 1) { |
| 611 | CODE_PROBE(true, "shardMerger cannot merge anything"); |
| 612 | return brokenPromiseToReady(prevIter->value().stats->onChange()); |
no test coverage detected