first commit

CMakeLists.txt

+#
+# CMakeLists.txt  top-level cmake file for mdcs
+# 13-Oct-2017  mdorier@anl.gov
+#
+
+#
+#  general cmake flags:
+#    -DCMAKE_INSTALL_PREFIX=/usr/local     -- the prefix for installing
+#    -DCMAKE_BUILD_TYPE=type               -- type can be Debug, Release, ...
+#    -DCMAKE_PREFIX_PATH=/dir              -- external packages
+#
+#     note that CMAKE_PREFIX_PATH can be a list of directories:
+#      -DCMAKE_PREFIX_PATH='/dir1;/dir2;/dir3'
+#
+
+cmake_minimum_required (VERSION 3.0)
+project (thallium CXX)
+enable_testing ()
+
+macro (use_cxx14)
+  if (CMAKE_VERSION VERSION_LESS "3.1")
+    if (CMAKE_CXX_COMPILER_ID STREQUAL "GNU")
+      set (CMAKE_CXX_FLAGS "--std=gnu++14 ${CMAKE_CXX_FLAGS}")
+    else ()
+      set (CMAKE_CXX_FLAGS "--std=c++14 ${CMAKE_CXX_FLAGS}")
+    endif ()
+  else ()
+    set (CMAKE_CXX_STANDARD 14)
+  endif ()
+endmacro (use_cxx14)
+
+use_cxx14 ()
+
+# add our cmake module directory to the path
+set (CMAKE_MODULE_PATH ${CMAKE_MODULE_PATH}
+     "${CMAKE_CURRENT_SOURCE_DIR}/cmake")
+
+# link shared lib with full rpath
+set (CMAKE_INSTALL_RPATH "${CMAKE_INSTALL_PREFIX}/lib")
+set (CMAKE_INSTALL_RPATH_USE_LINK_PATH TRUE)
+
+# setup cache variables for ccmake
+if (NOT CMAKE_BUILD_TYPE)
+    set (CMAKE_BUILD_TYPE Release
+         CACHE STRING "Choose the type of build." FORCE)
+    set_property (CACHE CMAKE_BUILD_TYPE PROPERTY STRINGS 
+                  "Debug" "Release" "RelWithDebInfo" "MinSizeRel")
+endif ()
+set (CMAKE_PREFIX_PATH "" CACHE STRING "External dependencies path")
+set (BUILD_SHARED_LIBS "OFF" CACHE BOOL "Build a shared library")
+
+# packages we depend on
+include (xpkg-import)
+find_package (mercury CONFIG REQUIRED)
+xpkg_import_module (margo REQUIRED margo)
+
+add_subdirectory (src)
+add_subdirectory (test)

include/thallium.hpp

+#ifndef __THALLIUM_HPP
+#define __THALLIUM_HPP
+
+#include <thallium/margo_engine.hpp>
+#include <thallium/endpoint.hpp>
+#include <thallium/remote_procedure.hpp>
+#include <thallium/callable_remote_procedure.hpp>
+#include <thallium/serialization.hpp>
+
+#endif

include/thallium/buffer.hpp

+#ifndef __THALLIUM_BUFFER_HPP
+#define __THALLIUM_BUFFER_HPP
+
+#include <vector>
+
+namespace thallium {
+
+using buffer = std::vector<char>;
+
+}
+
+#endif

include/thallium/callable_remote_procedure.hpp

