client: switch ll_* methods to Inode* arguments

This commit implements a set of API changes proposed by
"David Zafman" <david.zafman@inktank.com>.

The principle change is consistently use Ceph Inode pointers as
context for inode operations, in place of materialized snapid,
inode# (vinodeno_t) tuples.

This provides a strictly handle-based interface to users.  They can either
do lookups themselves, or call a lookup method to get an Inode* from a
vinodeno_t.

(Build fixed in later commits, e.g., SyntheticClient.cc.)

Signed-off-by: Matt Benjamin <matt@linuxbox.com>

2 files changed, 332 insertions, 188 deletions

diff --git a/src/client/Client.cc b/src/client/Client.cc
index cbc771ae595..e5bd866790c 100644
--- a/src/client/Client.cc
+++ b/src/client/Client.cc
@@ -13,7 +13,6 @@
  */
 
 
-
 // unix-ey fs stuff
 #include <unistd.h>
 #include <sys/types.h>
@@ -3654,6 +3653,18 @@ int Client::check_permissions(Inode *in, int flags, int uid, int gid)
   return 0;
 }
 
+vinodeno_t Client::_get_vino(Inode *in)
+{
+  /* The caller must hold the client lock */
+  return vinodeno_t(in->ino, in->snapid);
+}
+
+inodeno_t Client::_get_inodeno(Inode *in)
+{
+  /* The caller must hold the client lock */
+  return in->ino;
+}
+
 
 // -------------------
 // MOUNT
@@ -6314,9 +6325,11 @@ int Client::statfs(const char *path, struct statvfs *stbuf)
   return rval;
 }
 
-int Client::ll_statfs(vinodeno_t vino, struct statvfs *stbuf)
+int Client::ll_statfs(Inode *in, struct statvfs *stbuf)
 {
-  tout(cct) << "ll_statfs" << std::endl;
+  /* Since the only thing this does is wrap a call to statfs, and
+     statfs takes a lock, it doesn't seem we have a need to split it
+     out. */
   return statfs(0, stbuf);
 }
 
@@ -6483,34 +6496,19 @@ Inode *Client::open_snapdir(Inode *diri)
   return in;
 }
 
