Revert "reduce the number of calls to operator< made by lower_bound and upper…"

fdbclient/CMakeLists.txt

@@ -71,7 +71,6 @@ set(FDBCLIENT_SRCS
   Tuple.h
   VersionedMap.actor.h
   VersionedMap.h
-  VersionedMap.cpp
   WriteMap.h
   json_spirit/json_spirit_error_position.h
   json_spirit/json_spirit_reader_template.h

fdbclient/RYWIterator.cpp

@@ -72,7 +72,7 @@ RYWIterator& RYWIterator::operator++() {
 	if (end_key_cmp <= 0) ++cache;
 	if (end_key_cmp >= 0) ++writes;
 	begin_key_cmp = -end_key_cmp;
-	end_key_cmp = cache.endKey().compare(writes.endKey());
+	end_key_cmp = cache.endKey().cmp(writes.endKey());
 	return *this;
 }
 
@@ -80,7 +80,7 @@ RYWIterator& RYWIterator::operator--() {
 	if (begin_key_cmp >= 0) --cache;
 	if (begin_key_cmp <= 0) --writes;
 	end_key_cmp = -begin_key_cmp;
-	begin_key_cmp = cache.beginKey().compare(writes.beginKey());
+	begin_key_cmp = cache.beginKey().cmp(writes.beginKey());
 	return *this;
 }
 
@@ -117,8 +117,8 @@ void RYWIterator::dbg() {
 }
 
 void RYWIterator::updateCmp() {
-	begin_key_cmp = cache.beginKey().compare(writes.beginKey());
+	begin_key_cmp = cache.beginKey().cmp(writes.beginKey());
-	end_key_cmp = cache.endKey().compare(writes.endKey());
+	end_key_cmp = cache.endKey().cmp(writes.endKey());
 }
 
 void testESR() {
@@ -157,13 +157,13 @@ void testESR() {
 				printf("Error: '%s' cmp '%s' = %d\n", printable(ssrs[i]).c_str(), printable(ssrs[j]).c_str(), c2);
 				return;
 			}
-
+			
 			/*
 			int c = ssrs[i] < ssrs[j] ? -1 : ssrs[i] == ssrs[j] ? 0 : 1;
-			int c2 = srs[i].compare(srs[j]);
+			int c2 = srs[i].cmp(srs[j]);
 			if ( c != (0<c2)-(c2<0) ) {
-			    printf("Error: '%s' cmp '%s' = %d\n", printable(ssrs[i]).c_str(), printable(ssrs[j]).c_str(), c2);
+				printf("Error: '%s' cmp '%s' = %d\n", printable(ssrs[i]).c_str(), printable(ssrs[j]).c_str(), c2);
-			    return;
+				return;
 			}*/
 
 			/*
@@ -413,8 +413,8 @@ TEST_CASE("/fdbclient/WriteMap/setVersionstampedKey") {
 	it.skip(allKeys.begin);
 
 	ASSERT(it.beginKey() < allKeys.end);
-	ASSERT(it.beginKey().compare(LiteralStringRef("")) == 0);
+	ASSERT(it.beginKey().cmp(LiteralStringRef("")) == 0);
-	ASSERT(it.endKey().compare(LiteralStringRef("stamp:XXXXXXXX\x06\x00\x00\x00")) == 0);
+	ASSERT(it.endKey().cmp(LiteralStringRef("stamp:XXXXXXXX\x06\x00\x00\x00")) == 0);
 	ASSERT(!it.is_cleared_range());
 	ASSERT(!it.is_conflict_range());
 	ASSERT(!it.is_operation());
@@ -423,8 +423,8 @@ TEST_CASE("/fdbclient/WriteMap/setVersionstampedKey") {
 	++it;
 
 	ASSERT(it.beginKey() < allKeys.end);
-	ASSERT(it.beginKey().compare(LiteralStringRef("stamp:XXXXXXXX\x06\x00\x00\x00")) == 0);
+	ASSERT(it.beginKey().cmp(LiteralStringRef("stamp:XXXXXXXX\x06\x00\x00\x00")) == 0);
-	ASSERT(it.endKey().compare(LiteralStringRef("stamp:XXXXXXXX\x06\x00\x00\x00\x00")) == 0);
+	ASSERT(it.endKey().cmp(LiteralStringRef("stamp:XXXXXXXX\x06\x00\x00\x00\x00")) == 0);
 	ASSERT(!it.is_cleared_range());
 	ASSERT(it.is_conflict_range());
 	ASSERT(it.is_operation());
@@ -434,8 +434,8 @@ TEST_CASE("/fdbclient/WriteMap/setVersionstampedKey") {
 	++it;
 
 	ASSERT(it.beginKey() < allKeys.end);
-	ASSERT(it.beginKey().compare(LiteralStringRef("stamp:XXXXXXXX\x06\x00\x00\x00\x00")) == 0);
+	ASSERT(it.beginKey().cmp(LiteralStringRef("stamp:XXXXXXXX\x06\x00\x00\x00\x00")) == 0);
-	ASSERT(it.endKey().compare(LiteralStringRef("stamp:ZZZZZZZZZZ")) == 0);
+	ASSERT(it.endKey().cmp(LiteralStringRef("stamp:ZZZZZZZZZZ")) == 0);
 	ASSERT(!it.is_cleared_range());
 	ASSERT(!it.is_conflict_range());
 	ASSERT(!it.is_operation());
@@ -444,8 +444,8 @@ TEST_CASE("/fdbclient/WriteMap/setVersionstampedKey") {
 	++it;
 
 	ASSERT(it.beginKey() < allKeys.end);
-	ASSERT(it.beginKey().compare(LiteralStringRef("stamp:ZZZZZZZZZZ")) == 0);
+	ASSERT(it.beginKey().cmp(LiteralStringRef("stamp:ZZZZZZZZZZ")) == 0);
-	ASSERT(it.endKey().compare(LiteralStringRef("stamp:ZZZZZZZZZZ\x00")) == 0);
+	ASSERT(it.endKey().cmp(LiteralStringRef("stamp:ZZZZZZZZZZ\x00")) == 0);
 	ASSERT(!it.is_cleared_range());
 	ASSERT(it.is_conflict_range());
 	ASSERT(it.is_operation());
@@ -455,8 +455,8 @@ TEST_CASE("/fdbclient/WriteMap/setVersionstampedKey") {
 	++it;
 
 	ASSERT(it.beginKey() < allKeys.end);
-	ASSERT(it.beginKey().compare(LiteralStringRef("stamp:ZZZZZZZZZZ\x00")) == 0);
+	ASSERT(it.beginKey().cmp(LiteralStringRef("stamp:ZZZZZZZZZZ\x00")) == 0);
-	ASSERT(it.endKey().compare(LiteralStringRef("\xff\xff")) == 0);
+	ASSERT(it.endKey().cmp(LiteralStringRef("\xff\xff")) == 0);
 	ASSERT(!it.is_cleared_range());
 	ASSERT(!it.is_conflict_range());
 	ASSERT(!it.is_operation());
@@ -486,8 +486,8 @@ TEST_CASE("/fdbclient/WriteMap/setVersionstampedValue") {
 	it.skip(allKeys.begin);
 
 	ASSERT(it.beginKey() < allKeys.end);
-	ASSERT(it.beginKey().compare(LiteralStringRef("")) == 0);
+	ASSERT(it.beginKey().cmp(LiteralStringRef("")) == 0);
-	ASSERT(it.endKey().compare(LiteralStringRef("stamp")) == 0);
+	ASSERT(it.endKey().cmp(LiteralStringRef("stamp")) == 0);
 	ASSERT(!it.is_cleared_range());
 	ASSERT(!it.is_conflict_range());
 	ASSERT(!it.is_operation());
@@ -496,8 +496,8 @@ TEST_CASE("/fdbclient/WriteMap/setVersionstampedValue") {
 	++it;
 
 	ASSERT(it.beginKey() < allKeys.end);
-	ASSERT(it.beginKey().compare(LiteralStringRef("stamp")) == 0);
+	ASSERT(it.beginKey().cmp(LiteralStringRef("stamp")) == 0);
-	ASSERT(it.endKey().compare(LiteralStringRef("stamp\x00")) == 0);
+	ASSERT(it.endKey().cmp(LiteralStringRef("stamp\x00")) == 0);
 	ASSERT(!it.is_cleared_range());
 	ASSERT(it.is_conflict_range());
 	ASSERT(it.is_operation());
@@ -507,8 +507,8 @@ TEST_CASE("/fdbclient/WriteMap/setVersionstampedValue") {
 	++it;
 
 	ASSERT(it.beginKey() < allKeys.end);
-	ASSERT(it.beginKey().compare(LiteralStringRef("stamp\x00")) == 0);
+	ASSERT(it.beginKey().cmp(LiteralStringRef("stamp\x00")) == 0);
-	ASSERT(it.endKey().compare(LiteralStringRef("stamp123")) == 0);
+	ASSERT(it.endKey().cmp(LiteralStringRef("stamp123")) == 0);
 	ASSERT(!it.is_cleared_range());
 	ASSERT(!it.is_conflict_range());
 	ASSERT(!it.is_operation());
@@ -517,8 +517,8 @@ TEST_CASE("/fdbclient/WriteMap/setVersionstampedValue") {
 	++it;
 
 	ASSERT(it.beginKey() < allKeys.end);
-	ASSERT(it.beginKey().compare(LiteralStringRef("stamp123")) == 0);
+	ASSERT(it.beginKey().cmp(LiteralStringRef("stamp123")) == 0);
-	ASSERT(it.endKey().compare(LiteralStringRef("stamp123\x00")) == 0);
+	ASSERT(it.endKey().cmp(LiteralStringRef("stamp123\x00")) == 0);
 	ASSERT(!it.is_cleared_range());
 	ASSERT(it.is_conflict_range());
 	ASSERT(it.is_operation());
@@ -528,8 +528,8 @@ TEST_CASE("/fdbclient/WriteMap/setVersionstampedValue") {
 	++it;
 
 	ASSERT(it.beginKey() < allKeys.end);
-	ASSERT(it.beginKey().compare(LiteralStringRef("stamp123\x00")) == 0);
+	ASSERT(it.beginKey().cmp(LiteralStringRef("stamp123\x00")) == 0);
-	ASSERT(it.endKey().compare(LiteralStringRef("\xff\xff")) == 0);
+	ASSERT(it.endKey().cmp(LiteralStringRef("\xff\xff")) == 0);
 	ASSERT(!it.is_cleared_range());
 	ASSERT(!it.is_conflict_range());
 	ASSERT(!it.is_operation());

fdbclient/SnapshotCache.h

@@ -71,7 +71,7 @@ struct ExtStringRef {
 
 	int size() const { return base.size() + extra_zero_bytes; }
 
-	int compare(ExtStringRef const& rhs) const {
+	int cmp(ExtStringRef const& rhs) const {
 		int cbl = std::min(base.size(), rhs.base.size());
 		if (cbl > 0) {
 			int c = memcmp(base.begin(), rhs.base.begin(), cbl);
@@ -82,7 +82,7 @@ struct ExtStringRef {
 			if (base[i]) return 1;
 		for(int i=cbl; i<rhs.base.size(); i++)
 			if (rhs.base[i]) return -1;
-		return ::compare(size(), rhs.size());
+		return size() - rhs.size();
 	}
 
 	bool startsWith( const ExtStringRef& s ) const { 
@@ -114,21 +114,13 @@ private:
 	int extra_zero_bytes;
 };
 inline bool operator == (const ExtStringRef& lhs, const ExtStringRef& rhs ) {
-	return lhs.size() == rhs.size() && !lhs.compare(rhs);
+	return lhs.size() == rhs.size() && !lhs.cmp(rhs);
 }
 inline bool operator != (const ExtStringRef& lhs, const ExtStringRef& rhs ) { return !(lhs==rhs); }
-inline bool operator<(const ExtStringRef& lhs, const ExtStringRef& rhs) {
+inline bool operator < ( const ExtStringRef& lhs, const ExtStringRef& rhs ) { return lhs.cmp(rhs)<0; }
-	return lhs.compare(rhs) < 0;
+inline bool operator > ( const ExtStringRef& lhs, const ExtStringRef& rhs ) { return lhs.cmp(rhs)>0; }
-}
+inline bool operator <= ( const ExtStringRef& lhs, const ExtStringRef& rhs ) { return lhs.cmp(rhs)<=0; }
-inline bool operator>(const ExtStringRef& lhs, const ExtStringRef& rhs) {
+inline bool operator >= ( const ExtStringRef& lhs, const ExtStringRef& rhs ) { return lhs.cmp(rhs)>=0; }
-	return lhs.compare(rhs) > 0;
-}
-inline bool operator<=(const ExtStringRef& lhs, const ExtStringRef& rhs) {
-	return lhs.compare(rhs) <= 0;
-}
-inline bool operator>=(const ExtStringRef& lhs, const ExtStringRef& rhs) {
-	return lhs.compare(rhs) >= 0;
-}
 
 template<>
 struct Traceable<ExtStringRef> : std::true_type {
@@ -160,10 +152,25 @@ private:
 		{
 			values.push_back( arena, kv );
 		}
-		int compare(Entry const& r) const { return ::compare(beginKey, r.beginKey); }
 		bool operator < (Entry const& r) const {
 			return beginKey < r.beginKey;
 		}
+		bool operator < (StringRef const& r) const {
+			return beginKey < r;
+		}
+		bool operator <= (Entry const& r) const {
+			return beginKey <= r.beginKey;
+		}
+		bool operator <= (StringRef const& r) const {
+			return beginKey <= r;
+		}
+		bool operator == (Entry const& r) const {
+			return beginKey == r.beginKey;
+		}
+		bool operator == (StringRef const& r) const {
+			return beginKey == r;
+		}
+
 		int segments() const { return 2*(values.size()+1); }
 	};
 

fdbclient/VersionedMap.cpp

@@ -1,56 +0,0 @@
-#include "fdbclient/VersionedMap.h"
-#include "flow/TreeBenchmark.h"
-#include "flow/UnitTest.h"
-
-template <typename K>
-struct VersionedMapHarness {
-	using map = VersionedMap<K, int>;
-	using key_type = K;
-
-	struct result {
-		typename map::iterator it;
-
-		result(typename map::iterator it) : it(it) {}
-
-		result& operator++() {
-			++it;
-			return *this;
-		}
-
-		const K& operator*() const { return it.key(); }
-
-		const K& operator->() const { return it.key(); }
-
-		bool operator==(result const& k) const { return it == k.it; }
-	};
-
-	map s;
-
-	void insert(K const& k) { s.insert(k, 1); }
-	result find(K const& k) const { return result(s.atLatest().find(k)); }
-	result not_found() const { return result(s.atLatest().end()); }
-	result begin() const { return result(s.atLatest().begin()); }
-	result end() const { return result(s.atLatest().end()); }
-	result lower_bound(K const& k) const { return result(s.atLatest().lower_bound(k)); }
-	result upper_bound(K const& k) const { return result(s.atLatest().upper_bound(k)); }
-	void erase(K const& k) { s.erase(k); }
-};
-
-TEST_CASE("performance/map/int/VersionedMap") {
-    VersionedMapHarness<int> tree;
-
-	treeBenchmark(tree, *randomInt);
-
-	return Void();
-}
-
-TEST_CASE("performance/map/StringRef/VersionedMap") {
-	Arena arena;
-    VersionedMapHarness<StringRef> tree;
-    
-	treeBenchmark(tree, [&arena]() { return randomStr(arena); });
-
-	return Void();
-}
-
-void forceLinkVersionedMapTests() {}

fdbclient/VersionedMap.h

@@ -67,63 +67,7 @@ namespace PTreeImpl {
 		PTree(PTree const&);
 	};
 
-    template <class T>
+	template<class T>
-    class PTreeFinger {
-	    using PTreeFingerEntry = PTree<T> const*;
-	    // This finger size supports trees with up to exp(96/4.3) ~= 4,964,514,749 entries.
-	    // see also: check().
-	    static constexpr size_t N = 96;
-	    PTreeFingerEntry entries_[N];
-	    size_t size_ = 0;
-	    size_t bound_sz_ = 0;
-
-	public:
-	    PTreeFinger() {}
-
-	    // Explicit copy constructors ensure we copy the live values in entries_.
-	    PTreeFinger(PTreeFinger const& f) { *this = f; }
-	    PTreeFinger(PTreeFinger&& f) { *this = f; }
-
-	    PTreeFinger& operator=(PTreeFinger const& f) {
-		    size_ = f.size_;
-		    bound_sz_ = f.bound_sz_;
-		    std::copy(f.entries_, f.entries_ + size_, entries_);
-		    return *this;
-	    }
-
-	    PTreeFinger& operator=(PTreeFinger&& f) {
-		    size_ = std::exchange(f.size_, 0);
-		    bound_sz_ = f.bound_sz_;
-		    std::copy(f.entries_, f.entries_ + size_, entries_);
-		    return *this;
-	    }
-
-	    size_t size() const { return size_; }
-	    PTree<T> const* back() const { return entries_[size_ - 1]; }
-	    void pop_back() { size_--; }
-	    void clear() { size_ = 0; }
-	    PTree<T> const* operator[](size_t i) const { return entries_[i]; }
-
-	    void resize(size_t sz) {
-		    size_ = sz;
-		    ASSERT(size_ < N);
-	    }
-
-	    void push_back(PTree<T> const* node) {
-		    entries_[size_++] = { node };
-		    ASSERT(size_ < N);
-	    }
-
-	    void push_for_bound(PTree<T> const* node, bool less) {
-		    push_back(node);
-		    bound_sz_ = less ? size_ : bound_sz_;
-	    }
-
-	    // remove the end of the finger so that the last entry is less than the probe
-	    void trim_to_bound() { size_ = bound_sz_; }
-    };
-
-    template<class T>
 	static Reference<PTree<T>> update( Reference<PTree<T>> const& node, bool which, Reference<PTree<T>> const& ptr, Version at ) {
 		if (ptr.getPtr() == node->child(which, at).getPtr()/* && node->replacedVersion <= at*/) {
 			return node;
@@ -165,41 +109,38 @@ namespace PTreeImpl {
 	template<class T, class X>
 	bool contains(const Reference<PTree<T>>& p, Version at, const X& x) {
 		if (!p) return false;
-		int cmp = compare(x, p->data);
+		bool less = x < p->data;
-		bool less = cmp < 0;
+		if (!less && !(p->data<x)) return true;  // x == p->data
-		if (cmp == 0) return true;
 		return contains(p->child(!less, at), at, x);
 	}
 
-	// TODO: Remove the number of invocations of operator<, and replace with something closer to memcmp.
+	template<class T, class X>
-	// and same for upper_bound.
+	void lower_bound(const Reference<PTree<T>>& p, Version at, const X& x, std::vector<const PTree<T>*>& f){
-    template <class T, class X>
+		if (!p) {
-    void lower_bound(const Reference<PTree<T>>& p, Version at, const X& x, PTreeFinger<T>& f) {
+			while (f.size() && !(x < f.back()->data))
-	    if (!p) {
+				f.pop_back();
-		    f.trim_to_bound();
+			return;
-		    return;
-		}
-		int cmp = compare(x, p->data);
-		bool less = cmp < 0;
-	    f.push_for_bound(p.getPtr(), less);
-	    if (cmp == 0) return;
-	    lower_bound(p->child(!less, at), at, x, f);
-    }
-
-    template <class T, class X>
-    void upper_bound(const Reference<PTree<T>>& p, Version at, const X& x, PTreeFinger<T>& f) {
-	    if (!p) {
-		    f.trim_to_bound();
-		    return;
 		}
+		f.push_back(p.getPtr());
 		bool less = x < p->data;
-	    f.push_for_bound(p.getPtr(), less);
+		if (!less && !(p->data<x)) return;  // x == p->data
-	    upper_bound(p->child(!less, at), at, x, f);
+		lower_bound(p->child(!less, at), at, x, f);
-    }
+	}
 
-    template <class T, bool forward>
+	template<class T, class X>
-    void move(Version at, PTreeFinger<T>& f) {
+	void upper_bound(const Reference<PTree<T>>& p, Version at, const X& x, std::vector<const PTree<T>*>& f){
-	    ASSERT(f.size());
+		if (!p) {
+			while (f.size() && !(x < f.back()->data))
+				f.pop_back();
+			return;
+		}
+		f.push_back(p.getPtr());
+		upper_bound(p->child(!(x < p->data), at), at, x, f);
+	}
+	
+	template<class T, bool forward>
+	void move(Version at, std::vector<const PTree<T>*>& f){
+		ASSERT(f.size());
 		const PTree<T> *n;
 		n = f.back();
 		if (n->child(forward, at)){
@@ -214,11 +155,11 @@ namespace PTreeImpl {
 				f.pop_back();
 			} while (f.size() && f.back()->child(forward, at).getPtr() == n);
 		}
-    }
+	}
 
-    template <class T, bool forward>
+	template<class T, bool forward>
-    int halfMove(Version at, PTreeFinger<T>& f) {
+	int halfMove(Version at, std::vector<const PTree<T>*>& f) {
-	    // Post: f[:return_value] is the finger that would have been returned by move<forward>(at,f), and f[:original_length_of_f] is unmodified
+		// Post: f[:return_value] is the finger that would have been returned by move<forward>(at,f), and f[:original_length_of_f] is unmodified
 		ASSERT(f.size());
 		const PTree<T> *n;
 		n = f.back();
@@ -237,35 +178,35 @@ namespace PTreeImpl {
 			} while (s && f[s-1]->child(forward, at).getPtr() == n);
 			return s;
 		}
-    }
+	}
 
-    template <class T>
+	template<class T>
-    void next(Version at, PTreeFinger<T>& f) {
+	void next(Version at, std::vector<const PTree<T>*>& f){
-	    move<T,true>(at, f);
+		move<T,true>(at, f);
-    }
+	}
+	
+	template<class T>
+	void previous(Version at, std::vector<const PTree<T>*>& f){
+		move<T,false>(at, f);
+	}
 
-    template <class T>
+	template<class T>
-    void previous(Version at, PTreeFinger<T>& f) {
+	int halfNext(Version at, std::vector<const PTree<T>*>& f){
-	    move<T,false>(at, f);
+		return halfMove<T,true>(at, f);
-    }
+	}
+	
+	template<class T>
+	int halfPrevious(Version at, std::vector<const PTree<T>*>& f){
+		return halfMove<T,false>(at, f);
+	}
 
-    template <class T>
+	template<class T>
-    int halfNext(Version at, PTreeFinger<T>& f) {
+	T get(std::vector<const PTree<T>*>& f){
-	    return halfMove<T,true>(at, f);
+		ASSERT(f.size());
-    }
-
-    template <class T>
-    int halfPrevious(Version at, PTreeFinger<T>& f) {
-	    return halfMove<T,false>(at, f);
-    }
-
-    template <class T>
-    T get(PTreeFinger<T>& f) {
-	    ASSERT(f.size());
 		return f.back()->data;
-    }
+	}
 
-    // Modifies p to point to a PTree with x inserted
+	// Modifies p to point to a PTree with x inserted
 	template<class T>
 	void insert(Reference<PTree<T>>& p, Version at, const T& x) {
 		if (!p){
@@ -294,24 +235,24 @@ namespace PTreeImpl {
 		return lastNode(p->right(at), at);
 	}
 
-    template <class T, bool last>
+	template<class T, bool last>
-    void firstOrLastFinger(const Reference<PTree<T>>& p, Version at, PTreeFinger<T>& f) {
+	void firstOrLastFinger(const Reference<PTree<T>>& p, Version at, std::vector<const PTree<T>*>& f) {
-	    if (!p) return;
+		if (!p) return;
 		f.push_back(p.getPtr());
 		firstOrLastFinger<T, last>(p->child(last, at), at, f);
-    }
+	}
+	
+	template<class T>
+	void first(const Reference<PTree<T>>& p, Version at, std::vector<const PTree<T>*>& f) {
+		return firstOrLastFinger<T, false>(p, at, f);
+	}
 
-    template <class T>
+	template<class T>
-    void first(const Reference<PTree<T>>& p, Version at, PTreeFinger<T>& f) {
+	void last(const Reference<PTree<T>>& p, Version at, std::vector<const PTree<T>*>& f) {
-	    return firstOrLastFinger<T, false>(p, at, f);
+		return firstOrLastFinger<T, true>(p, at, f);
-    }
+	}
 
-    template <class T>
+	// modifies p to point to a PTree with the root of p removed
-    void last(const Reference<PTree<T>>& p, Version at, PTreeFinger<T>& f) {
-	    return firstOrLastFinger<T, true>(p, at, f);
-    }
-
-    // modifies p to point to a PTree with the root of p removed
 	template<class T>
 	void removeRoot(Reference<PTree<T>>& p, Version at) {
 		if (!p->right(at))
@@ -331,27 +272,24 @@ namespace PTreeImpl {
 	template<class T, class X>
 	void remove(Reference<PTree<T>>& p, Version at, const X& x) {
 		if (!p) ASSERT(false); // attempt to remove item not present in PTree
-		int cmp = compare(x, p->data);
+		if (x < p->data) {
-		if (cmp < 0) {
 			Reference<PTree<T>> child = p->child(0, at);
 			remove(child, at, x);
 			p = update(p, 0, child, at);
-		} else if (cmp > 0) {
+		} else if (p->data < x) {
 			Reference<PTree<T>> child = p->child(1, at);
 			remove(child, at, x);
 			p = update(p, 1, child, at);
-		} else {
+		} else
 			removeRoot(p, at);
-		}
 	}
 
 	template<class T, class X>
 	void remove(Reference<PTree<T>>& p, Version at, const X& begin, const X& end) {
 		if (!p) return;
 		int beginDir, endDir;
-		int beginCmp = compare(begin, p->data);
+		if (begin < p->data) beginDir = -1;
-		if (beginCmp < 0) beginDir = -1;
+		else if (p->data < begin) beginDir = +1;
-		else if (beginCmp > 0) beginDir = +1;
 		else beginDir = 0;
 		if (!(p->data < end)) endDir = -1;
 		else endDir = +1;
@@ -426,9 +364,7 @@ namespace PTreeImpl {
 		if (!right) return left;
 
 		Reference<PTree<T>> r = Reference<PTree<T>>(new PTree<T>(lastNode(left, at)->data, at));
-		if (EXPENSIVE_VALIDATION) {
+		ASSERT( r->data < firstNode(right, at)->data);
-			ASSERT( r->data < firstNode(right, at)->data);
-		}
 		Reference<PTree<T>> a = left;
 		remove(a, at, r->data);
 
@@ -577,7 +513,6 @@ class VersionedMap : NonCopyable {
 //private:
 public:
 	typedef PTreeImpl::PTree<MapPair<K,std::pair<T,Version>>> PTreeT;
-	typedef PTreeImpl::PTreeFinger<MapPair<K, std::pair<T, Version>>> PTreeFingerT;
 	typedef Reference< PTreeT > Tree;
 
 	Version oldestVersion, latestVersion;
@@ -654,7 +589,7 @@ public:
 
 		UNSTOPPABLE_ASSERT(r->first == newOldestVersion);
 
-		std::vector<Tree> toFree;
+		vector<Tree> toFree;
 		toFree.reserve(10000);
 		auto newBegin = r;
 		Tree *lastRoot = nullptr;
@@ -744,7 +679,7 @@ public:
 		friend class VersionedMap<K,T>;
 		Tree root;
 		Version at;
-		PTreeFingerT finger;
+		vector< PTreeT const* > finger;
 	};
 
 	class ViewAtVersion {

fdbclient/WriteMap.h

@@ -107,35 +107,18 @@ struct WriteMapEntry {
 
 	WriteMapEntry( KeyRef const& key, OperationStack && stack, bool following_keys_cleared, bool following_keys_conflict, bool is_conflict, bool following_keys_unreadable, bool is_unreadable ) : key(key), stack(std::move(stack)), following_keys_cleared(following_keys_cleared), following_keys_conflict(following_keys_conflict), is_conflict(is_conflict), following_keys_unreadable(following_keys_unreadable), is_unreadable(is_unreadable) {}
 
-	int compare(StringRef const& r) const { return key.compare(r); }
-
-	int compare(ExtStringRef const& r) const { return -r.compare(key); }
-
 	std::string toString() const { return printable(key); }
 };
 
-inline int compare(StringRef const& l, WriteMapEntry const& r) {
-	return l.compare(r.key);
-}
-
-inline int compare(ExtStringRef const& l, WriteMapEntry const& r) {
-	return l.compare(r.key);
-}
-
 inline bool operator < ( const WriteMapEntry& lhs, const WriteMapEntry& rhs ) { return lhs.key < rhs.key; }
 inline bool operator < ( const WriteMapEntry& lhs, const StringRef& rhs ) { return lhs.key < rhs; }
 inline bool operator < ( const StringRef& lhs, const WriteMapEntry& rhs ) { return lhs < rhs.key; }
-inline bool operator<(const WriteMapEntry& lhs, const ExtStringRef& rhs) {
+inline bool operator < ( const WriteMapEntry& lhs, const ExtStringRef& rhs ) { return rhs.cmp(lhs.key)>0; }
-	return rhs.compare(lhs.key) > 0;
+inline bool operator < ( const ExtStringRef& lhs, const WriteMapEntry& rhs ) { return lhs.cmp(rhs.key)<0; }
-}
-inline bool operator<(const ExtStringRef& lhs, const WriteMapEntry& rhs) {
-	return lhs.compare(rhs.key) < 0;
-}
 
 class WriteMap {
 private:
-	typedef PTreeImpl::PTree<WriteMapEntry> PTreeT;
+	typedef PTreeImpl::PTree< WriteMapEntry > PTreeT;
-	typedef PTreeImpl::PTreeFinger<WriteMapEntry> PTreeFingerT;
 	typedef Reference<PTreeT> Tree;
 
 public:
@@ -391,7 +374,7 @@ public:
 		Tree tree;
 		Version at;
 		int beginLen, endLen;
-		PTreeFingerT finger;
+		vector< PTreeT const* > finger;
 		bool offset;  // false-> the operation stack at entry(); true-> the following cleared or unmodified range
 	};
 

fdbrpc/FailureMonitor.h

@@ -23,6 +23,7 @@
 #pragma once
 
 #include "flow/flow.h"
+#include "flow/IndexedSet.h"
 #include "fdbrpc/FlowTransport.h" // Endpoint
 #include <unordered_map>
 

fdbserver/workloads/UnitTests.actor.cpp

@@ -25,7 +25,6 @@
 void forceLinkIndexedSetTests();
 void forceLinkDequeTests();
 void forceLinkFlowTests();
-void forceLinkVersionedMapTests();
 
 struct UnitTestWorkload : TestWorkload {
 	bool enabled;
@@ -44,7 +43,6 @@ struct UnitTestWorkload : TestWorkload {
 		forceLinkIndexedSetTests();
 		forceLinkDequeTests();
 		forceLinkFlowTests();
-		forceLinkVersionedMapTests();
 	}
 
 	virtual std::string description() { return "UnitTests"; }

flow/Arena.h

@@ -530,12 +530,11 @@ public:
 	int expectedSize() const { return size(); }
 
 	int compare(StringRef const& other) const {
-		size_t minSize = std::min(size(), other.size());
+		if (std::min(size(), other.size()) > 0) {
-		if (minSize != 0) {
+			int c = memcmp(begin(), other.begin(), std::min(size(), other.size()));
-			int c = memcmp(begin(), other.begin(), minSize);
 			if (c != 0) return c;
 		}
-		return ::compare(size(), other.size());
+		return size() - other.size();
 	}
 
 	// Removes bytes from begin up to and including the sep string, returns StringRef of the part before sep

flow/CMakeLists.txt

@@ -61,7 +61,6 @@ set(FLOW_SRCS
   ThreadSafeQueue.h
   Trace.cpp
   Trace.h
-  TreeBenchmark.h
   UnitTest.cpp
   UnitTest.h
   XmlTraceLogFormatter.cpp

flow/IKeyValueContainer.h

@@ -41,22 +41,11 @@ struct KeyValueMapPair {
 	KeyValueMapPair(KeyRef key, ValueRef value)
 	  : arena(key.expectedSize() + value.expectedSize()), key(arena, key), value(arena, value) {}
 
-	int compare(KeyValueMapPair const& r) const { return ::compare(key, r.key); }
-
-	template <class CompatibleWithKey>
-	int compare(CompatibleWithKey const& r) const {
-		return ::compare(key, r);
-	}
-
 	bool operator<(KeyValueMapPair const& r) const { return key < r.key; }
 	bool operator==(KeyValueMapPair const& r) const { return key == r.key; }
 	bool operator!=(KeyValueMapPair const& r) const { return key != r.key; }
 };
 
-template <class CompatibleWithKey>
-int compare(CompatibleWithKey const& l, KeyValueMapPair const& r) {
-	return ::compare(l, r.key);
-}
 template <class CompatibleWithKey>
 bool operator<(KeyValueMapPair const& l, CompatibleWithKey const& r) {
 	return l.key < r;

flow/IRandom.h

@@ -34,27 +34,6 @@
 #endif
 #include <functional>
 
-// Until we move to C++20, we'll need something to take the place of operator<=>.
-// This is as good a place as any, I guess.
-
-template <typename T>
-typename std::enable_if<std::is_integral<T>::value, int>::type compare(T l, T r) {
-	return l < r ? -1 : l == r ? 0 : 1;
-}
-
-template <typename T, typename U>
-typename std::enable_if<!std::is_integral<T>::value, int>::type compare(T const& l, U const& r) {
-	return l.compare(r);
-}
-
-template <class K, class V>
-int compare(std::pair<K, V> const& l, std::pair<K, V> const& r) {
-	if (int cmp = compare(l.first, r.first)) {
-		return cmp;
-	}
-	return compare(l.second, r.second);
-}
-
 class UID {
 	uint64_t part[2];
 public:
@@ -65,12 +44,6 @@ public:
 	std::string shortString() const;
 	bool isValid() const { return part[0] || part[1]; }
 
-	bool compare(const UID& r) const {
-		if (int cmp = ::compare(part[0], r.part[0])) {
-			return cmp;
-		}
-		return ::compare(part[1], r.part[1]);
-	}
 	bool operator == ( const UID& r ) const { return part[0]==r.part[0] && part[1]==r.part[1]; }
 	bool operator != ( const UID& r ) const { return part[0]!=r.part[0] || part[1]!=r.part[1]; }
 	bool operator < ( const UID& r ) const { return part[0] < r.part[0] || (part[0] == r.part[0] && part[1] < r.part[1]); }

flow/IndexedSet.cpp

@@ -31,8 +31,8 @@
 #include <cstring>
 #include <deque>
 #include <random>
-#include "flow/TreeBenchmark.h"
 #include "flow/UnitTest.h"
+
 template <class Node>
 int ISGetHeight(Node* n){
 	if (!n) return 0;
@@ -137,121 +137,7 @@ TEST_CASE("/flow/IndexedSet/erase 400k of 1M") {
 	return Void();
 }
 
-TEST_CASE("/flow/IndexedSet/random ops") {
+/*TEST_CASE("/flow/IndexedSet/performance") {
-	for (int t = 0; t < 100; t++) {
-		IndexedSet<int, int> is;
-		int rr = deterministicRandom()->randomInt(0, 600) * deterministicRandom()->randomInt(0, 600);
-		for (int n = 0; n < rr; n++) {
-			if (deterministicRandom()->random01() < (double)is.sumTo(is.end()) / rr * 2)
-				is.erase(is.lower_bound(deterministicRandom()->randomInt(0, 10000000)));
-			else
-				is.insert(deterministicRandom()->randomInt(0, 10000000), 3);
-		}
-
-		int b = deterministicRandom()->randomInt(0, 10000000);
-		// int e = b + deterministicRandom()->randomInt(0, 10);
-		int e = deterministicRandom()->randomInt(0, 10000000);
-		if (e < b) std::swap(b, e);
-		auto ib = is.lower_bound(b);
-		auto ie = is.lower_bound(e);
-
-		int original_count = is.sumTo(is.end()) / 3;
-		int original_incount = is.sumRange(ib, ie) / 3;
-
-		// printf("\n#%d Erasing %d of %d items\n", t, original_incount, original_count);
-
-		is.erase(ib, ie);
-		is.testonly_assertBalanced();
-
-		int count = 0, incount = 0;
-		for (auto i : is) {
-			++count;
-			if (i >= b && i < e) {
-				// printf("Remaining item: %d (%d - %d)\n", i, b, e);
-				incount++;
-			}
-		}
-
-		// printf("%d items remain, totalling %d\n", count, is.sumTo(is.end()));
-		// printf("%d items remain in erased range\n", incount);
-
-		ASSERT(incount == 0);
-		ASSERT(count == original_count - original_incount);
-		ASSERT(is.sumTo(is.end()) == count * 3);
-	}
-	return Void();
-}
-
-TEST_CASE("/flow/IndexedSet/strings") {
-	Map<std::string, int> myMap;
-	std::map<std::string, int> aMap;
-	myMap["Hello"] = 1;
-	myMap["Planet"] = 5;
-	for (auto i = myMap.begin(); i != myMap.end(); ++i) aMap[i->key] = i->value;
-
-	ASSERT(myMap.find(std::string("Hello"))->value == 1);
-	ASSERT(myMap.find(std::string("World")) == myMap.end());
-	ASSERT(myMap["Hello"] == 1);
-
-	auto a = myMap.upper_bound("A")->key;
-	auto x = myMap.lower_bound("M")->key;
-
-	ASSERT((a + x) == (std::string) "HelloPlanet");
-
-	return Void();
-}
-
-template <typename K>
-struct IndexedSetHarness {
-	using map = IndexedSet<K, int>;
-	using result = typename map::iterator;
-	using key_type = K;
-
-	map s;
-
-	void insert(K const& k) { s.insert(K(k), 1); }
-	result find(K const& k) const { return s.find(k); }
-	result not_found() const { return s.end(); }
-	result begin() const { return s.begin(); }
-	result end() const { return s.end(); }
-	result lower_bound(K const& k) const { return s.lower_bound(k); }
-	result upper_bound(K const& k) const { return s.upper_bound(k); }
-	void erase(K const& k) { s.erase(k); }
-};
-
-TEST_CASE("performance/map/StringRef/IndexedSet") {
-	Arena arena;
-
-	IndexedSetHarness<StringRef> is;
-	treeBenchmark(is, [&arena]() { return randomStr(arena); });
-
-	return Void();
-}
-
-TEST_CASE("performance/map/StringRef/StdMap") {
-	Arena arena;
-
-	MapHarness<StringRef> is;
-	treeBenchmark(is, [&arena]() { return randomStr(arena); });
-
-	return Void();
-}
-
-TEST_CASE("performance/map/int/IndexedSet") {
-	IndexedSetHarness<int> is;
-	treeBenchmark(is, &randomInt);
-
-	return Void();
-}
-
-TEST_CASE("performance/map/int/StdMap") {
-	MapHarness<int> is;
-	treeBenchmark(is, &randomInt);
-
-	return Void();
-}
-
-TEST_CASE("performance/flow/IndexedSet/integers") {
 	std::vector<int> x;
 	for (int i = 0; i<1000000; i++)
 		x.push_back(deterministicRandom()->randomInt(0, 10000000));
@@ -265,6 +151,7 @@ TEST_CASE("performance/flow/IndexedSet/integers") {
 	double end = timer();
 	double kps = x.size() / 1000.0 / (end - start);
 	printf("%0.1f Kinsert/sec\n", kps);
+	ASSERT(kps >= 500);                                           //< Or something?
 
 	start = timer();
 	for (int i = 0; i<x.size(); i++)
@@ -272,6 +159,7 @@ TEST_CASE("performance/flow/IndexedSet/integers") {
 	end = timer();
 	kps = x.size() / 1000.0 / (end - start);
 	printf("%0.1f Kfind/sec\n", kps);
+	ASSERT(kps >= 500);
 
 	{
 		//std::set<int> ss;
@@ -316,41 +204,87 @@ TEST_CASE("performance/flow/IndexedSet/integers") {
 
 	printf("%0.1f Kerase/sec\n", x.size() / 1000.0 / (end - start));
 	is.testonly_assertBalanced();
-	for (int i = 0; i < x.size() / 2; i++) {
+	for (int i = 0; i<x.size() / 2; i++)
 		ASSERT(is.find(x[i]) == is.end());
-	}
+}*/
 
+TEST_CASE("/flow/IndexedSet/random ops") {
+	for (int t = 0; t<100; t++) {
+		IndexedSet<int, int> is;
+		int rr = deterministicRandom()->randomInt(0, 600) * deterministicRandom()->randomInt(0, 600);
+		for (int n = 0; n<rr; n++) {
+			if (deterministicRandom()->random01() < (double)is.sumTo(is.end()) / rr * 2)
+				is.erase(is.lower_bound(deterministicRandom()->randomInt(0, 10000000)));
+			else
+				is.insert(deterministicRandom()->randomInt(0, 10000000), 3);
+		}
+
+		int b = deterministicRandom()->randomInt(0, 10000000);
+		//int e = b + deterministicRandom()->randomInt(0, 10);
+		int e = deterministicRandom()->randomInt(0, 10000000);
+		if (e<b) std::swap(b, e);
+		auto ib = is.lower_bound(b);
+		auto ie = is.lower_bound(e);
+
+		int original_count = is.sumTo(is.end())/3;
+		int original_incount = is.sumRange(ib, ie)/3;
+
+		//printf("\n#%d Erasing %d of %d items\n", t, original_incount, original_count);
+
+		is.erase(ib, ie);
+		is.testonly_assertBalanced();
+
+		int count = 0, incount = 0;
+		for (auto i : is) {
+			++count;
+			if (i >= b && i < e) { 
+				//printf("Remaining item: %d (%d - %d)\n", i, b, e); 
+				incount++; 
+			}
+		}
+
+		//printf("%d items remain, totalling %d\n", count, is.sumTo(is.end()));
+		//printf("%d items remain in erased range\n", incount);
+
+		ASSERT(incount == 0);
+		ASSERT(count == original_count - original_incount);
+		ASSERT(is.sumTo(is.end()) == count*3);
+	}
 	return Void();
 }
 
-TEST_CASE("performance/flow/IndexedSet/strings") {
+TEST_CASE("/flow/IndexedSet/strings") {
-	constexpr size_t count = 1000000;
 	Map< std::string, int > myMap;
 	std::map< std::string, int > aMap;
-	double start, end;
+	myMap["Hello"] = 1;
-	int tt = 0;
+	myMap["Planet"] = 5;
+	for (auto i = myMap.begin(); i != myMap.end(); ++i)
+		aMap[i->key] = i->value;
 
-	std::string const hello{ "Hello" };
+	ASSERT(myMap.find("Hello")->value == 1);
-	myMap[hello] = 1;
+	ASSERT(myMap.find("World") == myMap.end());
-	aMap["Hello"] = 1;
+	ASSERT(myMap["Hello"] == 1);
 
-	start = timer();
+	auto a = myMap.upper_bound("A")->key;
+	auto x = myMap.lower_bound("M")->key;
 
-	for (size_t i = 0; i < count; i++) {
+	ASSERT((a + x) == (std::string)"HelloPlanet");
-		tt += myMap.find(hello)->value;
-	}
-	end = timer();
 
-	ASSERT(tt == count);
+	/* This was a performance test:
 
-	printf("%0.1f Map.KfindStr/sec\n", count / 1000.0 / (end - start));
+		double start = timer();
+		volatile int tt=0;
+		for(int i=0; i<1000000; i++)
+		tt += myMap.find( "Hello" )->value;
+		double end = timer();
+		printf("%0.1f Map.KfindStr/sec\n", 1000000/1000.0/(end-start));
 
-	start = timer();
+		start = timer();
-	for (size_t i = 0; i < count; i++) {
+		for(int i=0; i<1000000; i++)
-		aMap.find(hello);
+		aMap.find( "Hello" );
-	}
+		end = timer();
-	end = timer();
+		printf("%0.1f std::map.KfindStr/sec\n", 1000000/1000.0/(end-start));
-	printf("%0.1f std::map.KfindStr/sec\n", count / 1000.0 / (end - start));
+	*/
 
 	return Void();
 }
@@ -406,7 +340,6 @@ TEST_CASE("/flow/IndexedSet/data constructor and destructor calls match") {
 		~Counter() { count--; }
 		Counter(const Counter& r) :value(r.value) { count++; }
 		void operator=(const Counter& r) { value = r.value; }
-		int compare(const Counter& r) const { return ::compare(value, r.value); }
 		bool operator<(const Counter& r) const { return value < r.value; }
 	};
 	IndexedSet<Counter, NoMetric> mySet;

flow/IndexedSet.h

@@ -22,7 +22,6 @@
 #define FLOW_INDEXEDSET_H
 #pragma once
 
-#include "flow/Arena.h"
 #include "flow/Platform.h"
 #include "flow/FastAlloc.h"
 #include "flow/Trace.h"
@@ -200,7 +199,7 @@ private:
 
 	Node *root;
 
-	Metric eraseHalf(Node* start, Node* end, int eraseDir, int& heightDelta, std::vector<Node*>& toFree);
+	Metric eraseHalf( Node* start, Node* end, int eraseDir, int& heightDelta, std::vector<Node*>& toFree );
 	void erase( iterator begin, iterator end, std::vector<Node*>& toFree );
 
 	void replacePointer( Node* oldNode, Node* newNode ) {
@@ -253,11 +252,6 @@ public:
 	MapPair(MapPair&& r) BOOST_NOEXCEPT  : key(std::move(r.key)), value(std::move(r.value)) {}
 	void operator=(MapPair&& r) BOOST_NOEXCEPT { key = std::move(r.key); value = std::move(r.value); }
 
-	int compare(MapPair<Key, Value> const& r) const { return ::compare(key, r.key); }
-	template <class CompatibleWithKey>
-	int compare(CompatibleWithKey const& r) const {
-		return ::compare(key, r);
-	}
 	bool operator<(MapPair<Key,Value> const& r) const { return key < r.key; }
 	bool operator<=(MapPair<Key,Value> const& r) const { return key <= r.key; }
 	bool operator==(MapPair<Key,Value> const& r) const { return key == r.key; }
@@ -266,11 +260,6 @@ public:
 //private: MapPair( const MapPair& );
 };
 
-template <class Key, class Value, class CompatibleWithKey>
-inline int compare(CompatibleWithKey const& l, MapPair<Key, Value> const& r) {
-	return compare(l, r.key);
-}
-
 template <class Key, class Value>
 inline MapPair<typename std::decay<Key>::type, typename std::decay<Value>::type> mapPair(Key&& key, Value&& value) { return MapPair<typename std::decay<Key>::type, typename std::decay<Value>::type>(std::forward<Key>(key), std::forward<Value>(value)); }
 
@@ -625,8 +614,8 @@ typename IndexedSet<T,Metric>::iterator IndexedSet<T,Metric>::insert(T_&& data,
 	int d; // direction
 	// traverse to find insert point
 	while (true){
-		int cmp = compare(data, t->data);
+		d = t->data < data;
-		if (cmp == 0) {
+		if (!d && !(data < t->data)) {	// t->data == data
 			Node *returnNode = t;
 			if(replaceExisting) {
 				t->data = std::forward<T_>(data);
@@ -644,7 +633,6 @@ typename IndexedSet<T,Metric>::iterator IndexedSet<T,Metric>::insert(T_&& data,
 
 			return returnNode;
 		}
-		d = cmp > 0;
 		Node *nextT = t->child[d];
 		if (!nextT) break;
 		t = nextT;
@@ -701,7 +689,7 @@ int IndexedSet<T,Metric>::insert(const std::vector<std::pair<T,Metric>>& dataVec
 		int d = 1; // direction
 		if(blockStart == NULL || (blockEnd != NULL && data >= blockEnd->data)) {
 			blockEnd = NULL;
-			if (root == NULL) {
+			if (root == NULL){
 				root = new Node(std::move(data), metric);
 				num_inserted++;
 				blockStart = root;
@@ -711,12 +699,11 @@ int IndexedSet<T,Metric>::insert(const std::vector<std::pair<T,Metric>>& dataVec
 			Node *t = root;
 			// traverse to find insert point
 			bool foundNode = false;
-			while (true) {
+			while (true){
-				int cmp = compare(data, t->data);
+				d = t->data < data;
-				d = cmp >= 0;
 				if (!d)
 					blockEnd = t;
-				if (cmp == 0) {
+				if (!d && !(data < t->data)) {	// t->data == data
 					Node *returnNode = t;
 					if(replaceExisting) {
 						num_inserted++;
@@ -797,8 +784,7 @@ int IndexedSet<T,Metric>::insert(const std::vector<std::pair<T,Metric>>& dataVec
 }
 
 template <class T, class Metric>
-Metric IndexedSet<T, Metric>::eraseHalf(Node* start, Node* end, int eraseDir, int& heightDelta,
+Metric IndexedSet<T,Metric>::eraseHalf( Node* start, Node* end, int eraseDir, int& heightDelta, std::vector<Node*>& toFree ) {
-                                        std::vector<Node*>& toFree) {
 	// Removes all nodes between start (inclusive) and end (exclusive) from the set, where start is equal to end or one of its descendants
 	// eraseDir 1 means erase the right half (nodes > at) of the left subtree of end.  eraseDir 0 means the left half of the right subtree
 	// toFree is extended with the roots of completely removed subtrees
@@ -890,8 +876,8 @@ void IndexedSet<T,Metric>::erase( typename IndexedSet<T,Metric>::iterator begin,
 	
 	// Erase all matching nodes that descend from subRoot, by first erasing descendants of subRoot->child[0] and then erasing the descendants of subRoot->child[1]
 	// subRoot is not removed from the tree at this time
-	metricDelta = metricDelta + eraseHalf(first, subRoot, 1, leftHeightDelta, toFree);
+	metricDelta = metricDelta + eraseHalf( first, subRoot, 1, leftHeightDelta, toFree );
-	metricDelta = metricDelta + eraseHalf(last, subRoot, 0, rightHeightDelta, toFree);
+	metricDelta = metricDelta + eraseHalf( last, subRoot, 0, rightHeightDelta, toFree );
 
 	// Change in the height of subRoot due to past activity, before subRoot is rebalanced. subRoot->balance already reflects changes in height to its children.
 	int heightDelta = leftHeightDelta + rightHeightDelta; 
@@ -1009,9 +995,10 @@ template <class Key>
 typename IndexedSet<T,Metric>::iterator IndexedSet<T,Metric>::find(const Key &key) const {
 	Node* t = root;
 	while (t){
-		int cmp = compare(key, t->data);
+		int d = t->data < key;
-		if (cmp == 0) return iterator(t);
+		if (!d && !(key < t->data)) // t->data == key
-		t = t->child[cmp > 0];
+			return iterator(t);
+		t = t->child[d];
 	}
 	return end();
 }
@@ -1022,15 +1009,14 @@ template <class Key>
 typename IndexedSet<T,Metric>::iterator IndexedSet<T,Metric>::lower_bound(const Key &key) const {
 	Node* t = root;
 	if (!t) return iterator();
-	bool less;
 	while (true) {
-		less = t->data < key;
+		Node *n = t->child[ t->data < key ];
-		Node* n = t->child[less];
 		if (!n) break;
 		t = n;
 	}
 
-	if (less) moveIterator<1>(t);
+	if (t->data < key)
+		moveIterator<1>(t);
 
 	return iterator(t);
 }
@@ -1041,15 +1027,14 @@ template <class Key>
 typename IndexedSet<T,Metric>::iterator IndexedSet<T,Metric>::upper_bound(const Key &key) const {
 	Node* t = root;
 	if (!t) return iterator();
-	bool not_less;
 	while (true) {
-		not_less = !(key < t->data);
+		Node *n = t->child[ !(key < t->data) ];
-		Node* n = t->child[not_less];
 		if (!n) break;
 		t = n;
 	}
 
-	if (not_less) moveIterator<1>(t);
+	if (!(key < t->data))
+		moveIterator<1>(t);
 
 	return iterator(t);
 }

flow/TDMetric.actor.h

@@ -28,6 +28,7 @@
         #define FLOW_TDMETRIC_ACTOR_H
 
 #include "flow/flow.h"
+#include "flow/IndexedSet.h"
 #include "flow/network.h"
 #include "flow/Knobs.h"
 #include "flow/genericactors.actor.h"
@@ -55,21 +56,9 @@ struct MetricNameRef {
 	int expectedSize() const {
 		return type.expectedSize() + name.expectedSize();
 	}
-
-	inline int compare(MetricNameRef const& r) const {
-		int cmp;
-		if ((cmp = type.compare(r.type))) {
-			return cmp;
-		}
-		if ((cmp = name.compare(r.name))) {
-			return cmp;
-		}
-		return id.compare(r.id);
-	}
 };
 
 extern std::string reduceFilename(std::string const &filename);
-
 inline bool operator < (const MetricNameRef& l, const MetricNameRef& r ) {
 	int cmp = l.type.compare(r.type);
 	if(cmp == 0) {

flow/TreeBenchmark.h

@@ -1,123 +0,0 @@
-/*
- * IndexedSet.h
- *
- * This source file is part of the FoundationDB open source project
- *
- * Copyright 2020-2020 Apple Inc. and the FoundationDB project authors
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-#ifndef FLOW_TREEBENCHMARK_H
-#define FLOW_TREEBENCHMARK_H
-#pragma once
-
-#include "flow/flow.h"
-
-struct opTimer {
-	double start = timer();
-	char const* name;
-	int opCount;
-
-	opTimer(char const* name, int opCount) : name(name), opCount(opCount) {}
-
-	~opTimer() { printf("%s: %0.1f Kop/s\n", name, (opCount / 1000.0) / (timer() - start)); }
-};
-
-template <typename F, typename T>
-void timedRun(char const* name, T& t, F f) {
-	opTimer timer(name, t.size());
-	for (auto& i : t) {
-		f(i);
-	}
-}
-
-template <typename K>
-struct MapHarness {
-	using map = std::map<K, int>;
-	using key_type = K;
-
-	struct result {
-		typename map::const_iterator it;
-
-		result(typename map::const_iterator it) : it(it) {}
-
-		result& operator++() {
-			it++;
-			return *this;
-		}
-
-		const K& operator*() const { return (*it).first; }
-
-		const K& operator->() const { return it->first; }
-
-		bool operator==(result const& k) const { return it == k.it; }
-	};
-
-	map s;
-
-	void insert(K const& k) { s.insert(std::pair<K, int>(k, 1)); }
-	result find(K const& k) const { return result(s.find(k)); }
-	result not_found() const { return result(s.end()); }
-	result begin() const { return result(s.begin()); }
-	result end() const { return result(s.end()); }
-	result lower_bound(K const& k) const { return result(s.lower_bound(k)); }
-	result upper_bound(K const& k) const { return result(s.upper_bound(k)); }
-	void erase(K const& k) { s.erase(k); }
-};
-
-template <typename T, typename F>
-void treeBenchmark(T& tree, F generateKey) {
-	using key = typename T::key_type;
-
-	int keyCount = 1000000;
-
-	std::vector<key> keys;
-	for (int i = 0; i < keyCount; i++) {
-		keys.push_back(generateKey());
-	}
-
-	timedRun("insert", keys, [&tree](key const& k) { tree.insert(k); });
-	timedRun("find", keys, [&tree](key const& k) { ASSERT(tree.find(k) != tree.not_found()); });
-	timedRun("lower_bound", keys, [&tree](key const & k) { ASSERT(tree.lower_bound(k) != tree.not_found()); });
-	timedRun("upper_bound", keys, [&tree](key const & k) { tree.upper_bound(k); });
-
-
-	std::sort(keys.begin(), keys.end());
-	keys.resize(std::unique(keys.begin(), keys.end()) - keys.begin());
-
-	auto iter = tree.lower_bound(*keys.begin());
-	timedRun("scan", keys, [&tree, &iter](key const& k) {
-		ASSERT(k == *iter);
-		++iter;
-	});
-	ASSERT(iter == tree.end());
-
-	timedRun("find (sorted)", keys, [&tree](key const& k) { ASSERT(tree.find(k) != tree.end()); });
-
-	std::random_shuffle(keys.begin(), keys.end());
-
-	timedRun("erase", keys, [&tree](key const& k) { tree.erase(k); });
-	ASSERT(tree.begin() == tree.end());
-}
-
-static inline StringRef randomStr(Arena& arena) {
-	size_t keySz = 100;
-	return StringRef(arena, deterministicRandom()->randomAlphaNumeric(keySz));
-}
-
-static inline int randomInt() {
-	return deterministicRandom()->randomInt(0, INT32_MAX);
-}
-
-#endif // FLOW_TREEBENCHMARK_H