Versionstamp value mutation that takes a position - Part I

fdbclient/Atomic.h

@@ -279,4 +279,19 @@ static void transformSetVersionstampedValue( MutationRef& mutation, Version vers
 	mutation.type = MutationRef::SetValue;
 }
 
-#endif
+static void transformSetVersionstampedValuePos( MutationRef& mutation, Version version, uint16_t transactionNumber ) {
+	if (mutation.param2.size() >= 4) {
+		int32_t pos;
+		memcpy(&pos, mutation.param2.end() - sizeof(int32_t), sizeof(int32_t));
+		pos = littleEndian32(pos);
+		mutation.param2 = mutation.param2.substr(0, mutation.param2.size() - 4);
+
+		if (pos >= 0 && pos + 10 <= mutation.param2.size()) {
+			placeVersionstamp( mutateString(mutation.param2) + pos, version, transactionNumber );
+		}
+	}
+
+	mutation.type = MutationRef::SetValue;
+}
+
+#endif

fdbclient/CommitTransaction.h

@@ -24,11 +24,11 @@
 
 #include "FDBTypes.h"
 
-static const char * typeString[] = { "SetValue", "ClearRange", "AddValue", "DebugKeyRange", "DebugKey", "NoOp", "And", "Or", "Xor", "AppendIfFits", "AvailableForReuse", "Reserved_For_LogProtocolMessage", "Max", "Min", "SetVersionstampedKey", "SetVersionstampedValue", "ByteMin", "ByteMax", "MinV2", "AndV2" };
+static const char * typeString[] = { "SetValue", "ClearRange", "AddValue", "DebugKeyRange", "DebugKey", "NoOp", "And", "Or", "Xor", "AppendIfFits", "AvailableForReuse", "Reserved_For_LogProtocolMessage", "Max", "Min", "SetVersionstampedKey", "SetVersionstampedValue", "ByteMin", "ByteMax", "MinV2", "AndV2", "SetVersionstampedValuePos" };
 
 struct MutationRef { 
 	static const int OVERHEAD_BYTES = 12; //12 is the size of Header in MutationList entries
-	enum Type : uint8_t { SetValue=0, ClearRange, AddValue, DebugKeyRange, DebugKey, NoOp, And, Or, Xor, AppendIfFits, AvailableForReuse, Reserved_For_LogProtocolMessage /* See fdbserver/LogProtocolMessage.h */, Max, Min, SetVersionstampedKey, SetVersionstampedValue, ByteMin, ByteMax, MinV2, AndV2, MAX_ATOMIC_OP };
+	enum Type : uint8_t { SetValue=0, ClearRange, AddValue, DebugKeyRange, DebugKey, NoOp, And, Or, Xor, AppendIfFits, AvailableForReuse, Reserved_For_LogProtocolMessage /* See fdbserver/LogProtocolMessage.h */, Max, Min, SetVersionstampedKey, SetVersionstampedValue, ByteMin, ByteMax, MinV2, AndV2, SetVersionstampedValuePos, MAX_ATOMIC_OP };
 	// This is stored this way for serialization purposes.
 	uint8_t type;
 	StringRef param1, param2;
@@ -55,9 +55,9 @@ struct MutationRef {
 
 	// These masks define which mutation types have particular properties (they are used to implement isSingleKeyMutation() etc)
 	enum { 
-		ATOMIC_MASK = (1 << AddValue) | (1 << And) | (1 << Or) | (1 << Xor) | (1 << AppendIfFits) | (1 << Max) | (1 << Min) | (1 << SetVersionstampedKey) | (1 << SetVersionstampedValue) | (1 << ByteMin) | (1 << ByteMax) | (1 << MinV2) | (1 << AndV2),
+		ATOMIC_MASK = (1 << AddValue) | (1 << And) | (1 << Or) | (1 << Xor) | (1 << AppendIfFits) | (1 << Max) | (1 << Min) | (1 << SetVersionstampedKey) | (1 << SetVersionstampedValue) | (1 << ByteMin) | (1 << ByteMax) | (1 << MinV2) | (1 << AndV2) | (1 << SetVersionstampedValuePos),
 		SINGLE_KEY_MASK = ATOMIC_MASK | (1<<SetValue),
-		NON_ASSOCIATIVE_MASK = (1 << AddValue) | (1 << Or) | (1 << Xor) | (1 << Max) | (1 << Min) | (1 << SetVersionstampedKey) | (1 << SetVersionstampedValue) | (1 << MinV2)
+		NON_ASSOCIATIVE_MASK = (1 << AddValue) | (1 << Or) | (1 << Xor) | (1 << Max) | (1 << Min) | (1 << SetVersionstampedKey) | (1 << SetVersionstampedValue) | (1 << MinV2) | (1 << SetVersionstampedValuePos)
 	};
 };
 

fdbclient/RYWIterator.h

@@ -85,22 +85,34 @@ class RandomTestImpl {
 		return ValueRef(arena, value);
 	}
 
+	static ValueRef getRandomVersionstampValuePos(Arena& arena) {
+		int len = g_random->randomInt(10, 98);
+		std::string value = std::string(len, 'x');
+		int32_t pos = g_random->randomInt(0, len - 9);
+		if (g_random->random01() < 0.01) {
+			pos = value.size() - 10;
+		}
+		pos = littleEndian32(pos);
+		value += std::string((const char*)&pos, sizeof(int32_t));
+		return ValueRef(arena, value);
+	}
+
 	static ValueRef getRandomVersionstampKey(Arena& arena) {
 		int idx = g_random->randomInt(0, 100);
 		std::string key = format("%010d", idx / 3);
 		if (idx % 3 >= 1)
 			key += '\x00';
 		if (idx % 3 >= 2)
 			key += '\x00';
-		int pos = key.size() - g_random->randomInt(0, 3);
+		int16_t pos = key.size() - g_random->randomInt(0, 3);
 		if (g_random->random01() < 0.01) {
 			pos = 0;
 		}
 		key = key.substr(0, pos);
 		key += "XXXXXXXXYY";
 		key += std::string(g_random->randomInt(0, 3), 'z');
-		key += (char)(pos & 0xFF);
-		key += (char)((pos >> 8) & 0xFF);
+		pos = littleEndian16(pos);
+		key += std::string((const char*)&pos, sizeof(int16_t));
 		return ValueRef(arena, key);
 	}
 
@@ -132,4 +144,4 @@ class RandomTestImpl {
 void testESR();
 void testSnapshotCache();
 
-#endif
+#endif

fdbclient/ReadYourWrites.actor.cpp

@@ -1423,6 +1423,7 @@ void ReadYourWritesTransaction::writeRangeToNativeTransaction( KeyRangeRef const
 					case MutationRef::Min:
 					case MutationRef::SetVersionstampedKey:
 					case MutationRef::SetVersionstampedValue:
+					case MutationRef::SetVersionstampedValuePos:
 					case MutationRef::ByteMin:
 					case MutationRef::ByteMax:
 					case MutationRef::MinV2:
@@ -1495,8 +1496,18 @@ void ReadYourWritesTransaction::atomicOp( const KeyRef& key, const ValueRef& ope
 		}
 	}
 
-	if (operationType == MutationRef::SetVersionstampedValue && operand.size() < 10)
+	if(operationType == MutationRef::SetVersionstampedValue && operand.size() < 10)
 		throw client_invalid_operation();
+
+	if(operationType == MutationRef::SetVersionstampedValuePos) {
+		if(operand.size() < 4)
+			throw client_invalid_operation();
+		int32_t pos;
+		memcpy(&pos, operand.end() - sizeof(int32_t), sizeof(int32_t));
+		pos = littleEndian32(pos);
+		if (pos < 0 || pos + 10 > operand.size() - 4)
+			throw client_invalid_operation();
+	}
 
 	if (tr.apiVersionAtLeast(510)) {
 		if (operationType == MutationRef::Min)
@@ -1898,4 +1909,4 @@ void ReadYourWritesTransaction::debugLogRetries(Optional<Error> error) {
 			transactionDebugInfo->lastRetryLogTime = now();
 		}
 	}
-}
+}

