Merge remote branch 'origin/wip-librbd-locking'

Conflicts:
	qa/workunits/rbd/copy.sh

Reviewed-by: Sage Weil <sage.weil@inktank.com>

4 files changed, 149 insertions, 120 deletions

diff --git a/src/cls/lock/cls_lock.cc b/src/cls/lock/cls_lock.cc
index 77b6ef958ec..b8641789519 100644
--- a/src/cls/lock/cls_lock.cc
+++ b/src/cls/lock/cls_lock.cc
@@ -355,7 +355,7 @@ static int break_lock(cls_method_context_t hctx,
 }
 
 
- /**
+/**
  * Retrieve lock info: lockers, tag, exclusive
  *
  * Input:
@@ -400,7 +400,7 @@ static int get_info(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 }
 
 
- /**
+/**
  * Retrieve a list of locks for this object
  *
  * Input:
@@ -410,7 +410,8 @@ static int get_info(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
  * @param out contains encoded cls_list_locks_reply
  *
  * @return 0 on success, -errno on failure.
- */static int list_locks(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
+ */
+static int list_locks(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 {
   CLS_LOG(20, "list_locks");
 
diff --git a/src/cls/lock/cls_lock_client.cc b/src/cls/lock/cls_lock_client.cc
index 8d47a841313..54af41cd049 100644
--- a/src/cls/lock/cls_lock_client.cc
+++ b/src/cls/lock/cls_lock_client.cc
@@ -32,11 +32,11 @@ namespace rados {
   namespace cls {
     namespace lock {
 
-      void lock(ObjectWriteOperation& rados_op,
-                string& name, ClsLockType type,
-                string& cookie, string& tag,
-                string description,
-                utime_t& duration, uint8_t flags)
+      void lock(ObjectWriteOperation *rados_op,
+                const string& name, ClsLockType type,
+                const string& cookie, const string& tag,
+                const string& description,
+                const utime_t& duration, uint8_t flags)
       {
         cls_lock_lock_op op;
         op.name = name;
@@ -48,22 +48,23 @@ namespace rados {
         op.flags = flags;
         bufferlist in;
         ::encode(op, in);
-        rados_op.exec("lock", "lock", in);
+        rados_op->exec("lock", "lock", in);
       }
 
-      int lock(IoCtx& ioctx,
-               string& oid,
-               string& name, ClsLockType type,
-               string& cookie, string& tag,
-               string description, utime_t& duration, uint8_t flags)
+      int lock(IoCtx *ioctx,
+               const string& oid,
+               const string& name, ClsLockType type,
+               const string& cookie, const string& tag,
+               const string& description, const utime_t& duration,
+	       uint8_t flags)
       {
         ObjectWriteOperation op;
-        lock(op, name, type, cookie, tag, description, duration, flags);
-        return ioctx.operate(oid, &op);
+        lock(&op, name, type, cookie, tag, description, duration, flags);
+        return ioctx->operate(oid, &op);
       }
 
-      void unlock(ObjectWriteOperation& rados_op,
-                  string& name, string& cookie)
+      void unlock(ObjectWriteOperation *rados_op,
+                  const string& name, const string& cookie)
       {
         cls_lock_unlock_op op;
         op.name = name;
@@ -71,20 +72,20 @@ namespace rados {
         bufferlist in;
         ::encode(op, in);
 
-        rados_op.exec("lock", "unlock", in);
+        rados_op->exec("lock", "unlock", in);
       }
 
-      int unlock(IoCtx& ioctx, string& oid,
-                 string& name, string& cookie)
+      int unlock(IoCtx *ioctx, const string& oid,
+                 const string& name, const string& cookie)
       {
         ObjectWriteOperation op;
-        unlock(op, name, cookie);
-        return ioctx.operate(oid, &op);
+        unlock(&op, name, cookie);
+        return ioctx->operate(oid, &op);
       }
 
-      void break_lock(ObjectWriteOperation& rados_op,
-                      string& name, string& cookie,
-                      entity_name_t& locker)
+      void break_lock(ObjectWriteOperation *rados_op,
+                      const string& name, const string& cookie,
+                      const entity_name_t& locker)
       {
         cls_lock_break_op op;
         op.name = name;
@@ -92,22 +93,22 @@ namespace rados {
         op.locker = locker;
         bufferlist in;
         ::encode(op, in);
-        rados_op.exec("lock", "break_lock", in);
+        rados_op->exec("lock", "break_lock", in);
       }
 
-      int break_lock(IoCtx& ioctx, string& oid,
-                     string& name, string& cookie,
-                     entity_name_t& locker)
+      int break_lock(IoCtx *ioctx, const string& oid,
+                     const string& name, const string& cookie,
+                     const entity_name_t& locker)
       {
         ObjectWriteOperation op;
-        break_lock(op, name, cookie, locker);
-        return ioctx.operate(oid, &op);
+        break_lock(&op, name, cookie, locker);
+        return ioctx->operate(oid, &op);
       }
 
-      int list_locks(IoCtx& ioctx, string& oid, list<string> *locks)
+      int list_locks(IoCtx *ioctx, const string& oid, list<string> *locks)
       {
         bufferlist in, out;
-        int r = ioctx.exec(oid, "lock", "list_locks", in, out);
+        int r = ioctx->exec(oid, "lock", "list_locks", in, out);
         if (r < 0)
           return r;
 
@@ -116,7 +117,7 @@ namespace rados {
         try {
           ::decode(ret, iter);
         } catch (buffer::error& err) {
-          cerr << __func__ << ": failed to decode response" << std::endl;
+	  return -EBADMSG;
         }
 
         *locks = ret.locks;
@@ -124,33 +125,33 @@ namespace rados {
         return 0;
       }
 
-      int get_lock_info(IoCtx& ioctx, string& oid, string& lock,
-                        map<locker_id_t, locker_info_t> *lockers,
-                        ClsLockType *lock_type,
-                        string *tag)
+      void get_lock_info_start(ObjectReadOperation *rados_op,
+			       const string& name)
       {
-        bufferlist in, out;
+        bufferlist in;
         cls_lock_get_info_op op;
-        op.name = lock;
+        op.name = name;
         ::encode(op, in);
-        int r = ioctx.exec(oid, "lock", "get_info", in, out);
-        if (r < 0)
-          return r;
+        rados_op->exec("lock", "get_info", in);
+      }
 
+      int get_lock_info_finish(bufferlist::iterator *iter,
+			       map<locker_id_t, locker_info_t> *lockers,
+			       ClsLockType *type, string *tag)
+      {
         cls_lock_get_info_reply ret;
-        bufferlist::iterator iter = out.begin();
         try {
-          ::decode(ret, iter);
+          ::decode(ret, *iter);
         } catch (buffer::error& err) {
-          cerr << __func__ << ": failed to decode response" << std::endl;
+	  return -EBADMSG;
         }
 
         if (lockers) {
           *lockers = ret.lockers;
         }
 
-        if (lock_type) {
-          *lock_type = ret.lock_type;
+        if (type) {
+          *type = ret.lock_type;
         }
 
         if (tag) {
@@ -160,46 +161,60 @@ namespace rados {
         return 0;
       }
 
-      void Lock::lock_shared(ObjectWriteOperation& op)
+      int get_lock_info(IoCtx *ioctx, const string& oid, const string& name,
+                        map<locker_id_t, locker_info_t> *lockers,
+                        ClsLockType *type, string *tag)
+      {
+        ObjectReadOperation op;
+        get_lock_info_start(&op, name);
+	bufferlist out;
+        int r = ioctx->operate(oid, &op, &out);
+	if (r < 0)
+	  return r;
+	bufferlist::iterator it = out.begin();
+	return get_lock_info_finish(&it, lockers, type, tag);
+      }
+
+      void Lock::lock_shared(ObjectWriteOperation *op)
       {
         lock(op, name, LOCK_SHARED,
              cookie, tag, description, duration, flags);
       }
 
-      int Lock::lock_shared(IoCtx& ioctx, string& oid)
+      int Lock::lock_shared(IoCtx *ioctx, const string& oid)
       {
         return lock(ioctx, oid, name, LOCK_SHARED,
                     cookie, tag, description, duration, flags);
       }
 
-      void Lock::lock_exclusive(ObjectWriteOperation& op)
+      void Lock::lock_exclusive(ObjectWriteOperation *op)
       {
         lock(op, name, LOCK_EXCLUSIVE,
              cookie, tag, description, duration, flags);
       }
 
-      int Lock::lock_exclusive(IoCtx& ioctx, string& oid)
+      int Lock::lock_exclusive(IoCtx *ioctx, const string& oid)
       {
         return lock(ioctx, oid, name, LOCK_EXCLUSIVE,
                     cookie, tag, description, duration, flags);
       }
 
-      void Lock::unlock(ObjectWriteOperation& op)
+      void Lock::unlock(ObjectWriteOperation *op)
       {
 	rados::cls::lock::unlock(op, name, cookie);
       }
 
-      int Lock::unlock(IoCtx& ioctx, string& oid)
+      int Lock::unlock(IoCtx *ioctx, const string& oid)
       {
         return rados::cls::lock::unlock(ioctx, oid, name, cookie);
       }
 
-      void Lock::break_lock(ObjectWriteOperation& op, entity_name_t& locker)
+      void Lock::break_lock(ObjectWriteOperation *op, const entity_name_t& locker)
       {
 	rados::cls::lock::break_lock(op, name, cookie, locker);
       }
 
-      int Lock::break_lock(IoCtx& ioctx, string& oid, entity_name_t& locker)
+      int Lock::break_lock(IoCtx *ioctx, const string& oid, const entity_name_t& locker)
       {
           return rados::cls::lock::break_lock(ioctx, oid, name, cookie, locker);
       }
diff --git a/src/cls/lock/cls_lock_client.h b/src/cls/lock/cls_lock_client.h
index 4cb08e86920..4e2144c79b4 100644
--- a/src/cls/lock/cls_lock_client.h
+++ b/src/cls/lock/cls_lock_client.h
@@ -1,3 +1,6 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
 #ifndef CEPH_CLS_LOCK_CLIENT_H
 #define CEPH_CLS_LOCK_CLIENT_H
 
@@ -12,72 +15,82 @@ namespace rados {
   namespace cls {
     namespace lock {
 
-      extern void lock(librados::ObjectWriteOperation& rados_op,
-                       std::string& name, ClsLockType type,
-                       std::string& cookie, std::string& tag,
-                       std::string description, utime_t& duration, uint8_t flags);
-
-      extern int lock(librados::IoCtx& ioctx,
-                      std::string& oid,
-                      std::string& name, ClsLockType type,
-                      std::string& cookie, std::string& tag,
-                      std::string description, utime_t& duration, uint8_t flags);
-
-      extern void unlock(librados::ObjectWriteOperation& rados_op,
-                         std::string& name, std::string& cookie);
-
-      extern int unlock(librados::IoCtx& ioctx, std::string& oid,
-                        std::string& name, std::string& cookie);
-
-      extern void break_lock(librados::ObjectWriteOperation& op,
-                             std::string& name, std::string& cookie,
-                             entity_name_t& locker);
-
-      extern int break_lock(librados::IoCtx& ioctx, std::string& oid,
-                            std::string& name, std::string& cookie,
-                            entity_name_t& locker);
-
-      extern int list_locks(librados::IoCtx& ioctx, std::string& oid, list<std::string> *locks);
-      extern int get_lock_info(librados::IoCtx& ioctx, std::string& oid, std::string& lock,
-                               map<locker_id_t, locker_info_t> *lockers,
-                               ClsLockType *lock_type,
-                               std::string *tag);
+      extern void lock(librados::ObjectWriteOperation *rados_op,
+		       const std::string& name, ClsLockType type,
+		       const std::string& cookie, const std::string& tag,
+		       const std::string& description, const utime_t& duration,
+		       uint8_t flags);
+
+      extern int lock(librados::IoCtx *ioctx,
+		      const std::string& oid,
+		      const std::string& name, ClsLockType type,
+		      const std::string& cookie, const std::string& tag,
+		      const std::string& description, const utime_t& duration,
+		      uint8_t flags);
+
+      extern void unlock(librados::ObjectWriteOperation *rados_op,
+			 const std::string& name, const std::string& cookie);
+
+      extern int unlock(librados::IoCtx *ioctx, const std::string& oid,
+			const std::string& name, const std::string& cookie);
+
+      extern void break_lock(librados::ObjectWriteOperation *op,
+			     const std::string& name, const std::string& cookie,
+			     const entity_name_t& locker);
+
+      extern int break_lock(librados::IoCtx *ioctx, const std::string& oid,
+			    const std::string& name, const std::string& cookie,
+			    const entity_name_t& locker);
+
+      extern int list_locks(librados::IoCtx *ioctx, const std::string& oid,
+			    list<std::string> *locks);
+      extern void get_lock_info_start(librados::ObjectReadOperation *rados_op,
+				      const std::string& name);
+      extern int get_lock_info_finish(ceph::bufferlist::iterator *out,
+				      map<locker_id_t, locker_info_t> *lockers,
+				      ClsLockType *type, std::string *tag);
+
+      extern int get_lock_info(librados::IoCtx *ioctx, const std::string& oid,
+			       const std::string& name,
+			       map<locker_id_t, locker_info_t> *lockers,
+			       ClsLockType *type, std::string *tag);
 
       class Lock {
-        std::string name;
-        std::string cookie;
-        std::string tag;
-        std::string description;
-        utime_t duration;
-        uint8_t flags;
+	std::string name;
+	std::string cookie;
+	std::string tag;
+	std::string description;
+	utime_t duration;
+	uint8_t flags;
 
       public:
 
-        Lock(const std::string& _n) : name(_n), flags(0) {}
-
-        void set_cookie(const std::string& c) { cookie = c; }
-        void set_tag(const std::string& t) { tag = t; }
-        void set_description(const std::string& desc) { description = desc; }
-        void set_duration(const utime_t& e) { duration = e; }
-        void set_renew(bool renew) {
-          if (renew) {
-            flags |= LOCK_FLAG_RENEW;
-          } else {
-            flags &= ~LOCK_FLAG_RENEW;
-          }
-        }
-
-        /* ObjectWriteOperation */
-        void lock_exclusive(librados::ObjectWriteOperation& ioctx);
-        void lock_shared(librados::ObjectWriteOperation& ioctx);
-        void unlock(librados::ObjectWriteOperation& ioctx);
-        void break_lock(librados::ObjectWriteOperation& ioctx, entity_name_t& locker);
-
-        /* IoCtx*/
-        int lock_exclusive(librados::IoCtx& ioctx, std::string& oid);
-        int lock_shared(librados::IoCtx& ioctx, std::string& oid);
-        int unlock(librados::IoCtx& ioctx, std::string& oid);
-        int break_lock(librados::IoCtx& ioctx, std::string& oid, entity_name_t& locker);
+	Lock(const std::string& _n) : name(_n), flags(0) {}
+
+	void set_cookie(const std::string& c) { cookie = c; }
+	void set_tag(const std::string& t) { tag = t; }
+	void set_description(const std::string& desc) { description = desc; }
+	void set_duration(const utime_t& e) { duration = e; }
+	void set_renew(bool renew) {
+	  if (renew) {
+	    flags |= LOCK_FLAG_RENEW;
+	  } else {
+	    flags &= ~LOCK_FLAG_RENEW;
+	  }
+	}
+
+	/* ObjectWriteOperation */
+	void lock_exclusive(librados::ObjectWriteOperation *ioctx);
+	void lock_shared(librados::ObjectWriteOperation *ioctx);
+	void unlock(librados::ObjectWriteOperation *ioctx);
+	void break_lock(librados::ObjectWriteOperation *ioctx, const entity_name_t& locker);
+
+	/* IoCtx */
+	int lock_exclusive(librados::IoCtx *ioctx, const std::string& oid);
+	int lock_shared(librados::IoCtx *ioctx, const std::string& oid);
+	int unlock(librados::IoCtx *ioctx, const std::string& oid);
+	int break_lock(librados::IoCtx *ioctx, const std::string& oid,
+		       const entity_name_t& locker);
       };
 
     } // namespace lock
@@ -85,4 +98,3 @@ namespace rados {
 } // namespace rados
 
 #endif
-
diff --git a/src/cls/lock/cls_lock_types.h b/src/cls/lock/cls_lock_types.h
index b68ec0a8098..aa60ed142c7 100644
--- a/src/cls/lock/cls_lock_types.h
+++ b/src/cls/lock/cls_lock_types.h
@@ -1,9 +1,10 @@
 #ifndef CEPH_CLS_LOCK_TYPES_H
 #define CEPH_CLS_LOCK_TYPES_H
 
-#include "include/types.h"
 #include "include/encoding.h"
+#include "include/types.h"
 #include "include/utime.h"
+#include "msg/msg_types.h"
 
 /* lock flags */
 #define LOCK_FLAG_RENEW 0x1        /* idempotent lock acquire */