ifs: Migrating rpc client to preload library (WIP) (ec5b79bc) · Commits · hpc / gekkofs

ifs/configure.hpp

+1 −0

Original line number	Diff line number	Diff line
		@@ -28,6 +28,7 @@
		// RPC configuration
		#define RPCPORT 4433
		#define RPC_TIMEOUT 150000
		#define RPC_PROTOCOL "bmi+tcp"

		// Debug configurations
		//#define RPC_TEST

ifs/include/preload/preload.hpp

+3 −3

Original line number	Diff line number	Diff line
		@@ -57,11 +57,11 @@ void register_client_ipcs();

		bool init_ipc_client();

		hg_class_t* ld_mercury_class();
		hg_class_t* ld_mercury_ipc_class();

		hg_context_t* ld_mercury_context();
		hg_context_t* ld_mercury_ipc_context();

		margo_instance_id ld_margo_id();
		margo_instance_id ld_margo_ipc_id();

		hg_addr_t daemon_addr();

ifs/main.hpp

+1 −1

Original line number	Diff line number	Diff line
		@@ -42,7 +42,7 @@ extern "C" {

		namespace bfs = boost::filesystem;

		#define RPC_PROTOCOL "bmi+tcp"

		#define ADAFS_ROOT_INODE static_cast<ino_t>(1)
		#define INVALID_INODE static_cast<ino_t>(0)
		#define ADAFS_DATA (static_cast<FsData*>(FsData::getInstance()))

ifs/src/daemon/adafs_daemon.cpp

+1 −0

Original line number	Diff line number	Diff line
		@@ -334,6 +334,7 @@ bool init_rpc_server() {
		void register_server_ipcs() {
		auto hg_class = RPC_DATA->server_ipc_hg_class();
		// preload IPCs
		MERCURY_REGISTER(hg_class, "rpc_minimal", rpc_minimal_in_t, rpc_minimal_out_t, rpc_minimal_handler);
		MERCURY_REGISTER(hg_class, "ipc_srv_fs_config", ipc_config_in_t, ipc_config_out_t, ipc_srv_fs_config_handler);
		MERCURY_REGISTER(hg_class, "ipc_srv_open", ipc_open_in_t, ipc_res_out_t, ipc_srv_open_handler);
		MERCURY_REGISTER(hg_class, "ipc_srv_stat", ipc_stat_in_t, ipc_stat_out_t, ipc_srv_stat_handler);

ifs/src/preload/margo_ipc.cpp

+10 −10

Original line number	Diff line number	Diff line
		@@ -19,7 +19,7 @@ void send_minimal_ipc(const hg_id_t minimal_id) {
		printf("minimal RPC is running...\n");

		/* create handle */
		auto ret = HG_Create(ld_mercury_context(), daemon_addr(), minimal_id, &handle);
		auto ret = HG_Create(ld_mercury_ipc_context(), daemon_addr(), minimal_id, &handle);
		assert(ret == HG_SUCCESS);

		/* Send rpc. Note that we are also transmitting the bulk handle in the
		@@ -27,7 +27,7 @@ void send_minimal_ipc(const hg_id_t minimal_id) {
		*/
		in.input = 42;
		printf("About to send RPC\n");
		margo_forward(ld_margo_id(), handle, &in);
		margo_forward(ld_margo_ipc_id(), handle, &in);
		printf("Waiting for response\n");
		/* decode response */
		ret = HG_Get_output(handle, &out);
		@@ -48,7 +48,7 @@ bool ipc_send_get_fs_config(const hg_id_t ipc_get_config_id) {
		ipc_config_out_t out;
		// fill in
		in.dummy = 0; // XXX should be removed. havent checked yet how empty input with margo works
		auto ret = HG_Create(ld_mercury_context(), daemon_addr(), ipc_get_config_id, &handle);
		auto ret = HG_Create(ld_mercury_ipc_context(), daemon_addr(), ipc_get_config_id, &handle);
		if (ret != HG_SUCCESS) {
		DAEMON_DEBUG0(debug_fd, "creating handle FAILED\n");
		return 1;
		@@ -56,7 +56,7 @@ bool ipc_send_get_fs_config(const hg_id_t ipc_get_config_id) {
		DAEMON_DEBUG0(debug_fd, "About to send get config IPC to daemon\n");
		int send_ret = HG_FALSE;
		for (int i = 0; i < max_retries; ++i) {
		send_ret = margo_forward_timed(ld_margo_id(), handle, &in, RPC_TIMEOUT);
		send_ret = margo_forward_timed(ld_margo_ipc_id(), handle, &in, RPC_TIMEOUT);
		if (send_ret == HG_SUCCESS) {
		break;
		}
		@@ -102,7 +102,7 @@ int ipc_send_open(const char* path, int flags, const mode_t mode, const hg_id_t
		in.flags = flags;
		in.path = path;
		hg_bool_t success = HG_FALSE;
		auto ret = HG_Create(ld_mercury_context(), daemon_addr(), ipc_open_id, &handle);
		auto ret = HG_Create(ld_mercury_ipc_context(), daemon_addr(), ipc_open_id, &handle);
		if (ret != HG_SUCCESS) {
		DAEMON_DEBUG0(debug_fd, "creating handle FAILED\n");
		return 1;
		@@ -110,7 +110,7 @@ int ipc_send_open(const char* path, int flags, const mode_t mode, const hg_id_t
		DAEMON_DEBUG0(debug_fd, "About to send open IPC to daemon\n");
		int send_ret = HG_FALSE;
		for (int i = 0; i < max_retries; ++i) {
		send_ret = margo_forward_timed(ld_margo_id(), handle, &in, RPC_TIMEOUT);
		send_ret = margo_forward_timed(ld_margo_ipc_id(), handle, &in, RPC_TIMEOUT);
		if (send_ret == HG_SUCCESS) {
		break;
		}
		@@ -142,7 +142,7 @@ int ipc_send_stat(const char* path, struct stat* attr, const hg_id_t ipc_stat_id
		// fill in
		in.path = path;
		hg_bool_t success = HG_FALSE;
		auto ret = HG_Create(ld_mercury_context(), daemon_addr(), ipc_stat_id, &handle);
		auto ret = HG_Create(ld_mercury_ipc_context(), daemon_addr(), ipc_stat_id, &handle);
		if (ret != HG_SUCCESS) {
		DAEMON_DEBUG0(debug_fd, "creating handle FAILED\n");
		return 1;
		@@ -150,7 +150,7 @@ int ipc_send_stat(const char* path, struct stat* attr, const hg_id_t ipc_stat_id
		DAEMON_DEBUG0(debug_fd, "About to send stat IPC to daemon\n");
		int send_ret = HG_FALSE;
		for (int i = 0; i < max_retries; ++i) {
		send_ret = margo_forward_timed(ld_margo_id(), handle, &in, RPC_TIMEOUT);
		send_ret = margo_forward_timed(ld_margo_ipc_id(), handle, &in, RPC_TIMEOUT);
		if (send_ret == HG_SUCCESS) {
		break;
		}
		@@ -185,7 +185,7 @@ int ipc_send_unlink(const char* path, const hg_id_t ipc_unlink_id) {
		// fill in
		in.path = path;
		hg_bool_t success = HG_FALSE;
		auto ret = HG_Create(ld_mercury_context(), daemon_addr(), ipc_unlink_id, &handle);
		auto ret = HG_Create(ld_mercury_ipc_context(), daemon_addr(), ipc_unlink_id, &handle);
		if (ret != HG_SUCCESS) {
		DAEMON_DEBUG0(debug_fd, "creating handle FAILED\n");
		return 1;
		@@ -193,7 +193,7 @@ int ipc_send_unlink(const char* path, const hg_id_t ipc_unlink_id) {
		DAEMON_DEBUG0(debug_fd, "About to send unlink IPC to daemon\n");
		int send_ret = HG_FALSE;
		for (int i = 0; i < max_retries; ++i) {
		send_ret = margo_forward_timed(ld_margo_id(), handle, &in, RPC_TIMEOUT);
		send_ret = margo_forward_timed(ld_margo_ipc_id(), handle, &in, RPC_TIMEOUT);
		if (send_ret == HG_SUCCESS) {
		break;
		}