Refactor to broadcastCall and allow member functions to be RPC functions

p2p-rm/include/ftl/p2p-rm/blob.hpp

@@ -19,6 +19,7 @@ struct Blob {
 	char *data_;
 	size_t size_;
 	std::string uri_;
+	std::string owner_;
 	uint32_t blobid_;
 	
 	void finished();

p2p-rm/include/ftl/p2p-rm/cluster.hpp

@@ -95,17 +95,51 @@ class Cluster {
 	/**
 	 * Connect to a new peer node using the specified socket.
 	 */
-	void addPeer(std::shared_ptr<ftl::net::Socket> &s);
+	void addPeer(std::shared_ptr<ftl::net::Socket> &s, bool incoming=false);
 
 	/**
 	 * Connect to a new peer using a URL string.
 	 */
 	void addPeer(const char *url);
 	
-	std::string getOwner(const char *uri);
+	/**
+	 * Allow member functions to be used for RPC calls by binding with 'this'.
+	 */
+	template <typename R, typename C, typename ...Args>
+	auto bind(R(C::*f)(Args...)) {
+	  return [this,f](Args... args) -> R { return (this->*f)(std::forward<Args>(args)...); };
+	}
+	
+	std::string getOwner(const std::string &uri);
+	
+	/**
+	 * Make an RPC call to all connected peers and put into a results vector.
+	 * This function blocks until all peers have responded or an error /
+	 * timeout occurs. The return value indicates the number of failed peers, or
+	 * is 0 if all returned.
+	 */
+	template <typename T, typename... ARGS>
+	int broadcastCall(const std::string &name, std::vector<T> &results,
+			ARGS... args) {
+		int count = 0;
+		auto f = [&count,&results](msgpack::object &r) {
+			count--;
+			results.push_back(r.as<T>());
+		};
+
+		for (auto p : peers_) {
+			count++;
+			p->async_call(name, f, std::forward<ARGS>(args)...);
+		}
+		
+		// TODO Limit in case of no return.
+		while (count > 0) {
+			ftl::net::wait();
+		}
+		return count;
+	}
 	
 	private:
-	static int rpcid_;
 	std::string root_;
 	std::shared_ptr<ftl::net::Listener> listener_;
 	std::vector<std::shared_ptr<ftl::net::Socket>> peers_;
@@ -115,6 +149,7 @@ class Cluster {
 	ftl::rm::Blob *_lookup(const char *uri);
 	Blob *_create(const char *uri, char *addr, size_t size, size_t count,
 		ftl::rm::flags_t flags, const std::string &tname);
+	void _registerRPC(ftl::net::Socket &s);
 };
 
 };

p2p-rm/src/blob.cpp

@@ -2,6 +2,8 @@
 #include <ftl/net/socket.hpp>
 #include <ftl/p2p-rm/protocol.hpp>
 
+#include <iostream>
+
 struct SyncHeader {
 	uint32_t blobid;
 	uint32_t offset;
@@ -10,7 +12,10 @@ struct SyncHeader {
 
 void ftl::rm::_sync(const Blob &blob, size_t offset, size_t size) {
 	// Sanity check
-	if (offset + size > blob.size_) throw -1;
+	if (offset + size > blob.size_) {
+		LOG(ERROR) << "Memory overrun during sync";
+		return;
+	}
 
 	// TODO Delay send to collate many write operations?
 

p2p-rm/src/cluster.cpp

@@ -10,24 +10,25 @@
 #include <map>
 #include <string>
 #include <iostream>
+#include <vector>
 
 using ftl::rm::Cluster;
 using ftl::net::Listener;
 using std::map;
+using std::vector;
 using std::shared_ptr;
+using std::string;
 using ftl::URI;
 using ftl::rm::Blob;
 using ftl::net::Socket;
 
-int Cluster::rpcid_ = 0;
-
 Cluster::Cluster(const URI &uri, shared_ptr<Listener> l) : listener_(l) {
 	//auto me = this;
 	root_ = uri.getHost();
 
 	if (l != nullptr) {
 		l->onConnection([&](shared_ptr<Socket> &s) {
-			addPeer(s);		
+			addPeer(s, true);		
 		});
 	}
 }
@@ -43,21 +44,27 @@ void Cluster::reset() {
 	blobs_.clear();
 }
 
-void Cluster::addPeer(shared_ptr<Socket> &p) {
-	LOG(INFO) << "Peer added: " << p->getURI();
-	//auto me = this;
+void Cluster::_registerRPC(Socket &s) {
+	//s.bind("getowner", [this](const std::string &u) { getOwner(u.c_str()); });
+	s.bind("getowner", bind(&Cluster::getOwner));
+}
+
+void Cluster::addPeer(shared_ptr<Socket> &p, bool incoming) {
+	LOG(INFO) << ((incoming) ? "Incoming peer added: " : "Peer added: ") << p->getURI();
+
 	peers_.push_back(p);
-	/*p->onMessage([&](int service, const std::string &data) {
-		std::cout << "MSG " << service << std::endl;
-	});*/
-	p->bind("getowner", [](const std::string &uri) -> std::string {
-		std::cout << "GETOWNER" << std::endl;
-		return "";
-	});
+	_registerRPC(*p);
 	
-	// TODO Check ownership of my blobs.
-	for (auto b : blobs_) {
-		getOwner(b.first.c_str());
+	if (!incoming) {
+		p->onConnect([this](Socket &s) {
+			for (auto b : blobs_) {
+				auto o = s.call<string>("getowner", b.first);
+				if (o.size() > 0) {
+					b.second->owner_ = o;
+					LOG(INFO) << "Lost ownership of " << b.first.c_str() << " to " << o;
+				}
+			}
+		});
 	}
 }
 
@@ -77,34 +84,23 @@ Blob *Cluster::_lookup(const char *uri) {
 	std::cout << "Blob Found for " << u.getBaseURI() << " = " << (b != nullptr) << std::endl;
 
 	if (!b) {
-		// Whoops, need to map it first!
+		LOG(WARNING) << "Unmapped memory requested: " << uri;
 	}
 
 	return b;
 }
 
-std::string Cluster::getOwner(const char *uri) {
-	std::string result;
-	int count = 0;
-
-	auto f = [&](msgpack::object &r) {
-		count--;
-		std::string res = r.as<std::string>();
-		if (res.size() > 0) result = res;
-	};
-
-	for (auto p : peers_) {
-		count++;
-		LOG(INFO) << "Request owner of " << uri << " from " << p->getURI();
-		p->async_call("getowner", f, std::string(uri));
-	}
+std::string Cluster::getOwner(const std::string &uri) {
+	vector<string> results;
 	
-	// TODO Limit in case of no return.
-	while (count > 0) {
-		ftl::net::wait();
-	}
+	std::cout << "GETOWNER" << std::endl;
+	if (blobs_.count(uri) != 0) return blobs_[uri]->owner_;
+
+	broadcastCall("getowner", results, uri);
 	
-	return result;
+	// TODO Verify all results are equal or empty
+	if (results.size() == 0) return "";
+	return results[0];
 }
 
 Blob *Cluster::_create(const char *uri, char *addr, size_t size, size_t count,
@@ -121,14 +117,17 @@ Blob *Cluster::_create(const char *uri, char *addr, size_t size, size_t count,
 	b->data_ = addr;
 	b->size_ = size;
 	b->uri_ = std::string(uri);
+	b->owner_ = "";
 	blobs_[u.getBaseURI()] = b;
 
 	std::string o = getOwner(uri);
 	if (o.size() == 0) {
 		// I am the owner!
 		std::cout << "I own " << uri << std::endl;
+		b->owner_ = "me";
 	} else {
 		std::cout << "I do not own " << uri << std::endl;
+		b->owner_ = o;
 	}
 	
 	//std::cout << owners << std::endl;

p2p-rm/test/CMakeLists.txt

@@ -6,6 +6,7 @@ add_executable(mapped_ptr EXCLUDE_FROM_ALL
 	./mapped_ptr.cpp
 	../src/blob.cpp
 )
+target_link_libraries(mapped_ptr gflags glog)
 
 add_executable(p2p_rm EXCLUDE_FROM_ALL
 	./tests.cpp
@@ -15,7 +16,7 @@ add_executable(p2p_rm EXCLUDE_FROM_ALL
 	
 	./p2p-rm.cpp
 )
-target_link_libraries(p2p_rm uriparser)
+target_link_libraries(p2p_rm uriparser ftlnet gflags glog)
 
 add_executable(peer_test EXCLUDE_FROM_ALL
 	./peer.cpp

p2p-rm/test/mapped_ptr.cpp

@@ -22,6 +22,7 @@ SCENARIO( "Reading from a remote pointer", "[remote_ptr]" ) {
 	// Make a dummy blob
 	auto blob = new ftl::rm::Blob();
 	blob->data_ = (char*)(new int[5]);
+	blob->size_ = 5*sizeof(int);
 	((int*)(blob->data_))[0] = 55;
 	((int*)(blob->data_))[1] = 66;
 	
@@ -37,6 +38,7 @@ SCENARIO( "Writing to a remote pointer", "[remote_ptr]" ) {
 	// Make a dummy blob
 	auto blob = new ftl::rm::Blob();
 	blob->data_ = (char*)(new int[5]);
+	blob->size_ = 5*sizeof(int);
 	((int*)(blob->data_))[0] = 55;
 	((int*)(blob->data_))[1] = 66;
 	
@@ -63,6 +65,7 @@ SCENARIO( "Writing to readonly pointer fails", "[remote_ptr]" ) {
 	// Make a dummy blob
 	auto blob = new ftl::rm::Blob();
 	blob->data_ = (char*)(new int[5]);
+	blob->size_ = 5*sizeof(int);
 	((int*)(blob->data_))[0] = 55;
 	((int*)(blob->data_))[1] = 66;
 	

p2p-rm/test/p2p-rm.cpp

@@ -7,7 +7,7 @@
 
 // --- Mock Socket Send
 
-static std::vector<uint32_t> msgs;
+/*static std::vector<uint32_t> msgs;
 
 int ftl::net::Socket::send2(uint32_t service, const std::string &data1, const std::string &data2) {
 	msgs.push_back(service);
@@ -29,12 +29,6 @@ int ftl::net::Socket::send(uint32_t service, const std::string &data) {
 	msgs.push_back(service);
 	std::cout << "SEND (" << service << ")" << std::endl;
 
-	/*if (service == P2P_RPC_CALL) {
-		// UNPACK
-		// PACK RETURN
-		message(P2P_RPC_RETURN, rdata);
-	}*/
-
 	return 0;
 }
 
@@ -44,7 +38,7 @@ bool ftl::net::wait() {
 
 std::shared_ptr<ftl::net::Socket> ftl::net::connect(const char *url) {
 	return nullptr;
-}
+}*/
 
 // --- End Mock Socket Send
 
@@ -103,7 +97,8 @@ SCENARIO( "Cluster::map()", "[map]" ) {
 SCENARIO( "Getting a read_ref", "[get]" ) {
 	auto cluster = ftl::rm::cluster("ftl://utu.fi", nullptr);
 	// Add fake peer
-	cluster->addPeer(std::shared_ptr<ftl::net::Socket>(new ftl::net::Socket(0)));
+	auto p = std::make_shared<ftl::net::Socket>(0);
+	cluster->addPeer(p);
 	
 	int data = 89;
 	int data2 = 99;