+#ifndef __THALLIUM_CALLABLE_REMOTE_PROCEDURE_HPP
+#define __THALLIUM_CALLABLE_REMOTE_PROCEDURE_HPP
+
+#include <tuple>
+#include <cstdint>
+#include <margo.h>
+#include <thallium/buffer.hpp>
+
+namespace thallium {
+
+class margo_engine;
+class remote_procedure;
+class endpoint;
+
+class callable_remote_procedure {
+
+	friend class remote_procedure;
+
+private:
+	hg_handle_t m_handle;
+
+	callable_remote_procedure(hg_id_t id, const endpoint& ep);
+
+public:
+
+	callable_remote_procedure(const callable_remote_procedure& other) {
+		if(m_handle != HG_HANDLE_NULL) {
+			margo_destroy(m_handle);
+		}
+		m_handle = other.m_handle;
+		if(m_handle != HG_HANDLE_NULL) {
+			margo_ref_incr(m_handle);
+		}
+	}
+
+	callable_remote_procedure(callable_remote_procedure&& other) {
+		if(m_handle != HG_HANDLE_NULL) {
+            margo_destroy(m_handle);
+        }
+		m_handle = other.m_handle;
+		other.m_handle = HG_HANDLE_NULL;
+	}
+
+	callable_remote_procedure& operator=(const callable_remote_procedure& other) {
+		if(&other == this) return *this;
+		if(m_handle != HG_HANDLE_NULL) {
+			margo_destroy(m_handle);
+		}
+		m_handle = other.m_handle;
+		margo_ref_incr(m_handle);
+		return *this;
+	}
+
+	callable_remote_procedure& operator=(callable_remote_procedure&& other) {
+		if(&other == this) return *this;
+		if(m_handle != HG_HANDLE_NULL) {
+			margo_destroy(m_handle);
+		}
+		m_handle = other.m_handle;
+		other.m_handle = HG_HANDLE_NULL;
+		return *this;
+	}
+
+	~callable_remote_procedure() {
+		if(m_handle != HG_HANDLE_NULL) {
+			margo_destroy(m_handle);
+		}
+	}
+
+	template<typename ... T>
+	auto operator()(T&& ... t) const {
+		// TODO throw an exception if handle is null
+//		buffer input;
+//		BufferOutputArchive arch(input);
+//		serialize_many(arch, std::forward<T>(t)...);
+//		serialize(std::forward<T>(t)
+		//auto input = std::tie(t...);
+		margo_forward(m_handle, nullptr);//&input);
+
+/*
+		Buffer output;
+*/
+		//margo_get_output(m_handle, &output);
+		return true;//Pack(std::move(output));
+	}
+};
+
+}
+
+#endif

include/thallium/endpoint.hpp

+#ifndef __THALLIUM_ENDPOINT_HPP
+#define __THALLIUM_ENDPOINT_HPP
+
+#include <cstdint>
+#include <string>
+#include <margo.h>
+
+namespace thallium {
+
+class margo_engine;
+
+class endpoint {
+
+	friend class margo_engine;
+	friend class callable_remote_procedure;
+
+private:
+
+	margo_engine& m_margo;
+	hg_addr_t     m_addr;
+
+	endpoint(margo_engine& m, hg_addr_t addr)
+	: m_margo(m), m_addr(addr) {}
+
+public:
+
+	endpoint(const endpoint& other);
+
+	endpoint(endpoint&& other)
+	: m_margo(other.m_margo), m_addr(other.m_addr) {
+		other.m_addr = HG_ADDR_NULL;
+	}
+
+	endpoint& operator=(const endpoint& other);
+
+	endpoint& operator=(endpoint&& other);
+	
+	~endpoint();
+
+	operator std::string() const; 
+};
+
+}
+
+#endif

include/thallium/function_cast.hpp

+#ifndef __THALLIUM_FUNCTION_CAST_HPP
+#define __THALLIUM_FUNCTION_CAST_HPP
+
+#include <cstdint>
+
+namespace thallium {
+
+template<typename F>
+F* function_cast(void* f) {
+	return reinterpret_cast<F*>(reinterpret_cast<std::intptr_t>(f));
+}
+
+template<typename F>
+void* void_cast(F&& fun) {
+	return reinterpret_cast<void*>(reinterpret_cast<std::intptr_t>(fun));
+}
+
+}
+
+#endif

include/thallium/function_traits.hpp

