Merge pull request ceph#4753 from ceph/wip-client-inode-refs

client: hold reference for returned inode

Reviewed-by: John Spray <[email protected]>

src/client/Client.h

@@ -47,12 +47,13 @@ using std::fstream;
 #include "common/Mutex.h"
 #include "common/Timer.h"
 #include "common/Finisher.h"
-
 #include "common/compiler_extensions.h"
 #include "common/cmdparse.h"
 
 #include "osdc/ObjectCacher.h"
 
+#include "InodeRef.h"
+
 class MDSMap;
 class MonClient;
 
@@ -117,7 +118,6 @@ struct DirEntry {
   DirEntry(const string &n, struct stat& s, int stm) : d_name(n), st(s), stmask(stm) {}
 };
 
-struct Inode;
 struct Cap;
 class Dir;
 class Dentry;
@@ -167,7 +167,7 @@ struct dir_result_t {
   }
 
 
-  Inode *inode;
+  InodeRef inode;
 
   int64_t offset;        // high bits: frag_t, low bits: an offset
 
@@ -180,7 +180,7 @@ struct dir_result_t {
   int start_shared_gen;  // dir shared_gen at start of readdir
 
   frag_t buffer_frag;
-  vector<pair<string,Inode*> > *buffer;
+  vector<pair<string,InodeRef> > *buffer;
 
   string at_cache_name;  // last entry we successfully returned
 
@@ -300,13 +300,13 @@ class Client : public Dispatcher, public md_config_obs_t {
 
   int make_request(MetaRequest *req, int uid, int gid,
 		   //MClientRequest *req, int uid, int gid,
-		   Inode **ptarget = 0, bool *pcreated = 0,
+		   InodeRef *ptarget = 0, bool *pcreated = 0,
 		   int use_mds=-1, bufferlist *pdirbl=0);
   void put_request(MetaRequest *request);
   void unregister_request(MetaRequest *request);
 
   int verify_reply_trace(int r, MetaRequest *request, MClientReply *reply,
-			 Inode **ptarget, bool *pcreated, int uid, int gid);
+			 InodeRef *ptarget, bool *pcreated, int uid, int gid);
   void encode_cap_releases(MetaRequest *request, mds_rank_t mds);
   int encode_inode_release(Inode *in, MetaRequest *req,
 			   mds_rank_t mds, int drop,
@@ -336,7 +336,7 @@ class Client : public Dispatcher, public md_config_obs_t {
 
 public:
   entity_name_t get_myname() { return messenger->get_myname(); } 
-  void sync_write_commit(Inode *in);
+  void sync_write_commit(InodeRef& in);
 
 protected:
   Filer                 *filer;     
@@ -347,7 +347,7 @@ class Client : public Dispatcher, public md_config_obs_t {
   // cache
   ceph::unordered_map<vinodeno_t, Inode*> inode_map;
   Inode*                 root;
-  map<Inode*, Inode*>    root_parents;
+  map<Inode*, InodeRef>  root_parents;
   Inode*                 root_ancestor;
   LRU                    lru;    // lru list of Dentry's in our local metadata cache.
 
@@ -424,6 +424,7 @@ class Client : public Dispatcher, public md_config_obs_t {
   friend class C_Client_SyncCommit; // Asserts on client_lock
   friend class C_Client_RequestInterrupt;
   friend class C_Client_Remount;
+  friend void intrusive_ptr_release(Inode *in);
 
   //int get_cache_size() { return lru.lru_get_size(); }
   //void set_cache_size(int m) { lru.lru_set_max(m); }
@@ -437,9 +438,12 @@ class Client : public Dispatcher, public md_config_obs_t {
   void unlink(Dentry *dn, bool keepdir, bool keepdentry);
 
   // path traversal for high-level interface
-  Inode *cwd;
-  int path_walk(const filepath& fp, Inode **end, bool followsym=true);
+  InodeRef cwd;
+  int path_walk(const filepath& fp, InodeRef *end, bool followsym=true);
   int fill_stat(Inode *in, struct stat *st, frag_info_t *dirstat=0, nest_info_t *rstat=0);
+  int fill_stat(InodeRef& in, struct stat *st, frag_info_t *dirstat=0, nest_info_t *rstat=0) {
+    return fill_stat(in.get(), st, dirstat, rstat);
+  }
   void touch_dn(Dentry *dn);
 
   // trim cache.
@@ -570,7 +574,7 @@ class Client : public Dispatcher, public md_config_obs_t {
   void _schedule_invalidate_callback(Inode *in, int64_t off, int64_t len, bool keep_caps);
   void _invalidate_inode_cache(Inode *in);
   void _invalidate_inode_cache(Inode *in, int64_t off, int64_t len);
-  void _async_invalidate(Inode *in, int64_t off, int64_t len, bool keep_caps);
+  void _async_invalidate(InodeRef& in, int64_t off, int64_t len, bool keep_caps);
   void _release(Inode *in);
 
   /**
@@ -654,27 +658,34 @@ class Client : public Dispatcher, public md_config_obs_t {
 
   // internal interface
   //   call these with client_lock held!
-  int _do_lookup(Inode *dir, const string& name, Inode **target);
-  int _lookup(Inode *dir, const string& dname, Inode **target);
+  int _do_lookup(Inode *dir, const string& name, InodeRef *target);
+  int _lookup(Inode *dir, const string& dname, InodeRef *target);
 
-  int _link(Inode *in, Inode *dir, const char *name, int uid=-1, int gid=-1, Inode **inp = 0);
+  int _link(Inode *in, Inode *dir, const char *name, int uid=-1, int gid=-1, InodeRef *inp = 0);
   int _unlink(Inode *dir, const char *name, int uid=-1, int gid=-1);
   int _rename(Inode *olddir, const char *oname, Inode *ndir, const char *nname, int uid=-1, int gid=-1);
-  int _mkdir(Inode *dir, const char *name, mode_t mode, int uid=-1, int gid=-1, Inode **inp = 0);
+  int _mkdir(Inode *dir, const char *name, mode_t mode, int uid=-1, int gid=-1, InodeRef *inp = 0);
   int _rmdir(Inode *dir, const char *name, int uid=-1, int gid=-1);
-  int _symlink(Inode *dir, const char *name, const char *target, int uid=-1, int gid=-1, Inode **inp = 0);
-  int _mknod(Inode *dir, const char *name, mode_t mode, dev_t rdev, int uid=-1, int gid=-1, Inode **inp = 0);
-  int _setattr(Inode *in, struct stat *attr, int mask, int uid=-1, int gid=-1, Inode **inp = 0);
+  int _symlink(Inode *dir, const char *name, const char *target, int uid=-1, int gid=-1, InodeRef *inp = 0);
+  int _mknod(Inode *dir, const char *name, mode_t mode, dev_t rdev, int uid=-1, int gid=-1, InodeRef *inp = 0);
+  int _setattr(Inode *in, struct stat *attr, int mask, int uid=-1, int gid=-1, InodeRef *inp = 0);
+  int _setattr(InodeRef &in, struct stat *attr, int mask, int uid=-1, int gid=-1, InodeRef *inp = 0) {
+    return _setattr(in.get(), attr, mask, uid, gid, inp);
+  }
   int _getattr(Inode *in, int mask, int uid=-1, int gid=-1, bool force=false);
+  int _getattr(InodeRef &in, int mask, int uid=-1, int gid=-1, bool force=false) {
+    return _getattr(in.get(), mask, uid, gid, force);
+  }
   int _readlink(Inode *in, char *buf, size_t size);
   int _getxattr(Inode *in, const char *name, void *value, size_t len, int uid=-1, int gid=-1);
   int _listxattr(Inode *in, char *names, size_t len, int uid=-1, int gid=-1);
   int _setxattr(Inode *in, const char *name, const void *value, size_t len, int flags, int uid=-1, int gid=-1);
   int _removexattr(Inode *in, const char *nm, int uid=-1, int gid=-1);
   int _open(Inode *in, int flags, mode_t mode, Fh **fhp, int uid=-1, int gid=-1);
-  int _create(Inode *in, const char *name, int flags, mode_t mode, Inode **inp, Fh **fhp,
+  int _create(Inode *in, const char *name, int flags, mode_t mode, InodeRef *inp, Fh **fhp,
               int stripe_unit, int stripe_count, int object_size, const char *data_pool,
 	      bool *created = NULL, int uid=-1, int gid=-1);
+
   loff_t _lseek(Fh *fh, loff_t offset, int whence);
   int _read(Fh *fh, int64_t offset, uint64_t size, bufferlist *bl);
   int _write(Fh *fh, int64_t offset, uint64_t size, const char *buf,
@@ -973,6 +984,7 @@ class Client : public Dispatcher, public md_config_obs_t {
   int ll_getlk(Fh *fh, struct flock *fl, uint64_t owner);
   int ll_setlk(Fh *fh, struct flock *fl, uint64_t owner, int sleep, void *fuse_req);
   int ll_flock(Fh *fh, int cmd, uint64_t owner, void *fuse_req);
+  int ll_file_layout(Fh *fh, ceph_file_layout *layout);
   void ll_interrupt(void *d);
   int ll_get_stripe_osd(struct Inode *in, uint64_t blockno,
 			ceph_file_layout* layout);

src/client/Dentry.h

@@ -5,17 +5,18 @@
 #include "include/xlist.h"
 
 #include "mds/mdstypes.h"
+#include "InodeRef.h"
 
 class Dir;
 struct Inode;
 
 class Dentry : public LRUObject {
  public:
-  string  name;                      // sort of lame
+  string   name;                      // sort of lame
   //const char *name;
-  Dir     *dir;
-  Inode   *inode;
-  int     ref;                       // 1 if there's a dir beneath me.
+  Dir	   *dir;
+  InodeRef inode;
+  int	   ref;                       // 1 if there's a dir beneath me.
   uint64_t offset;
   mds_rank_t lease_mds;
   utime_t lease_ttl;
@@ -47,7 +48,7 @@ class Dentry : public LRUObject {
   void dump(Formatter *f) const;
 
   Dentry() :
-    dir(0), inode(0), ref(1), offset(0),
+    dir(0), ref(1), offset(0),
     lease_mds(-1), lease_gen(0), lease_seq(0), cap_shared_gen(0),
     item_dentry_list(this)  { }
 private:

src/client/Fh.h

@@ -3,16 +3,16 @@
 
 #include "common/Readahead.h"
 #include "include/types.h"
+#include "InodeRef.h"
 
-struct Inode;
 class Cond;
 class ceph_lock_state_t;
 
 // file handle for any open file state
 
 struct Fh {
+  InodeRef  inode;
   int	    _ref;
-  Inode     *inode;
   loff_t    pos;
   int       mds;        // have to talk to mds we opened with (for now)
   int       mode;       // the mode i opened the file with
@@ -27,7 +27,7 @@ struct Fh {
   ceph_lock_state_t *fcntl_locks;
   ceph_lock_state_t *flock_locks;
 
-  Fh() : _ref(1), inode(0), pos(0), mds(0), mode(0), flags(0), pos_locked(false),
+  Fh() : _ref(1), pos(0), mds(0), mode(0), flags(0), pos_locked(false),
       readahead(), fcntl_locks(NULL), flock_locks(NULL) {}
   void get() { ++_ref; }
   int put() { return --_ref; }

src/client/Inode.cc

@@ -1,10 +1,11 @@
 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
 // vim: ts=8 sw=2 smarttab
 
-#include "MetaSession.h"
+#include "Client.h"
 #include "Inode.h"
 #include "Dentry.h"
 #include "Dir.h"
+#include "MetaSession.h"
 #include "ClientSnapRealm.h"
 
 ostream& operator<<(ostream &out, Inode &in)
@@ -126,7 +127,7 @@ int Inode::put_cap_ref(int cap)
     if (cap & 1) {
       int c = 1 << n;
       if (cap_refs[c] <= 0) {
-	lderr(cct) << "put_cap_ref " << ccap_string(c) << " went negative on " << *this << dendl;
+	lderr(client->cct) << "put_cap_ref " << ccap_string(c) << " went negative on " << *this << dendl;
 	assert(cap_refs[c] > 0);
       }
       if (--cap_refs[c] == 0)
@@ -151,7 +152,7 @@ bool Inode::cap_is_valid(Cap* cap)
     << "cap expire  " << cap->session->cap_ttl << std::endl
     << "cur time    " << ceph_clock_now(cct) << std::endl;*/
   if ((cap->session->cap_gen <= cap->gen)
-      && (ceph_clock_now(cct) < cap->session->cap_ttl)) {
+      && (ceph_clock_now(client->cct) < cap->session->cap_ttl)) {
     return true;
   }
   return false;
@@ -268,7 +269,7 @@ Dir *Inode::open_dir()
 {
   if (!dir) {
     dir = new Dir(this);
-    lsubdout(cct, mds, 15) << "open_dir " << dir << " on " << this << dendl;
+    lsubdout(client->cct, client, 15) << "open_dir " << dir << " on " << this << dendl;
     assert(dn_set.size() < 2); // dirs can't be hard-linked
     if (!dn_set.empty())
       (*dn_set.begin())->get();      // pin dentry
@@ -307,6 +308,21 @@ bool Inode::check_mode(uid_t ruid, gid_t rgid, gid_t *sgids, int sgids_count, ui
   return (mode & fmode) == fmode;
 }
 
+void Inode::get() {
+  _ref++;
+  lsubdout(client->cct, client, 15) << "inode.get on " << this << " " <<  ino << '.' << snapid
+				    << " now " << _ref << dendl;
+}
+
+//private method to put a reference; see Client::put_inode()
+int Inode::_put(int n) {
+  _ref -= n;
+  lsubdout(client->cct, client, 15) << "inode.put on " << this << " " << ino << '.' << snapid
+				    << " now " << _ref << dendl;
+  assert(_ref >= 0);
+  return _ref;
+}
+
 
 void Inode::dump(Formatter *f) const
 {

src/client/Inode.h

@@ -13,6 +13,9 @@
 #include "osdc/ObjectCacher.h"
 #include "include/assert.h"
 
+#include "InodeRef.h"
+
+class Client;
 struct MetaSession;
 class Dentry;
 class Dir;
@@ -42,7 +45,7 @@ struct Cap {
 
 struct CapSnap {
   //snapid_t follows;  // map key
-  Inode *in;
+  InodeRef in;
   SnapContext context;
   int issued, dirty;
 
@@ -151,7 +154,7 @@ class QuotaTree {
 #define I_DIR_ORDERED 2
 
 struct Inode {
-  CephContext *cct;
+  Client *client;
 
   // -- the actual inode --
   inodeno_t ino;
@@ -240,7 +243,7 @@ struct Inode {
 
   SnapRealm *snaprealm;
   xlist<Inode*>::item snaprealm_item;
-  Inode *snapdir_parent;  // only if we are a snapdir inode
+  InodeRef snapdir_parent;  // only if we are a snapdir inode
   map<snapid_t,CapSnap*> cap_snaps;   // pending flush to mds
 
   //int open_by_mode[CEPH_FILE_MODE_NUM];
@@ -271,19 +274,8 @@ struct Inode {
   void make_long_path(filepath& p);
   void make_nosnap_relative_path(filepath& p);
 
-  void get() {
-    _ref++;
-    lsubdout(cct, mds, 15) << "inode.get on " << this << " " <<  ino << '.' << snapid
-                           << " now " << _ref << dendl;
-  }
-  /// private method to put a reference; see Client::put_inode()
-  int _put(int n=1) {
-    _ref -= n; 
-    lsubdout(cct, mds, 15) << "inode.put on " << this << " " << ino << '.' << snapid
-                           << " now " << _ref << dendl;
-    assert(_ref >= 0);
-    return _ref;
-  }
+  void get();
+  int _put(int n=1);
 
   int get_num_ref() {
     return _ref;
@@ -303,8 +295,8 @@ struct Inode {
 
   xlist<MetaRequest*> unsafe_dir_ops;
 
-  Inode(CephContext *cct_, vinodeno_t vino, ceph_file_layout *newlayout)
-    : cct(cct_), ino(vino.ino), snapid(vino.snapid),
+  Inode(Client *c, vinodeno_t vino, ceph_file_layout *newlayout)
+    : client(c), ino(vino.ino), snapid(vino.snapid),
       rdev(0), mode(0), uid(0), gid(0), nlink(0),
       size(0), truncate_seq(1), truncate_size(-1),
       time_warp_seq(0), max_size(0), version(0), xattr_version(0),
@@ -315,7 +307,7 @@ struct Inode {
       dirty_caps(0), flushing_caps(0), flushing_cap_seq(0), shared_gen(0), cache_gen(0),
       snap_caps(0), snap_cap_refs(0),
       cap_item(this), flushing_cap_item(this), last_flush_tid(0),
-      snaprealm(0), snaprealm_item(this), snapdir_parent(0),
+      snaprealm(0), snaprealm_item(this),
       oset((void *)this, newlayout->fl_pg_pool, ino),
       reported_size(0), wanted_max_size(0), requested_max_size(0),
       _ref(0), ll_ref(0), dir(0), dn_set(),

src/client/InodeRef.h

@@ -0,0 +1,12 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_CLIENT_INODEREF_H
+#define CEPH_CLIENT_INODEREF_H
+
+#include <boost/intrusive_ptr.hpp>
+class Inode;
+void intrusive_ptr_add_ref(Inode *in);
+void intrusive_ptr_release(Inode *in);
+typedef boost::intrusive_ptr<Inode> InodeRef;
+#endif