Stored set

8 files changed, 412 insertions, 0 deletions

diff --git a/include/erebos/merge.h b/include/erebos/merge.h
new file mode 100644
index 0000000..bef8212
--- /dev/null
+++ b/include/erebos/merge.h
@@ -0,0 +1,20 @@
+#pragma once
+
+#include <erebos/storage.h>
+
+namespace erebos
+{
+
+template<class T> struct Mergeable
+{
+};
+
+template<> struct Mergeable<vector<Stored<Object>>>
+{
+	using Component = Object;
+
+	static vector<Stored<Object>> components(const vector<Stored<Object>> & x) { return x; }
+	static vector<Stored<Object>> merge(const vector<Stored<Object>> & x) { return x; }
+};
+
+}
diff --git a/include/erebos/set.h b/include/erebos/set.h
new file mode 100644
index 0000000..f625cb0
--- /dev/null
+++ b/include/erebos/set.h
@@ -0,0 +1,96 @@
+#pragma once
+
+#include <erebos/merge.h>
+#include <erebos/storage.h>
+
+namespace erebos
+{
+
+class SetViewBase;
+template<class T> class SetView;
+
+class SetBase
+{
+protected:
+	struct Priv;
+
+	SetBase();
+	SetBase(const vector<Ref> &);
+	SetBase(shared_ptr<const Priv>);
+	
+	shared_ptr<const Priv> add(Storage &, const vector<Ref> &) const;
+
+	vector<vector<Ref>> toList() const;
+
+public:
+	vector<Digest> digests() const;
+
+protected:
+	shared_ptr<const Priv> p;
+};
+
+template<class T>
+class Set : public SetBase
+{
+	Set(shared_ptr<const Priv> p): SetBase(p) {};
+public:
+	Set() = default;
+	Set(const vector<Ref> & refs): SetBase(move(refs)) {}
+	Set(const Set<T> &) = default;
+	Set(Set<T> &&) = default;
+	Set & operator=(const Set<T> &) = default;
+	Set & operator=(Set<T> &&) = default;
+
+	static Set<T> load(const vector<Ref> & refs) { return Set<T>(move(refs)); }
+
+	Set<T> add(Storage &, const T &) const;
+
+	template<class F>
+	SetView<T> view(F && cmp) const;
+};
+
+template<class T>
+class SetView
+{
+public:
+	template<class F>
+	SetView(F && cmp, const vector<vector<Ref>> & refs);
+
+	typename vector<T>::const_iterator begin() const { return items.begin(); }
+	typename vector<T>::const_iterator end() const { return items.end(); }
+
+private:
+	vector<T> items;
+};
+
+template<class T>
+Set<T> Set<T>::add(Storage & st, const T & x) const
+{
+	return Set<T>(SetBase::add(st, storedRefs(Mergeable<T>::components(x))));
+}
+
+template<class T>
+template<class F>
+SetView<T> Set<T>::view(F && cmp) const
+{
+	return SetView<T>(std::move(cmp), toList());
+}
+
+template<class T>
+template<class F>
+SetView<T>::SetView(F && cmp, const vector<vector<Ref>> & refs)
+{
+	items.reserve(refs.size());
+	for (const auto & crefs : refs) {
+		vector<Stored<typename Mergeable<T>::Component>> comps;
+		comps.reserve(crefs.size());
+		for (const auto & r : crefs)
+			comps.push_back(Stored<typename Mergeable<T>::Component>::load(r));
+
+		filterAncestors(comps);
+		items.push_back(Mergeable<T>::merge(comps));
+	}
+	std::sort(items.begin(), items.end(), cmp);
+}
+
+}
diff --git a/include/erebos/storage.h b/include/erebos/storage.h
index 15ee0bb..735b399 100644
--- a/include/erebos/storage.h
+++ b/include/erebos/storage.h
@@ -668,6 +668,16 @@ WatchedHead<T>::~WatchedHead()
 				T::headTypeId, Head<T>::id(), watcherId);
 }
 
+template<class T>
+vector<Ref> storedRefs(const vector<Stored<T>> & v)
+{
+	vector<Ref> res;
+	res.reserve(v.size());
+	for (const auto & x : v)
+		res.push_back(x.ref());
+	return res;
+}
+
 }
 
 namespace std
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index b68860a..72094b8 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -13,6 +13,7 @@ add_library(erebos
 	pairing.cpp
 	pubkey.cpp
 	service.cpp
+	set.cpp
 	state.cpp
 	storage.cpp
 	sync.cpp
diff --git a/src/main.cpp b/src/main.cpp
index 4dc4582..95d3eeb 100644
--- a/src/main.cpp
+++ b/src/main.cpp
@@ -1,6 +1,7 @@
 #include <erebos/attach.h>
 #include <erebos/identity.h>
 #include <erebos/network.h>
+#include <erebos/set.h>
 #include <erebos/storage.h>
 #include <erebos/sync.h>
 
@@ -162,6 +163,39 @@ void storedRoots(const vector<string> & args)
 	printLine(ss.str());
 }
 
