librados, objecter: add operations to support rgw atomic get

diff --git a/src/include/rados/librados.h b/src/include/rados/librados.h
index f3c52161afe6872ce137a60647d75f7527236a8e..cf2a9426d02ab0e0164471200a0d90a65b74a3a8 100644 (file)
--- a/src/include/rados/librados.h
+++ b/src/include/rados/librados.h
@@ -26,6 +26,17 @@ extern "C" {
 
 #define LIBRADOS_SUPPORTS_WATCH 1
 
+/* xattr comparison */
+enum {
+       LIBRADOS_CMPXATTR_OP_NOP = 0,
+       LIBRADOS_CMPXATTR_OP_EQ  = 1,
+       LIBRADOS_CMPXATTR_OP_NE  = 2,
+       LIBRADOS_CMPXATTR_OP_GT  = 3,
+       LIBRADOS_CMPXATTR_OP_GTE = 4,
+       LIBRADOS_CMPXATTR_OP_LT  = 5,
+       LIBRADOS_CMPXATTR_OP_LTE = 6
+};
+
 struct CephContext;
 
 typedef void *rados_t;

diff --git a/src/include/rados/librados.hpp b/src/include/rados/librados.hpp
index 6fd78ef68a6c7ae55cdbe2deacbd7d5bbee07de4..3fab292fe6d6e6f1609b6db5036cbcb399fae3e5 100644 (file)
--- a/src/include/rados/librados.hpp
+++ b/src/include/rados/librados.hpp
@@ -121,7 +121,8 @@ namespace librados
     void zero(uint64_t off, uint64_t len);
     void rmxattr(const char *name);
     void setxattr(const char *name, const bufferlist& bl);
-    void cmpxattr(const char *name, const bufferlist& bl);
+    void cmpxattr(const char *name, uint8_t op, const bufferlist& bl);
+    void cmpxattr(const char *name, uint8_t op, uint64_t v);
     void tmap_update(const bufferlist& cmdbl);
     void clone_range(uint64_t dst_off,
                      const std::string& src_oid, uint64_t src_off,
@@ -139,7 +140,28 @@ namespace librados
     friend class Rados;
   };
 
-  /* IoCtx : This is a context in which we can perform I/O.
+  /*
+   * ObjectReadOperation : compount object operation that return value
+   * Batch multiple object operations into a single request, to be applied
+   * atomically.
+   */
+  class ObjectReadOperation
+  {
+  public:
+    ObjectReadOperation();
+    ~ObjectReadOperation();
+
+    void stat();
+    void getxattr(const char *name);
+    void getxattrs();
+
+  private:
+    ObjectOperationImpl *impl;
+    ObjectReadOperation(const ObjectReadOperation& rhs);
+    ObjectReadOperation& operator=(const ObjectReadOperation& rhs);
+    friend class IoCtx;
+    friend class Rados;
+  };  /* IoCtx : This is a context in which we can perform I/O.
    * It includes a Pool,
    *
    * Typical use (error checking omitted):
@@ -245,6 +267,7 @@ namespace librados
 
     // compound object operations
     int operate(const std::string& oid, ObjectOperation *op, bufferlist *pbl);
+    int operate(const std::string& oid, ObjectReadOperation *op, bufferlist *pbl);
     int aio_operate(const std::string& oid, AioCompletion *c, ObjectOperation *op, bufferlist *pbl);
 
     // watch/notify

diff --git a/src/librados.cc b/src/librados.cc
index 5986544f9eab1a8e20111ab07ea7e7ebf1055f1d..edec39c80e241aef4bae13daa4bd1fff83b7c021 100644 (file)
--- a/src/librados.cc
+++ b/src/librados.cc
@@ -132,6 +132,23 @@ struct librados::IoCtxImpl {
   }
 };
 
+void librados::ObjectReadOperation::stat()
+{
+  ::ObjectOperation *o = (::ObjectOperation *)impl;
+  o->add_op(CEPH_OSD_OP_STAT);
+}
+
+void librados::ObjectReadOperation::getxattr(const char *name)
+{
+  ::ObjectOperation *o = (::ObjectOperation *)impl;
+  o->getxattr(name);
+}
+
+void librados::ObjectReadOperation::getxattrs()
+{
+  ::ObjectOperation *o = (::ObjectOperation *)impl;
+  o->getxattrs();
+}
 
 void librados::ObjectOperation::create(bool exclusive)
 {
@@ -190,10 +207,18 @@ void librados::ObjectOperation::setxattr(const char *name, const bufferlist& v)
   o->setxattr(name, v);
 }
 
-void librados::ObjectOperation::cmpxattr(const char *name, const bufferlist& v)
+void librados::ObjectOperation::cmpxattr(const char *name, uint8_t op, const bufferlist& v)
+{
+  ::ObjectOperation *o = (::ObjectOperation *)impl;
+  o->cmpxattr(name, op, CEPH_OSD_CMPXATTR_MODE_STRING, v);
+}
+
+void librados::ObjectOperation::cmpxattr(const char *name, uint8_t op, uint64_t v)
 {
   ::ObjectOperation *o = (::ObjectOperation *)impl;
-  o->cmpxattr(name, v);
+  bufferlist bl;
+  ::encode(v, bl);
+  o->cmpxattr(name, op, CEPH_OSD_CMPXATTR_MODE_U64, bl);
 }
 
 void librados::ObjectOperation::tmap_update(const bufferlist& cmdbl)
@@ -553,6 +578,7 @@ public:
   int list(Objecter::ListContext *context, int max_entries);
 
   int operate(IoCtxImpl& io, const object_t& oid, ::ObjectOperation *o, bufferlist *pbl);
+  int operate_read(IoCtxImpl& io, const object_t& oid, ::ObjectOperation *o, bufferlist *pbl);
   int aio_operate(IoCtxImpl& io, const object_t& oid, ::ObjectOperation *o, AioCompletionImpl *c, bufferlist *pbl);
 
   struct C_aio_Ack : public Context {
@@ -1642,6 +1668,39 @@ operate(IoCtxImpl& io, const object_t& oid, ::ObjectOperation *o, bufferlist *pb
   return r;
 }
 
+int librados::RadosClient::
+operate_read(IoCtxImpl& io, const object_t& oid, ::ObjectOperation *o, bufferlist *pbl)
+{
+  utime_t ut = ceph_clock_now(cct);
+
+  /* can't write to a snapshot */
+  if (io.snap_seq != CEPH_NOSNAP)
+    return -EINVAL;
+
+  Mutex mylock("RadosClient::mutate::mylock");
+  Cond cond;
+  bool done;
+  int r;
+  eversion_t ver;
+
+  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
+
+  lock.Lock();
+  objecter->read(oid, io.oloc,
+                  *o, io.snap_seq, pbl, 0,
+                  onack, &ver);
+  lock.Unlock();
+
+  mylock.Lock();
+  while (!done)
+    cond.Wait(mylock);
+  mylock.Unlock();
+
+  set_sync_op_version(io, ver);
+
+  return r;
+}
+
 int librados::RadosClient::
 aio_operate(IoCtxImpl& io, const object_t& oid, ::ObjectOperation *o, AioCompletionImpl *c,
            bufferlist *pbl)
