Clear pages fault

lazyfs/config/default.toml

@@ -13,5 +13,13 @@ blocks_per_page=1
 [filesystem]
 log_all_operations=true
 logfile=""
+[[injection]]
+type="clear-page"
+from="/home/marta/lazyfs/root/ex.txt"
+timing="after"
+op="write"
+occurrence=999
+pages="first"
+return=true
 
-# /home/kianda/Desktop/root/ex.txt
+# /home/marta/lazyfs/root/ex.txt

lazyfs/include/lazyfs/lazyfs.hpp

@@ -201,6 +201,14 @@ class LazyFS : public Fusepp::Fuse<LazyFS> {
      */
     void command_fault_clear_cache ();
 
+    /**
+     * @brief Fifo: (fault) Clear the cached pages requested
+     * @param path Path of the file
+     * @param parts Pages to be removed
+     *
+     */
+    void command_fault_clear_page (string path, string parts);
+
     /**
      * @brief Fifo: (info) Display the cache usage
      *

lazyfs/src/lazyfs.cpp

@@ -125,6 +125,7 @@ bool LazyFS::trigger_crash_fault (string opname,
         for (auto fault : v_faults) {
             faults::ReorderF* faultR = dynamic_cast<faults::ReorderF*>(fault);
             if (faultR && faultR->op == opname && ret) ret = faultR->ret;
+
             faults::SplitWriteF* faultS = dynamic_cast<faults::SplitWriteF*>(fault);
             if (faultS && opname == "write" && ret) ret = faultS->ret;
         }
@@ -195,6 +196,7 @@ bool LazyFS::trigger_configured_clear_fault (string opname,
 
         for (auto fault : v_faults) {
             faults::ClearF* clear_fault = dynamic_cast<faults::ClearF*>(fault);
+            faults::ClearP* page_fault = dynamic_cast<faults::ClearP*>(fault);
 
             if (clear_fault && clear_fault->op == opname) {
 
@@ -232,6 +234,39 @@ bool LazyFS::trigger_configured_clear_fault (string opname,
                     }
                 }
             }
+
+            if (page_fault && page_fault->op == opname) {
+
+                bool is_multi_path = this_ ()->fs_op_multi_path.find (opname) != this_ ()->fs_op_multi_path.end ();
+
+                if ((is_multi_path && to_path == page_fault->to) || !is_multi_path) {
+
+                    int current_count = page_fault->counter.load();
+                    if (optiming == "before") {
+                        current_count = page_fault->counter.fetch_add(1);
+                    }
+
+                    if (page_fault->timing == optiming) {
+
+                        if (current_count == page_fault->occurrence) {
+
+                            spdlog::critical ("Triggered fault condition (op={},timing={})", opname, optiming);
+
+                            this->injecting_fault_lock.lock();
+                            this_ ()->command_unsynced_data_report (this->injecting_fault);
+                            this->injecting_fault_lock.unlock();
+
+                            this_ ()->command_fault_clear_page (from_path, page_fault->pages);
+
+                            if (optiming == "after" && page_fault->ret) return true;
+
+                            pid_t lazyfs_pid = getpid ();
+                            spdlog::critical ("Killing LazyFS pid {}!", lazyfs_pid);
+                            kill (lazyfs_pid, SIGKILL);
+                        }
+                    }
+                }
+            }
         }
     }
     return false;
@@ -464,6 +499,19 @@ void LazyFS::command_fault_clear_cache () {
     spdlog::warn ("[lazyfs.cmds]: cache is cleared.");
 }
 
+void LazyFS::command_fault_clear_page (string path, string parts) {
+
+    //std::unique_lock<std::shared_mutex> lock (cache_command_lock);
+
+    spdlog::warn ("[lazyfs.cmds]: clear page request submitted...");
+
+    string owner (path);
+
+    FSCache->partial_file_sync (owner, path.data(), parts);
+
+    spdlog::warn ("[lazyfs.cmds]: pages requested cleared.");
+}
+
 void LazyFS::command_display_cache_usage () {
 
     std::unique_lock<std::shared_mutex> lock (cache_command_lock);

libs/libpcache/include/cache/cache.hpp

@@ -276,6 +276,17 @@ class Cache {
      */
     void full_checkpoint ();
 
+    /**
+     * @brief Performs a partial checkpoint for uncached data based on input from user
+     *
+     * @param owner the content id
+     * @param path original path name
+     * @param parts parts from file that will be removed
+     * @return int true if item was removed
+     * 
+     */
+    int partial_file_sync (string owner, char* path, string parts);
+
     /**
      * @brief Gets the list of files that have unsynced data, mapped to
      *        the size cached (number of bytes).

libs/libpcache/include/cache/engine/backends/custom/custom_cache.hpp

@@ -175,6 +175,7 @@ class CustomCacheEngine : public PageCacheEngine {
     double get_engine_usage ();
     bool remove_cached_blocks (string content_owner_id);
     bool sync_pages (string owner, off_t size, char* orig_path);
+    bool partial_sync_pages (string owner, off_t size, char* orig_path, string parts);
     void make_block_readable_to_offset (string cid, int page_id, int block_id, int offset);
     bool rename_owner_pages (string old_owner, string new_owner);
     bool truncate_cached_blocks (string content_owner_id,

libs/libpcache/include/cache/engine/page_cache.hpp

@@ -141,6 +141,20 @@ class PageCacheEngine {
      */
     virtual bool sync_pages (string owner, off_t size, char* orig_path) = 0;
 
+    /**
+     * @brief Syncs the specified blocks associated with an owner with the underlying filesystem.
+     * It calls pwritev for the dirty data chunks but fsync is not issued, for performance
+     * reasons.
+     *
+     * @param owner the content id
+     * @param size final file size
+     * @param orig_path original file name to sync data
+     * @param parts blocks to sync
+     * @return true the pages were synced
+     * @return false the content was not found
+     */
+    virtual bool partial_sync_pages (string owner, off_t size, char* orig_path, string parts) = 0;
+
     /**
      * @brief Renames the content associated with an owner to the new owner
      *

libs/libpcache/include/faults/faults.hpp

@@ -310,11 +310,6 @@ class ClearP : public Fault {
     */
     std::atomic_int counter;
 
-    /**
-     * @brief If the fault is a crash fault.
-    */
-    bool crash;
-
     /**
      * @brief Pages that will be cleared.
     */
@@ -349,7 +344,7 @@ class ClearP : public Fault {
      * @param pages Pages to clear.
      * @param ret If the current system call is finished before crashing.
     */
-    ClearP(string timing, string op, string from, string to, int occurrence, bool crash, string pages, bool ret);
+    ClearP(string timing, string op, string from, string to, int occurrence, string pages, bool ret);
 
     ~ClearP ();
 

libs/libpcache/src/cache.cpp

@@ -560,6 +560,17 @@ void Cache::full_checkpoint () {
         this->sync_owner (it.second, false, (char*)it.first.c_str ());
 }
 
+int Cache::partial_file_sync (string owner, char* path, string parts) {
+
+    string inode = get_original_inode (owner);
+    off_t last_size = get_content_metadata (inode)->size;
+
+    int res = this->engine->partial_sync_pages (inode, last_size, path, parts);
+    _get_content_ptr (inode)->set_data_sync_flag (true);
+
+    return res;
+}
+
 std::vector<tuple<string, size_t, vector<tuple<int, pair<int, int>, int>>>>
 Cache::report_unsynced_data () {
 

libs/libpcache/src/config/config.cpp

@@ -459,13 +459,6 @@ unordered_map<string,vector<faults::Fault*>> Config::load_config (string filenam
                     error_msg += "\tKey 'pages' for some injection of type \"clear-page\" is not defined in the configuration file.\n";
                 } else 
                     pages = toml::find<string>(injection,"pages");
-
-                bool crash = false;
-                if (!injection.contains("crash")) {
-                    valid_fault = false;
-                    error_msg += "\tKey 'crash' for some injection of type \"clear\" is not defined in the configuration file.\n";
-                } else 
-                    crash = toml::find<bool>(injection,"crash");
 
                 bool ret;
                 if (injection.contains("return")) {
@@ -477,7 +470,7 @@ unordered_map<string,vector<faults::Fault*>> Config::load_config (string filenam
                 faults::ClearP * fault = NULL;
                 vector<string> errors;
                 if (valid_fault) {
-                    fault = new faults::ClearP(timing,op,from,to,occurrence,crash,pages,ret);
+                    fault = new faults::ClearP(timing,op,from,to,occurrence,pages,ret);
                     errors = fault->validate();
                 }
 

libs/libpcache/src/engine/backends/custom/custom_cache.cpp

@@ -566,6 +566,140 @@ bool CustomCacheEngine::sync_pages (string owner, off_t size, char* orig_path) {
     return 0;
 }
 
+bool CustomCacheEngine::partial_sync_pages (string owner, off_t last_size, char* orig_path, string parts) {
+
+    std::unique_lock<std::shared_mutex> lock (lock_cache_mtx);
+
+    int fd = open (orig_path, O_WRONLY);
+
+    if (this->owner_pages_mapping.find (owner) != this->owner_pages_mapping.end ()) {
+
+        off_t wrote_bytes = 0;
+        off_t page_streak = 0;
+
+        auto& iterate_blocks = this->owner_ordered_pages_mapping.at (owner);
+
+        map<int, tuple<int, Page*, pair<int, int>, bool>> new_iterate_blocks;
+
+        int size = iterate_blocks.size();
+        spdlog::warn("SIZE = {}", size);
+
+        if (parts == "first") {
+
+            auto cit = iterate_blocks.begin();
+
+            auto pptr = std::get<1> (cit->second);
+            if (pptr->is_page_dirty ()) {
+                new_iterate_blocks.insert ({cit->first, cit->second});
+                pptr->set_page_as_dirty (false);
+            }
+
+        } else if (parts == "last") {
+
+            auto cit = iterate_blocks.rbegin();
+
+            auto pptr = std::get<1> (cit->second);
+            if (pptr->is_page_dirty ()) {
+                new_iterate_blocks.insert ({cit->first, cit->second});
+                pptr->set_page_as_dirty (false);
+            }
+
+        } else if (parts == "first-half") {
+
+            auto cit = iterate_blocks.begin ();
+            for (int i = 0; i < size/2; i++) {
+                auto pptr = std::get<1> (cit->second);
+                if (pptr->is_page_dirty ()) {
+                    new_iterate_blocks.insert ({cit->first, cit->second});
+                    pptr->set_page_as_dirty (false);
+                }
+                cit++;
+            }
+
+        } else if (parts == "last-half") {
+
+            auto cit = iterate_blocks.rbegin ();
+            for (int i = size/2; i <= size; i++) {
+                auto pptr = std::get<1> (cit->second);
+                if (pptr->is_page_dirty ()) {
+                    new_iterate_blocks.insert ({cit->first, cit->second});
+                    pptr->set_page_as_dirty (false);
+                }
+                cit++;
+            }
+
+        }
+
+        off_t page_streak_last_offset =
+            (new_iterate_blocks.begin ()->first) * this->config->IO_BLOCK_SIZE;
+
+        vector<tuple<int, Page*, pair<int, int>, bool>> page_chunk;
+        page_chunk.reserve (new_iterate_blocks.size ());
+
+        for (auto it = new_iterate_blocks.begin (); it != new_iterate_blocks.end (); it++) {
+
+            auto current_block_id     = it->first;
+            auto const& next_block_id = std::next (it, 1)->first;
+
+            if ((page_streak < (__IOV_MAX - 1)) && (it != prev (new_iterate_blocks.end (), 1)) &&
+                (current_block_id == (next_block_id - 1))) {
+
+                page_streak++;
+
+                page_chunk.push_back (it->second);
+
+            } else {
+
+                page_streak++;
+
+                page_chunk.push_back (it->second);
+
+                struct iovec iov[page_streak];
+
+                page_streak_last_offset =
+                    (current_block_id - page_streak + 1) * this->config->IO_BLOCK_SIZE;
+
+                for (int p_id = 0; p_id < page_streak; p_id++) {
+
+                    int streak_block = current_block_id - page_streak + p_id + 1;
+
+                    auto const& streak_pair = page_chunk[p_id];
+                    // auto const& streak_pair     = iterate_blocks.at (streak_block);
+                    Page* page_ptr = get<1> (streak_pair);
+                    // auto const& block_data_offs = page_ptr->get_block_offsets (streak_block);
+                    auto const& block_data_offs = get<2> (streak_pair);
+                    iov[p_id].iov_base          = page_ptr->data + block_data_offs.first;
+                    if (p_id == page_streak - 1) {
+                        iov[p_id].iov_len =
+                            page_ptr->allocated_block_ids.get_readable_to (streak_block) + 1;
+                    } else {
+                        iov[p_id].iov_len = this->config->IO_BLOCK_SIZE;
+                    }
+
+                    // mark block as clean
+                    std::get<3> (iterate_blocks.at (streak_block)) = true;
+                }
+
+                wrote_bytes += pwritev (fd, iov, page_streak, page_streak_last_offset);
+
+                page_streak = 0;
+
+                page_chunk.clear ();
+
+                page_streak_last_offset = (current_block_id + 1) * this->config->IO_BLOCK_SIZE;
+            }
+        }
+    }
+
+    if (ftruncate (fd, last_size) < 0) {
+        spdlog::info ("ftruncate: failed");
+    }
+
+    close (fd);
+
+    return 0;
+}
+
 bool CustomCacheEngine::rename_owner_pages (string old_owner, string new_owner) {
 
     std::unique_lock<std::shared_mutex> lock (lock_cache_mtx);

libs/libpcache/src/faults.cpp

@@ -189,13 +189,12 @@ const unordered_set<string> ClearP::allow_crash_fs_operations = {"unlink",
 
 const unordered_set<string> ClearP::fs_op_multi_path = {"rename", "link", "symlink"};
 
-ClearP::ClearP (string timing, string op, string from, string to, int occurrence, bool crash, string pages, bool ret) : Fault(CLEAR) {
+ClearP::ClearP (string timing, string op, string from, string to, int occurrence, string pages, bool ret) : Fault(CLEAR) {
     this->timing = timing;
     this->op = op;
     this->from = from;
     this->to = to;
     this->occurrence = occurrence;
-    this->crash = crash;
     this->pages = pages;
     (this->counter).store(0);
     this->ret = ret;