-int Client::ll_lookup(vinodeno_t parent, const char *name, struct stat *attr, int uid, int gid)
+int Client::ll_lookup(Inode *parent, const char *name, struct stat *attr,
+		      Inode **out, int uid, int gid)
 {
   Mutex::Locker lock(client_lock);
   ldout(cct, 3) << "ll_lookup " << parent << " " << name << dendl;
   tout(cct) << "ll_lookup" << std::endl;
-  tout(cct) << parent.ino.val << std::endl;
   tout(cct) << name << std::endl;
 
-  string dname = name;
-  Inode *diri = 0;
-  Inode *in = 0;
+  string dname(name);
+  Inode *in;
   int r = 0;
 
-  if (inode_map.count(parent) == 0) {
-    ldout(cct, 1) << "ll_lookup " << parent << " " << name << " -> ENOENT (parent DNE... WTF)" << dendl;
-    r = -ENOENT;
-    attr->st_ino = 0;
-    goto out;
-  }
-  diri = inode_map[parent];
-  if (!diri->is_dir()) {
-    ldout(cct, 1) << "ll_lookup " << parent << " " << name << " -> ENOTDIR (parent not a dir... WTF)" << dendl;
-    r = -ENOTDIR;
-    attr->st_ino = 0;
-    goto out;
-  }
-
-  r = _lookup(diri, dname.c_str(), &in);
+  r = _lookup(parent, dname.c_str(), &in);
   if (r < 0) {
     attr->st_ino = 0;
     goto out;
@@ -6524,9 +6522,37 @@ int Client::ll_lookup(vinodeno_t parent, const char *name, struct stat *attr, in
   ldout(cct, 3) << "ll_lookup " << parent << " " << name
 	  << " -> " << r << " (" << hex << attr->st_ino << dec << ")" << dendl;
   tout(cct) << attr->st_ino << std::endl;
+  *out = in;
   return r;
 }
 
+int Client::ll_walk(const char* name, Inode **i, struct stat *attr)
+{
+  Mutex::Locker lock(client_lock);
+  filepath fp(name, 0);
+  Inode *destination = NULL;
+  int rc;
+
+  ldout(cct, 3) << "ll_walk" << name << dendl;
+  tout(cct) << "ll_walk" << std::endl;
+  tout(cct) << name << std::endl;
+
+  rc = path_walk(fp, &destination, false);
+  if (rc < 0)
+    {
+      attr->st_ino = 0;
+      *i = NULL;
+      return rc;
+    }
+  else
+    {
+      fill_stat(destination, attr);
+      *i = destination;
+      return 0;
+    }
+}
+
+
 void Client::_ll_get(Inode *in)
 {
   if (in->ll_ref == 0)
@@ -6562,45 +6588,61 @@ void Client::_ll_drop_pins()
   }
 }
 
-bool Client::ll_forget(vinodeno_t vino, int num)
+bool Client::ll_forget(Inode *in, int count)
 {
   Mutex::Locker lock(client_lock);
-  ldout(cct, 3) << "ll_forget " << vino << " " << num << dendl;
+  inodeno_t ino = _get_inodeno(in);
+
+  ldout(cct, 3) << "ll_forget " << ino << " " << count << dendl;
   tout(cct) << "ll_forget" << std::endl;
-  tout(cct) << vino.ino.val << std::endl;
-  tout(cct) << num << std::endl;
+  tout(cct) << ino.val << std::endl;
+  tout(cct) << count << std::endl;
 
-  if (vino.ino == 1) return true;  // ignore forget on root.
+  if (ino == 1) return true;  // ignore forget on root.
 
   bool last = false;
-  if (inode_map.count(vino) == 0) {
-    ldout(cct, 1) << "WARNING: ll_forget on " << vino << " " << num 
-	    << ", which I don't have" << dendl;
-  } else {
-    Inode *in = inode_map[vino];
-    assert(in);
-    if (in->ll_ref < num) {
-      ldout(cct, 1) << "WARNING: ll_forget on " << vino << " " << num << ", which only has ll_ref=" << in->ll_ref << dendl;
-      _ll_put(in, in->ll_ref);
+  if (in->ll_ref < count) {
+    ldout(cct, 1) << "WARNING: ll_forget on " << ino << " " << count
+		  << ", which only has ll_ref=" << in->ll_ref << dendl;
+    _ll_put(in, in->ll_ref);
       last = true;
     } else {
-      if (_ll_put(in, num) == 0)
-	last = true;
-    }
+    if (_ll_put(in, count) == 0)
+      last = true;
   }
+
   return last;
 }
 
-Inode *Client::_ll_get_inode(vinodeno_t vino)
+bool Client::ll_put(Inode *in)
 {
-  assert(inode_map.count(vino));
-  return inode_map[vino];
+  /* ll_forget already takes the lock */
+  return ll_forget(in, 1);
 }
 
+snapid_t Client::ll_get_snapid(Inode *in)
+{
+  Mutex::Locker lock(client_lock);
+  return in->snapid;
+}
+
+Inode *Client::ll_get_inode(vinodeno_t vino)
+{
+  Mutex::Locker lock(client_lock);
+  hash_map<vinodeno_t,Inode*>::iterator p = inode_map.find(vino);
+  if (p == inode_map.end())
+    return NULL;
+  Inode *in = p->second;
+  _ll_get(in);
+  return in;
+}
 
-int Client::ll_getattr(vinodeno_t vino, struct stat *attr, int uid, int gid)
+int Client::ll_getattr(Inode *in, struct stat *attr, int uid, int gid)
 {
   Mutex::Locker lock(client_lock);
+
+  vinodeno_t vino = _get_vino(in);
+
   ldout(cct, 3) << "ll_getattr " << vino << dendl;
   tout(cct) << "ll_getattr" << std::endl;
   tout(cct) << vino.ino.val << std::endl;
@@ -6612,7 +6654,6 @@ int Client::ll_getattr(vinodeno_t vino, struct stat *attr, int uid, int gid)
     return 0;
   }
 
-  Inode *in = _ll_get_inode(vino);
   int res;
   if (vino.snapid < CEPH_NOSNAP)
     res = 0;
@@ -6624,10 +6665,15 @@ int Client::ll_getattr(vinodeno_t vino, struct stat *attr, int uid, int gid)
   return res;
 }
 
-int Client::ll_setattr(vinodeno_t vino, struct stat *attr, int mask, int uid, int gid)
+int Client::ll_setattr(Inode *in, struct stat *attr, int mask, int uid,
+		       int gid)
 {
   Mutex::Locker lock(client_lock);
-  ldout(cct, 3) << "ll_setattr " << vino << " mask " << hex << mask << dec << dendl;
+
+  vinodeno_t vino = _get_vino(in);
+
+  ldout(cct, 3) << "ll_setattr " << vino << " mask " << hex << mask << dec
+		<< dendl;
   tout(cct) << "ll_setattr" << std::endl;
   tout(cct) << vino.ino.val << std::endl;
   tout(cct) << attr->st_mode << std::endl;
@@ -6638,7 +6684,6 @@ int Client::ll_setattr(vinodeno_t vino, struct stat *attr, int mask, int uid, in
   tout(cct) << attr->st_atime << std::endl;
   tout(cct) << mask << std::endl;
 
-  Inode *in = _ll_get_inode(vino);
   Inode *target = in;
   int res = _setattr(in, attr, mask, uid, gid, &target);
   if (res == 0) {
@@ -6802,15 +6847,18 @@ int Client::_getxattr(Inode *in, const char *name, void *value, size_t size,
   return r;
 }
 
-int Client::ll_getxattr(vinodeno_t vino, const char *name, void *value, size_t size, int uid, int gid)
+int Client::ll_getxattr(Inode *in, const char *name, void *value,
+			size_t size, int uid, int gid)
 {
   Mutex::Locker lock(client_lock);
+
+  vinodeno_t vino = _get_vino(in);
+
   ldout(cct, 3) << "ll_getxattr " << vino << " " << name << " size " << size << dendl;
   tout(cct) << "ll_getxattr" << std::endl;
   tout(cct) << vino.ino.val << std::endl;
   tout(cct) << name << std::endl;
 
-  Inode *in = _ll_get_inode(vino);
   return _getxattr(in, name, value, size, uid, gid);
 }
 
@@ -6855,20 +6903,23 @@ int Client::_listxattr(Inode *in, char *name, size_t size, int uid, int gid)
   return r;
 }
 
-int Client::ll_listxattr(vinodeno_t vino, char *names, size_t size, int uid, int gid)
+int Client::ll_listxattr(Inode *in, char *names, size_t size, int uid,
+			 int gid)
 {
   Mutex::Locker lock(client_lock);
+
+  vinodeno_t vino = _get_vino(in);
+
   ldout(cct, 3) << "ll_listxattr " << vino << " size " << size << dendl;
   tout(cct) << "ll_listxattr" << std::endl;
   tout(cct) << vino.ino.val << std::endl;
   tout(cct) << size << std::endl;
 
-  Inode *in = _ll_get_inode(vino);
   return _listxattr(in, names, size, uid, gid);
 }
 
-int Client::_setxattr(Inode *in, const char *name, const void *value, size_t size, int flags,
-		      int uid, int gid)
+int Client::_setxattr(Inode *in, const char *name, const void *value,
+		      size_t size, int flags, int uid, int gid)
 {
   if (in->snapid != CEPH_NOSNAP) {
     return -EROFS;
@@ -6901,16 +6952,18 @@ int Client::_setxattr(Inode *in, const char *name, const void *value, size_t siz
   return res;
 }
 
-int Client::ll_setxattr(vinodeno_t vino, const char *name, const void *value, size_t size, int flags,
-			int uid, int gid)
+int Client::ll_setxattr(Inode *in, const char *name, const void *value,
+			size_t size, int flags, int uid, int gid)
 {
   Mutex::Locker lock(client_lock);
+
+  vinodeno_t vino = _get_vino(in);
+
   ldout(cct, 3) << "ll_setxattr " << vino << " " << name << " size " << size << dendl;
   tout(cct) << "ll_setxattr" << std::endl;
   tout(cct) << vino.ino.val << std::endl;
   tout(cct) << name << std::endl;
 
-  Inode *in = _ll_get_inode(vino);
   return _setxattr(in, name, value, size, flags, uid, gid);
 }
 
@@ -6942,27 +6995,31 @@ int Client::_removexattr(Inode *in, const char *name, int uid, int gid)
 }
 
 
-int Client::ll_removexattr(vinodeno_t vino, const char *name, int uid, int gid)
+int Client::ll_removexattr(Inode *in, const char *name, int uid, int gid)
 {
   Mutex::Locker lock(client_lock);
+
+  vinodeno_t vino = _get_vino(in);
+
   ldout(cct, 3) << "ll_removexattr " << vino << " " << name << dendl;
   tout(cct) << "ll_removexattr" << std::endl;
   tout(cct) << vino.ino.val << std::endl;
   tout(cct) << name << std::endl;
 
-  Inode *in = _ll_get_inode(vino);
   return _removexattr(in, name, uid, gid);
 }
 
 
-int Client::ll_readlink(vinodeno_t vino, const char **value, int uid, int gid)
+int Client::ll_readlink(Inode *in, const char **value, int uid, int gid)
 {
   Mutex::Locker lock(client_lock);
+
+  vinodeno_t vino = _get_vino(in);
+
   ldout(cct, 3) << "ll_readlink " << vino << dendl;
   tout(cct) << "ll_readlink" << std::endl;
   tout(cct) << vino.ino.val << std::endl;
 
-  Inode *in = _ll_get_inode(vino);
   set<Dentry*>::iterator dn = in->dn_set.begin();
   while (dn != in->dn_set.end()) {
     touch_dn(*dn);
@@ -6976,12 +7033,14 @@ int Client::ll_readlink(vinodeno_t vino, const char **value, int uid, int gid)
     *value = "";
     r = -EINVAL;
   }
-  ldout(cct, 3) << "ll_readlink " << vino << " = " << r << " (" << *value << ")" << dendl;
+  ldout(cct, 3) << "ll_readlink " << vino << " = " << r << " (" << *value
+		<< ")" << dendl;
   return r;
 }
 
-int Client::_mknod(Inode *dir, const char *name, mode_t mode, dev_t rdev, int uid, int gid, Inode **inp)
-{ 
+int Client::_mknod(Inode *dir, const char *name, mode_t mode, dev_t rdev,
+		   int uid, int gid, Inode **inp)
+{
   ldout(cct, 3) << "_mknod(" << dir->ino << " " << name << ", 0" << oct
 		<< mode << dec << ", " << rdev << ", uid " << uid << ", gid "
 		<< gid << ")" << dendl;
@@ -7023,31 +7082,38 @@ int Client::_mknod(Inode *dir, const char *name, mode_t mode, dev_t rdev, int ui
   return res;
 }
 
-int Client::ll_mknod(vinodeno_t parent, const char *name, mode_t mode, dev_t rdev, struct stat *attr, int uid, int gid)
+int Client::ll_mknod(Inode *parent, const char *name, mode_t mode,
+		     dev_t rdev, struct stat *attr, Inode **out,
+		     int uid, int gid)
 {
   Mutex::Locker lock(client_lock);
-  ldout(cct, 3) << "ll_mknod " << parent << " " << name << dendl;
+
+  vinodeno_t vparent = _get_vino(parent);
+
+  ldout(cct, 3) << "ll_mknod " << vparent << " " << name << dendl;
   tout(cct) << "ll_mknod" << std::endl;
-  tout(cct) << parent.ino.val << std::endl;
+  tout(cct) << vparent.ino.val << std::endl;
   tout(cct) << name << std::endl;
   tout(cct) << mode << std::endl;
   tout(cct) << rdev << std::endl;
 
-  Inode *diri = _ll_get_inode(parent);
-  Inode *in = 0;
-  int r = _mknod(diri, name, mode, rdev, uid, gid, &in);
+  Inode *in = NULL;
+  int r = _mknod(parent, name, mode, rdev, uid, gid, &in);
   if (r == 0) {
     fill_stat(in, attr);
     _ll_get(in);
   }
   tout(cct) << attr->st_ino << std::endl;
-  ldout(cct, 3) << "ll_mknod " << parent << " " << name
+  ldout(cct, 3) << "ll_mknod " << vparent << " " << name
 	  << " = " << r << " (" << hex << attr->st_ino << dec << ")" << dendl;
+  *out = in;
   return r;
 }
 
-int Client::_create(Inode *dir, const char *name, int flags, mode_t mode, Inode **inp, Fh **fhp,
-    int stripe_unit, int stripe_count, int object_size, const char *data_pool, bool *created, int uid, int gid)
+int Client::_create(Inode *dir, const char *name, int flags, mode_t mode,
+		    Inode **inp, Fh **fhp, int stripe_unit, int stripe_count,
+		    int object_size, const char *data_pool, bool *created,
+		    int uid, int gid)
 {
   ldout(cct, 3) << "_create(" << dir->ino << " " << name << ", 0" << oct <<
     mode << dec << ")" << dendl;
@@ -7102,8 +7168,11 @@ int Client::_create(Inode *dir, const char *name, int flags, mode_t mode, Inode
     goto reply_error;
   }
 
-  (*inp)->get_open_ref(cmode);
-  *fhp = _create_fh(*inp, flags, cmode);
+  /* If the caller passed a value in fhp, do the open */
+  if(fhp) {
+    (*inp)->get_open_ref(cmode);
+    *fhp = _create_fh(*inp, flags, cmode);
+  }
 
  reply_error:
   trim_cache();
@@ -7166,26 +7235,29 @@ int Client::_mkdir(Inode *dir, const char *name, mode_t mode, int uid, int gid,
   return res;
 }
 
-int Client::ll_mkdir(vinodeno_t parent, const char *name, mode_t mode, struct stat *attr, int uid, int gid)
+int Client::ll_mkdir(Inode *parent, const char *name, mode_t mode,
+		     struct stat *attr, Inode **out, int uid, int gid)
 {
   Mutex::Locker lock(client_lock);
-  ldout(cct, 3) << "ll_mkdir " << parent << " " << name << dendl;
+
+  vinodeno_t vparent = _get_vino(parent);
+
+  ldout(cct, 3) << "ll_mkdir " << vparent << " " << name << dendl;
   tout(cct) << "ll_mkdir" << std::endl;
-  tout(cct) << parent.ino.val << std::endl;
+  tout(cct) << vparent.ino.val << std::endl;
   tout(cct) << name << std::endl;
   tout(cct) << mode << std::endl;
 
-  Inode *diri = _ll_get_inode(parent);
-
-  Inode *in = 0;
-  int r = _mkdir(diri, name, mode, uid, gid, &in);
+  Inode *in = NULL;
+  int r = _mkdir(parent, name, mode, uid, gid, &in);
   if (r == 0) {
     fill_stat(in, attr);
     _ll_get(in);
   }
   tout(cct) << attr->st_ino << std::endl;
-  ldout(cct, 3) << "ll_mkdir " << parent << " " << name
+  ldout(cct, 3) << "ll_mkdir " << vparent << " " << name
 	  << " = " << r << " (" << hex << attr->st_ino << dec << ")" << dendl;
+  *out = in;
   return r;
 }
 
@@ -7231,25 +7303,30 @@ int Client::_symlink(Inode *dir, const char *name, const char *target, int uid,
   return res;
 }
 
-int Client::ll_symlink(vinodeno_t parent, const char *name, const char *value, struct stat *attr, int uid, int gid)
+int Client::ll_symlink(Inode *parent, const char *name, const char *value,
+		       struct stat *attr, Inode **out, int uid, int gid)
 {
   Mutex::Locker lock(client_lock);
-  ldout(cct, 3) << "ll_symlink " << parent << " " << name << " -> " << value << dendl;
+
+  vinodeno_t vparent = _get_vino(parent);
+
+  ldout(cct, 3) << "ll_symlink " << vparent << " " << name << " -> " << value
+		<< dendl;
   tout(cct) << "ll_symlink" << std::endl;
-  tout(cct) << parent.ino.val << std::endl;
+  tout(cct) << vparent.ino.val << std::endl;
   tout(cct) << name << std::endl;
   tout(cct) << value << std::endl;
 
-  Inode *diri = _ll_get_inode(parent);
-  Inode *in = 0;
-  int r = _symlink(diri, name, value, uid, gid, &in);
+  Inode *in = NULL;
+  int r = _symlink(parent, name, value, uid, gid, &in);
   if (r == 0) {
     fill_stat(in, attr);
     _ll_get(in);
   }
   tout(cct) << attr->st_ino << std::endl;
-  ldout(cct, 3) << "ll_symlink " << parent << " " << name
+  ldout(cct, 3) << "ll_symlink " << vparent << " " << name
 	  << " = " << r << " (" << hex << attr->st_ino << dec << ")" << dendl;
+  *out = in;
   return r;
 }
 
@@ -7303,16 +7380,18 @@ int Client::_unlink(Inode *dir, const char *name, int uid, int gid)
   return res;
 }
 
-int Client::ll_unlink(vinodeno_t vino, const char *name, int uid, int gid)
+int Client::ll_unlink(Inode *in, const char *name, int uid, int gid)
 {
   Mutex::Locker lock(client_lock);
+
+  vinodeno_t vino = _get_vino(in);
+
   ldout(cct, 3) << "ll_unlink " << vino << " " << name << dendl;
   tout(cct) << "ll_unlink" << std::endl;
   tout(cct) << vino.ino.val << std::endl;
   tout(cct) << name << std::endl;
 
-  Inode *diri = _ll_get_inode(vino);
-  return _unlink(diri, name, uid, gid);
+  return _unlink(in, name, uid, gid);
 }
 
 int Client::_rmdir(Inode *dir, const char *name, int uid, int gid)
@@ -7365,16 +7444,18 @@ int Client::_rmdir(Inode *dir, const char *name, int uid, int gid)
   return res;
 }
 
-int Client::ll_rmdir(vinodeno_t vino, const char *name, int uid, int gid)
+int Client::ll_rmdir(Inode *in, const char *name, int uid, int gid)
 {
   Mutex::Locker lock(client_lock);
+
+  vinodeno_t vino = _get_vino(in);
+
   ldout(cct, 3) << "ll_rmdir " << vino << " " << name << dendl;
   tout(cct) << "ll_rmdir" << std::endl;
   tout(cct) << vino.ino.val << std::endl;
   tout(cct) << name << std::endl;
 
-  Inode *diri = _ll_get_inode(vino);
-  return _rmdir(diri, name, uid, gid);
+  return _rmdir(in, name, uid, gid);
 }
 
 int Client::_rename(Inode *fromdir, const char *fromname, Inode *todir, const char *toname, int uid, int gid)
@@ -7448,20 +7529,23 @@ int Client::_rename(Inode *fromdir, const char *fromname, Inode *todir, const ch
   return res;
 }
 
-int Client::ll_rename(vinodeno_t parent, const char *name, vinodeno_t newparent, const char *newname, int uid, int gid)
+int Client::ll_rename(Inode *parent, const char *name, Inode *newparent,
+		      const char *newname, int uid, int gid)
 {
   Mutex::Locker lock(client_lock);
-  ldout(cct, 3) << "ll_rename " << parent << " " << name << " to "
-	  << newparent << " " << newname << dendl;
+
+  vinodeno_t vparent = _get_vino(parent);
+  vinodeno_t vnewparent = _get_vino(newparent);
+
+  ldout(cct, 3) << "ll_rename " << vparent << " " << name << " to "
+	  << vnewparent << " " << newname << dendl;
   tout(cct) << "ll_rename" << std::endl;
-  tout(cct) << parent.ino.val << std::endl;
+  tout(cct) << vparent.ino.val << std::endl;
   tout(cct) << name << std::endl;
-  tout(cct) << newparent.ino.val << std::endl;
+  tout(cct) << vnewparent.ino.val << std::endl;
   tout(cct) << newname << std::endl;
 
-  Inode *fromdiri = _ll_get_inode(parent);
-  Inode *todiri = _ll_get_inode(newparent);
-  return _rename(fromdiri, name, todiri, newname, uid, gid);
+  return _rename(parent, name, newparent, newname, uid, gid);
 }
 
 int Client::_link(Inode *in, Inode *dir, const char *newname, int uid, int gid, Inode **inp)
@@ -7505,83 +7589,76 @@ int Client::_link(Inode *in, Inode *dir, const char *newname, int uid, int gid,
   return res;
 }
 
-int Client::ll_link(vinodeno_t vino, vinodeno_t newparent, const char *newname, struct stat *attr, int uid, int gid)
+int Client::ll_link(Inode *parent, Inode *newparent, const char *newname,
+		    struct stat *attr, int uid, int gid)
 {
   Mutex::Locker lock(client_lock);
-  ldout(cct, 3) << "ll_link " << vino << " to " << newparent << " " << newname << dendl;
+
+  vinodeno_t vparent = _get_vino(parent);
+  vinodeno_t vnewparent = _get_vino(newparent);
+
+  ldout(cct, 3) << "ll_link " << parent << " to " << vnewparent << " " <<
+    newname << dendl;
   tout(cct) << "ll_link" << std::endl;
-  tout(cct) << vino.ino.val << std::endl;
-  tout(cct) << newparent << std::endl;
+  tout(cct) << vparent.ino.val << std::endl;
+  tout(cct) << vnewparent << std::endl;
   tout(cct) << newname << std::endl;
 
-  Inode *old = _ll_get_inode(vino);
-  Inode *diri = _ll_get_inode(newparent);
-
-  int r = _link(old, diri, newname, uid, gid, &old);
+  int r = _link(parent, newparent, newname, uid, gid, &parent);
   if (r == 0) {
-    Inode *in = _ll_get_inode(vino);
-    fill_stat(in, attr);
-    _ll_get(in);
+    fill_stat(parent, attr);
+    _ll_get(parent);
   }
   return r;
 }
 
-int Client::ll_describe_layout(Fh *fh, ceph_file_layout* lp)
+int Client::ll_opendir(Inode *in, dir_result_t** dirpp, int uid, int gid)
 {
   Mutex::Locker lock(client_lock);
-  ldout(cct, 3) << "ll_describe_layout " << fh << " " << fh->inode->ino << dendl;
-  tout(cct) << "ll_describe_layout" << std::endl;
 
-  Inode *in = fh->inode;
-  *lp = in->layout;
+  vinodeno_t vino = _get_vino(in);
 
-  return 0;
-}
-
-int Client::ll_opendir(vinodeno_t vino, void **dirpp, int uid, int gid)
-{
-  Mutex::Locker lock(client_lock);
   ldout(cct, 3) << "ll_opendir " << vino << dendl;
   tout(cct) << "ll_opendir" << std::endl;
   tout(cct) << vino.ino.val << std::endl;
-  
-  Inode *diri = inode_map[vino];
-  assert(diri);
 
   int r = 0;
   if (vino.snapid == CEPH_SNAPDIR) {
-    *dirpp = new dir_result_t(diri);
+    *dirpp = new dir_result_t(in);
   } else {
-    r = _opendir(diri, (dir_result_t**)dirpp);
+    r = _opendir(in, dirpp);
   }
 
   tout(cct) << (unsigned long)*dirpp << std::endl;
 
-  ldout(cct, 3) << "ll_opendir " << vino << " = " << r << " (" << *dirpp << ")" << dendl;
+  ldout(cct, 3) << "ll_opendir " << vino << " = " << r << " (" << *dirpp << ")"
+		<< dendl;
   return r;
 }
 
-void Client::ll_releasedir(void *dirp)
+int Client::ll_releasedir(dir_result_t *dirp)
 {
   Mutex::Locker lock(client_lock);
   ldout(cct, 3) << "ll_releasedir " << dirp << dendl;
   tout(cct) << "ll_releasedir" << std::endl;
   tout(cct) << (unsigned long)dirp << std::endl;
-  _closedir(static_cast<dir_result_t*>(dirp));
+  _closedir(dirp);
+  return 0;
 }
 
-int Client::ll_open(vinodeno_t vino, int flags, Fh **fhp, int uid, int gid)
+int Client::ll_open(Inode *in, int flags, Fh **fhp, int uid, int gid)
 {
   assert(!(flags & O_CREAT));
 
   Mutex::Locker lock(client_lock);
+
+  vinodeno_t vino = _get_vino(in);
+
   ldout(cct, 3) << "ll_open " << vino << " " << flags << dendl;
   tout(cct) << "ll_open" << std::endl;
   tout(cct) << vino.ino.val << std::endl;
   tout(cct) << flags << std::endl;
 
-  Inode *in = _ll_get_inode(vino);
-
   int r;
   if (uid < 0) {
     uid = geteuid();
@@ -7591,41 +7668,47 @@ int Client::ll_open(vinodeno_t vino, int flags, Fh **fhp, int uid, int gid)
   if (r < 0)
     goto out;
 
-  r = _open(in, flags, 0, fhp, uid, gid);
+  r = _open(in, flags, 0, fhp /* may be NULL */, uid, gid);
 
  out:
-  tout(cct) << (unsigned long)*fhp << std::endl;
-  ldout(cct, 3) << "ll_open " << vino << " " << flags << " = " << r << " (" << *fhp << ")" << dendl;
+  Fh *fhptr = fhp ? *fhp : NULL;
+  tout(cct) << (unsigned long)fhptr << std::endl;
+  ldout(cct, 3) << "ll_open " << vino << " " << flags << " = " << r << " (" <<
+    fhptr << ")" << dendl;
   return r;
 }
 
-int Client::ll_create(vinodeno_t parent, const char *name, mode_t mode, int flags,
-		      struct stat *attr, Fh **fhp, int uid, int gid)
+int Client::ll_create(Inode *parent, const char *name, mode_t mode,
+		      int flags, struct stat *attr, Inode **outp, Fh **fhp,
+		      int uid, int gid)
 {
   Mutex::Locker lock(client_lock);
-  ldout(cct, 3) << "ll_create " << parent << " " << name << " 0" << oct << mode << dec << " " << flags << ", uid " << uid << ", gid " << gid << dendl;
+
+  vinodeno_t vparent = _get_vino(parent);
+
+  ldout(cct, 3) << "ll_create " << vparent << " " << name << " 0" << oct <<
+    mode << dec << " " << flags << ", uid " << uid << ", gid " << gid << dendl;
   tout(cct) << "ll_create" << std::endl;
-  tout(cct) << parent.ino.val << std::endl;
+  tout(cct) << vparent.ino.val << std::endl;
   tout(cct) << name << std::endl;
   tout(cct) << mode << std::endl;
   tout(cct) << flags << std::endl;
 
-  *fhp = NULL;
-
   bool created = false;
   Inode *in = NULL;
-  Inode *dir = _ll_get_inode(parent);
-  int r = _lookup(dir, name, &in);
+  int r = _lookup(parent, name, &in);
+
   if (r == 0 && (flags & O_CREAT) && (flags & O_EXCL))
     return -EEXIST;
-  if (r == -ENOENT && (flags & O_CREAT)) {
-    r = _create(dir, name, flags, mode, &in, fhp,
-	        0, 0, 0,
-		NULL, &created, uid, gid);
+
+   if (r == -ENOENT && (flags & O_CREAT)) {
+     r = _create(parent, name, flags, mode, &in, fhp /* may be NULL */,
+	        0, 0, 0, NULL, &created, uid, gid);
     if (r < 0)
       goto out;
 
-    in = (*fhp)->inode;
+    if ((!in) && fhp)
+      in = (*fhp)->inode;
   }
 
   if (r < 0)
@@ -7633,18 +7716,17 @@ int Client::ll_create(vinodeno_t parent, const char *name, mode_t mode, int flag
 
   assert(in);
   fill_stat(in, attr);
-  _ll_get(in);
 
   ldout(cct, 20) << "ll_create created = " << created << dendl;
   if (!created) {
     r = check_permissions(in, flags, uid, gid);
     if (r < 0) {
-      if (*fhp) {
+      if (fhp && *fhp) {
 	_release_fh(*fhp);
       }
       goto out;
     }
-    if (*fhp == NULL) {
+    if (fhp && (*fhp == NULL)) {
       r = _open(in, flags, mode, fhp);
       if (r < 0)
 	goto out;
@@ -7655,13 +7737,32 @@ out:
   if (r < 0)
     attr->st_ino = 0;
 
-  tout(cct) << (unsigned long)*fhp << std::endl;
+  Fh *fhptr = fhp ? *fhp : NULL;
+  tout(cct) << (unsigned long)fhptr << std::endl;
   tout(cct) << attr->st_ino << std::endl;
-  ldout(cct, 3) << "ll_create " << parent << " " << name << " 0" << oct << mode << dec << " " << flags
-	  << " = " << r << " (" << *fhp << " " << hex << attr->st_ino << dec << ")" << dendl;
+  ldout(cct, 3) << "ll_create " << parent << " " << name << " 0" << oct <<
+    mode << dec << " " << flags << " = " << r << " (" << fhptr << " " <<
+    hex << attr->st_ino << dec << ")" << dendl;
+
+  // passing an Inode in outp requires an additional ref
+  if (outp) {
+    _ll_get(in);
+    *outp = in;
+  }
+
   return r;
 }
 
+loff_t Client::ll_lseek(Fh *fh, loff_t offset, int whence)
+{
+  Mutex::Locker lock(client_lock);
+  tout(cct) << "ll_lseek" << std::endl;
+  tout(cct) << offset << std::endl;
+  tout(cct) << whence << std::endl;
+
+  return _lseek(fh, offset, whence);
+}
+
 int Client::ll_read(Fh *fh, loff_t off, loff_t len, bufferlist *bl)
 {
   Mutex::Locker lock(client_lock);
@@ -7828,10 +7929,6 @@ int Client::ll_release(Fh *fh)
 }
 
 
-
-
-
-
 // =========================================
 // layout
 
diff --git a/src/client/Client.h b/src/client/Client.h
index 6bdeffa4b83..b9d83cc2a70 100644
--- a/src/client/Client.h
+++ b/src/client/Client.h
@@ -20,9 +20,11 @@
 
 // stl
 #include <string>
+#include <memory>
 #include <set>
 #include <map>
 #include <fstream>
+#include <exception>
 using std::set;
 using std::map;
 using std::fstream;
@@ -573,6 +575,9 @@ private:
 
   int check_permissions(Inode *in, int flags, int uid, int gid);
 
+  vinodeno_t _get_vino(Inode *in);
+  inodeno_t _get_inodeno(Inode *in);
+
 public:
   int mount(const std::string &mount_root);
   void unmount();
@@ -708,39 +713,81 @@ public:
   int get_caps_issued(int fd);
   int get_caps_issued(const char *path);
 
-  // low-level interface
-  int ll_lookup(vinodeno_t parent, const char *name, struct stat *attr, int uid = -1, int gid = -1);
-  bool ll_forget(vinodeno_t vino, int count);
-  Inode *_ll_get_inode(vinodeno_t vino);
-  int ll_getattr(vinodeno_t vino, struct stat *st, int uid = -1, int gid = -1);
-  int ll_setattr(vinodeno_t vino, struct stat *st, int mask, int uid = -1, int gid = -1);
-  int ll_getxattr(vinodeno_t vino, const char *name, void *value, size_t size, int uid=-1, int gid=-1);
-  int ll_setxattr(vinodeno_t vino, const char *name, const void *value, size_t size, int flags, int uid=-1, int gid=-1);
-  int ll_removexattr(vinodeno_t vino, const char *name, int uid=-1, int gid=-1);
-  int ll_listxattr(vinodeno_t vino, char *list, size_t size, int uid=-1, int gid=-1);
-  int ll_opendir(vinodeno_t vino, void **dirpp, int uid = -1, int gid = -1);
-  void ll_releasedir(void *dirp);
-  int ll_readlink(vinodeno_t vino, const char **value, int uid = -1, int gid = -1);
-  int ll_mknod(vinodeno_t vino, const char *name, mode_t mode, dev_t rdev, struct stat *attr, int uid = -1, int gid = -1);
-  int ll_mkdir(vinodeno_t vino, const char *name, mode_t mode, struct stat *attr, int uid = -1, int gid = -1);
-  int ll_symlink(vinodeno_t vino, const char *name, const char *value, struct stat *attr, int uid = -1, int gid = -1);
-  int ll_unlink(vinodeno_t vino, const char *name, int uid = -1, int gid = -1);
-  int ll_rmdir(vinodeno_t vino, const char *name, int uid = -1, int gid = -1);
-  int ll_rename(vinodeno_t parent, const char *name, vinodeno_t newparent, const char *newname, int uid = -1, int gid = -1);
-  int ll_link(vinodeno_t vino, vinodeno_t newparent, const char *newname, struct stat *attr, int uid = -1, int gid = -1);
-  int ll_describe_layout(Fh *fh, ceph_file_layout* layout);
-  int ll_open(vinodeno_t vino, int flags, Fh **fh, int uid = -1, int gid = -1);
-  int ll_create(vinodeno_t parent, const char *name, mode_t mode, int flags, struct stat *attr, Fh **fh, int uid = -1, int gid = -1);
+  // low-level interface v2
+  inodeno_t ll_get_inodeno(Inode *in) {
+    Mutex::Locker lock(client_lock);
+    return _get_inodeno(in);
+  }
+  snapid_t ll_get_snapid(Inode *in);
+  vinodeno_t ll_get_vino(Inode *in) {
+    Mutex::Locker lock(client_lock);
+    return _get_vino(in);
+  }
+  Inode *ll_get_inode(vinodeno_t vino);
+  int ll_lookup(Inode *parent, const char *name, struct stat *attr,
+		Inode **out, int uid = -1, int gid = -1);
+  bool ll_forget(Inode *in, int count);
+  bool ll_put(Inode *in);
+  int ll_getattr(Inode *in, struct stat *st, int uid = -1, int gid = -1);
+  int ll_setattr(Inode *in, struct stat *st, int mask, int uid = -1,
+		 int gid = -1);
+  int ll_getxattr(Inode *in, const char *name, void *value, size_t size,
+		  int uid=-1, int gid=-1);
+  int ll_setxattr(Inode *in, const char *name, const void *value, size_t size,
+		  int flags, int uid=-1, int gid=-1);
+  int ll_removexattr(Inode *in, const char *name, int uid=-1, int gid=-1);
+  int ll_listxattr(Inode *in, char *list, size_t size, int uid=-1, int gid=-1);
+  int ll_opendir(Inode *in, dir_result_t **dirpp, int uid = -1, int gid = -1);
+  int ll_releasedir(dir_result_t* dirp);
+  int ll_readlink(Inode *in, const char **value, int uid = -1, int gid = -1);
+  int ll_mknod(Inode *in, const char *name, mode_t mode, dev_t rdev,
+	       struct stat *attr, Inode **out, int uid = -1, int gid = -1);
+  int ll_mkdir(Inode *in, const char *name, mode_t mode, struct stat *attr,
+	       Inode **out, int uid = -1, int gid = -1);
+  int ll_symlink(Inode *in, const char *name, const char *value,
+		 struct stat *attr, Inode **out, int uid = -1, int gid = -1);
+  int ll_unlink(Inode *in, const char *name, int uid = -1, int gid = -1);
+  int ll_rmdir(Inode *in, const char *name, int uid = -1, int gid = -1);
+  int ll_rename(Inode *parent, const char *name, Inode *newparent,
+		const char *newname, int uid = -1, int gid = -1);
+  int ll_link(Inode *in, Inode *newparent, const char *newname,
+	      struct stat *attr, int uid = -1, int gid = -1);
+  int ll_open(Inode *in, int flags, Fh **fh, int uid = -1, int gid = -1);
+  int ll_create(Inode *parent, const char *name, mode_t mode, int flags,
+		struct stat *attr, Inode **out, Fh **fhp, int uid = -1,
+		int gid = -1);
+  int ll_read_block(Inode *in, uint64_t blockid, char *buf,  uint64_t offset,
+		    uint64_t length, ceph_file_layout* layout);
+
+  int ll_write_block(Inode *in, uint64_t blockid,
+		     char* buf, uint64_t offset,
+		     uint64_t length, ceph_file_layout* layout,
+		     uint64_t snapseq, uint32_t sync);
+  int ll_commit_blocks(Inode *in, uint64_t offset, uint64_t length);
+
+  int ll_statfs(Inode *in, struct statvfs *stbuf);
+  int ll_walk(const char* name, Inode **i, struct stat *attr); // XXX in?
+  int ll_listxattr_chunks(Inode *in, char *names, size_t size,
+			  int *cookie, int *eol, int uid, int gid);
+  uint32_t ll_stripe_unit(Inode *in);
+  int ll_file_layout(Inode *in, ceph_file_layout *layout);
+  uint64_t ll_snap_seq(Inode *in);
+
   int ll_read(Fh *fh, loff_t off, loff_t len, bufferlist *bl);
   int ll_write(Fh *fh, loff_t off, loff_t len, const char *data);
+  loff_t ll_lseek(Fh *fh, loff_t offset, int whence);
   int ll_flush(Fh *fh);
   int ll_fsync(Fh *fh, bool syncdataonly);
   int ll_fallocate(Fh *fh, int mode, loff_t offset, loff_t length);
   int ll_release(Fh *fh);
-  int ll_statfs(vinodeno_t vino, struct statvfs *stbuf);
+  int ll_get_stripe_osd(struct Inode *in, uint64_t blockno,
+			ceph_file_layout* layout);
+  uint64_t ll_get_internal_offset(struct Inode *in, uint64_t blockno);
 
+  int ll_num_osds(void);
+  int ll_osdaddr(int osd, uint32_t *addr);
+  int ll_osdaddr(int osd, char* buf, size_t size);
   void ll_register_ino_invalidate_cb(client_ino_callback_t cb, void *handle);
-
   void ll_register_getgroups_cb(client_getgroups_callback_t cb, void *handle);
 };