@@ -2747,6 +2806,12 @@ int librados::IoCtx::operate(const std::string& oid, librados::ObjectOperation *
   return io_ctx_impl->client->operate(*io_ctx_impl, obj, (::ObjectOperation*)o->impl, pbl);
 }
 
+int librados::IoCtx::operate(const std::string& oid, librados::ObjectReadOperation *o, bufferlist *pbl)
+{
+  object_t obj(oid);
+  return io_ctx_impl->client->operate_read(*io_ctx_impl, obj, (::ObjectOperation*)o->impl, pbl);
+}
+
 int librados::IoCtx::aio_operate(const std::string& oid, AioCompletion *c, librados::ObjectOperation *o, bufferlist *pbl)
 {
   object_t obj(oid);
@@ -3195,6 +3260,17 @@ librados::ObjectOperation::~ObjectOperation()
   delete o;
 }
 
+librados::ObjectReadOperation::ObjectReadOperation()
+{
+  impl = (ObjectOperationImpl *)new ::ObjectOperation;
+}
+
+librados::ObjectReadOperation::~ObjectReadOperation()
+{
+  ::ObjectOperation *o = (::ObjectOperation *)impl;
+  delete o;
+}
+
 ///////////////////////////// C API //////////////////////////////
 extern "C" int rados_create(rados_t *pcluster, const char * const id)
 {

diff --git a/src/osdc/Objecter.h b/src/osdc/Objecter.h
index 873ef68b74c4c5eee4d4063828b090b3877bbebc..bc28e72280565772116bc79e6dc8464949ed169a 100644 (file)
--- a/src/osdc/Objecter.h
+++ b/src/osdc/Objecter.h
@@ -84,6 +84,18 @@ struct ObjectOperation {
       ops[s].data.append(name);
     ops[s].data.append(data);
   }
+  void add_xattr_cmp(int op, const char *name, uint8_t cmp_op, uint8_t cmp_mode, const bufferlist& data) {
+    int s = ops.size();
+    ops.resize(s+1);
+    ops[s].op.op = op;
+    ops[s].op.xattr.name_len = (name ? strlen(name) : 0);
+    ops[s].op.xattr.value_len = data.length();
+    ops[s].op.xattr.cmp_op = cmp_op;
+    ops[s].op.xattr.cmp_mode = cmp_mode;
+    if (name)
+      ops[s].data.append(name);
+    ops[s].data.append(data);
+  }
   void add_call(int op, const char *cname, const char *method, bufferlist &indata) {
     int s = ops.size();
     ops.resize(s+1);
@@ -198,8 +210,8 @@ struct ObjectOperation {
     bl.append(s);
     add_xattr(CEPH_OSD_OP_SETXATTR, name, bl);
   }
-  void cmpxattr(const char *name, const bufferlist& bl) {
-    add_xattr(CEPH_OSD_OP_CMPXATTR, name, bl);
+  void cmpxattr(const char *name, uint8_t cmp_op, uint8_t cmp_mode, const bufferlist& bl) {
+    add_xattr_cmp(CEPH_OSD_OP_CMPXATTR, name, cmp_op, cmp_mode, bl);
   }
   void rmxattr(const char *name) {
     bufferlist bl;