Unverified Commit c2ee0f60 authored by Tommaso Tocci's avatar Tommaso Tocci
Browse files

MetadataDB: use exceptions on remove()

parent f5441bd0

ifs/include/daemon/adafs_ops/metadentry.hpp

+1 −3
Original line number Diff line number Diff line
@@ -15,9 +15,7 @@ std::string get_metadentry_str(const std::string& path); 


Metadata get_metadentry(const std::string& path);



int remove_metadentry(const std::string& path);



int remove_node(const std::string& path);

void remove_node(const std::string& path);



size_t get_metadentry_size(const std::string& path);

ifs/include/daemon/backend/metadata/db.hpp

+1 −1
Original line number Diff line number Diff line
@@ -22,7 +22,7 @@ class MetadataDB { 


        std::string get(const std::string& key) const;

        bool put(const std::string& key, const std::string& val);

        bool remove(const std::string& key);

        void remove(const std::string& key);

        bool exists(const std::string& key);

        bool update(const std::string& old_key, const std::string& new_key, const std::string& val);

        bool update_size(const std::string& key, size_t size, off64_t offset, bool append);

ifs/src/daemon/adafs_ops/metadentry.cpp

+2 −17
Original line number Diff line number Diff line
@@ -56,29 +56,14 @@ Metadata get_metadentry(const std::string& path) { 
    return {path, get_metadentry_str(path)};

}



/**

 * Wrapper to remove a KV store entry with the path as key

 * @param path

 * @return

 */

int remove_metadentry(const string& path) {

    return ADAFS_DATA->mdb()->remove(path) ? 0 : -1;

}



/**

 * Remove metadentry if exists and try to remove all chunks for path

 * @param path

 * @return

 */

int remove_node(const string& path) {

    int err = 0; // assume we succeed

    Metadata md{};

    // If metadentry exists, try to remove it

    if (get_metadentry(path, md) == 0) {

        err = remove_metadentry(path);

    }

void remove_node(const string& path) {

    ADAFS_DATA->mdb()->remove(path); // remove metadentry

    destroy_chunk_space(path); // destroys all chunks for the path on this node

    return err;

}



/**

ifs/src/daemon/backend/metadata/db.cpp

+5 −2
Original line number Diff line number Diff line
@@ -52,8 +52,11 @@ bool MetadataDB::put(const std::string& key, const std::string& val) { 
    return s.ok();

}



bool MetadataDB::remove(const std::string& key) {

    return db->Delete(write_opts, key).ok();

void MetadataDB::remove(const std::string& key) {

    auto s = db->Delete(write_opts, key);

    if(!s.ok()){

        MetadataDB::throw_rdb_status_excpt(s);

    }

}



bool MetadataDB::exists(const std::string& key) {

ifs/src/daemon/handler/h_metadentry.cpp

+18 −2
Original line number Diff line number Diff line
@@ -133,8 +133,24 @@ static hg_return_t rpc_srv_rm_node(hg_handle_t handle) { 
    assert(ret == HG_SUCCESS);

    ADAFS_DATA->spdlogger()->debug("Got remove node RPC with path {}", in.path);



    // Remove metadentry if exists on the node but also remove all chunks for that path

    out.err = remove_node(in.path);

    try {

        // Remove metadentry if exists on the node

        // and remove all chunks for that file

        remove_node(in.path);

        out.err = 0;

    } catch (const NotFoundException& e) {

        /* The metadentry was not found on this node,

         * this is not an error. At least one node involved in this

         * broadcast operation will find and delete the entry on its local

         * MetadataDB.

         * TODO: send the metadentry remove only to the node that actually

         * has it.

         */

        out.err = 0;

    } catch (const std::exception& e) {

        ADAFS_DATA->spdlogger()->error("{}() Failed to remove node: {}", __func__, e.what());

        out.err = EBUSY;

    }



    ADAFS_DATA->spdlogger()->debug("Sending output {}", out.err);

    auto hret = margo_respond(handle, &out);