From eb64eede8db1ed43bd9e4884d46fff59a537b8a5 Mon Sep 17 00:00:00 2001 From: Alex Miller Date: Wed, 15 Jan 2020 19:15:35 -0800 Subject: [PATCH] Make a smaller range inaccessable after writing a versionstamped key A transaction's read version is the lower bound of what a transaction's commit version could be. Thus, we can narrow the conflict range of a versionstamped key, and thus reduce the amount of the keyspace that is rendered inaccessable, by filling in the read version on the versionstamped key and using that as the lower bound of the conflict range. This allows reads to still be done to versionstampled keys lower than the read version of the current transaction. --- fdbclient/Atomic.h | 22 +++++++++---------- fdbclient/NativeAPI.actor.cpp | 8 +++++++ fdbclient/NativeAPI.actor.h | 1 + fdbclient/ReadYourWrites.actor.cpp | 7 +++++- fdbclient/ReadYourWrites.h | 1 + fdbserver/workloads/Unreadable.actor.cpp | 2 +- fdbserver/workloads/WriteDuringRead.actor.cpp | 2 +- 7 files changed, 29 insertions(+), 14 deletions(-) diff --git a/fdbclient/Atomic.h b/fdbclient/Atomic.h index d9aecbe8a3..25ad8ff691 100644 --- a/fdbclient/Atomic.h +++ b/fdbclient/Atomic.h @@ -229,10 +229,19 @@ static Optional doCompareAndClear(const Optional& existingVa return existingValueOptional; // No change required. } +static void placeVersionstamp( uint8_t* destination, Version version, uint16_t transactionNumber ) { + version = bigEndian64(version); + transactionNumber = bigEndian16(transactionNumber); + static_assert( sizeof(version) == 8, "version size mismatch" ); + memcpy( destination, &version, sizeof(version) ); + static_assert( sizeof(transactionNumber) == 2, "txn num size mismatch"); + memcpy( destination + sizeof(version), &transactionNumber, sizeof(transactionNumber) ); +} + /* * Returns the range corresponding to the specified versionstamp key. */ -static KeyRangeRef getVersionstampKeyRange(Arena& arena, const KeyRef &key, const KeyRef &maxKey) { +static KeyRangeRef getVersionstampKeyRange(Arena& arena, const KeyRef &key, Version minVersion, const KeyRef &maxKey) { KeyRef begin(arena, key); KeyRef end(arena, key); @@ -249,21 +258,12 @@ static KeyRangeRef getVersionstampKeyRange(Arena& arena, const KeyRef &key, cons if (pos < 0 || pos + 10 > begin.size()) throw client_invalid_operation(); - memset(mutateString(begin) + pos, 0, 10); + placeVersionstamp(mutateString(begin)+pos, minVersion, 0); memset(mutateString(end) + pos, '\xff', 10); return KeyRangeRef(begin, std::min(end, maxKey)); } -static void placeVersionstamp( uint8_t* destination, Version version, uint16_t transactionNumber ) { - version = bigEndian64(version); - transactionNumber = bigEndian16(transactionNumber); - static_assert( sizeof(version) == 8, "version size mismatch" ); - memcpy( destination, &version, sizeof(version) ); - static_assert( sizeof(transactionNumber) == 2, "txn num size mismatch"); - memcpy( destination + sizeof(version), &transactionNumber, sizeof(transactionNumber) ); -} - static void transformVersionstampMutation( MutationRef& mutation, StringRef MutationRef::* param, Version version, uint16_t transactionNumber ) { if ((mutation.*param).size() >= 4) { int32_t pos; diff --git a/fdbclient/NativeAPI.actor.cpp b/fdbclient/NativeAPI.actor.cpp index 79c9fdfef7..c17bb53149 100644 --- a/fdbclient/NativeAPI.actor.cpp +++ b/fdbclient/NativeAPI.actor.cpp @@ -3114,6 +3114,14 @@ Future Transaction::getReadVersion(uint32_t flags) { return readVersion; } +Optional Transaction::getCachedReadVersion() { + if (readVersion.isValid() && readVersion.isReady() && !readVersion.isError()) { + return readVersion.get(); + } else { + return Optional(); + } +} + Future> Transaction::getVersionstamp() { if(committing.isValid()) { return transaction_invalid_version(); diff --git a/fdbclient/NativeAPI.actor.h b/fdbclient/NativeAPI.actor.h index 96d5daadd9..0eb470969f 100644 --- a/fdbclient/NativeAPI.actor.h +++ b/fdbclient/NativeAPI.actor.h @@ -212,6 +212,7 @@ public: void setVersion( Version v ); Future getReadVersion() { return getReadVersion(0); } Future getRawReadVersion(); + Optional getCachedReadVersion(); [[nodiscard]] Future> get(const Key& key, bool snapshot = false); [[nodiscard]] Future watch(Reference watch); diff --git a/fdbclient/ReadYourWrites.actor.cpp b/fdbclient/ReadYourWrites.actor.cpp index 1d9efbef2d..a2cd09609a 100644 --- a/fdbclient/ReadYourWrites.actor.cpp +++ b/fdbclient/ReadYourWrites.actor.cpp @@ -1565,11 +1565,16 @@ void ReadYourWritesTransaction::atomicOp( const KeyRef& key, const ValueRef& ope } if(operationType == MutationRef::SetVersionstampedKey) { - KeyRangeRef range = getVersionstampKeyRange(arena, k, getMaxReadKey()); // this does validation of the key and needs to be performed before the readYourWritesDisabled path + // this does validation of the key and needs to be performed before the readYourWritesDisabled path + KeyRangeRef range = getVersionstampKeyRange(arena, k, tr.getCachedReadVersion().orDefault(0), getMaxReadKey()); if(!options.readYourWritesDisabled) { writeRangeToNativeTransaction(range); writes.addUnmodifiedAndUnreadableRange(range); } + // k is the unversionstamped key provided by the user. If we've filled in a minimum bound + // for the versionstamp, we need to make sure that's reflected when we insert it into the + // WriteMap below. + k = range.begin; } if(operationType == MutationRef::SetVersionstampedValue) { diff --git a/fdbclient/ReadYourWrites.h b/fdbclient/ReadYourWrites.h index 363c9f4639..fc766617bf 100644 --- a/fdbclient/ReadYourWrites.h +++ b/fdbclient/ReadYourWrites.h @@ -69,6 +69,7 @@ public: void setVersion( Version v ) { tr.setVersion(v); } Future getReadVersion(); + Optional getCachedReadVersion() { return tr.getCachedReadVersion(); } Future< Optional > get( const Key& key, bool snapshot = false ); Future< Key > getKey( const KeySelector& key, bool snapshot = false ); Future< Standalone > getRange( const KeySelector& begin, const KeySelector& end, int limit, bool snapshot = false, bool reverse = false ); diff --git a/fdbserver/workloads/Unreadable.actor.cpp b/fdbserver/workloads/Unreadable.actor.cpp index 92f30e4a29..455389d9d2 100644 --- a/fdbserver/workloads/Unreadable.actor.cpp +++ b/fdbserver/workloads/Unreadable.actor.cpp @@ -344,7 +344,7 @@ struct UnreadableWorkload : TestWorkload { key = RandomTestImpl::getRandomVersionstampKey(arena); value = RandomTestImpl::getRandomValue(arena); tr.atomicOp(key, value, MutationRef::SetVersionstampedKey); - range = getVersionstampKeyRange(arena, key, allKeys.end); + range = getVersionstampKeyRange(arena, key, tr.getCachedReadVersion().orDefault(0), allKeys.end); unreadableMap.insert(range, true); //TraceEvent("RYWT_SetVersionstampKey").detail("Range", printable(range)); } diff --git a/fdbserver/workloads/WriteDuringRead.actor.cpp b/fdbserver/workloads/WriteDuringRead.actor.cpp index ab2ed87ec3..817d1042ea 100644 --- a/fdbserver/workloads/WriteDuringRead.actor.cpp +++ b/fdbserver/workloads/WriteDuringRead.actor.cpp @@ -724,7 +724,7 @@ struct WriteDuringReadWorkload : TestWorkload { if(!self->useSystemKeys && deterministicRandom()->random01() < 0.01) { Key versionStampKey = self->getRandomVersionStampKey(); Value value = self->getRandomValue(); - KeyRangeRef range = getVersionstampKeyRange(versionStampKey.arena(), versionStampKey, normalKeys.end); + KeyRangeRef range = getVersionstampKeyRange(versionStampKey.arena(), versionStampKey, tr.getCachedReadVersion().orDefault(0), normalKeys.end); self->changeCount.insert( range, changeNum++ ); //TraceEvent("WDRVersionStamp").detail("VersionStampKey", printable(versionStampKey)).detail("Range", printable(range)); tr.atomicOp( versionStampKey, value, MutationRef::SetVersionstampedKey );