| 6975 | } |
| 6976 | |
| 6977 | Future<Version> Transaction::getReadVersion(uint32_t flags) { |
| 6978 | if (!readVersion.isValid()) { |
| 6979 | if (!CLIENT_KNOBS->FORCE_GRV_CACHE_OFF && !trState->options.skipGrvCache && |
| 6980 | (deterministicRandom()->random01() <= CLIENT_KNOBS->DEBUG_USE_GRV_CACHE_CHANCE || |
| 6981 | trState->options.useGrvCache) && |
| 6982 | rkThrottlingCooledDown(getDatabase().getPtr(), trState->options.priority)) { |
| 6983 | // Upon our first request to use cached RVs, start the background updater |
| 6984 | if (!trState->cx->grvUpdateHandler.isValid()) { |
| 6985 | trState->cx->grvUpdateHandler = backgroundGrvUpdater(getDatabase().getPtr()); |
| 6986 | } |
| 6987 | Version rv = trState->cx->getCachedReadVersion(); |
| 6988 | double lastTime = trState->cx->getLastGrvTime(); |
| 6989 | double requestTime = now(); |
| 6990 | if (requestTime - lastTime <= CLIENT_KNOBS->MAX_VERSION_CACHE_LAG && rv != Version(0)) { |
| 6991 | ASSERT(!debug_checkVersionTime(rv, requestTime, "CheckStaleness")); |
| 6992 | readVersion = rv; |
| 6993 | return readVersion; |
| 6994 | } // else go through regular GRV path |
| 6995 | } |
| 6996 | ++trState->cx->transactionReadVersions; |
| 6997 | flags |= trState->options.getReadVersionFlags; |
| 6998 | switch (trState->options.priority) { |
| 6999 | case TransactionPriority::IMMEDIATE: |
| 7000 | flags |= GetReadVersionRequest::PRIORITY_SYSTEM_IMMEDIATE; |
| 7001 | ++trState->cx->transactionImmediateReadVersions; |
| 7002 | break; |
| 7003 | case TransactionPriority::DEFAULT: |
| 7004 | flags |= GetReadVersionRequest::PRIORITY_DEFAULT; |
| 7005 | ++trState->cx->transactionDefaultReadVersions; |
| 7006 | break; |
| 7007 | case TransactionPriority::BATCH: |
| 7008 | flags |= GetReadVersionRequest::PRIORITY_BATCH; |
| 7009 | ++trState->cx->transactionBatchReadVersions; |
| 7010 | break; |
| 7011 | default: |
| 7012 | ASSERT(false); |
| 7013 | } |
| 7014 | |
| 7015 | if (trState->options.tags.size() != 0) { |
| 7016 | double maxThrottleDelay = 0.0; |
| 7017 | bool canRecheck = false; |
| 7018 | |
| 7019 | auto& priorityThrottledTags = trState->cx->throttledTags[trState->options.priority]; |
| 7020 | for (auto& tag : trState->options.tags) { |
| 7021 | auto itr = priorityThrottledTags.find(tag); |
| 7022 | if (itr != priorityThrottledTags.end()) { |
| 7023 | if (!itr->second.expired()) { |
| 7024 | maxThrottleDelay = std::max(maxThrottleDelay, itr->second.throttleDuration()); |
| 7025 | canRecheck = itr->second.canRecheck(); |
| 7026 | } else { |
| 7027 | priorityThrottledTags.erase(itr); |
| 7028 | } |
| 7029 | } |
| 7030 | } |
| 7031 | |
| 7032 | if (maxThrottleDelay > 0.0 && !canRecheck) { // TODO: allow delaying? |
| 7033 | CODE_PROBE(true, "Throttling tag before GRV request"); |
| 7034 | ++trState->cx->transactionReadVersionsThrottled; |
nothing calls this directly
no test coverage detected