Addressed the review comments.

Author: nitin-deamon

turbonfs/inc/fcsm.h

@@ -108,7 +108,8 @@ class fcsm
     void ensure_commit(uint64_t write_off,
                        uint64_t write_len,
                        struct rpc_task *task = nullptr,
-                       std::atomic<bool> *conditional_variable = nullptr);
+                       std::atomic<bool> *conditional_variable = nullptr,
+                       bool commit_full = false);
 
     /**
      * Callbacks to be called when flush/commit successfully complete.

turbonfs/inc/nfs_inode.h

@@ -333,7 +333,7 @@ struct nfs_inode
      * Note: As of now, we are not using this flag as commit changes not yet
      *       integrated, so we are setting this flag to true.
      */
-    bool stable_write = false;
+    bool stable_write = true;
 
 public:
     /*

turbonfs/src/fcsm.cpp

@@ -341,10 +341,12 @@ void fcsm::ctgtq_cleanup()
 void fcsm::ensure_commit(uint64_t write_off,
                          uint64_t write_len,
                          struct rpc_task *task,
-                         std::atomic<bool> *conditional_variable)
+                         std::atomic<bool> *conditional_variable,
+                         bool commit_full)
 {
     assert(inode->is_flushing);
     assert(!inode->is_stable_write());
+    assert(!commit_full || task == nullptr);
 
     /*
      * If any of the flush/commit targets are waiting completion, state machine
@@ -383,13 +385,23 @@ void fcsm::ensure_commit(uint64_t write_off,
     uint64_t commit_bytes =
         inode->get_filecache()->get_bytes_to_commit();
 
+    /*
+     * If commit_full flag is true, wait_for_ongoing_flush()
+     * committed the commit_pending bytes. Now we need to flush
+     * dirty bytes and commit them.
+     */
+    if (commit_full) {
+        assert(commit_bytes == 0);
+        commit_bytes = inode->get_filecache()->get_bytes_to_flush();
+    }
+
     if (commit_bytes == 0) {
         /*
          * TODO: Make sure this doesn't result in small-blocks being written.
          */
-        const int64_t bytes =
-            inode->get_filecache()->get_bytes_to_flush() -
-            inode->get_filecache()->max_dirty_extent_bytes();
+        const int64_t bytes = (inode->get_filecache()->get_bytes_to_flush() -
+             inode->get_filecache()->max_dirty_extent_bytes());
+
         commit_bytes = std::max(bytes, (int64_t) 0);
     }
 
@@ -477,11 +489,21 @@ void fcsm::ensure_commit(uint64_t write_off,
             ctgtq.emplace(this,
                           0 /* target flush_seq */,
                           target_committed_seq_num /* target commit_seq */,
-                          task);
+                          task,
+                          conditional_variable);
         } else {
             if (task) {
                 task->reply_write(task->rpc_api->write_task.get_size());
             }
+
+            /*
+             * Flush_cache_and_wait() waiting for dirty_bytes to flushed,
+             * can't complete until all dirty bytes flushed. so add the
+             * flush target.
+             */
+            if (commit_full) {
+                ensure_flush(0, 0, nullptr, conditional_variable);
+            }
         }
 
         return;
