Loading include/client/rpc/hg_rpcs.hpp +113 −0 Original line number Diff line number Diff line Loading @@ -1163,6 +1163,119 @@ struct update_metadentry_size { }; }; #ifdef HAS_SYMLINKS //============================================================================== // definitions for mk_symlink struct mk_symlink { // forward declarations of public input/output types for this RPC class input; class output; // traits used so that the engine knows what to do with the RPC using self_type = mk_symlink; using handle_type = hermes::rpc_handle<self_type>; using input_type = input; using output_type = output; using mercury_input_type = rpc_mk_symlink_in_t; using mercury_output_type = rpc_err_out_t; // RPC public identifier constexpr static const uint64_t public_id = 3207004160; // RPC internal Mercury identifier constexpr static const hg_id_t mercury_id = public_id; // RPC name constexpr static const auto name = hg_tag::mk_symlink; // requires response? constexpr static const auto requires_response = true; // Mercury callback to serialize input arguments constexpr static const auto mercury_in_proc_cb = HG_GEN_PROC_NAME(rpc_mk_symlink_in_t); // Mercury callback to serialize output arguments constexpr static const auto mercury_out_proc_cb = HG_GEN_PROC_NAME(rpc_err_out_t); class input { template <typename ExecutionContext> friend hg_return_t hermes::detail::post_to_mercury(ExecutionContext*); public: input(const std::string& path, const std::string& target_path) : m_path(path), m_target_path(target_path) { } input(input&& rhs) = default; input(const input& other) = default; input& operator=(input&& rhs) = default; input& operator=(const input& other) = default; std::string path() const { return m_path; } std::string target_path() const { return m_target_path; } explicit input(const rpc_mk_symlink_in_t& other) : m_path(other.path), m_target_path(other.target_path) { } explicit operator rpc_mk_symlink_in_t() { return {m_path.c_str(), m_target_path.c_str()}; } private: std::string m_path; std::string m_target_path; }; class output { template <typename ExecutionContext> friend hg_return_t hermes::detail::post_to_mercury(ExecutionContext*); public: output() : m_err() {} output(int32_t err) : m_err(err) {} output(output&& rhs) = default; output(const output& other) = default; output& operator=(output&& rhs) = default; output& operator=(const output& other) = default; explicit output(const rpc_err_out_t& out) { m_err = out.err; } int32_t err() const { return m_err; } private: int32_t m_err; }; }; #endif // HAS_SYMLINKS } // namespace rpc } // namespace gkfs Loading src/client/preload.cpp +4 −0 Original line number Diff line number Diff line Loading @@ -199,6 +199,10 @@ bool init_hermes_client(const std::string& transport_prefix) { rpc_get_metadentry_size_id = gkfs::rpc::get_metadentry_size::public_id; rpc_update_metadentry_size_id = gkfs::rpc::update_metadentry::public_id; #ifdef HAS_SYMLINKS rpc_mk_symlink_id = gkfs::rpc::mk_symlink::public_id; #endif // HAS_SYMLINKS return true; } Loading src/client/rpc/hg_rpcs.cpp +6 −0 Original line number Diff line number Diff line Loading @@ -29,6 +29,12 @@ register_user_request_types() { (void) registered_requests().add<gkfs::rpc::update_metadentry>(); (void) registered_requests().add<gkfs::rpc::get_metadentry_size>(); (void) registered_requests().add<gkfs::rpc::update_metadentry_size>(); #ifdef HAS_SYMLINKS (void) registered_requests().add<gkfs::rpc::mk_symlink>(); #endif // HAS_SYMLINKS } }} // namespace hermes::detail src/client/rpc/ld_rpc_metadentry.cpp +27 −34 Original line number Diff line number Diff line Loading @@ -416,43 +416,36 @@ void get_dirents(OpenDir& open_dir){ #ifdef HAS_SYMLINKS int mk_symlink(const std::string& path, const std::string& target_path) { hg_handle_t handle; rpc_mk_symlink_in_t in{}; rpc_err_out_t out{}; int err = EUNKNOWN; // fill in in.path = path.c_str(); in.target_path = target_path.c_str(); // Create handle CTX->log()->debug("{}() Creating Mercury handle ...", __func__); auto ret = margo_create_wrap(rpc_mk_symlink_id, path, handle); if (ret != HG_SUCCESS) { errno = EBUSY; auto endp = CTX->hosts2().at( CTX->distributor()->locate_file_metadata(path)); try { CTX->log()->debug("{}() Sending RPC ...", __func__); // TODO(amiranda): add a post() with RPC_TIMEOUT to hermes so that we can // retry for RPC_TRIES (see old commits with margo) // TODO(amiranda): hermes will eventually provide a post(endpoint) // returning one result and a broadcast(endpoint_set) returning a // result_set. When that happens we can remove the .at(0) :/ auto out = ld_network_service->post<gkfs::rpc::mk_symlink>( endp, path, target_path).get().at(0); CTX->log()->debug("{}() Got response success: {}", __func__, out.err()); if(out.err() != 0) { errno = out.err(); return -1; } // Send rpc CTX->log()->debug("{}() About to send RPC ...", __func__); ret = margo_forward_timed_wrap(handle, &in); // Get response if (ret == HG_SUCCESS) { CTX->log()->trace("{}() Waiting for response", __func__); ret = margo_get_output(handle, &out); if (ret == HG_SUCCESS) { CTX->log()->debug("{}() Got response success: {}", __func__, out.err); err = out.err; } else { // something is wrong errno = EBUSY; return 0; } catch(const std::exception& ex) { CTX->log()->error("{}() while getting rpc output", __func__); } /* clean up resources consumed by this rpc */ margo_free_output(handle, &out); } else { CTX->log()->warn("{}() timed out"); errno = EBUSY; return -1; } margo_destroy(handle); return err; } #endif Loading Loading
include/client/rpc/hg_rpcs.hpp +113 −0 Original line number Diff line number Diff line Loading @@ -1163,6 +1163,119 @@ struct update_metadentry_size { }; }; #ifdef HAS_SYMLINKS //============================================================================== // definitions for mk_symlink struct mk_symlink { // forward declarations of public input/output types for this RPC class input; class output; // traits used so that the engine knows what to do with the RPC using self_type = mk_symlink; using handle_type = hermes::rpc_handle<self_type>; using input_type = input; using output_type = output; using mercury_input_type = rpc_mk_symlink_in_t; using mercury_output_type = rpc_err_out_t; // RPC public identifier constexpr static const uint64_t public_id = 3207004160; // RPC internal Mercury identifier constexpr static const hg_id_t mercury_id = public_id; // RPC name constexpr static const auto name = hg_tag::mk_symlink; // requires response? constexpr static const auto requires_response = true; // Mercury callback to serialize input arguments constexpr static const auto mercury_in_proc_cb = HG_GEN_PROC_NAME(rpc_mk_symlink_in_t); // Mercury callback to serialize output arguments constexpr static const auto mercury_out_proc_cb = HG_GEN_PROC_NAME(rpc_err_out_t); class input { template <typename ExecutionContext> friend hg_return_t hermes::detail::post_to_mercury(ExecutionContext*); public: input(const std::string& path, const std::string& target_path) : m_path(path), m_target_path(target_path) { } input(input&& rhs) = default; input(const input& other) = default; input& operator=(input&& rhs) = default; input& operator=(const input& other) = default; std::string path() const { return m_path; } std::string target_path() const { return m_target_path; } explicit input(const rpc_mk_symlink_in_t& other) : m_path(other.path), m_target_path(other.target_path) { } explicit operator rpc_mk_symlink_in_t() { return {m_path.c_str(), m_target_path.c_str()}; } private: std::string m_path; std::string m_target_path; }; class output { template <typename ExecutionContext> friend hg_return_t hermes::detail::post_to_mercury(ExecutionContext*); public: output() : m_err() {} output(int32_t err) : m_err(err) {} output(output&& rhs) = default; output(const output& other) = default; output& operator=(output&& rhs) = default; output& operator=(const output& other) = default; explicit output(const rpc_err_out_t& out) { m_err = out.err; } int32_t err() const { return m_err; } private: int32_t m_err; }; }; #endif // HAS_SYMLINKS } // namespace rpc } // namespace gkfs Loading
src/client/preload.cpp +4 −0 Original line number Diff line number Diff line Loading @@ -199,6 +199,10 @@ bool init_hermes_client(const std::string& transport_prefix) { rpc_get_metadentry_size_id = gkfs::rpc::get_metadentry_size::public_id; rpc_update_metadentry_size_id = gkfs::rpc::update_metadentry::public_id; #ifdef HAS_SYMLINKS rpc_mk_symlink_id = gkfs::rpc::mk_symlink::public_id; #endif // HAS_SYMLINKS return true; } Loading
src/client/rpc/hg_rpcs.cpp +6 −0 Original line number Diff line number Diff line Loading @@ -29,6 +29,12 @@ register_user_request_types() { (void) registered_requests().add<gkfs::rpc::update_metadentry>(); (void) registered_requests().add<gkfs::rpc::get_metadentry_size>(); (void) registered_requests().add<gkfs::rpc::update_metadentry_size>(); #ifdef HAS_SYMLINKS (void) registered_requests().add<gkfs::rpc::mk_symlink>(); #endif // HAS_SYMLINKS } }} // namespace hermes::detail
src/client/rpc/ld_rpc_metadentry.cpp +27 −34 Original line number Diff line number Diff line Loading @@ -416,43 +416,36 @@ void get_dirents(OpenDir& open_dir){ #ifdef HAS_SYMLINKS int mk_symlink(const std::string& path, const std::string& target_path) { hg_handle_t handle; rpc_mk_symlink_in_t in{}; rpc_err_out_t out{}; int err = EUNKNOWN; // fill in in.path = path.c_str(); in.target_path = target_path.c_str(); // Create handle CTX->log()->debug("{}() Creating Mercury handle ...", __func__); auto ret = margo_create_wrap(rpc_mk_symlink_id, path, handle); if (ret != HG_SUCCESS) { errno = EBUSY; auto endp = CTX->hosts2().at( CTX->distributor()->locate_file_metadata(path)); try { CTX->log()->debug("{}() Sending RPC ...", __func__); // TODO(amiranda): add a post() with RPC_TIMEOUT to hermes so that we can // retry for RPC_TRIES (see old commits with margo) // TODO(amiranda): hermes will eventually provide a post(endpoint) // returning one result and a broadcast(endpoint_set) returning a // result_set. When that happens we can remove the .at(0) :/ auto out = ld_network_service->post<gkfs::rpc::mk_symlink>( endp, path, target_path).get().at(0); CTX->log()->debug("{}() Got response success: {}", __func__, out.err()); if(out.err() != 0) { errno = out.err(); return -1; } // Send rpc CTX->log()->debug("{}() About to send RPC ...", __func__); ret = margo_forward_timed_wrap(handle, &in); // Get response if (ret == HG_SUCCESS) { CTX->log()->trace("{}() Waiting for response", __func__); ret = margo_get_output(handle, &out); if (ret == HG_SUCCESS) { CTX->log()->debug("{}() Got response success: {}", __func__, out.err); err = out.err; } else { // something is wrong errno = EBUSY; return 0; } catch(const std::exception& ex) { CTX->log()->error("{}() while getting rpc output", __func__); } /* clean up resources consumed by this rpc */ margo_free_output(handle, &out); } else { CTX->log()->warn("{}() timed out"); errno = EBUSY; return -1; } margo_destroy(handle); return err; } #endif Loading