+#ifndef __THALLIUM_FUNCTION_TRAITS_HPP
+#define __THALLIUM_FUNCTION_TRAITS_HPP
+
+namespace thallium {
+
+template<typename...>
+struct types {
+	using type=types;
+};
+
+template<class Sig> struct args;
+
+template<class R, class...Args>
+struct args<R(Args...)> : types<Args...> {};
+
+template<class Sig> using args_t = typename args<Sig>::type;
+
+/*
+EXAMPLE:
+template <class...Params>
+void some_function(types<Params...>) {
+    }
+
+    void* fopen(const char* filename, const char* mode);
+
+    int main(){
+      some_function(args_t<decltype(fopen)>{});
+    }
+
+}
+*/
+#endif

include/thallium/margo_engine.hpp

+#ifndef __THALLIUM_MARGO_ENGINE_HPP
+#define __THALLIUM_MARGO_ENGINE_HPP
+
+#include <iostream>
+#include <string>
+#include <margo.h>
+#include <thallium/function_cast.hpp>
+#include <thallium/buffer.hpp>
+#include <thallium/request.hpp>
+
+namespace thallium {
+
+class endpoint;
+class remote_procedure;
+
+class margo_engine {
+
+	friend class endpoint;
+	friend class callable_remote_procedure;
+
+private:
+
+	margo_instance_id m_mid;
+
+	template<typename F>
+	static hg_return_t rpc_callback(hg_handle_t handle) {
+		using G = std::remove_reference_t<F>;
+		const struct hg_info* info = margo_get_info(handle);
+		margo_instance_id mid = margo_hg_handle_get_instance(handle);
+		void* data = margo_registered_data(mid, info->id);
+		auto f = function_cast<G>(data);
+		request req(handle);
+		(*f)(req);
+		return HG_SUCCESS;
+	}
+
+public:
+
+	margo_engine(const std::string& addr, int mode, 
+	              bool use_progress_thread = false,
+	              std::int32_t rpc_thread_count = 0) {
+
+		m_mid = margo_init(addr.c_str(), mode,
+				use_progress_thread,
+				rpc_thread_count);
+		// TODO throw exception if m_mid is null
+	}
+
+	margo_engine(const margo_engine& other)            = delete;
+	margo_engine(margo_engine&& other)                 = delete;
+	margo_engine& operator=(margo_engine&& other)      = delete;
+	margo_engine& operator=(const margo_engine& other) = delete;
+
+	~margo_engine() {
+		// TODO throw an exception if following call fails
+		margo_wait_for_finalize(m_mid);
+	}
+
+	void finalize() {
+		// TODO throw an exception if the following call fails
+		margo_finalize(m_mid);
+	}
+
+	endpoint self() const;
+
+	template<typename F>
+	remote_procedure define(const std::string& name);
+
+	template<typename F>
+	remote_procedure define(const std::string& name, F&& fun);
+
+	endpoint lookup(const std::string& address) const;
+
+	operator std::string() const;
+};
+
+} // namespace thallium
+
+#include <thallium/remote_procedure.hpp>
+
+namespace thallium {
+
+template<typename F>
+remote_procedure margo_engine::define(const std::string& name) {
+    // TODO throw an exception if the following call fails
+    hg_id_t id = margo_register_name(m_mid, name.c_str(),
+                    nullptr, //proc_vector<char>,
+                    nullptr, //proc_vector<char>,
+                    nullptr);
+    margo_registered_disable_response(m_mid, id, HG_TRUE);
+    return remote_procedure(id);
+}
+
+template<typename F>
+remote_procedure margo_engine::define(const std::string& name, F&& fun) {
+    // TODO throw an exception if the following call fails
+    hg_id_t id = margo_register_name(m_mid, name.c_str(),
+                    nullptr,//serialize<Args...>,
+                    nullptr, //proc_vector<char>,
+                    rpc_callback<decltype(fun)>);
+    margo_register_data(m_mid, id, void_cast(&fun), nullptr);
+    margo_registered_disable_response(m_mid, id, HG_TRUE);
+    return remote_procedure(id);
+}
+
+}
+
+#endif

include/thallium/proc_basic.hpp

+#ifndef __THALLIUM_PROC_BASIC_HPP
+#define __THALLIUM_PROC_BASIC_HPP
+
+#include <type_traits>
+#include <mercury_proc.h>
+
+namespace thallium {
+
+namespace proc {
+
+template<typename T, 
+	typename std::enable_if_t<
+		std::is_arithmetic<
+			std::remove_reference_t<T>
+		>::value> = 0>
+hg_return_t process_type(hg_proc_t proc, void *data, std::size_t& size) {
+	T* t = static_cast<T*>(data);
+	size = sizeof(*t);
+	return hg_proc_memcpy(proc, data, sizeof(*t));
+}
+
+} // namespace proc
+
+} // namespace thallium
+
+#endif

include/thallium/proc_buffer.hpp

+#ifndef __THALLIUM_PROC_BUFFER_HPP
+#define __THALLIUM_PROC_BUFFER_HPP
+
+#include <vector>
+#include <mercury_proc.h>
+#include <thallium/buffer.hpp>
+
+namespace thallium {
+
+namespace proc {
+
+//using namespace std;
+
+template <>
+hg_return_t process_type<buffer>(hg_proc_t proc, void* data, std::size_t& size) {
+	buffer *vec = static_cast<buffer*>(data);
+	std::size_t num_T = 0;
+	size = 0;
+	hg_return_t hret;
+	switch(hg_proc_get_op(proc)) {
+		case HG_ENCODE:
+			num_T = vec->size();
+			hret = hg_proc_memcpy(proc, &num_T, sizeof(num_T));
+			if (hret != HG_SUCCESS) return hret;
+			size += sizeof(num_T);
+			if (num_T > 0)
+				hret = hg_proc_memcpy(proc, vec->data(), num_T);
+			if (hret != HG_SUCCESS) return hret;
+			size += num_T;
+            break;
+		case HG_DECODE:
+			hret = hg_proc_memcpy(proc, &num_T, sizeof(num_T));
+			if (hret != HG_SUCCESS) return hret;
+			size += sizeof(num_T);
+			if (num_T > 0) {
+				vec->resize(num_T);
+				hret = hg_proc_memcpy(proc, vec->data(), num_T);
+				if (hret != HG_SUCCESS) return hret;
+				size += num_T;
+			}
+			break;
+		case HG_FREE:
+			return HG_SUCCESS;
+	} 
+	return HG_SUCCESS;
+}
+
+} // namespace proc
+
+} // namespace thallium
+
+#endif

include/thallium/proc_tuple.hpp

+#ifndef __THALLIUM_PROC_TUPLE_HPP
+#define __THALLIUM_PROC_TUPLE_HPP
+
+#include <type_traits>
+#include <tuple>
+#include <mercury_proc.h>
+
+namespace thallium {
+
+namespace proc {
+
+
+template<std::size_t N, typename ... T>
+struct tuple_proc {
+
+	static hg_return_t apply(hg_proc_t proc, std::tuple<T...>& t, std::size_t& size) {
+		size = 0;
+		std::size_t s;
+		hg_return_t hret = tuple_proc<N-1>::apply(proc, t, s);
+		if(hret != HG_SUCCESS) return hret;
+		size += s;
+		void* data = &(std::get<N-1>(t));
+		hret = process_type<std::remove_reference_t<decltype(std::get<N-1>(t))>>(proc, data, s);
+		if(hret != HG_SUCCESS) return hret;
+		size += s;
+		return HG_SUCCESS;
+    }   
+};  
+
+template<typename ... T>
+struct tuple_proc<0, T...> {
+
+    static hg_return_t apply(hg_proc_t proc, std::tuple<T...>& t, std::size_t& size) {
+		size = 0;
+		return HG_SUCCESS;
+	}
+};
+
+using namespace std;
+
+template<template <typename ... T> class tuple, typename ... T>
+hg_return_t process_type(hg_proc_t proc, void* data, std::size_t& size) {
+	return tuple_proc<sizeof...(T),T...>::apply(proc, data, size);
+}
+
+} // namespace proc
+
+} // namespace thallium
+
+#endif

include/thallium/proc_vector.hpp

+#ifndef __THALLIUM_PROC_VECTOR_HPP
+#define __THALLIUM_PROC_VECTOR_HPP
+
+#include <vector>
+#include <mercury_proc.h>
+
+namespace thallium {
+
+namespace proc {
+
+using namespace std;
+
+template<template <typename T> class vector, class T,
+		typename std::enable_if_t<
+			std::is_arithmetic<
+				std::remove_reference_t<T>
+			>::value
+		> = 0>
+hg_return_t process_type(hg_proc_t proc, void* data, std::size_t& size) {
+	std::vector<T> *vec = static_cast<std::vector<T>*>(data);
+	std::size_t num_T = 0; 
+	hg_return_t hret;
+	size = 0;
+	switch(hg_proc_get_op(proc)) {
+		case HG_ENCODE:
+			num_T = vec->size();
+			hret = hg_proc_memcpy(proc, &num_T, sizeof(num_T));
+			if (hret != HG_SUCCESS) return hret;
+			size += sizeof(num_T);
+			if (num_T > 0)
+				hret = hg_proc_memcpy(proc, vec->data(), num_T*sizeof(T));
+			if (hret != HG_SUCCESS) return hret;
+			size += num_T*sizeof(T);
+            break;
+		case HG_DECODE:
+			hret = hg_proc_memcpy(proc, &num_T, sizeof(num_T));
+			if (hret != HG_SUCCESS) return hret;
+			size += sizeof(num_T);
+			if (num_T > 0) {
+				vec->resize(num_T);
+				hret = hg_proc_memcpy(proc, vec->data(), num_T*sizeof(T));
+				if (hret != HG_SUCCESS) return hret;
+				size += num_T*sizeof(T);
+			}
+			break;
+		case HG_FREE:
+			return HG_SUCCESS;
+	} 
+	return HG_SUCCESS;
+}
+
+template<template <typename T> class vector, class T>
+hg_return_t process_type(hg_proc_t proc, void* data, std::size_t& size) {
+	std::vector<T> *vec = static_cast<std::vector<T>*>(data);
+	std::size_t num_T = 0; 
+	hg_return_t hret;
+	size = 0;
+	switch(hg_proc_get_op(proc)) {
+		case HG_ENCODE:
+			num_T = vec->size();
+			hret = hg_proc_memcpy(proc, &num_T, sizeof(num_T));
+			if (hret != HG_SUCCESS) return hret;
+			size += sizeof(num_T);
+			char* ptr = vec->data();
+			for(auto i=0; i < num_T; i++) {
+				std::size_t s;
+				hret = process_type<T>(proc, static_cast<void*>(ptr), s);
+				if (hret != HG_SUCCESS) return hret;
+				ptr += s;
+				size += s;
+			}
+            break;
+		case HG_DECODE:
+			hret = hg_proc_memcpy(proc, &num_T, sizeof(num_T));
+			if (hret != HG_SUCCESS) return hret;
+			size += sizeof(num_T);
+			if (num_T > 0) {
+				vec->resize(num_T);
+				char* ptr = vec->data();
+				for(auto i=0; i < num_T; i++) {
+					std::size_t s;
+					hret = process_type<T>(proc, static_cast<void*>(ptr), s);
+					if (hret != HG_SUCCESS) return hret;
+					ptr += s;
+					size += s;
+				}
+			}
+			break;
+		case HG_FREE:
+			return HG_SUCCESS;
+	} 
+	return HG_SUCCESS;
+}
+
+} // namespace proc
+
+} // namespace thallium
+
+#endif

include/thallium/remote_procedure.hpp

+#ifndef __THALLIUM_REMOTE_PROCEDURE_HPP
+#define __THALLIUM_REMOTE_PROCEDURE_HPP
+
+#include <margo.h>
+
+namespace thallium {
+
+class margo_engine;
+class endpoint;
+class callable_remote_procedure;
+
+class remote_procedure {
+
+	friend class margo_engine;
+
+private:
+	hg_id_t m_id;
+
+	remote_procedure(hg_id_t id)
+	: m_id(id) {}
+
+public:
+
+	remote_procedure(const remote_procedure& other)            = default;
+	remote_procedure(remote_procedure&& other)                 = default;
+	remote_procedure& operator=(const remote_procedure& other) = default;
+	remote_procedure& operator=(remote_procedure&& other)      = default;
+	~remote_procedure()                                        = default;
+
+	callable_remote_procedure on(const endpoint& ep) const;
+
+	callable_remote_procedure operator,(const endpoint& ep) const;
+};
+
+}
+
+#endif

include/thallium/request.hpp

+#ifndef __THALLIUM_REQUEST_HPP
+#define __THALLIUM_REQUEST_HPP
+
+#include <margo.h>
+
+namespace thallium {
+
+class margo_engine;
+
+class request {
+
+	friend class margo_engine;
+
+private:
+
+	hg_handle_t m_handle;
+
+	request(hg_handle_t h)
+	: m_handle(h) {}
+
+public:
+
+	request(const request& other)
+	: m_handle(other.m_handle) {
+		margo_ref_incr(m_handle);
+	}
+
+	request(request&& other) 
+	: m_handle(other.m_handle) {
+		other.m_handle = HG_HANDLE_NULL;
+	}
+
+	request& operator=(const request& other) {
+		if(m_handle == other.m_handle) return *this;
+		margo_destroy(m_handle);
+		m_handle = other.m_handle;
+		margo_ref_incr(m_handle);
+		return *this;
+	}
+
+	request& operator=(request&& other) {
+		if(m_handle == other.m_handle) return *this;
+		margo_destroy(m_handle);
+		m_handle = other.m_handle;
+		other.m_handle = HG_HANDLE_NULL;
+		return *this;
+	}
+
+	~request() {
+		margo_destroy(m_handle);
+	}
+
+	template<typename T>
+	void respond(T&& t) {
+		// TODO serialize
+		if(m_handle != HG_HANDLE_NULL) {
+			margo_respond(m_handle, nullptr);