Loading include/client/rpc/forward_data.hpp +5 −3 Viewed Changes for include/client/rpc/forward_data.hpp: 5 added lines, 3 removed lines. Original line number Diff line number Diff line Loading @@ -24,14 +24,16 @@ struct ChunkStat { unsigned long chunk_free; }; ssize_t forward_write(const std::string& path, const void* buf, bool append_flag, off64_t in_offset, // TODO once we have LEAF, remove all the error code returns and throw them as an exception. std::pair<int, ssize_t> forward_write(const std::string& path, const void* buf, bool append_flag, off64_t in_offset, size_t write_size, int64_t updated_metadentry_size); ssize_t forward_read(const std::string& path, void* buf, off64_t offset, size_t read_size); std::pair<int, ssize_t> forward_read(const std::string& path, void* buf, off64_t offset, size_t read_size); int forward_truncate(const std::string& path, size_t current_size, size_t new_size); ChunkStat forward_get_chunk_stat(); std::pair<int, ChunkStat> forward_get_chunk_stat(); } // namespace rpc } // namespace gkfs Loading include/client/rpc/forward_metadata.hpp +7 −4 Viewed Changes for include/client/rpc/forward_metadata.hpp: 7 added lines, 4 removed lines. Original line number Diff line number Diff line Loading @@ -16,6 +16,7 @@ #define GEKKOFS_CLIENT_FORWARD_METADATA_HPP #include <string> #include <memory> /* Forward declaration */ namespace gkfs { Loading @@ -27,6 +28,8 @@ struct MetadentryUpdateFlags; class Metadata; } // TODO once we have LEAF, remove all the error code returns and throw them as an exception. namespace rpc { int forward_create(const std::string& path, mode_t mode); Loading @@ -40,12 +43,12 @@ int forward_decr_size(const std::string& path, size_t length); int forward_update_metadentry(const std::string& path, const gkfs::metadata::Metadata& md, const gkfs::metadata::MetadentryUpdateFlags& md_flags); int forward_update_metadentry_size(const std::string& path, size_t size, off64_t offset, bool append_flag, off64_t& ret_size); std::pair<int, off64_t> forward_update_metadentry_size(const std::string& path, size_t size, off64_t offset, bool append_flag); int forward_get_metadentry_size(const std::string& path, off64_t& ret_size); std::pair<int, off64_t> forward_get_metadentry_size(const std::string& path); void forward_get_dirents(gkfs::filemap::OpenDir& open_dir); std::pair<int, std::shared_ptr<gkfs::filemap::OpenDir>> forward_get_dirents(const std::string& path); #ifdef HAS_SYMLINKS Loading src/client/rpc/forward_data.cpp +80 −58 Viewed Changes for src/client/rpc/forward_data.cpp: 80 added lines, 58 removed lines. Original line number Diff line number Diff line Loading @@ -26,13 +26,25 @@ using namespace std; namespace gkfs { namespace rpc { /* * This file includes all metadata RPC calls. * NOTE: No errno is defined here! */ // TODO If we decide to keep this functionality with one segment, the function can be merged mostly. // Code is mostly redundant /** * Sends an RPC request to a specific node to pull all chunks that belong to him * Send an RPC request to write from a buffer. * @param path * @param buf * @param append_flag * @param in_offset * @param write_size * @param updated_metadentry_size * @return pair<error code, written size> */ ssize_t forward_write(const string& path, const void* buf, const bool append_flag, pair<int, ssize_t> forward_write(const string& path, const void* buf, const bool append_flag, const off64_t in_offset, const size_t write_size, const int64_t updated_metadentry_size) { Loading Loading @@ -90,8 +102,7 @@ ssize_t forward_write(const string& path, const void* buf, const bool append_fla } catch (const std::exception& ex) { LOG(ERROR, "Failed to expose buffers for RMA"); errno = EBUSY; return -1; return make_pair(EBUSY, 0); } std::vector<hermes::rpc_handle<gkfs::rpc::write_data>> handles; Loading Loading @@ -152,15 +163,14 @@ ssize_t forward_write(const string& path, const void* buf, const bool append_fla } catch (const std::exception& ex) { LOG(ERROR, "Unable to send non-blocking rpc for " "path \"{}\" [peer: {}]", path, target); errno = EBUSY; return -1; return make_pair(EBUSY, 0); } } // Wait for RPC responses and then get response and add it to out_size // which is the written size All potential outputs are served to free // resources regardless of errors, although an errorcode is set. bool error = false; auto err = 0; ssize_t out_size = 0; std::size_t idx = 0; Loading @@ -172,8 +182,7 @@ ssize_t forward_write(const string& path, const void* buf, const bool append_fla if (out.err() != 0) { LOG(ERROR, "Daemon reported error: {}", out.err()); error = true; errno = out.err(); err = out.err(); } out_size += static_cast<size_t>(out.io_size()); Loading @@ -181,20 +190,30 @@ ssize_t forward_write(const string& path, const void* buf, const bool append_fla } catch (const std::exception& ex) { LOG(ERROR, "Failed to get rpc output for path \"{}\" [peer: {}]", path, targets[idx]); error = true; errno = EIO; err = EIO; } ++idx; idx++; } return error ? -1 : out_size; /* * Typically file systems return the size even if only a part of it was written. * In our case, we do not keep track which daemon fully wrote its workload. Thus, we always return size 0 on error. */ if (err) return make_pair(err, 0); else return make_pair(0, out_size); } /** * Sends an RPC request to a specific node to push all chunks that belong to him * Send an RPC request to read to a buffer. * @param path * @param buf * @param offset * @param read_size * @return pair<error code, read size> */ ssize_t forward_read(const string& path, void* buf, const off64_t offset, const size_t read_size) { pair<int, ssize_t> forward_read(const string& path, void* buf, const off64_t offset, const size_t read_size) { // Calculate chunkid boundaries and numbers so that daemons know in which // interval to look for chunks Loading Loading @@ -246,8 +265,7 @@ ssize_t forward_read(const string& path, void* buf, const off64_t offset, const } catch (const std::exception& ex) { LOG(ERROR, "Failed to expose buffers for RMA"); errno = EBUSY; return -1; return make_pair(EBUSY, 0); } std::vector<hermes::rpc_handle<gkfs::rpc::read_data>> handles; Loading Loading @@ -309,15 +327,14 @@ ssize_t forward_read(const string& path, void* buf, const off64_t offset, const } catch (const std::exception& ex) { LOG(ERROR, "Unable to send non-blocking rpc for path \"{}\" " "[peer: {}]", path, target); errno = EBUSY; return -1; return make_pair(EBUSY, 0); } } // Wait for RPC responses and then get response and add it to out_size // which is the read size. All potential outputs are served to free // resources regardless of errors, although an errorcode is set. bool error = false; auto err = 0; ssize_t out_size = 0; std::size_t idx = 0; Loading @@ -329,8 +346,7 @@ ssize_t forward_read(const string& path, void* buf, const off64_t offset, const if (out.err() != 0) { LOG(ERROR, "Daemon reported error: {}", out.err()); error = true; errno = out.err(); err = out.err(); } out_size += static_cast<size_t>(out.io_size()); Loading @@ -338,20 +354,30 @@ ssize_t forward_read(const string& path, void* buf, const off64_t offset, const } catch (const std::exception& ex) { LOG(ERROR, "Failed to get rpc output for path \"{}\" [peer: {}]", path, targets[idx]); error = true; errno = EIO; err = EIO; } ++idx; idx++; } return error ? -1 : out_size; /* * Typically file systems return the size even if only a part of it was read. * In our case, we do not keep track which daemon fully read its workload. Thus, we always return size 0 on error. */ if (err) return make_pair(err, 0); else return make_pair(0, out_size); } /** * Send an RPC request to truncate a file to given new size * @param path * @param current_size * @param new_size * @return error code */ int forward_truncate(const std::string& path, size_t current_size, size_t new_size) { assert(current_size > new_size); bool error = false; // Find out which data servers need to delete data chunks in order to // contact only them Loading @@ -366,6 +392,8 @@ int forward_truncate(const std::string& path, size_t current_size, size_t new_si std::vector<hermes::rpc_handle<gkfs::rpc::trunc_data>> handles; auto err = 0; for (const auto& host: hosts) { auto endp = CTX->hosts().at(host); Loading @@ -386,44 +414,41 @@ int forward_truncate(const std::string& path, size_t current_size, size_t new_si // TODO(amiranda): we should cancel all previously posted requests // here, unfortunately, Hermes does not support it yet :/ LOG(ERROR, "Failed to send request to host: {}", host); errno = EIO; return -1; err = EIO; break; // We need to gather all responses so we can't return here } } // Wait for RPC responses and then get response for (const auto& h : handles) { try { // XXX We might need a timeout here to not wait forever for an // output that never comes? auto out = h.get().at(0); if (out.err() != 0) { if (out.err()) { LOG(ERROR, "received error response: {}", out.err()); error = true; errno = EIO; err = EIO; } } catch (const std::exception& ex) { LOG(ERROR, "while getting rpc output"); error = true; errno = EIO; err = EIO; } } return error ? -1 : 0; return err ? err : 0; } /** * Performs a chunk stat RPC to all hosts * @return rpc::ChunkStat * @throws std::runtime_error * Send an RPC request to chunk stat all hosts * @return pair<error code, rpc::ChunkStat> */ ChunkStat forward_get_chunk_stat() { pair<int, ChunkStat> forward_get_chunk_stat() { std::vector<hermes::rpc_handle<gkfs::rpc::chunk_stat>> handles; auto err = 0; for (const auto& endp : CTX->hosts()) { try { LOG(DEBUG, "Sending RPC to host: {}", endp.to_string()); Loading @@ -441,7 +466,8 @@ ChunkStat forward_get_chunk_stat() { // TODO(amiranda): we should cancel all previously posted requests // here, unfortunately, Hermes does not support it yet :/ LOG(ERROR, "Failed to send request to host: {}", endp.to_string()); throw std::runtime_error("Failed to forward non-blocking rpc request"); err = EBUSY; break; // We need to gather all responses so we can't return here } } Loading @@ -449,8 +475,6 @@ ChunkStat forward_get_chunk_stat() { unsigned long chunk_total = 0; unsigned long chunk_free = 0; int error = 0; // wait for RPC responses for (std::size_t i = 0; i < handles.size(); ++i) { Loading @@ -461,10 +485,10 @@ ChunkStat forward_get_chunk_stat() { // output that never comes? out = handles[i].get().at(0); if (out.err() != 0) { error = out.err(); if (out.err()) { err = out.err(); LOG(ERROR, "Host '{}' reported err code '{}' during stat chunk.", CTX->hosts().at(i).to_string(), error); err); // we don't break here to ensure all responses are processed continue; } Loading @@ -472,16 +496,14 @@ ChunkStat forward_get_chunk_stat() { chunk_total += out.chunk_total(); chunk_free += out.chunk_free(); } catch (const std::exception& ex) { errno = EBUSY; throw std::runtime_error(fmt::format("Failed to get RPC output from host: {}", i)); LOG(ERROR, "Failed to get RPC output from host: {}", i); err = EBUSY; } } if (error != 0) { errno = error; throw std::runtime_error("chunk stat failed on one host"); } return {chunk_size, chunk_total, chunk_free}; if (err) return make_pair(err, ChunkStat{}); else return make_pair(0, ChunkStat{chunk_size, chunk_total, chunk_free}); } } // namespace rpc Loading src/client/rpc/forward_metadata.cpp +138 −129 Viewed Changes for src/client/rpc/forward_metadata.cpp: 138 added lines, 129 removed lines. Original line number Diff line number Diff line Loading @@ -27,9 +27,19 @@ using namespace std; namespace gkfs { namespace rpc { /* * This file includes all metadata RPC calls. * NOTE: No errno is defined here! */ /** * Send an RPC for a create request * @param path * @param mode * @return error code */ int forward_create(const std::string& path, const mode_t mode) { int err = EUNKNOWN; auto endp = CTX->hosts().at(CTX->distributor()->locate_file_metadata(path)); try { Loading @@ -40,23 +50,21 @@ int forward_create(const std::string& path, const mode_t mode) { // 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::create>(endp, path, mode).get().at(0); err = out.err(); LOG(DEBUG, "Got response success: {}", err); if (out.err()) { errno = out.err(); return -1; } LOG(DEBUG, "Got response success: {}", out.err()); return out.err() ? out.err() : 0; } catch (const std::exception& ex) { LOG(ERROR, "while getting rpc output"); errno = EBUSY; return -1; return EBUSY; } return err; } /** * Send an RPC for a stat request * @param path * @param attr * @return error code */ int forward_stat(const std::string& path, string& attr) { auto endp = CTX->hosts().at(CTX->distributor()->locate_file_metadata(path)); Loading @@ -71,30 +79,32 @@ int forward_stat(const std::string& path, string& attr) { auto out = ld_network_service->post<gkfs::rpc::stat>(endp, path).get().at(0); LOG(DEBUG, "Got response success: {}", out.err()); if (out.err() != 0) { errno = out.err(); return -1; } if (out.err()) return out.err(); attr = out.db_val(); return 0; } catch (const std::exception& ex) { LOG(ERROR, "while getting rpc output"); errno = EBUSY; return -1; return EBUSY; } return 0; } /** * Send an RPC for a remove request. This removes metadata and all data chunks possible distributed across many daemons. * Optimizations are in place for small files (file_size / chunk_size) < number_of_daemons where no broadcast to all * daemons is used to remove all chunks. Otherwise, a broadcast to all daemons is used. * @param path * @param remove_metadentry_only * @param size * @return error code */ int forward_remove(const std::string& path, const bool remove_metadentry_only, const ssize_t size) { // if only the metadentry should be removed, send one rpc to the // metadentry's responsible node to remove the metadata // else, send an rpc to all hosts and thus broadcast chunk_removal. if (remove_metadentry_only) { auto endp = CTX->hosts().at(CTX->distributor()->locate_file_metadata(path)); try { Loading @@ -109,20 +119,11 @@ int forward_remove(const std::string& path, const bool remove_metadentry_only, c LOG(DEBUG, "Got response success: {}", out.err()); if (out.err() != 0) { errno = out.err(); return -1; } return 0; return out.err() ? out.err() : 0; } catch (const std::exception& ex) { LOG(ERROR, "while getting rpc output"); errno = EBUSY; return -1; return EBUSY; } return 0; } std::vector<hermes::rpc_handle<gkfs::rpc::remove>> handles; Loading Loading @@ -155,9 +156,8 @@ int forward_remove(const std::string& path, const bool remove_metadentry_only, c handles.emplace_back(ld_network_service->post<gkfs::rpc::remove>(endp_chnk, in)); } } catch (const std::exception& ex) { LOG(ERROR, "Failed to send reduced remove requests"); throw std::runtime_error( "Failed to forward non-blocking rpc request"); LOG(ERROR, "Failed to forward non-blocking rpc request reduced remove requests"); return EBUSY; } } else { // "Big" files for (const auto& endp : CTX->hosts()) { Loading @@ -171,26 +171,21 @@ int forward_remove(const std::string& path, const bool remove_metadentry_only, c // 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) :/ // // handles.emplace_back(ld_network_service->post<gkfs::rpc::remove>(endp, in)); } catch (const std::exception& ex) { // TODO(amiranda): we should cancel all previously posted requests // here, unfortunately, Hermes does not support it yet :/ LOG(ERROR, "Failed to send request to host: {}", LOG(ERROR, "Failed to forward non-blocking rpc request to host: {}", endp.to_string()); throw std::runtime_error( "Failed to forward non-blocking rpc request"); return EBUSY; } } } // wait for RPC responses bool got_error = false; auto err = 0; for (const auto& h : handles) { try { // XXX We might need a timeout here to not wait forever for an // output that never comes? Loading @@ -198,25 +193,27 @@ int forward_remove(const std::string& path, const bool remove_metadentry_only, c if (out.err() != 0) { LOG(ERROR, "received error response: {}", out.err()); got_error = true; errno = out.err(); err = out.err(); } } catch (const std::exception& ex) { LOG(ERROR, "while getting rpc output"); got_error = true; errno = EBUSY; err = EBUSY; } } return got_error ? -1 : 0; return err; } /** * Send an RPC for a decrement file size request. This is for example used during a truncate() call. * @param path * @param length * @return error code */ int forward_decr_size(const std::string& path, size_t length) { auto endp = CTX->hosts().at(CTX->distributor()->locate_file_metadata(path)); try { LOG(DEBUG, "Sending RPC ..."); // TODO(amiranda): add a post() with RPC_TIMEOUT to hermes so that we can // retry for RPC_TRIES (see old commits with margo) Loading @@ -227,27 +224,28 @@ int forward_decr_size(const std::string& path, size_t length) { LOG(DEBUG, "Got response success: {}", out.err()); if (out.err() != 0) { errno = out.err(); return -1; } return 0; return out.err() ? out.err() : 0; } catch (const std::exception& ex) { LOG(ERROR, "while getting rpc output"); errno = EBUSY; return -1; return EBUSY; } } /** * Send an RPC for an update metadentry request. * NOTE: Currently unused. * @param path * @param md * @param md_flags * @return error code */ int forward_update_metadentry(const string& path, const gkfs::metadata::Metadata& md, const gkfs::metadata::MetadentryUpdateFlags& md_flags) { auto endp = CTX->hosts().at(CTX->distributor()->locate_file_metadata(path)); try { LOG(DEBUG, "Sending RPC ..."); // TODO(amiranda): add a post() with RPC_TIMEOUT to hermes so that we can // retry for RPC_TRIES (see old commits with margo) Loading Loading @@ -276,28 +274,27 @@ int forward_update_metadentry(const string& path, const gkfs::metadata::Metadata LOG(DEBUG, "Got response success: {}", out.err()); if (out.err() != 0) { errno = out.err(); return -1; } return 0; return out.err() ? out.err() : 0; } catch (const std::exception& ex) { LOG(ERROR, "while getting rpc output"); errno = EBUSY; return -1; return EBUSY; } } int forward_update_metadentry_size(const string& path, const size_t size, const off64_t offset, const bool append_flag, off64_t& ret_size) { /** * Send an RPC request for an update to the file size. * This is called during a write() call or similar * @param path * @param size * @param offset * @param append_flag * @return pair<error code, size after update> */ pair<int, off64_t> forward_update_metadentry_size(const string& path, const size_t size, const off64_t offset, const bool append_flag) { auto endp = CTX->hosts().at(CTX->distributor()->locate_file_metadata(path)); try { LOG(DEBUG, "Sending RPC ..."); // TODO(amiranda): add a post() with RPC_TIMEOUT to hermes so that we can // retry for RPC_TRIES (see old commits with margo) Loading @@ -310,30 +307,27 @@ forward_update_metadentry_size(const string& path, const size_t size, const off6 LOG(DEBUG, "Got response success: {}", out.err()); if (out.err() != 0) { errno = out.err(); return -1; } ret_size = out.ret_size(); return out.err(); return 0; if (out.err()) return make_pair(out.err(), 0); else return make_pair(0, out.ret_size()); } catch (const std::exception& ex) { LOG(ERROR, "while getting rpc output"); errno = EBUSY; ret_size = 0; return EUNKNOWN; return make_pair(EBUSY, 0); } } int forward_get_metadentry_size(const std::string& path, off64_t& ret_size) { /** * Send an RPC request to get the current file size. * This is called during a lseek() call * @param path * @return pair<error code, file size> */ pair<int, off64_t> forward_get_metadentry_size(const std::string& path) { auto endp = CTX->hosts().at(CTX->distributor()->locate_file_metadata(path)); try { LOG(DEBUG, "Sending RPC ..."); // TODO(amiranda): add a post() with RPC_TIMEOUT to hermes so that we can // retry for RPC_TRIES (see old commits with margo) Loading @@ -344,24 +338,26 @@ int forward_get_metadentry_size(const std::string& path, off64_t& ret_size) { LOG(DEBUG, "Got response success: {}", out.err()); ret_size = out.ret_size(); return out.err(); if (out.err()) return make_pair(out.err(), 0); else return make_pair(0, out.ret_size()); } catch (const std::exception& ex) { LOG(ERROR, "while getting rpc output"); errno = EBUSY; ret_size = 0; return EUNKNOWN; return make_pair(EBUSY, 0); } } /** * Sends an RPC request to a specific node to push all chunks that belong to him * Send an RPC request to receive all entries of a directory. * @param open_dir * @return error code */ void forward_get_dirents(gkfs::filemap::OpenDir& open_dir) { pair<int, shared_ptr<gkfs::filemap::OpenDir>> forward_get_dirents(const string& path) { LOG(DEBUG, "{}() enter for path '{}'", __func__, path) auto const root_dir = open_dir.path(); auto const targets = CTX->distributor()->locate_directory_metadata(root_dir); auto const targets = CTX->distributor()->locate_directory_metadata(path); /* preallocate receiving buffer. The actual size is not known yet. * Loading Loading @@ -389,33 +385,39 @@ void forward_get_dirents(gkfs::filemap::OpenDir& open_dir) { }, hermes::access_mode::write_only)); } catch (const std::exception& ex) { throw std::runtime_error("Failed to expose buffers for RMA"); LOG(ERROR, "{}() Failed to expose buffers for RMA. err '{}'", __func__, ex.what()); return make_pair(EBUSY, nullptr); } } auto err = 0; // send RPCs std::vector<hermes::rpc_handle<gkfs::rpc::get_dirents>> handles; for (std::size_t i = 0; i < targets.size(); ++i) { LOG(DEBUG, "target_host: {}", targets[i]); // Setup rpc input parameters for each host auto endp = CTX->hosts().at(targets[i]); gkfs::rpc::get_dirents::input in(root_dir, exposed_buffers[i]); gkfs::rpc::get_dirents::input in(path, exposed_buffers[i]); try { LOG(DEBUG, "Sending RPC to host: {}", targets[i]); LOG(DEBUG, "{}() Sending RPC to host: '{}'", __func__, targets[i]); handles.emplace_back(ld_network_service->post<gkfs::rpc::get_dirents>(endp, in)); } catch (const std::exception& ex) { LOG(ERROR, "Unable to send non-blocking get_dirents() " "on {} [peer: {}]", root_dir, targets[i]); throw std::runtime_error("Failed to post non-blocking RPC request"); LOG(ERROR, "{}() Unable to send non-blocking get_dirents() on {} [peer: {}] err '{}'", __func__, path, targets[i], ex.what()); err = EBUSY; break; // we need to gather responses from already sent RPCS } } LOG(INFO, "{}() path '{}' send rpc_srv_get_dirents() rpc to '{}' targets. per_host_buff_size '{}' Waiting on reply next and deserialize", __func__, path, targets.size(), per_host_buff_size); auto send_error = err != 0; auto open_dir = make_shared<gkfs::filemap::OpenDir>(path); // wait for RPC responses for (std::size_t i = 0; i < handles.size(); ++i) { Loading @@ -425,17 +427,25 @@ void forward_get_dirents(gkfs::filemap::OpenDir& open_dir) { // XXX We might need a timeout here to not wait forever for an // output that never comes? out = handles[i].get().at(0); // skip processing dirent data if there was an error during send // In this case all responses are gathered but their contents skipped if (send_error) continue; if (out.err() != 0) { throw std::runtime_error( fmt::format("Failed to retrieve dir entries from " "host '{}'. Error '{}', path '{}'", targets[i], strerror(out.err()), root_dir)); LOG(ERROR, "{}() Failed to retrieve dir entries from host '{}'. Error '{}', path '{}'", __func__, targets[i], strerror(out.err()), path); err = out.err(); // We need to gather all responses before exiting continue; } } catch (const std::exception& ex) { throw std::runtime_error( fmt::format("Failed to get rpc output.. [path: {}, " "target host: {}]", root_dir, targets[i])); LOG(ERROR, "{}() Failed to get rpc output.. [path: {}, target host: {}] err '{}'", __func__, path, targets[i], ex.what()); err = EBUSY; // We need to gather all responses before exiting continue; } // each server wrote information to its pre-defined region in Loading @@ -445,8 +455,7 @@ void forward_get_dirents(gkfs::filemap::OpenDir& open_dir) { void* base_ptr = exposed_buffers[i].begin()->data(); bool* bool_ptr = reinterpret_cast<bool*>(base_ptr); char* names_ptr = reinterpret_cast<char*>(base_ptr) + (out.dirents_size() * sizeof(bool)); char* names_ptr = reinterpret_cast<char*>(base_ptr) + (out.dirents_size() * sizeof(bool)); for (std::size_t j = 0; j < out.dirents_size(); j++) { Loading @@ -459,21 +468,28 @@ void forward_get_dirents(gkfs::filemap::OpenDir& open_dir) { assert(static_cast<unsigned long int>(names_ptr - reinterpret_cast<char*>(base_ptr)) < per_host_buff_size); auto name = std::string(names_ptr); // number of characters in entry + \0 terminator names_ptr += name.size() + 1; open_dir.add(name, ftype); open_dir->add(name, ftype); } } return make_pair(err, open_dir); } #ifdef HAS_SYMLINKS /** * Send an RPC request to create a symlink. * @param path * @param target_path * @return error code */ int forward_mk_symlink(const std::string& path, const std::string& target_path) { auto endp = CTX->hosts().at(CTX->distributor()->locate_file_metadata(path)); try { LOG(DEBUG, "Sending RPC ..."); // TODO(amiranda): add a post() with RPC_TIMEOUT to hermes so that we can // retry for RPC_TRIES (see old commits with margo) Loading @@ -484,17 +500,10 @@ int forward_mk_symlink(const std::string& path, const std::string& target_path) LOG(DEBUG, "Got response success: {}", out.err()); if (out.err() != 0) { errno = out.err(); return -1; } return 0; return out.err() ? out.err() : 0; } catch (const std::exception& ex) { LOG(ERROR, "while getting rpc output"); errno = EBUSY; return -1; return EBUSY; } } Loading src/client/gkfs_functions.cpp +330 −49 Viewed Changes for src/client/gkfs_functions.cpp: 330 added lines, 49 removed lines. Original line number Diff line number Diff line Loading @@ -61,6 +61,12 @@ struct linux_dirent64 { namespace { /** * Checks if metadata for parent directory exists (can be disabled with CREATE_CHECK_PARENTS). * errno may be set * @param path * @return 0 on success, -1 on failure */ int check_parent_dir(const std::string& path) { #if CREATE_CHECK_PARENTS auto p_comp = gkfs::path::dirname(path); Loading @@ -86,6 +92,14 @@ int check_parent_dir(const std::string& path) { namespace gkfs { namespace syscall { /** * gkfs wrapper for open() system calls * errno may be set * @param path * @param mode * @param flags * @return 0 on success, -1 on failure */ int gkfs_open(const std::string& path, mode_t mode, int flags) { if (flags & O_PATH) { Loading Loading @@ -171,6 +185,13 @@ int gkfs_open(const std::string& path, mode_t mode, int flags) { return CTX->file_map()->add(std::make_shared<gkfs::filemap::OpenFile>(path, flags)); } /** * Wrapper function for file/directory creation * errno may be set * @param path * @param mode * @return 0 on success, -1 on failure */ int gkfs_create(const std::string& path, mode_t mode) { //file type must be set Loading @@ -197,13 +218,19 @@ int gkfs_create(const std::string& path, mode_t mode) { if (check_parent_dir(path)) { return -1; } return gkfs::rpc::forward_create(path, mode); auto err = gkfs::rpc::forward_create(path, mode); if (err) { errno = err; return -1; } return 0; } /** * This sends internally a broadcast (i.e. n RPCs) to clean their chunk folders for that path * gkfs wrapper for unlink() system calls * errno may be set * @param path * @return * @return 0 on success, -1 on failure */ int gkfs_remove(const std::string& path) { auto md = gkfs::util::get_metadata(path); Loading @@ -211,9 +238,22 @@ int gkfs_remove(const std::string& path) { return -1; } bool has_data = S_ISREG(md->mode()) && (md->size() != 0); return gkfs::rpc::forward_remove(path, !has_data, md->size()); auto err = gkfs::rpc::forward_remove(path, !has_data, md->size()); if (err) { errno = err; return -1; } return 0; } /** * gkfs wrapper for access() system calls * errno may be set * @param path * @param mask * @param follow_links * @return 0 on success, -1 on failure */ int gkfs_access(const std::string& path, const int mask, bool follow_links) { auto md = gkfs::util::get_metadata(path, follow_links); if (!md) { Loading @@ -223,6 +263,14 @@ int gkfs_access(const std::string& path, const int mask, bool follow_links) { return 0; } /** * gkfs wrapper for stat() system calls * errno may be set * @param path * @param buf * @param follow_links * @return 0 on success, -1 on failure */ int gkfs_stat(const string& path, struct stat* buf, bool follow_links) { auto md = gkfs::util::get_metadata(path, follow_links); if (!md) { Loading @@ -233,6 +281,18 @@ int gkfs_stat(const string& path, struct stat* buf, bool follow_links) { } #ifdef STATX_TYPE /** * gkfs wrapper for statx() system calls * errno may be set * @param dirfs * @param path * @param flags * @param mask * @param buf * @param follow_links * @return 0 on success, -1 on failure */ int gkfs_statx(int dirfs, const std::string& path, int flags, unsigned int mask, struct statx* buf, bool follow_links) { auto md = gkfs::util::get_metadata(path, follow_links); if (!md) { Loading Loading @@ -270,14 +330,22 @@ int gkfs_statx(int dirfs, const std::string& path, int flags, unsigned int mask, } #endif /** * gkfs wrapper for statfs() system calls * errno may be set * @param buf * @return 0 on success, -1 on failure */ int gkfs_statfs(struct statfs* buf) { gkfs::rpc::ChunkStat blk_stat{}; try { blk_stat = gkfs::rpc::forward_get_chunk_stat(); } catch (const std::exception& e) { LOG(ERROR, "{}() Failure with error: '{}'", e.what()); auto ret = gkfs::rpc::forward_get_chunk_stat(); auto err = ret.first; if (err) { LOG(ERROR, "{}() Failure with error: '{}'", err); errno = err; return -1; } auto blk_stat = ret.second; buf->f_type = 0; buf->f_bsize = blk_stat.chunk_size; buf->f_blocks = blk_stat.chunk_total; Loading @@ -293,14 +361,24 @@ int gkfs_statfs(struct statfs* buf) { return 0; } /** * gkfs wrapper for statvfs() system calls * errno may be set * * NOTE: Currently unused. * * @param buf * @return 0 on success, -1 on failure */ int gkfs_statvfs(struct statvfs* buf) { gkfs::rpc::ChunkStat blk_stat{}; try { blk_stat = gkfs::rpc::forward_get_chunk_stat(); } catch (const std::exception& e) { LOG(ERROR, "{}() Failure with error: '{}'", e.what()); auto ret = gkfs::rpc::forward_get_chunk_stat(); auto err = ret.first; if (err) { LOG(ERROR, "{}() Failure with error: '{}'", err); errno = err; return -1; } auto blk_stat = ret.second; buf->f_bsize = blk_stat.chunk_size; buf->f_blocks = blk_stat.chunk_total; buf->f_bfree = blk_stat.chunk_free; Loading @@ -316,10 +394,26 @@ int gkfs_statvfs(struct statvfs* buf) { return 0; } /** * gkfs wrapper for lseek() system calls with available file descriptor * errno may be set * @param fd * @param offset * @param whence * @return 0 on success, -1 on failure */ off_t gkfs_lseek(unsigned int fd, off_t offset, unsigned int whence) { return gkfs_lseek(CTX->file_map()->get(fd), offset, whence); } /** * gkfs wrapper for lseek() system calls with available shared ptr to gkfs FileMap * errno may be set * @param gkfs_fd * @param offset * @param whence * @return 0 on success, -1 on failure */ off_t gkfs_lseek(shared_ptr<gkfs::filemap::OpenFile> gkfs_fd, off_t offset, unsigned int whence) { switch (whence) { case SEEK_SET: Loading @@ -329,15 +423,15 @@ off_t gkfs_lseek(shared_ptr<gkfs::filemap::OpenFile> gkfs_fd, off_t offset, unsi gkfs_fd->pos(gkfs_fd->pos() + offset); break; case SEEK_END: { off64_t file_size; auto err = gkfs::rpc::forward_get_metadentry_size(gkfs_fd->path(), file_size); if (err < 0) { errno = err; // Negative numbers are explicitly for error codes auto ret = gkfs::rpc::forward_get_metadentry_size(gkfs_fd->path()); auto err = ret.first; if (err) { errno = err; return -1; } if (offset < 0 and file_size < -offset) { auto file_size = ret.second; if (offset < 0 && file_size < -offset) { errno = EINVAL; return -1; } Loading @@ -362,6 +456,14 @@ off_t gkfs_lseek(shared_ptr<gkfs::filemap::OpenFile> gkfs_fd, off_t offset, unsi return gkfs_fd->pos(); } /** * wrapper function for gkfs_truncate * errno may be set * @param path * @param old_size * @param new_size * @return 0 on success, -1 on failure */ int gkfs_truncate(const std::string& path, off_t old_size, off_t new_size) { assert(new_size >= 0); assert(new_size <= old_size); Loading @@ -369,19 +471,29 @@ int gkfs_truncate(const std::string& path, off_t old_size, off_t new_size) { if (new_size == old_size) { return 0; } if (gkfs::rpc::forward_decr_size(path, new_size)) { auto err = gkfs::rpc::forward_decr_size(path, new_size); if (err) { LOG(DEBUG, "Failed to decrease size"); errno = err; return -1; } if (gkfs::rpc::forward_truncate(path, old_size, new_size)) { err = gkfs::rpc::forward_truncate(path, old_size, new_size); if (err) { LOG(DEBUG, "Failed to truncate data"); errno = err; return -1; } return 0; } /** * gkfs wrapper for truncate() system calls * errno may be set * @param path * @param length * @return 0 on success, -1 on failure */ int gkfs_truncate(const std::string& path, off_t length) { /* TODO CONCURRENCY: * At the moment we first ask the length to the metadata-server in order to Loading Loading @@ -410,14 +522,36 @@ int gkfs_truncate(const std::string& path, off_t length) { return gkfs_truncate(path, size, length); } /** * gkfs wrapper for dup() system calls * errno may be set * @param oldfd * @return file descriptor int or -1 on error */ int gkfs_dup(const int oldfd) { return CTX->file_map()->dup(oldfd); } /** * gkfs wrapper for dup2() system calls * errno may be set * @param oldfd * @param newfd * @return file descriptor int or -1 on error */ int gkfs_dup2(const int oldfd, const int newfd) { return CTX->file_map()->dup2(oldfd, newfd); } /** * Wrapper function for all gkfs write operations * errno may be set * @param file * @param buf * @param count * @param offset * @return written size or -1 on error */ ssize_t gkfs_pwrite(std::shared_ptr<gkfs::filemap::OpenFile> file, const char* buf, size_t count, off64_t offset) { if (file->type() != gkfs::filemap::FileType::regular) { assert(file->type() == gkfs::filemap::FileType::directory); Loading @@ -427,30 +561,47 @@ ssize_t gkfs_pwrite(std::shared_ptr<gkfs::filemap::OpenFile> file, const char* b } auto path = make_shared<string>(file->path()); auto append_flag = file->get_flag(gkfs::filemap::OpenFile_flags::append); ssize_t ret = 0; long updated_size = 0; ret = gkfs::rpc::forward_update_metadentry_size(*path, count, offset, append_flag, updated_size); if (ret != 0) { LOG(ERROR, "update_metadentry_size() failed with ret {}", ret); return ret; // ERR auto ret_update_size = gkfs::rpc::forward_update_metadentry_size(*path, count, offset, append_flag); auto err = ret_update_size.first; if (err) { LOG(ERROR, "update_metadentry_size() failed with err '{}'", err); errno = err; return -1; } ret = gkfs::rpc::forward_write(*path, buf, append_flag, offset, count, updated_size); if (ret < 0) { LOG(WARNING, "gkfs::rpc::forward_write() failed with ret {}", ret); auto updated_size = ret_update_size.second; auto ret_write = gkfs::rpc::forward_write(*path, buf, append_flag, offset, count, updated_size); err = ret_write.first; if (err) { LOG(WARNING, "gkfs::rpc::forward_write() failed with err '{}'", err); errno = err; return -1; } return ret; // return written size or -1 as error return ret_write.second; // return written size } /** * gkfs wrapper for pwrite() system calls * errno may be set * @param fd * @param buf * @param count * @param offset * @return written size or -1 on error */ ssize_t gkfs_pwrite_ws(int fd, const void* buf, size_t count, off64_t offset) { auto file = CTX->file_map()->get(fd); return gkfs_pwrite(file, reinterpret_cast<const char*>(buf), count, offset); } /* Write counts bytes starting from current file position * It also update the file position accordingly * * Same as write syscall. /** * gkfs wrapper for write() system calls * errno may be set * @param fd * @param buf * @param count * @return written size or -1 on error */ ssize_t gkfs_write(int fd, const void* buf, size_t count) { auto gkfs_fd = CTX->file_map()->get(fd); Loading @@ -465,6 +616,15 @@ ssize_t gkfs_write(int fd, const void* buf, size_t count) { return ret; } /** * gkfs wrapper for pwritev() system calls * errno may be set * @param fd * @param iov * @param iovcnt * @param offset * @return written size or -1 on error */ ssize_t gkfs_pwritev(int fd, const struct iovec* iov, int iovcnt, off_t offset) { auto file = CTX->file_map()->get(fd); Loading Loading @@ -495,6 +655,14 @@ ssize_t gkfs_pwritev(int fd, const struct iovec* iov, int iovcnt, off_t offset) return written; } /** * gkfs wrapper for writev() system calls * errno may be set * @param fd * @param iov * @param iovcnt * @return written size or -1 on error */ ssize_t gkfs_writev(int fd, const struct iovec* iov, int iovcnt) { auto gkfs_fd = CTX->file_map()->get(fd); Loading @@ -508,6 +676,14 @@ ssize_t gkfs_writev(int fd, const struct iovec* iov, int iovcnt) { return ret; } /** * Wrapper function for all gkfs read operations * @param file * @param buf * @param count * @param offset * @return read size or -1 on error */ ssize_t gkfs_pread(std::shared_ptr<gkfs::filemap::OpenFile> file, char* buf, size_t count, off64_t offset) { if (file->type() != gkfs::filemap::FileType::regular) { assert(file->type() == gkfs::filemap::FileType::directory); Loading @@ -521,13 +697,24 @@ ssize_t gkfs_pread(std::shared_ptr<gkfs::filemap::OpenFile> file, char* buf, siz memset(buf, 0, sizeof(char) * count); } auto ret = gkfs::rpc::forward_read(file->path(), buf, offset, count); if (ret < 0) { LOG(WARNING, "gkfs::rpc::forward_read() failed with ret {}", ret); auto err = ret.first; if (err) { LOG(WARNING, "gkfs::rpc::forward_read() failed with ret '{}'", err); errno = err; return -1; } // XXX check that we don't try to read past end of the file return ret; // return read size or -1 as error return ret.second; // return read size } /** * gkfs wrapper for read() system calls * errno may be set * @param fd * @param buf * @param count * @return read size or -1 on error */ ssize_t gkfs_read(int fd, void* buf, size_t count) { auto gkfs_fd = CTX->file_map()->get(fd); auto pos = gkfs_fd->pos(); //retrieve the current offset Loading @@ -539,6 +726,15 @@ ssize_t gkfs_read(int fd, void* buf, size_t count) { return ret; } /** * gkfs wrapper for preadv() system calls * errno may be set * @param fd * @param iov * @param iovcnt * @param offset * @return read size or -1 on error */ ssize_t gkfs_preadv(int fd, const struct iovec* iov, int iovcnt, off_t offset) { auto file = CTX->file_map()->get(fd); Loading Loading @@ -569,6 +765,14 @@ ssize_t gkfs_preadv(int fd, const struct iovec* iov, int iovcnt, off_t offset) { return read; } /** * gkfs wrapper for readv() system calls * errno may be set * @param fd * @param iov * @param iovcnt * @return read size or -1 on error */ ssize_t gkfs_readv(int fd, const struct iovec* iov, int iovcnt) { auto gkfs_fd = CTX->file_map()->get(fd); Loading @@ -582,11 +786,26 @@ ssize_t gkfs_readv(int fd, const struct iovec* iov, int iovcnt) { return ret; } /** * gkfs wrapper for pread() system calls * errno may be set * @param fd * @param buf * @param count * @param offset * @return read size or -1 on error */ ssize_t gkfs_pread_ws(int fd, void* buf, size_t count, off64_t offset) { auto gkfs_fd = CTX->file_map()->get(fd); return gkfs_pread(gkfs_fd, reinterpret_cast<char*>(buf), count, offset); } /** * wrapper function for opening directories * errno may be set * @param path * @return 0 on success or -1 on error */ int gkfs_opendir(const std::string& path) { auto md = gkfs::util::get_metadata(path); Loading @@ -599,11 +818,22 @@ int gkfs_opendir(const std::string& path) { return -1; } auto open_dir = std::make_shared<gkfs::filemap::OpenDir>(path); gkfs::rpc::forward_get_dirents(*open_dir); return CTX->file_map()->add(open_dir); auto ret = gkfs::rpc::forward_get_dirents(path); auto err = ret.first; if (err) { errno = err; return -1; } assert(ret.second); return CTX->file_map()->add(ret.second); } /** * gkfs wrapper for rmdir() system calls * errno may be set * @param path * @return 0 on success or -1 on error */ int gkfs_rmdir(const std::string& path) { auto md = gkfs::util::get_metadata(path); if (!md) { Loading @@ -617,15 +847,34 @@ int gkfs_rmdir(const std::string& path) { return -1; } auto open_dir = std::make_shared<gkfs::filemap::OpenDir>(path); gkfs::rpc::forward_get_dirents(*open_dir); auto ret = gkfs::rpc::forward_get_dirents(path); auto err = ret.first; if (err) { errno = err; return -1; } assert(ret.second); auto open_dir = ret.second; if (open_dir->size() != 0) { errno = ENOTEMPTY; return -1; } return gkfs::rpc::forward_remove(path, true, 0); err = gkfs::rpc::forward_remove(path, true, 0); if (err) { errno = err; return -1; } return 0; } /** * gkfs wrapper for getdents() system calls * errno may be set * @param fd * @param dirp * @param count * @return 0 on success or -1 on error */ int gkfs_getdents(unsigned int fd, struct linux_dirent* dirp, unsigned int count) { Loading Loading @@ -688,7 +937,14 @@ int gkfs_getdents(unsigned int fd, return written; } /** * gkfs wrapper for getdents64() system calls * errno may be set * @param fd * @param dirp * @param count * @return 0 on success or -1 on error */ int gkfs_getdents64(unsigned int fd, struct linux_dirent64* dirp, unsigned int count) { Loading Loading @@ -749,6 +1005,16 @@ int gkfs_getdents64(unsigned int fd, #ifdef HAS_SYMLINKS /** * gkfs wrapper for make symlink() system calls * errno may be set * * * NOTE: Currently unused * * @param path * @param target_path * @return 0 on success or -1 on error */ int gkfs_mk_symlink(const std::string& path, const std::string& target_path) { gkfs::preload::init_ld_env_if_needed(); /* The following check is not POSIX compliant. Loading Loading @@ -777,10 +1043,25 @@ int gkfs_mk_symlink(const std::string& path, const std::string& target_path) { errno = EEXIST; return -1; } return gkfs::rpc::forward_mk_symlink(path, target_path); auto err = gkfs::rpc::forward_mk_symlink(path, target_path); if (err) { errno = err; return -1; } return 0; } /** * gkfs wrapper for reading symlinks * errno may be set * * NOTE: Currently unused * * @param path * @param buf * @param bufsize * @return 0 on success or -1 on error */ int gkfs_readlink(const std::string& path, char* buf, int bufsize) { gkfs::preload::init_ld_env_if_needed(); auto md = gkfs::util::get_metadata(path, false); Loading Loading
include/client/rpc/forward_data.hpp +5 −3 Viewed Changes for include/client/rpc/forward_data.hpp: 5 added lines, 3 removed lines. Original line number Diff line number Diff line Loading @@ -24,14 +24,16 @@ struct ChunkStat { unsigned long chunk_free; }; ssize_t forward_write(const std::string& path, const void* buf, bool append_flag, off64_t in_offset, // TODO once we have LEAF, remove all the error code returns and throw them as an exception. std::pair<int, ssize_t> forward_write(const std::string& path, const void* buf, bool append_flag, off64_t in_offset, size_t write_size, int64_t updated_metadentry_size); ssize_t forward_read(const std::string& path, void* buf, off64_t offset, size_t read_size); std::pair<int, ssize_t> forward_read(const std::string& path, void* buf, off64_t offset, size_t read_size); int forward_truncate(const std::string& path, size_t current_size, size_t new_size); ChunkStat forward_get_chunk_stat(); std::pair<int, ChunkStat> forward_get_chunk_stat(); } // namespace rpc } // namespace gkfs Loading
include/client/rpc/forward_metadata.hpp +7 −4 Viewed Changes for include/client/rpc/forward_metadata.hpp: 7 added lines, 4 removed lines. Original line number Diff line number Diff line Loading @@ -16,6 +16,7 @@ #define GEKKOFS_CLIENT_FORWARD_METADATA_HPP #include <string> #include <memory> /* Forward declaration */ namespace gkfs { Loading @@ -27,6 +28,8 @@ struct MetadentryUpdateFlags; class Metadata; } // TODO once we have LEAF, remove all the error code returns and throw them as an exception. namespace rpc { int forward_create(const std::string& path, mode_t mode); Loading @@ -40,12 +43,12 @@ int forward_decr_size(const std::string& path, size_t length); int forward_update_metadentry(const std::string& path, const gkfs::metadata::Metadata& md, const gkfs::metadata::MetadentryUpdateFlags& md_flags); int forward_update_metadentry_size(const std::string& path, size_t size, off64_t offset, bool append_flag, off64_t& ret_size); std::pair<int, off64_t> forward_update_metadentry_size(const std::string& path, size_t size, off64_t offset, bool append_flag); int forward_get_metadentry_size(const std::string& path, off64_t& ret_size); std::pair<int, off64_t> forward_get_metadentry_size(const std::string& path); void forward_get_dirents(gkfs::filemap::OpenDir& open_dir); std::pair<int, std::shared_ptr<gkfs::filemap::OpenDir>> forward_get_dirents(const std::string& path); #ifdef HAS_SYMLINKS Loading
src/client/rpc/forward_data.cpp +80 −58 Viewed Changes for src/client/rpc/forward_data.cpp: 80 added lines, 58 removed lines. Original line number Diff line number Diff line Loading @@ -26,13 +26,25 @@ using namespace std; namespace gkfs { namespace rpc { /* * This file includes all metadata RPC calls. * NOTE: No errno is defined here! */ // TODO If we decide to keep this functionality with one segment, the function can be merged mostly. // Code is mostly redundant /** * Sends an RPC request to a specific node to pull all chunks that belong to him * Send an RPC request to write from a buffer. * @param path * @param buf * @param append_flag * @param in_offset * @param write_size * @param updated_metadentry_size * @return pair<error code, written size> */ ssize_t forward_write(const string& path, const void* buf, const bool append_flag, pair<int, ssize_t> forward_write(const string& path, const void* buf, const bool append_flag, const off64_t in_offset, const size_t write_size, const int64_t updated_metadentry_size) { Loading Loading @@ -90,8 +102,7 @@ ssize_t forward_write(const string& path, const void* buf, const bool append_fla } catch (const std::exception& ex) { LOG(ERROR, "Failed to expose buffers for RMA"); errno = EBUSY; return -1; return make_pair(EBUSY, 0); } std::vector<hermes::rpc_handle<gkfs::rpc::write_data>> handles; Loading Loading @@ -152,15 +163,14 @@ ssize_t forward_write(const string& path, const void* buf, const bool append_fla } catch (const std::exception& ex) { LOG(ERROR, "Unable to send non-blocking rpc for " "path \"{}\" [peer: {}]", path, target); errno = EBUSY; return -1; return make_pair(EBUSY, 0); } } // Wait for RPC responses and then get response and add it to out_size // which is the written size All potential outputs are served to free // resources regardless of errors, although an errorcode is set. bool error = false; auto err = 0; ssize_t out_size = 0; std::size_t idx = 0; Loading @@ -172,8 +182,7 @@ ssize_t forward_write(const string& path, const void* buf, const bool append_fla if (out.err() != 0) { LOG(ERROR, "Daemon reported error: {}", out.err()); error = true; errno = out.err(); err = out.err(); } out_size += static_cast<size_t>(out.io_size()); Loading @@ -181,20 +190,30 @@ ssize_t forward_write(const string& path, const void* buf, const bool append_fla } catch (const std::exception& ex) { LOG(ERROR, "Failed to get rpc output for path \"{}\" [peer: {}]", path, targets[idx]); error = true; errno = EIO; err = EIO; } ++idx; idx++; } return error ? -1 : out_size; /* * Typically file systems return the size even if only a part of it was written. * In our case, we do not keep track which daemon fully wrote its workload. Thus, we always return size 0 on error. */ if (err) return make_pair(err, 0); else return make_pair(0, out_size); } /** * Sends an RPC request to a specific node to push all chunks that belong to him * Send an RPC request to read to a buffer. * @param path * @param buf * @param offset * @param read_size * @return pair<error code, read size> */ ssize_t forward_read(const string& path, void* buf, const off64_t offset, const size_t read_size) { pair<int, ssize_t> forward_read(const string& path, void* buf, const off64_t offset, const size_t read_size) { // Calculate chunkid boundaries and numbers so that daemons know in which // interval to look for chunks Loading Loading @@ -246,8 +265,7 @@ ssize_t forward_read(const string& path, void* buf, const off64_t offset, const } catch (const std::exception& ex) { LOG(ERROR, "Failed to expose buffers for RMA"); errno = EBUSY; return -1; return make_pair(EBUSY, 0); } std::vector<hermes::rpc_handle<gkfs::rpc::read_data>> handles; Loading Loading @@ -309,15 +327,14 @@ ssize_t forward_read(const string& path, void* buf, const off64_t offset, const } catch (const std::exception& ex) { LOG(ERROR, "Unable to send non-blocking rpc for path \"{}\" " "[peer: {}]", path, target); errno = EBUSY; return -1; return make_pair(EBUSY, 0); } } // Wait for RPC responses and then get response and add it to out_size // which is the read size. All potential outputs are served to free // resources regardless of errors, although an errorcode is set. bool error = false; auto err = 0; ssize_t out_size = 0; std::size_t idx = 0; Loading @@ -329,8 +346,7 @@ ssize_t forward_read(const string& path, void* buf, const off64_t offset, const if (out.err() != 0) { LOG(ERROR, "Daemon reported error: {}", out.err()); error = true; errno = out.err(); err = out.err(); } out_size += static_cast<size_t>(out.io_size()); Loading @@ -338,20 +354,30 @@ ssize_t forward_read(const string& path, void* buf, const off64_t offset, const } catch (const std::exception& ex) { LOG(ERROR, "Failed to get rpc output for path \"{}\" [peer: {}]", path, targets[idx]); error = true; errno = EIO; err = EIO; } ++idx; idx++; } return error ? -1 : out_size; /* * Typically file systems return the size even if only a part of it was read. * In our case, we do not keep track which daemon fully read its workload. Thus, we always return size 0 on error. */ if (err) return make_pair(err, 0); else return make_pair(0, out_size); } /** * Send an RPC request to truncate a file to given new size * @param path * @param current_size * @param new_size * @return error code */ int forward_truncate(const std::string& path, size_t current_size, size_t new_size) { assert(current_size > new_size); bool error = false; // Find out which data servers need to delete data chunks in order to // contact only them Loading @@ -366,6 +392,8 @@ int forward_truncate(const std::string& path, size_t current_size, size_t new_si std::vector<hermes::rpc_handle<gkfs::rpc::trunc_data>> handles; auto err = 0; for (const auto& host: hosts) { auto endp = CTX->hosts().at(host); Loading @@ -386,44 +414,41 @@ int forward_truncate(const std::string& path, size_t current_size, size_t new_si // TODO(amiranda): we should cancel all previously posted requests // here, unfortunately, Hermes does not support it yet :/ LOG(ERROR, "Failed to send request to host: {}", host); errno = EIO; return -1; err = EIO; break; // We need to gather all responses so we can't return here } } // Wait for RPC responses and then get response for (const auto& h : handles) { try { // XXX We might need a timeout here to not wait forever for an // output that never comes? auto out = h.get().at(0); if (out.err() != 0) { if (out.err()) { LOG(ERROR, "received error response: {}", out.err()); error = true; errno = EIO; err = EIO; } } catch (const std::exception& ex) { LOG(ERROR, "while getting rpc output"); error = true; errno = EIO; err = EIO; } } return error ? -1 : 0; return err ? err : 0; } /** * Performs a chunk stat RPC to all hosts * @return rpc::ChunkStat * @throws std::runtime_error * Send an RPC request to chunk stat all hosts * @return pair<error code, rpc::ChunkStat> */ ChunkStat forward_get_chunk_stat() { pair<int, ChunkStat> forward_get_chunk_stat() { std::vector<hermes::rpc_handle<gkfs::rpc::chunk_stat>> handles; auto err = 0; for (const auto& endp : CTX->hosts()) { try { LOG(DEBUG, "Sending RPC to host: {}", endp.to_string()); Loading @@ -441,7 +466,8 @@ ChunkStat forward_get_chunk_stat() { // TODO(amiranda): we should cancel all previously posted requests // here, unfortunately, Hermes does not support it yet :/ LOG(ERROR, "Failed to send request to host: {}", endp.to_string()); throw std::runtime_error("Failed to forward non-blocking rpc request"); err = EBUSY; break; // We need to gather all responses so we can't return here } } Loading @@ -449,8 +475,6 @@ ChunkStat forward_get_chunk_stat() { unsigned long chunk_total = 0; unsigned long chunk_free = 0; int error = 0; // wait for RPC responses for (std::size_t i = 0; i < handles.size(); ++i) { Loading @@ -461,10 +485,10 @@ ChunkStat forward_get_chunk_stat() { // output that never comes? out = handles[i].get().at(0); if (out.err() != 0) { error = out.err(); if (out.err()) { err = out.err(); LOG(ERROR, "Host '{}' reported err code '{}' during stat chunk.", CTX->hosts().at(i).to_string(), error); err); // we don't break here to ensure all responses are processed continue; } Loading @@ -472,16 +496,14 @@ ChunkStat forward_get_chunk_stat() { chunk_total += out.chunk_total(); chunk_free += out.chunk_free(); } catch (const std::exception& ex) { errno = EBUSY; throw std::runtime_error(fmt::format("Failed to get RPC output from host: {}", i)); LOG(ERROR, "Failed to get RPC output from host: {}", i); err = EBUSY; } } if (error != 0) { errno = error; throw std::runtime_error("chunk stat failed on one host"); } return {chunk_size, chunk_total, chunk_free}; if (err) return make_pair(err, ChunkStat{}); else return make_pair(0, ChunkStat{chunk_size, chunk_total, chunk_free}); } } // namespace rpc Loading
src/client/rpc/forward_metadata.cpp +138 −129 Viewed Changes for src/client/rpc/forward_metadata.cpp: 138 added lines, 129 removed lines. Original line number Diff line number Diff line Loading @@ -27,9 +27,19 @@ using namespace std; namespace gkfs { namespace rpc { /* * This file includes all metadata RPC calls. * NOTE: No errno is defined here! */ /** * Send an RPC for a create request * @param path * @param mode * @return error code */ int forward_create(const std::string& path, const mode_t mode) { int err = EUNKNOWN; auto endp = CTX->hosts().at(CTX->distributor()->locate_file_metadata(path)); try { Loading @@ -40,23 +50,21 @@ int forward_create(const std::string& path, const mode_t mode) { // 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::create>(endp, path, mode).get().at(0); err = out.err(); LOG(DEBUG, "Got response success: {}", err); if (out.err()) { errno = out.err(); return -1; } LOG(DEBUG, "Got response success: {}", out.err()); return out.err() ? out.err() : 0; } catch (const std::exception& ex) { LOG(ERROR, "while getting rpc output"); errno = EBUSY; return -1; return EBUSY; } return err; } /** * Send an RPC for a stat request * @param path * @param attr * @return error code */ int forward_stat(const std::string& path, string& attr) { auto endp = CTX->hosts().at(CTX->distributor()->locate_file_metadata(path)); Loading @@ -71,30 +79,32 @@ int forward_stat(const std::string& path, string& attr) { auto out = ld_network_service->post<gkfs::rpc::stat>(endp, path).get().at(0); LOG(DEBUG, "Got response success: {}", out.err()); if (out.err() != 0) { errno = out.err(); return -1; } if (out.err()) return out.err(); attr = out.db_val(); return 0; } catch (const std::exception& ex) { LOG(ERROR, "while getting rpc output"); errno = EBUSY; return -1; return EBUSY; } return 0; } /** * Send an RPC for a remove request. This removes metadata and all data chunks possible distributed across many daemons. * Optimizations are in place for small files (file_size / chunk_size) < number_of_daemons where no broadcast to all * daemons is used to remove all chunks. Otherwise, a broadcast to all daemons is used. * @param path * @param remove_metadentry_only * @param size * @return error code */ int forward_remove(const std::string& path, const bool remove_metadentry_only, const ssize_t size) { // if only the metadentry should be removed, send one rpc to the // metadentry's responsible node to remove the metadata // else, send an rpc to all hosts and thus broadcast chunk_removal. if (remove_metadentry_only) { auto endp = CTX->hosts().at(CTX->distributor()->locate_file_metadata(path)); try { Loading @@ -109,20 +119,11 @@ int forward_remove(const std::string& path, const bool remove_metadentry_only, c LOG(DEBUG, "Got response success: {}", out.err()); if (out.err() != 0) { errno = out.err(); return -1; } return 0; return out.err() ? out.err() : 0; } catch (const std::exception& ex) { LOG(ERROR, "while getting rpc output"); errno = EBUSY; return -1; return EBUSY; } return 0; } std::vector<hermes::rpc_handle<gkfs::rpc::remove>> handles; Loading Loading @@ -155,9 +156,8 @@ int forward_remove(const std::string& path, const bool remove_metadentry_only, c handles.emplace_back(ld_network_service->post<gkfs::rpc::remove>(endp_chnk, in)); } } catch (const std::exception& ex) { LOG(ERROR, "Failed to send reduced remove requests"); throw std::runtime_error( "Failed to forward non-blocking rpc request"); LOG(ERROR, "Failed to forward non-blocking rpc request reduced remove requests"); return EBUSY; } } else { // "Big" files for (const auto& endp : CTX->hosts()) { Loading @@ -171,26 +171,21 @@ int forward_remove(const std::string& path, const bool remove_metadentry_only, c // 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) :/ // // handles.emplace_back(ld_network_service->post<gkfs::rpc::remove>(endp, in)); } catch (const std::exception& ex) { // TODO(amiranda): we should cancel all previously posted requests // here, unfortunately, Hermes does not support it yet :/ LOG(ERROR, "Failed to send request to host: {}", LOG(ERROR, "Failed to forward non-blocking rpc request to host: {}", endp.to_string()); throw std::runtime_error( "Failed to forward non-blocking rpc request"); return EBUSY; } } } // wait for RPC responses bool got_error = false; auto err = 0; for (const auto& h : handles) { try { // XXX We might need a timeout here to not wait forever for an // output that never comes? Loading @@ -198,25 +193,27 @@ int forward_remove(const std::string& path, const bool remove_metadentry_only, c if (out.err() != 0) { LOG(ERROR, "received error response: {}", out.err()); got_error = true; errno = out.err(); err = out.err(); } } catch (const std::exception& ex) { LOG(ERROR, "while getting rpc output"); got_error = true; errno = EBUSY; err = EBUSY; } } return got_error ? -1 : 0; return err; } /** * Send an RPC for a decrement file size request. This is for example used during a truncate() call. * @param path * @param length * @return error code */ int forward_decr_size(const std::string& path, size_t length) { auto endp = CTX->hosts().at(CTX->distributor()->locate_file_metadata(path)); try { LOG(DEBUG, "Sending RPC ..."); // TODO(amiranda): add a post() with RPC_TIMEOUT to hermes so that we can // retry for RPC_TRIES (see old commits with margo) Loading @@ -227,27 +224,28 @@ int forward_decr_size(const std::string& path, size_t length) { LOG(DEBUG, "Got response success: {}", out.err()); if (out.err() != 0) { errno = out.err(); return -1; } return 0; return out.err() ? out.err() : 0; } catch (const std::exception& ex) { LOG(ERROR, "while getting rpc output"); errno = EBUSY; return -1; return EBUSY; } } /** * Send an RPC for an update metadentry request. * NOTE: Currently unused. * @param path * @param md * @param md_flags * @return error code */ int forward_update_metadentry(const string& path, const gkfs::metadata::Metadata& md, const gkfs::metadata::MetadentryUpdateFlags& md_flags) { auto endp = CTX->hosts().at(CTX->distributor()->locate_file_metadata(path)); try { LOG(DEBUG, "Sending RPC ..."); // TODO(amiranda): add a post() with RPC_TIMEOUT to hermes so that we can // retry for RPC_TRIES (see old commits with margo) Loading Loading @@ -276,28 +274,27 @@ int forward_update_metadentry(const string& path, const gkfs::metadata::Metadata LOG(DEBUG, "Got response success: {}", out.err()); if (out.err() != 0) { errno = out.err(); return -1; } return 0; return out.err() ? out.err() : 0; } catch (const std::exception& ex) { LOG(ERROR, "while getting rpc output"); errno = EBUSY; return -1; return EBUSY; } } int forward_update_metadentry_size(const string& path, const size_t size, const off64_t offset, const bool append_flag, off64_t& ret_size) { /** * Send an RPC request for an update to the file size. * This is called during a write() call or similar * @param path * @param size * @param offset * @param append_flag * @return pair<error code, size after update> */ pair<int, off64_t> forward_update_metadentry_size(const string& path, const size_t size, const off64_t offset, const bool append_flag) { auto endp = CTX->hosts().at(CTX->distributor()->locate_file_metadata(path)); try { LOG(DEBUG, "Sending RPC ..."); // TODO(amiranda): add a post() with RPC_TIMEOUT to hermes so that we can // retry for RPC_TRIES (see old commits with margo) Loading @@ -310,30 +307,27 @@ forward_update_metadentry_size(const string& path, const size_t size, const off6 LOG(DEBUG, "Got response success: {}", out.err()); if (out.err() != 0) { errno = out.err(); return -1; } ret_size = out.ret_size(); return out.err(); return 0; if (out.err()) return make_pair(out.err(), 0); else return make_pair(0, out.ret_size()); } catch (const std::exception& ex) { LOG(ERROR, "while getting rpc output"); errno = EBUSY; ret_size = 0; return EUNKNOWN; return make_pair(EBUSY, 0); } } int forward_get_metadentry_size(const std::string& path, off64_t& ret_size) { /** * Send an RPC request to get the current file size. * This is called during a lseek() call * @param path * @return pair<error code, file size> */ pair<int, off64_t> forward_get_metadentry_size(const std::string& path) { auto endp = CTX->hosts().at(CTX->distributor()->locate_file_metadata(path)); try { LOG(DEBUG, "Sending RPC ..."); // TODO(amiranda): add a post() with RPC_TIMEOUT to hermes so that we can // retry for RPC_TRIES (see old commits with margo) Loading @@ -344,24 +338,26 @@ int forward_get_metadentry_size(const std::string& path, off64_t& ret_size) { LOG(DEBUG, "Got response success: {}", out.err()); ret_size = out.ret_size(); return out.err(); if (out.err()) return make_pair(out.err(), 0); else return make_pair(0, out.ret_size()); } catch (const std::exception& ex) { LOG(ERROR, "while getting rpc output"); errno = EBUSY; ret_size = 0; return EUNKNOWN; return make_pair(EBUSY, 0); } } /** * Sends an RPC request to a specific node to push all chunks that belong to him * Send an RPC request to receive all entries of a directory. * @param open_dir * @return error code */ void forward_get_dirents(gkfs::filemap::OpenDir& open_dir) { pair<int, shared_ptr<gkfs::filemap::OpenDir>> forward_get_dirents(const string& path) { LOG(DEBUG, "{}() enter for path '{}'", __func__, path) auto const root_dir = open_dir.path(); auto const targets = CTX->distributor()->locate_directory_metadata(root_dir); auto const targets = CTX->distributor()->locate_directory_metadata(path); /* preallocate receiving buffer. The actual size is not known yet. * Loading Loading @@ -389,33 +385,39 @@ void forward_get_dirents(gkfs::filemap::OpenDir& open_dir) { }, hermes::access_mode::write_only)); } catch (const std::exception& ex) { throw std::runtime_error("Failed to expose buffers for RMA"); LOG(ERROR, "{}() Failed to expose buffers for RMA. err '{}'", __func__, ex.what()); return make_pair(EBUSY, nullptr); } } auto err = 0; // send RPCs std::vector<hermes::rpc_handle<gkfs::rpc::get_dirents>> handles; for (std::size_t i = 0; i < targets.size(); ++i) { LOG(DEBUG, "target_host: {}", targets[i]); // Setup rpc input parameters for each host auto endp = CTX->hosts().at(targets[i]); gkfs::rpc::get_dirents::input in(root_dir, exposed_buffers[i]); gkfs::rpc::get_dirents::input in(path, exposed_buffers[i]); try { LOG(DEBUG, "Sending RPC to host: {}", targets[i]); LOG(DEBUG, "{}() Sending RPC to host: '{}'", __func__, targets[i]); handles.emplace_back(ld_network_service->post<gkfs::rpc::get_dirents>(endp, in)); } catch (const std::exception& ex) { LOG(ERROR, "Unable to send non-blocking get_dirents() " "on {} [peer: {}]", root_dir, targets[i]); throw std::runtime_error("Failed to post non-blocking RPC request"); LOG(ERROR, "{}() Unable to send non-blocking get_dirents() on {} [peer: {}] err '{}'", __func__, path, targets[i], ex.what()); err = EBUSY; break; // we need to gather responses from already sent RPCS } } LOG(INFO, "{}() path '{}' send rpc_srv_get_dirents() rpc to '{}' targets. per_host_buff_size '{}' Waiting on reply next and deserialize", __func__, path, targets.size(), per_host_buff_size); auto send_error = err != 0; auto open_dir = make_shared<gkfs::filemap::OpenDir>(path); // wait for RPC responses for (std::size_t i = 0; i < handles.size(); ++i) { Loading @@ -425,17 +427,25 @@ void forward_get_dirents(gkfs::filemap::OpenDir& open_dir) { // XXX We might need a timeout here to not wait forever for an // output that never comes? out = handles[i].get().at(0); // skip processing dirent data if there was an error during send // In this case all responses are gathered but their contents skipped if (send_error) continue; if (out.err() != 0) { throw std::runtime_error( fmt::format("Failed to retrieve dir entries from " "host '{}'. Error '{}', path '{}'", targets[i], strerror(out.err()), root_dir)); LOG(ERROR, "{}() Failed to retrieve dir entries from host '{}'. Error '{}', path '{}'", __func__, targets[i], strerror(out.err()), path); err = out.err(); // We need to gather all responses before exiting continue; } } catch (const std::exception& ex) { throw std::runtime_error( fmt::format("Failed to get rpc output.. [path: {}, " "target host: {}]", root_dir, targets[i])); LOG(ERROR, "{}() Failed to get rpc output.. [path: {}, target host: {}] err '{}'", __func__, path, targets[i], ex.what()); err = EBUSY; // We need to gather all responses before exiting continue; } // each server wrote information to its pre-defined region in Loading @@ -445,8 +455,7 @@ void forward_get_dirents(gkfs::filemap::OpenDir& open_dir) { void* base_ptr = exposed_buffers[i].begin()->data(); bool* bool_ptr = reinterpret_cast<bool*>(base_ptr); char* names_ptr = reinterpret_cast<char*>(base_ptr) + (out.dirents_size() * sizeof(bool)); char* names_ptr = reinterpret_cast<char*>(base_ptr) + (out.dirents_size() * sizeof(bool)); for (std::size_t j = 0; j < out.dirents_size(); j++) { Loading @@ -459,21 +468,28 @@ void forward_get_dirents(gkfs::filemap::OpenDir& open_dir) { assert(static_cast<unsigned long int>(names_ptr - reinterpret_cast<char*>(base_ptr)) < per_host_buff_size); auto name = std::string(names_ptr); // number of characters in entry + \0 terminator names_ptr += name.size() + 1; open_dir.add(name, ftype); open_dir->add(name, ftype); } } return make_pair(err, open_dir); } #ifdef HAS_SYMLINKS /** * Send an RPC request to create a symlink. * @param path * @param target_path * @return error code */ int forward_mk_symlink(const std::string& path, const std::string& target_path) { auto endp = CTX->hosts().at(CTX->distributor()->locate_file_metadata(path)); try { LOG(DEBUG, "Sending RPC ..."); // TODO(amiranda): add a post() with RPC_TIMEOUT to hermes so that we can // retry for RPC_TRIES (see old commits with margo) Loading @@ -484,17 +500,10 @@ int forward_mk_symlink(const std::string& path, const std::string& target_path) LOG(DEBUG, "Got response success: {}", out.err()); if (out.err() != 0) { errno = out.err(); return -1; } return 0; return out.err() ? out.err() : 0; } catch (const std::exception& ex) { LOG(ERROR, "while getting rpc output"); errno = EBUSY; return -1; return EBUSY; } } Loading
src/client/gkfs_functions.cpp +330 −49 Viewed Changes for src/client/gkfs_functions.cpp: 330 added lines, 49 removed lines. Original line number Diff line number Diff line Loading @@ -61,6 +61,12 @@ struct linux_dirent64 { namespace { /** * Checks if metadata for parent directory exists (can be disabled with CREATE_CHECK_PARENTS). * errno may be set * @param path * @return 0 on success, -1 on failure */ int check_parent_dir(const std::string& path) { #if CREATE_CHECK_PARENTS auto p_comp = gkfs::path::dirname(path); Loading @@ -86,6 +92,14 @@ int check_parent_dir(const std::string& path) { namespace gkfs { namespace syscall { /** * gkfs wrapper for open() system calls * errno may be set * @param path * @param mode * @param flags * @return 0 on success, -1 on failure */ int gkfs_open(const std::string& path, mode_t mode, int flags) { if (flags & O_PATH) { Loading Loading @@ -171,6 +185,13 @@ int gkfs_open(const std::string& path, mode_t mode, int flags) { return CTX->file_map()->add(std::make_shared<gkfs::filemap::OpenFile>(path, flags)); } /** * Wrapper function for file/directory creation * errno may be set * @param path * @param mode * @return 0 on success, -1 on failure */ int gkfs_create(const std::string& path, mode_t mode) { //file type must be set Loading @@ -197,13 +218,19 @@ int gkfs_create(const std::string& path, mode_t mode) { if (check_parent_dir(path)) { return -1; } return gkfs::rpc::forward_create(path, mode); auto err = gkfs::rpc::forward_create(path, mode); if (err) { errno = err; return -1; } return 0; } /** * This sends internally a broadcast (i.e. n RPCs) to clean their chunk folders for that path * gkfs wrapper for unlink() system calls * errno may be set * @param path * @return * @return 0 on success, -1 on failure */ int gkfs_remove(const std::string& path) { auto md = gkfs::util::get_metadata(path); Loading @@ -211,9 +238,22 @@ int gkfs_remove(const std::string& path) { return -1; } bool has_data = S_ISREG(md->mode()) && (md->size() != 0); return gkfs::rpc::forward_remove(path, !has_data, md->size()); auto err = gkfs::rpc::forward_remove(path, !has_data, md->size()); if (err) { errno = err; return -1; } return 0; } /** * gkfs wrapper for access() system calls * errno may be set * @param path * @param mask * @param follow_links * @return 0 on success, -1 on failure */ int gkfs_access(const std::string& path, const int mask, bool follow_links) { auto md = gkfs::util::get_metadata(path, follow_links); if (!md) { Loading @@ -223,6 +263,14 @@ int gkfs_access(const std::string& path, const int mask, bool follow_links) { return 0; } /** * gkfs wrapper for stat() system calls * errno may be set * @param path * @param buf * @param follow_links * @return 0 on success, -1 on failure */ int gkfs_stat(const string& path, struct stat* buf, bool follow_links) { auto md = gkfs::util::get_metadata(path, follow_links); if (!md) { Loading @@ -233,6 +281,18 @@ int gkfs_stat(const string& path, struct stat* buf, bool follow_links) { } #ifdef STATX_TYPE /** * gkfs wrapper for statx() system calls * errno may be set * @param dirfs * @param path * @param flags * @param mask * @param buf * @param follow_links * @return 0 on success, -1 on failure */ int gkfs_statx(int dirfs, const std::string& path, int flags, unsigned int mask, struct statx* buf, bool follow_links) { auto md = gkfs::util::get_metadata(path, follow_links); if (!md) { Loading Loading @@ -270,14 +330,22 @@ int gkfs_statx(int dirfs, const std::string& path, int flags, unsigned int mask, } #endif /** * gkfs wrapper for statfs() system calls * errno may be set * @param buf * @return 0 on success, -1 on failure */ int gkfs_statfs(struct statfs* buf) { gkfs::rpc::ChunkStat blk_stat{}; try { blk_stat = gkfs::rpc::forward_get_chunk_stat(); } catch (const std::exception& e) { LOG(ERROR, "{}() Failure with error: '{}'", e.what()); auto ret = gkfs::rpc::forward_get_chunk_stat(); auto err = ret.first; if (err) { LOG(ERROR, "{}() Failure with error: '{}'", err); errno = err; return -1; } auto blk_stat = ret.second; buf->f_type = 0; buf->f_bsize = blk_stat.chunk_size; buf->f_blocks = blk_stat.chunk_total; Loading @@ -293,14 +361,24 @@ int gkfs_statfs(struct statfs* buf) { return 0; } /** * gkfs wrapper for statvfs() system calls * errno may be set * * NOTE: Currently unused. * * @param buf * @return 0 on success, -1 on failure */ int gkfs_statvfs(struct statvfs* buf) { gkfs::rpc::ChunkStat blk_stat{}; try { blk_stat = gkfs::rpc::forward_get_chunk_stat(); } catch (const std::exception& e) { LOG(ERROR, "{}() Failure with error: '{}'", e.what()); auto ret = gkfs::rpc::forward_get_chunk_stat(); auto err = ret.first; if (err) { LOG(ERROR, "{}() Failure with error: '{}'", err); errno = err; return -1; } auto blk_stat = ret.second; buf->f_bsize = blk_stat.chunk_size; buf->f_blocks = blk_stat.chunk_total; buf->f_bfree = blk_stat.chunk_free; Loading @@ -316,10 +394,26 @@ int gkfs_statvfs(struct statvfs* buf) { return 0; } /** * gkfs wrapper for lseek() system calls with available file descriptor * errno may be set * @param fd * @param offset * @param whence * @return 0 on success, -1 on failure */ off_t gkfs_lseek(unsigned int fd, off_t offset, unsigned int whence) { return gkfs_lseek(CTX->file_map()->get(fd), offset, whence); } /** * gkfs wrapper for lseek() system calls with available shared ptr to gkfs FileMap * errno may be set * @param gkfs_fd * @param offset * @param whence * @return 0 on success, -1 on failure */ off_t gkfs_lseek(shared_ptr<gkfs::filemap::OpenFile> gkfs_fd, off_t offset, unsigned int whence) { switch (whence) { case SEEK_SET: Loading @@ -329,15 +423,15 @@ off_t gkfs_lseek(shared_ptr<gkfs::filemap::OpenFile> gkfs_fd, off_t offset, unsi gkfs_fd->pos(gkfs_fd->pos() + offset); break; case SEEK_END: { off64_t file_size; auto err = gkfs::rpc::forward_get_metadentry_size(gkfs_fd->path(), file_size); if (err < 0) { errno = err; // Negative numbers are explicitly for error codes auto ret = gkfs::rpc::forward_get_metadentry_size(gkfs_fd->path()); auto err = ret.first; if (err) { errno = err; return -1; } if (offset < 0 and file_size < -offset) { auto file_size = ret.second; if (offset < 0 && file_size < -offset) { errno = EINVAL; return -1; } Loading @@ -362,6 +456,14 @@ off_t gkfs_lseek(shared_ptr<gkfs::filemap::OpenFile> gkfs_fd, off_t offset, unsi return gkfs_fd->pos(); } /** * wrapper function for gkfs_truncate * errno may be set * @param path * @param old_size * @param new_size * @return 0 on success, -1 on failure */ int gkfs_truncate(const std::string& path, off_t old_size, off_t new_size) { assert(new_size >= 0); assert(new_size <= old_size); Loading @@ -369,19 +471,29 @@ int gkfs_truncate(const std::string& path, off_t old_size, off_t new_size) { if (new_size == old_size) { return 0; } if (gkfs::rpc::forward_decr_size(path, new_size)) { auto err = gkfs::rpc::forward_decr_size(path, new_size); if (err) { LOG(DEBUG, "Failed to decrease size"); errno = err; return -1; } if (gkfs::rpc::forward_truncate(path, old_size, new_size)) { err = gkfs::rpc::forward_truncate(path, old_size, new_size); if (err) { LOG(DEBUG, "Failed to truncate data"); errno = err; return -1; } return 0; } /** * gkfs wrapper for truncate() system calls * errno may be set * @param path * @param length * @return 0 on success, -1 on failure */ int gkfs_truncate(const std::string& path, off_t length) { /* TODO CONCURRENCY: * At the moment we first ask the length to the metadata-server in order to Loading Loading @@ -410,14 +522,36 @@ int gkfs_truncate(const std::string& path, off_t length) { return gkfs_truncate(path, size, length); } /** * gkfs wrapper for dup() system calls * errno may be set * @param oldfd * @return file descriptor int or -1 on error */ int gkfs_dup(const int oldfd) { return CTX->file_map()->dup(oldfd); } /** * gkfs wrapper for dup2() system calls * errno may be set * @param oldfd * @param newfd * @return file descriptor int or -1 on error */ int gkfs_dup2(const int oldfd, const int newfd) { return CTX->file_map()->dup2(oldfd, newfd); } /** * Wrapper function for all gkfs write operations * errno may be set * @param file * @param buf * @param count * @param offset * @return written size or -1 on error */ ssize_t gkfs_pwrite(std::shared_ptr<gkfs::filemap::OpenFile> file, const char* buf, size_t count, off64_t offset) { if (file->type() != gkfs::filemap::FileType::regular) { assert(file->type() == gkfs::filemap::FileType::directory); Loading @@ -427,30 +561,47 @@ ssize_t gkfs_pwrite(std::shared_ptr<gkfs::filemap::OpenFile> file, const char* b } auto path = make_shared<string>(file->path()); auto append_flag = file->get_flag(gkfs::filemap::OpenFile_flags::append); ssize_t ret = 0; long updated_size = 0; ret = gkfs::rpc::forward_update_metadentry_size(*path, count, offset, append_flag, updated_size); if (ret != 0) { LOG(ERROR, "update_metadentry_size() failed with ret {}", ret); return ret; // ERR auto ret_update_size = gkfs::rpc::forward_update_metadentry_size(*path, count, offset, append_flag); auto err = ret_update_size.first; if (err) { LOG(ERROR, "update_metadentry_size() failed with err '{}'", err); errno = err; return -1; } ret = gkfs::rpc::forward_write(*path, buf, append_flag, offset, count, updated_size); if (ret < 0) { LOG(WARNING, "gkfs::rpc::forward_write() failed with ret {}", ret); auto updated_size = ret_update_size.second; auto ret_write = gkfs::rpc::forward_write(*path, buf, append_flag, offset, count, updated_size); err = ret_write.first; if (err) { LOG(WARNING, "gkfs::rpc::forward_write() failed with err '{}'", err); errno = err; return -1; } return ret; // return written size or -1 as error return ret_write.second; // return written size } /** * gkfs wrapper for pwrite() system calls * errno may be set * @param fd * @param buf * @param count * @param offset * @return written size or -1 on error */ ssize_t gkfs_pwrite_ws(int fd, const void* buf, size_t count, off64_t offset) { auto file = CTX->file_map()->get(fd); return gkfs_pwrite(file, reinterpret_cast<const char*>(buf), count, offset); } /* Write counts bytes starting from current file position * It also update the file position accordingly * * Same as write syscall. /** * gkfs wrapper for write() system calls * errno may be set * @param fd * @param buf * @param count * @return written size or -1 on error */ ssize_t gkfs_write(int fd, const void* buf, size_t count) { auto gkfs_fd = CTX->file_map()->get(fd); Loading @@ -465,6 +616,15 @@ ssize_t gkfs_write(int fd, const void* buf, size_t count) { return ret; } /** * gkfs wrapper for pwritev() system calls * errno may be set * @param fd * @param iov * @param iovcnt * @param offset * @return written size or -1 on error */ ssize_t gkfs_pwritev(int fd, const struct iovec* iov, int iovcnt, off_t offset) { auto file = CTX->file_map()->get(fd); Loading Loading @@ -495,6 +655,14 @@ ssize_t gkfs_pwritev(int fd, const struct iovec* iov, int iovcnt, off_t offset) return written; } /** * gkfs wrapper for writev() system calls * errno may be set * @param fd * @param iov * @param iovcnt * @return written size or -1 on error */ ssize_t gkfs_writev(int fd, const struct iovec* iov, int iovcnt) { auto gkfs_fd = CTX->file_map()->get(fd); Loading @@ -508,6 +676,14 @@ ssize_t gkfs_writev(int fd, const struct iovec* iov, int iovcnt) { return ret; } /** * Wrapper function for all gkfs read operations * @param file * @param buf * @param count * @param offset * @return read size or -1 on error */ ssize_t gkfs_pread(std::shared_ptr<gkfs::filemap::OpenFile> file, char* buf, size_t count, off64_t offset) { if (file->type() != gkfs::filemap::FileType::regular) { assert(file->type() == gkfs::filemap::FileType::directory); Loading @@ -521,13 +697,24 @@ ssize_t gkfs_pread(std::shared_ptr<gkfs::filemap::OpenFile> file, char* buf, siz memset(buf, 0, sizeof(char) * count); } auto ret = gkfs::rpc::forward_read(file->path(), buf, offset, count); if (ret < 0) { LOG(WARNING, "gkfs::rpc::forward_read() failed with ret {}", ret); auto err = ret.first; if (err) { LOG(WARNING, "gkfs::rpc::forward_read() failed with ret '{}'", err); errno = err; return -1; } // XXX check that we don't try to read past end of the file return ret; // return read size or -1 as error return ret.second; // return read size } /** * gkfs wrapper for read() system calls * errno may be set * @param fd * @param buf * @param count * @return read size or -1 on error */ ssize_t gkfs_read(int fd, void* buf, size_t count) { auto gkfs_fd = CTX->file_map()->get(fd); auto pos = gkfs_fd->pos(); //retrieve the current offset Loading @@ -539,6 +726,15 @@ ssize_t gkfs_read(int fd, void* buf, size_t count) { return ret; } /** * gkfs wrapper for preadv() system calls * errno may be set * @param fd * @param iov * @param iovcnt * @param offset * @return read size or -1 on error */ ssize_t gkfs_preadv(int fd, const struct iovec* iov, int iovcnt, off_t offset) { auto file = CTX->file_map()->get(fd); Loading Loading @@ -569,6 +765,14 @@ ssize_t gkfs_preadv(int fd, const struct iovec* iov, int iovcnt, off_t offset) { return read; } /** * gkfs wrapper for readv() system calls * errno may be set * @param fd * @param iov * @param iovcnt * @return read size or -1 on error */ ssize_t gkfs_readv(int fd, const struct iovec* iov, int iovcnt) { auto gkfs_fd = CTX->file_map()->get(fd); Loading @@ -582,11 +786,26 @@ ssize_t gkfs_readv(int fd, const struct iovec* iov, int iovcnt) { return ret; } /** * gkfs wrapper for pread() system calls * errno may be set * @param fd * @param buf * @param count * @param offset * @return read size or -1 on error */ ssize_t gkfs_pread_ws(int fd, void* buf, size_t count, off64_t offset) { auto gkfs_fd = CTX->file_map()->get(fd); return gkfs_pread(gkfs_fd, reinterpret_cast<char*>(buf), count, offset); } /** * wrapper function for opening directories * errno may be set * @param path * @return 0 on success or -1 on error */ int gkfs_opendir(const std::string& path) { auto md = gkfs::util::get_metadata(path); Loading @@ -599,11 +818,22 @@ int gkfs_opendir(const std::string& path) { return -1; } auto open_dir = std::make_shared<gkfs::filemap::OpenDir>(path); gkfs::rpc::forward_get_dirents(*open_dir); return CTX->file_map()->add(open_dir); auto ret = gkfs::rpc::forward_get_dirents(path); auto err = ret.first; if (err) { errno = err; return -1; } assert(ret.second); return CTX->file_map()->add(ret.second); } /** * gkfs wrapper for rmdir() system calls * errno may be set * @param path * @return 0 on success or -1 on error */ int gkfs_rmdir(const std::string& path) { auto md = gkfs::util::get_metadata(path); if (!md) { Loading @@ -617,15 +847,34 @@ int gkfs_rmdir(const std::string& path) { return -1; } auto open_dir = std::make_shared<gkfs::filemap::OpenDir>(path); gkfs::rpc::forward_get_dirents(*open_dir); auto ret = gkfs::rpc::forward_get_dirents(path); auto err = ret.first; if (err) { errno = err; return -1; } assert(ret.second); auto open_dir = ret.second; if (open_dir->size() != 0) { errno = ENOTEMPTY; return -1; } return gkfs::rpc::forward_remove(path, true, 0); err = gkfs::rpc::forward_remove(path, true, 0); if (err) { errno = err; return -1; } return 0; } /** * gkfs wrapper for getdents() system calls * errno may be set * @param fd * @param dirp * @param count * @return 0 on success or -1 on error */ int gkfs_getdents(unsigned int fd, struct linux_dirent* dirp, unsigned int count) { Loading Loading @@ -688,7 +937,14 @@ int gkfs_getdents(unsigned int fd, return written; } /** * gkfs wrapper for getdents64() system calls * errno may be set * @param fd * @param dirp * @param count * @return 0 on success or -1 on error */ int gkfs_getdents64(unsigned int fd, struct linux_dirent64* dirp, unsigned int count) { Loading Loading @@ -749,6 +1005,16 @@ int gkfs_getdents64(unsigned int fd, #ifdef HAS_SYMLINKS /** * gkfs wrapper for make symlink() system calls * errno may be set * * * NOTE: Currently unused * * @param path * @param target_path * @return 0 on success or -1 on error */ int gkfs_mk_symlink(const std::string& path, const std::string& target_path) { gkfs::preload::init_ld_env_if_needed(); /* The following check is not POSIX compliant. Loading Loading @@ -777,10 +1043,25 @@ int gkfs_mk_symlink(const std::string& path, const std::string& target_path) { errno = EEXIST; return -1; } return gkfs::rpc::forward_mk_symlink(path, target_path); auto err = gkfs::rpc::forward_mk_symlink(path, target_path); if (err) { errno = err; return -1; } return 0; } /** * gkfs wrapper for reading symlinks * errno may be set * * NOTE: Currently unused * * @param path * @param buf * @param bufsize * @return 0 on success or -1 on error */ int gkfs_readlink(const std::string& path, char* buf, int bufsize) { gkfs::preload::init_ld_env_if_needed(); auto md = gkfs::util::get_metadata(path, false); Loading