Stored set

4 files changed, 227 insertions, 0 deletions

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;
+};
+
+}