fdbclient/WriteMap.h

@@ -65,7 +65,7 @@ class OperationStack {
 		bool isDependent() const {
 			if( !size() )
 				return false;
-			return singletonOperation.type != MutationRef::SetValue && singletonOperation.type != MutationRef::ClearRange && singletonOperation.type != MutationRef::SetVersionstampedValue && singletonOperation.type != MutationRef::SetVersionstampedKey;
+			return singletonOperation.type != MutationRef::SetValue && singletonOperation.type != MutationRef::ClearRange && singletonOperation.type != MutationRef::SetVersionstampedValue && singletonOperation.type != MutationRef::SetVersionstampedValuePos && singletonOperation.type != MutationRef::SetVersionstampedKey;
 		}
 		const RYWMutation& top() const { return hasVector() ? optionalOperations.get().back() : singletonOperation; }
 		RYWMutation& operator[] (int n) { return (n==0) ? singletonOperation : optionalOperations.get()[n-1]; }
@@ -142,8 +142,8 @@ class WriteMap {
 		bool following_conflict = it.entry().following_keys_conflict;
 		bool is_conflict = addConflict || it.is_conflict_range();
 		bool following_unreadable = it.entry().following_keys_unreadable;
-		bool is_unreadable = it.is_unreadable() || operation == MutationRef::SetVersionstampedValue || operation == MutationRef::SetVersionstampedKey;
-		bool is_dependent = operation != MutationRef::SetValue && operation != MutationRef::SetVersionstampedValue && operation != MutationRef::SetVersionstampedKey;
+		bool is_unreadable = it.is_unreadable() || operation == MutationRef::SetVersionstampedValue || operation == MutationRef::SetVersionstampedValuePos || operation == MutationRef::SetVersionstampedKey;
+		bool is_dependent = operation != MutationRef::SetValue && operation != MutationRef::SetVersionstampedValue && operation!= MutationRef::SetVersionstampedValuePos && operation != MutationRef::SetVersionstampedKey;
 
 		if (it.entry().key != key) {
 			if( it.is_cleared_range() && is_dependent ) {
@@ -630,4 +630,4 @@ class WriteMap {
 
 	*/
 
-#endif
+#endif

fdbserver/MasterProxyServer.actor.cpp

@@ -248,6 +248,8 @@ struct ResolutionRequestBuilder {
 				trIn.write_conflict_ranges.push_back( requests[0].arena, singleKeyRange( m.param1, requests[0].arena ) );
 			} else if (m.type == MutationRef::SetVersionstampedValue ) {
 				transformSetVersionstampedValue( m, requests[0].version, transactionNumberInBatch );
+			} else if (m.type == MutationRef::SetVersionstampedValuePos ) {
+				transformSetVersionstampedValuePos( m, requests[0].version, transactionNumberInBatch );
 			}
 			if (isMetadataMutation(m)) {
 				isTXNStateTransaction = true;

fdbserver/workloads/FuzzApiCorrectness.actor.cpp

@@ -821,7 +821,8 @@ struct FuzzApiCorrectnessWorkload : TestWorkload {
 		Key key;
 		Value value;
 		uint8_t op;
-		int16_t pos;
+		int16_t pos16;
+		int32_t pos32;
 
 		TestAtomicOp(unsigned int id, FuzzApiCorrectnessWorkload *workload) : BaseTestCallback(id, workload, "TestAtomicOp") {
 			key = makeKey();
@@ -853,9 +854,13 @@ struct FuzzApiCorrectnessWorkload : TestWorkload {
 				op = g_random->randomInt(minval, maxval+1);
 			}
 
-			pos = -1;
+			pos16 = -1;
 			if(op == MutationRef::SetVersionstampedKey && key.size() >= 2) {
-				pos = littleEndian16(*(int16_t*)&key.end()[-2]);
+				pos16 = littleEndian16(*(int16_t*)&key.end()[-2]);
+			}
+			pos32 = -1;
+			if(op == MutationRef::SetVersionstampedValuePos && value.size() >= 4) {
+				pos32 = littleEndian32(*(int32_t*)&value.end()[-4]);
 			}
 
 			contract = {
@@ -867,7 +872,8 @@ struct FuzzApiCorrectnessWorkload : TestWorkload {
 							(key >= (workload->useSystemKeys ? systemKeys.end : normalKeys.end))) ),
 				std::make_pair( error_code_client_invalid_operation, ExceptionContract::requiredIf(
 							(op == MutationRef::SetVersionstampedValue && value.size() < 10) ||
-							(op == MutationRef::SetVersionstampedKey && (pos < 0 || pos + 10 > key.size() - 2))) )
+							(op == MutationRef::SetVersionstampedKey && (pos16 < 0 || pos16 + 10 > key.size() - 2)) ||
+							(op == MutationRef::SetVersionstampedValuePos && (pos32 < 0 || pos32 + 10 > value.size() - 4))) )
 			};
 		}
 
@@ -877,7 +883,7 @@ struct FuzzApiCorrectnessWorkload : TestWorkload {
 
 		void augmentTrace(TraceEvent &e) const {
 			base_type::augmentTrace(e);
-			e.detail("key", printable(key)).detail("value", printable(value)).detail("op", op).detail("pos", pos);
+			e.detail("key", printable(key)).detail("value", printable(value)).detail("op", op).detail("pos16", pos16).detail("pos32", pos32);
 		}
 	};
 

fdbserver/workloads/Unreadable.actor.cpp

@@ -308,7 +308,7 @@ struct UnreadableWorkload : TestWorkload {
 			setMap[normalKeys.end] = ValueRef();
 
 			for (; opCount < 500; opCount++) {
-				int r = g_random->randomInt(0, 19);
+				int r = g_random->randomInt(0, 20);
 				if (r <= 10) {
 					key = RandomTestImpl::getRandomKey(arena);
 					value = RandomTestImpl::getRandomValue(arena);
@@ -341,14 +341,21 @@ struct UnreadableWorkload : TestWorkload {
 					//TraceEvent("RYWT_setVersionstampValue").detail("key", printable(key));
 				}
 				else if (r == 15) {
+					key = RandomTestImpl::getRandomKey(arena);
+					value = RandomTestImpl::getRandomVersionstampValuePos(arena);
+					tr.atomicOp(key, value, MutationRef::SetVersionstampedValuePos);
+					unreadableMap.insert(key, true);
+					//TraceEvent("RYWT_setVersionstampValuePos").detail("key", printable(key));
+				}
+				else if (r == 16) {
 					key = RandomTestImpl::getRandomVersionstampKey(arena);
 					value = RandomTestImpl::getRandomValue(arena);
 					tr.atomicOp(key, value, MutationRef::SetVersionstampedKey);
 					range = getVersionstampKeyRange(arena, key, allKeys.end);
 					unreadableMap.insert(range, true);
 					//TraceEvent("RYWT_setVersionstampKey").detail("range", printable(range));
 				}
-				else if (r == 16) {
+				else if (r == 17) {
 					range = RandomTestImpl::getRandomRange(arena);
 					snapshot = g_random->random01() < 0.05;
 					reverse = g_random->random01() < 0.5;
@@ -378,7 +385,7 @@ struct UnreadableWorkload : TestWorkload {
 						setMap[normalKeys.end] = ValueRef();
 					}
 				}
-				else if (r == 17) {
+				else if (r == 18) {
 					begin = RandomTestImpl::getRandomKeySelector(arena);
 					end = RandomTestImpl::getRandomKeySelector(arena);
 					limit = g_random->randomInt(1, 100);   // maximum number of results to return from the db
@@ -425,7 +432,7 @@ struct UnreadableWorkload : TestWorkload {
 						setMap[normalKeys.end] = ValueRef();
 					}		
 				}
-				else if (r == 18) {
+				else if (r == 19) {
 					key = RandomTestImpl::getRandomKey(arena);
 					snapshot = g_random->random01() < 0.05;
 

fdbserver/workloads/VersionStamp.actor.cpp

@@ -232,11 +232,23 @@ struct VersionStampWorkload : TestWorkload {
 			state KeyRangeRef range(prefix, endOfRange);
 			state Standalone<StringRef> committedVersionStamp;
 			state Version committedVersion;
+
+			bool useVersionstampPos = (g_random->random01() < 0.5);
+			state MutationRef::Type valueMutation;
+			state Value versionStampValue;
+			if(useVersionstampPos) {
+				versionStampValue = LiteralStringRef("\x00\x00\x00\x00").withPrefix(value);
+				valueMutation = MutationRef::SetVersionstampedValuePos;
+			} else {
+				versionStampValue = value;
+				valueMutation = MutationRef::SetVersionstampedValue;
+			}
+
 			loop{
 				state bool error = false;
 				//TraceEvent("VST_commit_begin").detail("key", printable(key)).detail("vsKey", printable(versionStampKey)).detail("clear", printable(range));
 				try {
-					tr.atomicOp(key, value, MutationRef::SetVersionstampedValue);
+					tr.atomicOp(key, versionStampValue, valueMutation);
 					tr.clear(range);
 					tr.atomicOp(versionStampKey, value, MutationRef::SetVersionstampedKey);
 					state Future<Standalone<StringRef>> fTrVs = tr.getVersionstamp();