+void storedSetAdd(const vector<string> & args)
+{
+	auto iref = st.ref(Digest(args.at(0)));
+	if (!iref)
+		throw invalid_argument("ref " + args.at(0) + " not found");
+
+	auto set = args.size() > 1 ?
+		Set<vector<Stored<Object>>>::load({ *st.ref(Digest(args.at(1))) }) :
+		Set<vector<Stored<Object>>>();
+
+	ostringstream ss;
+	ss << "stored-set-add";
+	for (const auto & d : set.add(st, { Stored<Object>::load(*iref) }).digests())
+		ss << " " << string(d);
+	printLine(ss.str());
+}
+
+void storedSetList(const vector<string> & args)
+{
+	auto ref = st.ref(Digest(args.at(0)));
+	if (!ref)
+		throw invalid_argument("ref " + args.at(0) + " not found");
+
+	for (const auto & vec : Set<vector<Stored<Object>>>::load({ *ref }).view(std::less{})) {
+		ostringstream ss;
+		ss << "stored-set-item";
+		for (const auto & x : vec)
+			ss << " " << string(x.ref().digest());
+		printLine(ss.str());
+	}
+	printLine("stored-set-done");
+}
+
 void createIdentity(const vector<string> & args)
 {
 	optional<Identity> identity;
@@ -353,6 +387,8 @@ vector<Command> commands = {
 	{ "store", store },
 	{ "stored-generation", storedGeneration },
 	{ "stored-roots", storedRoots },
+	{ "stored-set-add", storedSetAdd },
+	{ "stored-set-list", storedSetList },
 	{ "create-identity", createIdentity },
 	{ "start-server", startServer },
 	{ "stop-server", stopServer },
diff --git a/src/set.cpp b/src/set.cpp
new file mode 100644
index 0000000..001bce3
--- /dev/null
+++ b/src/set.cpp
@@ -0,0 +1,171 @@
+#include "set.h"
+
+#include <unordered_map>
+#include <unordered_set>
+#include <utility>
+
+namespace erebos {
+
+using std::pair;
+using std::unordered_map;
+using std::unordered_set;
+using std::move;
+
+SetBase::SetBase():
+	p(make_shared<Priv>())
+{
+}
+
+SetBase::SetBase(const vector<Ref> & refs)
+{
+	vector<Stored<SetItem>> items;
+	for (const auto & r : refs)
+		items.push_back(Stored<SetItem>::load(r));
+
+	p = shared_ptr<Priv>(new Priv {
+		.items = move(items),
+	});
+}
+
+SetBase::SetBase(shared_ptr<const Priv> p_):
+	p(move(p_))
+{
+}
+
+shared_ptr<const SetBase::Priv> SetBase::add(Storage & st, const vector<Ref> & refs) const
+{
+	auto item = st.store(SetItem {
+		.prev = p->items,
+		.item = refs,
+	});
+
+	return shared_ptr<const Priv>(new Priv {
+		.items = { move(item) },
+	});
+}
+
+static void gatherSetItems(unordered_set<Digest> & seenSet, unordered_set<Digest> & seenElem,
+		vector<Ref> & res, const Stored<SetItem> & item)
+{
+	if (!seenElem.insert(item.ref().digest()).second)
+		return;
+
+	for (const auto & r : item->item)
+		if (seenSet.insert(r.digest()).second)
+			res.push_back(r);
+
+	for (const auto & p : item->prev)
+		gatherSetItems(seenSet, seenElem, res, p);
+}
+
+vector<vector<Ref>> SetBase::toList() const
+{
+	/* Splits the graph starting from all set item refs into connected
+	 * components (partitions), each such partition makes one set item,
+	 * merged together in the templated SetView constructor. */
+
+	// Gather all item references
+	vector<Ref> items;
+	{
+		unordered_set<Digest> seenSet, seenElem;
+		for (const auto & i : p->items)
+			gatherSetItems(seenSet, seenElem, items, i);
+	}
+
+	unordered_map<Digest, unsigned> partMap; // maps item ref to partition number
+	vector<unsigned> partMerge; // maps partitions to resulting one after partition merge
+
+	// Use (cached) root set for assigning partition numbers
+	for (const auto & item : items) {
+		const auto roots = item.roots();
+		unsigned part = partMerge.size();
+
+		// If any root has partition number already, pick the smallest one
+		for (const auto & rdgst : roots) {
+			auto it = partMap.find(rdgst);
+			if (it != partMap.end() && it->second < part)
+				part = it->second;
+		}
+
+		// Update partition number for the roots and if this item
+		// merges some partitions, also update the merge info
+		for (const auto & rdgst : roots) {
+			auto it = partMap.find(rdgst);
+			if (it == partMap.end()) {
+				partMap.emplace(rdgst, part);
+			} else if (it->second != part) {
+				partMerge[it->second] = part;
+				it->second = part;
+			}
+		}
+
+		// If no existing partition has been touched, mark a new one
+		if (part == partMerge.size())
+			partMerge.push_back(part);
+
+		// And store resulting partition number
+		partMap.emplace(item.digest(), part);
+	}
+
+	// Get all the refs for each partition
+	vector<vector<Ref>> res(partMerge.size());
+	for (const auto & item : items) {
+		unsigned part = partMap[item.digest()];
+		for (unsigned p = partMerge[part]; p != part; p = partMerge[p])
+			part = p;
+		res[part].push_back(item);
+	}
+
+	// Remove empty elements (merged partitions) from result list
+	res.erase(std::remove(res.begin(), res.end(), vector<Ref>()), res.end());
+
+	return res;
+}
+
+vector<Digest> SetBase::digests() const
+{
+	vector<Digest> res;
+	res.reserve(p->items.size());
+	for (const auto & i : p->items)
+		res.push_back(i.ref().digest());
+	return res;
+}
+
+SetItem SetItem::load(const Ref & ref)
+{
+	if (auto rec = ref->asRecord()) {
+		vector<Stored<SetItem>> prev;
+		for (auto p : rec->items("PREV"))
+			if (const auto & x = p.as<SetItem>())
+				prev.push_back(*x);
+
+		vector<Ref> item;
+		for (auto i : rec->items("item"))
+			if (const auto & x = i.asRef())
+				item.push_back(*x);
+
+		return SetItem {
+			.prev = std::move(prev),
+			.item = std::move(item),
+		};
+	}
+
+	return SetItem {
+		.prev = {},
+		.item = {},
+	};
+}
+
+Ref SetItem::store(const Storage & st) const
+{
+	vector<Record::Item> items;
+
+	for (const auto & p : prev)
+		items.emplace_back("PREV", p.ref());
+	for (const auto & r : item)
+		items.emplace_back("item", r);
+
+	return st.storeObject(Record(std::move(items)));
+}
+
+}
diff --git a/src/set.h b/src/set.h
new file mode 100644
index 0000000..ffbcbd6
--- /dev/null
+++ b/src/set.h
@@ -0,0 +1,19 @@
+#include <erebos/set.h>
+
+namespace erebos {
+
+struct SetItem
+{
+	static SetItem load(const Ref &);
+	Ref store(const Storage & st) const;
+
+	const vector<Stored<SetItem>> prev;
+	const vector<Ref> item;
+};
+
+struct SetBase::Priv
+{
+	vector<Stored<SetItem>> items;
+};
+
+}
diff --git a/test/storage.test b/test/storage.test
index 10f5d2c..12e267d 100644
--- a/test/storage.test
+++ b/test/storage.test
@@ -1,6 +1,9 @@
 test:
 	spawn on node1 as p1
 
+	# Root finding
+	###############
+
 	# Diamond history
 	send to p1:
 		"store rec"
@@ -91,3 +94,59 @@ test:
 
 	send to p1 "stored-roots $r2_2"
 	expect from p1 /stored-roots $r2_2 $r2_1/
+
+
+	# Set
+	#####
+
+	send to p1 "stored-set-add $r1"
+	expect from p1 /stored-set-add (blake2#[0-9a-f]*)/ capture s1
+	send to p1 "stored-set-add $r2 $s1"
+	expect from p1 /stored-set-add (blake2#[0-9a-f]*)/ capture s2
+	send to p1 "stored-set-add $r3 $s2"
+	expect from p1 /stored-set-add (blake2#[0-9a-f]*)/ capture s3
+	send to p1 "stored-set-add $r4 $s3"
+	expect from p1 /stored-set-add (blake2#[0-9a-f]*)/ capture s4
+
+	send to p1 "stored-set-list $s1"
+	expect from p1:
+		/stored-set-item $r1/
+		/stored-set-(.*)/ capture done1
+	guard done1 == "done"
+
+	send to p1 "stored-set-list $s2"
+	expect from p1:
+		/stored-set-item $r2/
+		/stored-set-(.*)/ capture done2
+	guard done2 == "done"
+
+	send to p1 "stored-set-list $s3"
+	expect from p1:
+		/stored-set-item $r2 $r3/
+		/stored-set-(.*)/ capture done3
+	guard done3 == "done"
+
+	send to p1 "stored-set-list $s4"
+	expect from p1:
+		/stored-set-item $r4/
+		/stored-set-(.*)/ capture done4
+	guard done4 == "done"
+
+
+	send to p1 "stored-set-add $r2_2 $s4"
+	expect from p1 /stored-set-add (blake2#[0-9a-f]*)/ capture s5
+	send to p1 "stored-set-add $r2_3 $s5"
+	expect from p1 /stored-set-add (blake2#[0-9a-f]*)/ capture s6
+
+	send to p1 "stored-set-list $s5"
+	expect from p1:
+		/stored-set-item $r4/
+		/stored-set-item $r2_2/
+		/stored-set-(.*)/ capture done5
+	guard done5 == "done"
+
+	send to p1 "stored-set-list $s6"
+	expect from p1:
+		/stored-set-item $r2_3/
+		/stored-set-(.*)/ capture done6
+	guard done6 == "done"