Add /fdbclient/multiversionclient/ to ctest, and fix thread safety

fdbclient/MultiVersionTransaction.actor.cpp

@@ -1987,6 +1987,9 @@ THREAD_FUNC runSingleAssignmentVarTest(void* arg) {
 			tf.validate();
 
 			tf.future.extractPtr(); // leaks
+			for (auto t : tf.threads) {
+				waitThread(t);
+			}
 		}
 
 		for (int numRuns = 0; numRuns < 25; ++numRuns) {
@@ -2057,12 +2060,14 @@ struct AbortableTest {
 
 TEST_CASE("/fdbclient/multiversionclient/AbortableSingleAssignmentVar") {
 	state volatile bool done = false;
-	g_network->startThread(runSingleAssignmentVarTest<AbortableTest>, (void*)&done);
+	state THREAD_HANDLE thread = g_network->startThread(runSingleAssignmentVarTest<AbortableTest>, (void*)&done);
 
 	while (!done) {
 		wait(delay(1.0));
 	}
 
+	waitThread(thread);
+
 	return Void();
 }
 
@@ -2134,20 +2139,24 @@ TEST_CASE("/fdbclient/multiversionclient/DLSingleAssignmentVar") {
 	state volatile bool done = false;
 
 	MultiVersionApi::api->callbackOnMainThread = true;
-	g_network->startThread(runSingleAssignmentVarTest<DLTest>, (void*)&done);
+	state THREAD_HANDLE thread = g_network->startThread(runSingleAssignmentVarTest<DLTest>, (void*)&done);
 
 	while (!done) {
 		wait(delay(1.0));
 	}
 
+	waitThread(thread);
+
 	done = false;
 	MultiVersionApi::api->callbackOnMainThread = false;
-	g_network->startThread(runSingleAssignmentVarTest<DLTest>, (void*)&done);
+	thread = g_network->startThread(runSingleAssignmentVarTest<DLTest>, (void*)&done);
 
 	while (!done) {
 		wait(delay(1.0));
 	}
 
+	waitThread(thread);
+
 	return Void();
 }
 
@@ -2172,12 +2181,14 @@ struct MapTest {
 
 TEST_CASE("/fdbclient/multiversionclient/MapSingleAssignmentVar") {
 	state volatile bool done = false;
-	g_network->startThread(runSingleAssignmentVarTest<MapTest>, (void*)&done);
+	state THREAD_HANDLE thread = g_network->startThread(runSingleAssignmentVarTest<MapTest>, (void*)&done);
 
 	while (!done) {
 		wait(delay(1.0));
 	}
 
+	waitThread(thread);
+
 	return Void();
 }
 
@@ -2209,11 +2220,13 @@ struct FlatMapTest {
 
 TEST_CASE("/fdbclient/multiversionclient/FlatMapSingleAssignmentVar") {
 	state volatile bool done = false;
-	g_network->startThread(runSingleAssignmentVarTest<FlatMapTest>, (void*)&done);
+	state THREAD_HANDLE thread = g_network->startThread(runSingleAssignmentVarTest<FlatMapTest>, (void*)&done);
 
 	while (!done) {
 		wait(delay(1.0));
 	}
 
+	waitThread(thread);
+
 	return Void();
 }

flow/FastRef.h

@@ -22,45 +22,39 @@
 #define FLOW_FASTREF_H
 #pragma once
 
+#include <atomic>
 #include <cstdint>
 
-#include "flow/Platform.h"
-
-#if VALGRIND
-#include <drd.h>
-#endif
-
 template <class Subclass>
 class ThreadSafeReferenceCounted {
 public:
 	ThreadSafeReferenceCounted() : referenceCount(1) {}
 	// NO virtual destructor!  Subclass should have a virtual destructor if it is not sealed.
-	void addref() const { interlockedIncrement(&referenceCount); }
+	void addref() const { referenceCount.fetch_add(1); }
 	// If return value is true, caller is responsible for destruction of object
 	bool delref_no_destroy() const {
-		if (interlockedDecrement(&referenceCount) != 0) {
-#ifdef VALGRIND
-			ANNOTATE_HAPPENS_BEFORE(&referenceCount);
-#endif
-			return false;
+		// The performance of this seems comparable to a version with less strict memory ordering (see e.g.
+		// https://www.boost.org/doc/libs/1_57_0/doc/html/atomic/usage_examples.html#boost_atomic.usage_examples.example_reference_counters),
+		// on both x86 and ARM, with gcc8.
+		if (referenceCount.fetch_sub(1) == 1) {
+			return true;
 		}
-#ifdef VALGRIND
-		ANNOTATE_HAPPENS_AFTER(&referenceCount);
-#endif
-		return true;
+		return false;
 	}
 	void delref() const {
 		if (delref_no_destroy())
 			delete (Subclass*)this;
 	}
-	void setrefCountUnsafe(int32_t count) const { referenceCount = count; }
-	int32_t debugGetReferenceCount() const { return referenceCount; } // Never use in production code, only for tracing
-	bool isSoleOwnerUnsafe() const { return referenceCount == 1; }
+	void setrefCountUnsafe(int32_t count) const { referenceCount.store(count); }
+	int32_t debugGetReferenceCount() const {
+		return referenceCount.load();
+	} // Never use in production code, only for tracing
+	bool isSoleOwnerUnsafe() const { return referenceCount.load() == 1; }
 
 private:
 	ThreadSafeReferenceCounted(const ThreadSafeReferenceCounted&) /* = delete*/;
 	void operator=(const ThreadSafeReferenceCounted&) /* = delete*/;
-	mutable volatile int32_t referenceCount;
+	mutable std::atomic<int32_t> referenceCount;
 };
 
 template <class Subclass>

tests/CMakeLists.txt

@@ -261,6 +261,10 @@ if(WITH_PYTHON)
   add_fdb_test(TEST_FILES status/separate_not_enough_servers.txt)
   add_fdb_test(TEST_FILES status/single_process_too_many_config_params.txt)
 
+  add_test(
+    NAME multiversion_client/unit_tests
+    COMMAND $<TARGET_FILE:fdbserver> -r unittests -f /fdbclient/multiversionclient/
+  )
 
   verify_testing()
   if (NOT OPEN_FOR_IDE AND NOT WIN32)