CalvinKirs commented on code in PR #66717:
URL: https://github.com/apache/doris/pull/66717#discussion_r3827276525
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCacheEntry.java:
##########
@@ -198,62 +552,738 @@ public MetaCacheEntryStats stats() {
failureCount,
totalLoadTime,
totalLoadCount == 0 ? 0D : (double) totalLoadTime /
totalLoadCount,
- cacheStats.evictionCount(),
+ MetaCacheWeightUtils.saturatedAdd(
+ cacheStats.evictionCount(), localEvictionCount.get()),
invalidateCount.get(),
lastLoadSuccessTimeMs.get(),
lastLoadFailureTimeMs.get(),
- lastError.get());
+ lastError.get(),
+ weightBounded,
+ weightBounded ? cacheSpec.getMaxWeight().getAsLong() : -1L,
+ weightBounded ? entryBudget.getUsedWeight() : -1L,
+ weightBounded ? MetaCacheWeightUtils.saturatedAdd(
+ automaticEvictionWeight.get(),
localEvictionWeight.get()) : -1L,
+ weightBounded ? weightAdmissionRejectedCount.get() : -1L,
+ weightBounded ? entryBudget.getCatalogMaxWeight() : -1L,
+ weightBounded ? entryBudget.getCatalogUsedWeight() : -1L,
+ weightBounded ? entryBudget.getGlobalMaxWeight() : -1L,
+ weightBounded ? entryBudget.getGlobalUsedWeight() : -1L,
+ weightBounded ? lastWeightRejectReason.get() : "");
+ }
+
+ public boolean isWeightBounded() {
+ return weightBounded;
+ }
+
+ /** True when this entry stores values at all (enabled with a positive
capacity or weight). */
+ public boolean isEffectivelyEnabled() {
+ return effectiveEnabled;
+ }
+
+ /**
+ * True when publication sizing can lead to a weighted admission: an entry
may be configured
+ * with a weight bound yet be ineffective (max-weight 0, disabled, zero
TTL or capacity), in
+ * which case preparing values for publication is pure waste.
+ */
+ public boolean isWeightAccounting() {
+ return weightBounded && effectiveEnabled;
+ }
+
+ private AdmissionResult admitWeightedValue(
+ K key, V value, @Nullable V expectedCurrent, boolean
requireExpected,
+ @Nullable KeyMutationToken expectedMutation, long
expectedReservationGeneration,
+ boolean advanceMutationOnAdmission) {
+ if (closed.get()) {
+ return AdmissionResult.DISABLED;
+ }
+ MetaCacheSizeEstimate estimate;
+ try {
+ estimate = Objects.requireNonNull(sizeEstimator.estimate(key,
value), "size estimate");
+ } catch (IllegalArgumentException e) {
+ throw e;
+ } catch (RuntimeException e) {
+ rejectWeight("invalid_estimate");
+ return AdmissionResult.REJECTED;
+ }
+ if (!estimate.isComplete()) {
+ rejectWeight(estimate.getIncompleteReason());
+ return AdmissionResult.REJECTED;
+ }
+
+ long estimatedPayloadBytes = estimate.getBytes();
+ // A retained non-null key/value plus Caffeine node can never consume
zero bytes. Treat a
+ // complete zero as an estimator contract violation so an omitted
formula cannot bypass
+ // every quota and admit an unbounded number of zero-weight entries.
+ if (estimatedPayloadBytes == 0L) {
+ rejectWeight("invalid_estimate");
+ return AdmissionResult.REJECTED;
+ }
+ long newWeight = MetaCacheWeightUtils.saturatedAdd(
+ estimatedPayloadBytes, FIXED_ENTRY_ACCOUNTING_OVERHEAD_BYTES);
+ synchronized (admissionLock) {
+ if (closed.get()) {
+ return AdmissionResult.DISABLED;
+ }
+ if (expectedMutation != null && !isKeyMutationCurrent(key,
expectedMutation)) {
+ return AdmissionResult.NOT_CURRENT;
+ }
+ V oldValue = data.asMap().get(key);
+ ReservationRecord record = reservations.get(key);
+ if (expectedReservationGeneration >= 0L
+ && (record == null || record.generation !=
expectedReservationGeneration)) {
+ return AdmissionResult.NOT_CURRENT;
+ }
+ if (requireExpected && oldValue != expectedCurrent) {
+ return AdmissionResult.NOT_CURRENT;
+ }
+ if (oldValue == null && record != null) {
+ // The previous generation was already removed by Caffeine; if
that removal was
+ // an eviction whose asynchronous cleanup has not run yet,
account it here.
+ if (pendingEvictionGenerations.remove(key, record.generation))
{
+ automaticEvictionWeight.accumulateAndGet(
+ record.weight, MetaCacheWeightUtils::saturatedAdd);
+ }
+ reservations.remove(key, record);
+ record.reservation.release();
+ record = null;
+ }
+ if (oldValue != null && (record == null || !record.published)) {
+ rejectWeight("missing_reservation");
+ return AdmissionResult.REJECTED;
+ }
+
+ if (record == null) {
+ Optional<AdmissionReservation> reservation =
reserveWithLocalEviction(key, newWeight);
+ if (!reservation.isPresent()) {
+ rejectWeight("budget_exceeded");
+ return AdmissionResult.REJECTED;
+ }
+ ReservationRecord newRecord = new ReservationRecord(
+ newWeight, reservation.get(),
nextReservationGeneration());
+ if (advanceMutationOnAdmission) {
+ advanceKeyMutation(key);
+ }
+ reservations.put(key, newRecord);
+ try {
+ beforeWeightedCachePutForTest(key, value);
+ data.put(key, value);
+ if (reservations.get(key) == newRecord &&
data.asMap().get(key) == value) {
+ newRecord.published = true;
+ notifyReplacement(key, null, value);
+ }
+ return AdmissionResult.ADMITTED;
+ } catch (RuntimeException | Error e) {
+ reservations.remove(key, newRecord);
+ newRecord.reservation.release();
+ throw e;
+ }
+ }
+
+ ReservationRecord previousRecord = record;
+ long reservedWeight = Math.max(previousRecord.weight, newWeight);
+ if (!resizeWithLocalEviction(key, previousRecord.reservation,
reservedWeight)) {
+ rejectWeight("budget_exceeded");
+ return AdmissionResult.REJECTED;
+ }
+ if (advanceMutationOnAdmission) {
+ advanceKeyMutation(key);
+ }
+ ReservationRecord newRecord = new ReservationRecord(
+ newWeight, previousRecord.reservation,
nextReservationGeneration());
+ reservations.put(key, newRecord);
+ try {
+ beforeWeightedCachePutForTest(key, value);
+ data.put(key, value);
+ boolean retained = reservations.get(key) == newRecord &&
data.asMap().get(key) == value;
+ if (retained) {
+ newRecord.published = true;
+ }
+ if (retained && reservedWeight != newWeight &&
!newRecord.reservation.tryResize(newWeight)) {
+ throw new IllegalStateException("failed to release cache
replacement reservation delta");
+ }
+ if (retained) {
+ notifyReplacement(key, oldValue, value);
+ }
+ return AdmissionResult.ADMITTED;
+ } catch (RuntimeException | Error e) {
+ if (reservations.replace(key, newRecord, previousRecord)) {
+ if (data.asMap().get(key) == null) {
+ reservations.remove(key, previousRecord);
+ previousRecord.reservation.release();
+ } else if
(!previousRecord.reservation.tryResize(previousRecord.weight)) {
+ throw new IllegalStateException("failed to roll back
cache replacement reservation", e);
+ }
+ }
+ throw e;
+ }
+ }
+ }
+
+ private Optional<AdmissionReservation> reserveWithLocalEviction(K
incomingKey, long bytes) {
+ if (bytes > entryBudget.getEffectiveMaxWeight()) {
+ return Optional.empty();
+ }
+ Optional<AdmissionReservation> reservation =
entryBudget.tryReserve(bytes);
+ while (!reservation.isPresent()) {
+ AtomicReference<Optional<AdmissionReservation>> retried =
+ new AtomicReference<>(Optional.empty());
+ int evicted = evictLocalColdest(incomingKey,
LOCAL_EVICTION_BATCH_SIZE, () -> {
+ Optional<AdmissionReservation> attempt =
entryBudget.tryReserve(bytes);
+ retried.set(attempt);
+ return attempt.isPresent();
+ });
+ reservation = retried.get();
+ if (reservation.isPresent()) {
+ break;
+ }
+ if (evicted == 0) {
+ entryBudget.requestPeerReclaim(bytes);
+ break;
+ }
+ }
+ return reservation;
+ }
+
+ private boolean resizeWithLocalEviction(K incomingKey,
AdmissionReservation reservation, long newBytes) {
+ if (newBytes > entryBudget.getEffectiveMaxWeight()) {
+ return false;
+ }
+ if (reservation.tryResize(newBytes)) {
+ return true;
+ }
+ while (true) {
+ AtomicBoolean resized = new AtomicBoolean();
+ int evicted = evictLocalColdest(incomingKey,
LOCAL_EVICTION_BATCH_SIZE, () -> {
+ if (reservation.tryResize(newBytes)) {
+ resized.set(true);
+ return true;
+ }
+ return false;
+ });
+ if (resized.get()) {
+ return true;
+ }
+ if (evicted == 0) {
+ entryBudget.requestPeerReclaim(Math.max(0L, newBytes -
reservation.getBytes()));
+ return false;
+ }
+ }
+ }
+
+ /**
+ * Evict up to {@code limit} coldest values, retrying the caller's goal
after every single
+ * eviction: a skewed cache where one large cold value creates all needed
headroom must not
+ * discard the rest of the selected batch.
+ */
+ private int evictLocalColdest(K incomingKey, int limit, BooleanSupplier
stopAfterEviction) {
+ if (!data.policy().eviction().isPresent()) {
+ return 0;
+ }
+ Map<K, V> coldest = data.policy().eviction().get().coldest(limit);
+ int evicted = 0;
+ for (Map.Entry<K, V> candidate : coldest.entrySet()) {
+ if (Objects.equals(candidate.getKey(), incomingKey)) {
+ continue;
+ }
+ V current = data.asMap().get(candidate.getKey());
+ ReservationRecord record = reservations.get(candidate.getKey());
+ long evictedWeight = record != null && record.published && current
!= null ? record.weight : 0L;
+ if (current == candidate.getValue() &&
data.asMap().remove(candidate.getKey(), current)) {
+ if (record != null) {
+ releaseReservation(candidate.getKey(), record.generation);
+ }
+ localEvictionCount.incrementAndGet();
+ localEvictionWeight.accumulateAndGet(evictedWeight,
MetaCacheWeightUtils::saturatedAdd);
+ evicted++;
+ if (stopAfterEviction.getAsBoolean()) {
+ return evicted;
+ }
+ }
+ }
+ return evicted;
+ }
+
+ private long reclaimForPeer(long targetBytes) {
+ if (targetBytes <= 0L || closed.get()) {
+ return 0L;
+ }
+ synchronized (admissionLock) {
+ long before = entryBudget.getUsedWeight();
+ long reclaimed = 0L;
+ while (reclaimed < targetBytes
+ && evictLocalColdest(null, LOCAL_EVICTION_BATCH_SIZE, () ->
+ Math.max(0L, before - entryBudget.getUsedWeight())
>= targetBytes) > 0) {
+ reclaimed = Math.max(0L, before - entryBudget.getUsedWeight());
+ }
+ return reclaimed;
+ }
+ }
+
+ private int weigh(K key, V value) {
+ ReservationRecord record = reservations.get(key);
+ // Every supported write path installs the reservation record before
calling data.put.
+ // Missing ownership is an invariant violation, so fail closed without
invoking an O(n)
+ // estimator from Caffeine's hot weigher callback.
+ long weight = record == null ? Integer.MAX_VALUE : record.weight;
+ return weight >= Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) weight;
+ }
+
+ private void onRemoval(@Nullable K key, @Nullable V value, RemovalCause
cause) {
+ if (key == null) {
+ return;
+ }
+ if (!weightBounded && !generationFencedRefresh && removalListener ==
null) {
+ return;
+ }
+ if (closed.get()) {
+ return;
+ }
+ // Replacement transfers the existing reservation to the newly
published generation. A
+ // soft-value collection instead reports a null value with COLLECTED
and must release it.
+ if (cause == RemovalCause.REPLACED) {
+ return;
+ }
+ if (removalListener != null) {
+ pendingRemovalNotifications.add(new RemovedToken<>(key,
removalToken(value)));
+ scheduleRemovalCleanup();
+ }
+ if (Thread.holdsLock(admissionLock)) {
+ // Other removals have already removed the Caffeine mapping and
can release their owner
+ // inline. A stale callback cannot release a replacement while its
mapping is visible.
+ if (data.asMap().get(key) == null) {
+ if (weightBounded) {
+ ReservationRecord record = reservations.get(key);
+ if (record != null) {
+ if (cause.wasEvicted()) {
+ automaticEvictionWeight.accumulateAndGet(
+ record.weight,
MetaCacheWeightUtils::saturatedAdd);
+ }
+ releaseReservation(key, record.generation);
+ }
+ } else {
+ RefreshRecord record = refreshRecords.get(key);
+ if (record != null) {
+ releaseRefreshRecord(key, record.generation);
+ }
+ }
+ }
+ return;
+ }
+ beforeRemovalOwnerSnapshotForTest(key);
+ long ownerGeneration = currentOwnerGeneration(key);
Review Comment:
Deferred by design for now: the delayed-collection race only affects the
exactness of eviction_weight telemetry (quota ownership stays correct, as
noted). Attributing the removed generation exactly would require tracking
per-value weights beyond the reservation map; we prefer to keep that out of the
admission hot path and revisit if the telemetry drift proves material in
practice.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