@@ -514,6 +536,12 @@ void fcsm::ensure_commit(uint64_t write_off,
 
         assert(committing_seq_num == (prev_committing_seq_num + bytes));
         assert(committing_seq_num > committed_seq_num);
+
+        /*
+         * Enqueue this target, on_commit_callback() completes
+         * this target. Otherwise task/conditional_variable
+         * waiting for this to complete never called.
+         */
         ctgtq.emplace(this,
                       0 /* target flush_seq */,
                       target_committed_seq_num /* target commit_seq */,
@@ -675,13 +703,21 @@ void fcsm::ensure_flush(uint64_t write_off,
     const uint64_t flushing_seq_num_before = flushing_seq_num;
     assert(flushed_seq_num <= flushing_seq_num);
 
-    // sync_membufs() will update flushing_seq_num() and mark fcsm running.
+    /*
+     * sync_membufs() will update flushing_seq_num() and mark fcsm running.
+     * Task is not passed to sync_membufs, but enqueued to ftgtq.
+     */
     inode->sync_membufs(bc_vec, false /* is_flush */, nullptr);
 
     assert(is_running());
     assert(flushing_seq_num == (flushing_seq_num_before + bytes));
     assert(flushed_seq_num <= flushing_seq_num);
 
+    /*
+     * Enqueue this target, on_flush_complete() completes
+     * this target. Otherwise task/conditional_variable
+     * waiting for this to complete never called.
+     */
     ftgtq.emplace(this,
                  target_flushed_seq_num /* target flush_seq */,
                  0 /* commit_seq */,

turbonfs/src/nfs_inode.cpp

@@ -34,7 +34,7 @@ nfs_inode::nfs_inode(const struct nfs_fh3 *filehandle,
     assert(write_error == 0);
 
     // We start doing unstable writes until proven o/w.
-    assert(stable_write == false);
+    assert(stable_write == true);
     assert(commit_state == commit_state_t::COMMIT_NOT_NEEDED);
 
 #ifndef ENABLE_NON_AZURE_NFS
@@ -554,8 +554,12 @@ bool nfs_inode::check_stable_write_required(off_t offset)
 void nfs_inode::commit_membufs(std::vector<bytes_chunk>& bc_vec)
 {
     assert(is_flushing);
-    // Commit is only called by FCSM, so it must be running.
-    // assert(get_fcsm()->is_running());
+
+    /*
+     * Commit is only called by FCSM, either callback of completion
+     * or inline as part of ensure_commit.
+     */
+    assert(get_fcsm()->is_running() || !get_fcsm()->fc_cb_count());
 
     set_commit_in_progress();
 
@@ -1349,12 +1353,17 @@ int nfs_inode::flush_cache_and_wait()
      */
     wait_for_ongoing_flush();
     std::atomic_bool complete = false;
+
     /*
      * Set the flush target for both stable/unstable write.
      * ensure_commit() doesn't set target for whole unflushed
      * bytes.
      */
-    get_fcsm()->ensure_flush(0, 0, nullptr, &complete);
+    if (is_stable_write()) {
+        get_fcsm()->ensure_flush(0, 0, nullptr, &complete);
+    } else {
+        get_fcsm()->ensure_commit(0, 0, nullptr, &complete, true);
+    }
     flush_unlock();
 
     // Wait for flush to complete.
@@ -1368,24 +1377,6 @@ int nfs_inode::flush_cache_and_wait()
         ::usleep(1000);
     }
 
-    flush_lock();
-    // Issue commit to commit_pending.
-    if (!is_stable_write()) {
-        complete = false;
-        get_fcsm()->ensure_commit(0, 0, nullptr, &complete);
-    }
-    flush_unlock();
-
-    iter = 0;
-    while (complete != true) {
-        if (++iter % 1000 == 0) {
-            AZLogWarn("[{}] flush_cache_and_wait() waiting for ongoing"
-                      " flush to complete still waiting, iter: {}",
-                      get_fuse_ino(), iter);
-        }
-        ::usleep(1000);
-    }
-
     return get_write_error();
 }
 

turbonfs/src/rpc_task.cpp

@@ -877,20 +877,6 @@ static void commit_callback(
     AZLogDebug("[{}] commit_callback, number of bc committed: {}",
                ino, bc_vec_ptr->size());
 
-#ifdef ENABLE_PARANOID
-    /*
-     * We always pick *all* commit-pending bcs for committing, and while commit
-     * is going on no other flush can run, hence new commit pending bcs cannot
-     * be added, so when an ongoing commit completes we we must not have any
-     * commit pending bcs in the cache.
-     */
-    {
-        std::vector<bytes_chunk> bc_vec =
-            inode->get_filecache()->get_commit_pending_bcs();
-        assert(bc_vec.empty());
-    }
-#endif
-
     uint64_t commit_bytes = 0;
     /*
      * Now that the request has completed, we can query libnfs for the
@@ -998,6 +984,20 @@ static void commit_callback(
         inode->set_stable_write();
     }
 
+#ifdef ENABLE_PARANOID
+    /*
+     * We always pick *all* commit-pending bcs for committing, and while commit
+     * is going on no other flush can run, hence new commit pending bcs cannot
+     * be added, so when an ongoing commit completes we we must not have any
+     * commit pending bcs in the cache.
+     */
+    {
+        std::vector<bytes_chunk> bc_vec =
+            inode->get_filecache()->get_commit_pending_bcs();
+        assert(bc_vec.empty());
+    }
+#endif
+
     if (status == 0) {
         assert(commit_bytes > 0);
         /*