Merge PR ceph#48038 into main

* refs/pull/48038/head:
	client test: Add fsync to ll_preadv_pwritev test
	libcephfs: Option to write + fsync via ceph_ll_nonblocking_readv_writev
	Client: Hook nonblocking fsync into the write path of ll_preadv_pwritev
	Client: Add non-blocking fsync
	Client/Inode: wait_for_caps fixups
	Client: change several waitfor_* to use Context list
	test: Add nonblocking I/O client test
	libcephfs: Add nonblocking readv/writev I/O interface
	Client: Add ll_preadv_pwritev to expose non-blocking I/O to libcephfs
	Client: Add non-blocking helper classes
	Client: Break some code into new methods in prep for non-blocking I/O
	Buffers: Add function to buffer.h to copy bufferlist to an iovec
	ObjectCacher: Prepare file_write path for non-blocking I/O

Reviewed-by: Venky Shankar <[email protected]>
Reviewed-by: Adam C. Emerson <[email protected]>

Author: vshankar

src/client/Client.cc

@@ -646,6 +646,11 @@ class Client : public Dispatcher, public md_config_obs_t {
   int ll_write(Fh *fh, loff_t off, loff_t len, const char *data);
   int64_t ll_readv(struct Fh *fh, const struct iovec *iov, int iovcnt, int64_t off);
   int64_t ll_writev(struct Fh *fh, const struct iovec *iov, int iovcnt, int64_t off);
+  int64_t ll_preadv_pwritev(struct Fh *fh, const struct iovec *iov, int iovcnt,
+                            int64_t offset, bool write,
+                            Context *onfinish = nullptr,
+                            bufferlist *blp = nullptr,
+                            bool do_fsync = false, bool syncdataonly = false);
   loff_t ll_lseek(Fh *fh, loff_t offset, int whence);
   int ll_flush(Fh *fh);
   int ll_fsync(Fh *fh, bool syncdataonly);
@@ -753,6 +758,7 @@ class Client : public Dispatcher, public md_config_obs_t {
   void flush_snaps(Inode *in);
   void get_cap_ref(Inode *in, int cap);
   void put_cap_ref(Inode *in, int cap);
+  void submit_sync_caps(Inode *in, ceph_tid_t want, Context *onfinish);
   void wait_sync_caps(Inode *in, ceph_tid_t want);
   void wait_sync_caps(ceph_tid_t want);
   void queue_cap_snap(Inode *in, SnapContext &old_snapc);
@@ -1019,8 +1025,12 @@ class Client : public Dispatcher, public md_config_obs_t {
   // helpers
   void wake_up_session_caps(MetaSession *s, bool reconnect);
 
+  void add_nonblocking_onfinish_to_context_list(std::list<Context*>& ls, Context *onfinish) {
+    ls.push_back(onfinish);
+  }
   void wait_on_context_list(std::list<Context*>& ls);
   void signal_context_list(std::list<Context*>& ls);
+  void signal_caps_inode(Inode *in);
 
   // -- metadata cache stuff
 
@@ -1245,6 +1255,258 @@ class Client : public Dispatcher, public md_config_obs_t {
   struct initialize_state_t initialize_state;
 
 private:
+  class C_Read_Finisher : public Context {
+  public:
+    bool iofinished;
+    void finish_io(int r);
+
+    C_Read_Finisher(Client *clnt, Context *onfinish, Context *iofinish,
+                    bool is_read_async, int have_caps, bool movepos,
+                    utime_t start, Fh *f, Inode *in, uint64_t fpos,
+                    int64_t offset, uint64_t size)
+      : clnt(clnt), onfinish(onfinish), iofinish(iofinish),
+        is_read_async(is_read_async), have_caps(have_caps), f(f), in(in),
+        start(start), fpos(fpos), offset(offset), size(size), movepos(movepos) {
+      iofinished = false;
+    }
+
+    void finish(int r) override {
+      // We need to override finish, but have nothing to do.
+    }
+
+  private:
+    Client *clnt;
+    Context *onfinish;
+    Context *iofinish;
+    bool is_read_async;
+    int have_caps;
+    Fh *f;
+    Inode *in;
+    utime_t start;
+    uint64_t fpos;
+    int64_t offset;
+    uint64_t size;
+    bool movepos;
+  };
+
+  struct CRF_iofinish : public Context {
+    Client::C_Read_Finisher *CRF;
+
+    CRF_iofinish()
+      : CRF(nullptr) {}
+
+    void finish(int r) override {
+      CRF->finish_io(r);
+    }
+
+    // For _read_async, we may not finish in one go, so be prepared for multiple
+    // calls to complete. All the handling though is in C_Read_Finisher.
+    void complete(int r) override {
+      finish(r);
+      if (CRF->iofinished)
+        delete this;
+    }
+  };
+
+  class C_Read_Sync_NonBlocking : public Context {
+  // When operating in non-blocking mode, what used to be done by _read_sync
+  // still needs to be handled, but it needs to be handled without blocking
+  // while still following the semantics. Note that _read_sync is actually
+  // asynchronous. it just uses condition variables to wait. Now instead, we use
+  // this Context class to synchronize the steps.
+  //
+  // The steps will be accomplished by complete/finish being called to complete
+  // each step, with complete only releasing this object once all is finally
+  // complete.
+  public:
+    C_Read_Sync_NonBlocking(Client *clnt, Context *onfinish, Fh *f, Inode *in,
+                            uint64_t fpos, uint64_t off, uint64_t len,
+                            bufferlist *bl, Filer *filer, int have_caps)
+      : clnt(clnt), onfinish(onfinish), f(f), in(in), off(off), len(len), bl(bl),
+        filer(filer), have_caps(have_caps)
+    {
+      left = len;
+      wanted = len;
+      read = 0;
+      pos = off;
+      fini = false;
+    }
+
+    void retry();
+
+  private:
+    Client *clnt;
+    Context *onfinish;
+    Fh *f;
+    Inode *in;
+    uint64_t off;
+    uint64_t len;
+    int left;
+    int wanted;
+    bufferlist *bl;
+    bufferlist tbl;
+    Filer *filer;
+    int have_caps;
+    int read;
+    uint64_t pos;
+    bool fini;
+
+    void finish(int r) override;
+
+    void complete(int r) override
+    {
+      finish(r);
+      if (fini)
+        delete this;
+    }
+  };
+
+  class C_Read_Async_Finisher : public Context {
+  public:
+    C_Read_Async_Finisher(Client *clnt, Context *onfinish, Fh *f, Inode *in,
+                          uint64_t fpos, uint64_t off, uint64_t len)
+      : clnt(clnt), onfinish(onfinish), f(f), in(in), off(off), len(len) {}
+
+  private:
+    Client *clnt;
+    Context *onfinish;
+    Fh *f;
+    Inode *in;
+    uint64_t off;
+    uint64_t len;
+
+    void finish(int r) override;
+  };
+
+  class C_Write_Finisher : public Context {
+  public:
+    void finish_io(int r);
+    void finish_onuninline(int r);
+    void finish_fsync(int r);
+
+    C_Write_Finisher(Client *clnt, Context *onfinish, bool dont_need_uninline,
+                     bool is_file_write, utime_t start, Fh *f, Inode *in,
+                     uint64_t fpos, int64_t offset, uint64_t size,
+                     bool do_fsync, bool syncdataonly)
+      : clnt(clnt), onfinish(onfinish),
+        is_file_write(is_file_write), start(start), f(f), in(in), fpos(fpos),
+        offset(offset), size(size), syncdataonly(syncdataonly) {
+      iofinished_r = 0;
+      onuninlinefinished_r = 0;
+      fsync_r = 0;
+      iofinished = false;
+      onuninlinefinished = dont_need_uninline;
+      fsync_finished = !do_fsync;
+    }
+
+    void finish(int r) override {
+      // We need to override finish, but have nothing to do.
+    }
+
+  private:
+    Client *clnt;
+    Context *onfinish;
+    bool is_file_write;
+    utime_t start;
+    Fh *f;
+    Inode *in;
+    uint64_t fpos;
+    int64_t offset;
+    uint64_t size;
+    bool syncdataonly;
+    int64_t iofinished_r;
+    int64_t onuninlinefinished_r;
+    int64_t fsync_r;
+    bool iofinished;
+    bool onuninlinefinished;
+    bool fsync_finished;
+    bool try_complete();
+  };
+
+  struct CWF_iofinish : public Context {
+    C_Write_Finisher *CWF;
+
+    CWF_iofinish()
+      : CWF(nullptr) {}
+
+    void finish(int r) override {
+      CWF->finish_io(r);
+    }
+  };
+
+  struct CWF_fsync_finish : public Context {
+    C_Write_Finisher *CWF;
+
+    CWF_fsync_finish(C_Write_Finisher *CWF)
+      : CWF(CWF) {}
+
+    void finish(int r) override {
+      CWF->finish_fsync(r);
+    }
+  };
+
+  struct C_nonblocking_fsync_state {
+    Client *clnt;
+
+    // nonblocking_fsync parms
+    Inode *in;
+    bool syncdataonly;
+    Context *onfinish;
+
+    // were local variables
+    ceph_tid_t flush_tid;
+    InodeRef tmp_ref;
+    utime_t start;
+    MetaRequest *req;
+
+    // we need to keep track of where we are
+    int progress;
+    bool flush_wait;
+    bool flush_completed;
+    int result;
+    bool waitfor_safe;
+
+    C_nonblocking_fsync_state(Client *clnt, Inode *in, bool syncdataonly, Context *onfinish)
+      : clnt(clnt), in(in), syncdataonly(syncdataonly), onfinish(onfinish) {
+      flush_tid = 0;
+      start = ceph_clock_now();
+      progress = 0;
+      flush_wait = false;
+      flush_completed = false;
+      result = 0;
+      waitfor_safe = false;
+    }
+
+    void advance();
+
+    void complete_flush(int r);
+  };
+
+  struct C_nonblocking_fsync_state_advancer : Context {
+    Client *clnt;
+    Client::C_nonblocking_fsync_state *state;
+
+    C_nonblocking_fsync_state_advancer(Client *clnt, Client::C_nonblocking_fsync_state *state)
+      : clnt(clnt), state(state) {
+    }
+
+    void finish(int r) override;
+  };
+
+  struct C_nonblocking_fsync_flush_finisher : Context {
+    Client *clnt;
+    Client::C_nonblocking_fsync_state *state;
+
+    C_nonblocking_fsync_flush_finisher(Client *clnt, Client::C_nonblocking_fsync_state *state)
+      : clnt(clnt), state(state) {
+    }
+
+    void finish(int r) override {
+      ceph_assert(ceph_mutex_is_locked_by_me(clnt->client_lock));
+      state->complete_flush(r);
+    }
+  };
+
   struct C_Readahead : public Context {
     C_Readahead(Client *c, Fh *f);
     ~C_Readahead() override;
@@ -1331,7 +1593,8 @@ class Client : public Dispatcher, public md_config_obs_t {
   std::pair<int, bool> _do_remount(bool retry_on_error);
 
   int _read_sync(Fh *f, uint64_t off, uint64_t len, bufferlist *bl, bool *checkeof);
-  int _read_async(Fh *f, uint64_t off, uint64_t len, bufferlist *bl);
+  int _read_async(Fh *f, uint64_t off, uint64_t len, bufferlist *bl,
+                  Context *onfinish);
 
   bool _dentry_valid(const Dentry *dn);
 
@@ -1399,15 +1662,25 @@ class Client : public Dispatcher, public md_config_obs_t {
               std::string alternate_name);
 
   loff_t _lseek(Fh *fh, loff_t offset, int whence);
-  int64_t _read(Fh *fh, int64_t offset, uint64_t size, bufferlist *bl);
+  int64_t _read(Fh *fh, int64_t offset, uint64_t size, bufferlist *bl,
+  		Context *onfinish = nullptr);
+  void do_readahead(Fh *f, Inode *in, uint64_t off, uint64_t len);
+  int64_t _write_success(Fh *fh, utime_t start, uint64_t fpos,
+          int64_t offset, uint64_t size, Inode *in);
   int64_t _write(Fh *fh, int64_t offset, uint64_t size, const char *buf,
-          const struct iovec *iov, int iovcnt);
+          const struct iovec *iov, int iovcnt, Context *onfinish = nullptr,
+          bool do_fsync = false, bool syncdataonly = false);
   int64_t _preadv_pwritev_locked(Fh *fh, const struct iovec *iov,
                                  unsigned iovcnt, int64_t offset,
-                                 bool write, bool clamp_to_int);
+                                 bool write, bool clamp_to_int,
+                                 Context *onfinish = nullptr,
+                                 bufferlist *blp = nullptr,
+                                 bool do_fsync = false, bool syncdataonly = false);
   int _preadv_pwritev(int fd, const struct iovec *iov, unsigned iovcnt,
-                      int64_t offset, bool write);
+                      int64_t offset, bool write, Context *onfinish = nullptr,
+                      bufferlist *blp = nullptr);
   int _flush(Fh *fh);
+  void nonblocking_fsync(Inode *in, bool syncdataonly, Context *onfinish);
   int _fsync(Fh *fh, bool syncdataonly);
   int _fsync(Inode *in, bool syncdataonly);
   int _sync_fs();

src/client/Client.h

@@ -241,8 +241,9 @@ struct Inode : RefCountedObject {
   std::map<frag_t,int> fragmap;  // known frag -> mds mappings
   std::map<frag_t, std::vector<mds_rank_t>> frag_repmap; // non-auth mds mappings
 
-  std::list<ceph::condition_variable*> waitfor_caps;
-  std::list<ceph::condition_variable*> waitfor_commit;
+  std::list<Context*> waitfor_caps;
+  std::list<Context*> waitfor_caps_pending;
+  std::list<Context*> waitfor_commit;
   std::list<ceph::condition_variable*> waitfor_deleg;
 
   Dentry *get_first_parent() {

src/client/Inode.h

@@ -70,7 +70,7 @@ struct MetaRequest {
 
   ceph::condition_variable *caller_cond = NULL;   // who to take up
   ceph::condition_variable *dispatch_cond = NULL; // who to kick back
-  std::list<ceph::condition_variable*> waitfor_safe;
+  std::list<Context*> waitfor_safe;
 
   InodeRef target;
   UserPerm perms;

src/client/MetaRequest.h

@@ -1286,6 +1286,23 @@ inline bufferhash& operator<<(bufferhash& l, const bufferlist &r) {
   return l;
 }
 
+static inline
+void copy_bufferlist_to_iovec(const struct iovec *iov, unsigned iovcnt,
+                              bufferlist *bl, int64_t r)
+{
+  auto iter = bl->cbegin();
+  for (unsigned j = 0, resid = r; j < iovcnt && resid > 0; j++) {
+         /*
+          * This piece of code aims to handle the case that bufferlist
+          * does not have enough data to fill in the iov
+          */
+         const auto round_size = std::min<unsigned>(resid, iov[j].iov_len);
+         iter.copy(round_size, reinterpret_cast<char*>(iov[j].iov_base));
+         resid -= round_size;
+         /* iter is self-updating */
+  }
+}
+
 } // namespace buffer
 
 } // namespace ceph