librados, osdc: Refactor IoCtxImpl to use operate()/operate_read()

Add ObjectOperation::write() that includes len instead of using bufferlist length
Have selfmanaged_snap_rollback_object() use mutate()

Signed-off-by: David Zafman <david.zafman@inktank.com>

diff --git a/src/librados/IoCtxImpl.cc b/src/librados/IoCtxImpl.cc
index 64c480c1a5574450371b1cf43cbc3767e7170c13..d2ef7cd217de44579340a689dfb43ca07581b719 100644 (file)
--- a/src/librados/IoCtxImpl.cc
+++ b/src/librados/IoCtxImpl.cc
@@ -210,16 +210,22 @@ int librados::IoCtxImpl::selfmanaged_snap_rollback_object(const object_t& oid,
                                                          ::SnapContext& snapc,
                                                          uint64_t snapid)
 {
+  utime_t ut = ceph_clock_now(client->cct);
   int reply;
+  eversion_t ver;
 
   Mutex mylock("IoCtxImpl::snap_rollback::mylock");
   Cond cond;
   bool done;
   Context *onack = new C_SafeCond(&mylock, &cond, &done, &reply);
 
+  ::ObjectOperation op;
+  prepare_assert_ops(&op);
+  op.rollback(snapid);
   lock->Lock();
-  objecter->rollback_object(oid, oloc, snapc, snapid,
-                           ceph_clock_now(client->cct), onack, NULL);
+  objecter->mutate(oid, oloc,
+                  op, snapc, ut, 0,
+                  onack, NULL, &ver);
   lock->Unlock();
 
   mylock.Lock();
@@ -379,68 +385,19 @@ int librados::IoCtxImpl::list(Objecter::ListContext *context, int max_entries)
 
 int librados::IoCtxImpl::create(const object_t& oid, bool exclusive)
 {
-  utime_t ut = ceph_clock_now(client->cct);
-
-  /* can't write to a snapshot */
-  if (snap_seq != CEPH_NOSNAP)
-    return -EROFS;
-
-  Mutex mylock("IoCtxImpl::create::mylock");
-  Cond cond;
-  bool done;
-  int r;
-  eversion_t ver;
-
-  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
-
-  lock->Lock();
-  objecter->create(oid, oloc,
-                 snapc, ut, 0, (exclusive ? CEPH_OSD_OP_FLAG_EXCL : 0),
-                 onack, NULL, &ver);
-  lock->Unlock();
-
-  mylock.Lock();
-  while (!done)
-    cond.Wait(mylock);
-  mylock.Unlock();
-
-  set_sync_op_version(ver);
-
-  return r;
+  ::ObjectOperation op;
+  prepare_assert_ops(&op);
+  op.create(exclusive);
+  return operate(oid, &op, NULL);
 }
 
 int librados::IoCtxImpl::create(const object_t& oid, bool exclusive,
                                const std::string& category)
 {
-  utime_t ut = ceph_clock_now(client->cct);
-
-  /* can't write to a snapshot */
-  if (snap_seq != CEPH_NOSNAP)
-    return -EROFS;
-
-  Mutex mylock("IoCtxImpl::create::mylock");
-  Cond cond;
-  bool done;
-  int r;
-  eversion_t ver;
-
-  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
-
-  ::ObjectOperation o;
-  o.create(exclusive ? CEPH_OSD_OP_FLAG_EXCL : 0, category);
-
-  lock->Lock();
-  objecter->mutate(oid, oloc, o, snapc, ut, 0, onack, NULL, &ver);
-  lock->Unlock();
-
-  mylock.Lock();
-  while (!done)
-    cond.Wait(mylock);
-  mylock.Unlock();
-
-  set_sync_op_version(ver);
-
-  return r;
+  ::ObjectOperation op;
+  prepare_assert_ops(&op);
+  op.create(exclusive, category);
+  return operate(oid, &op, NULL);
 }
 
 /*
@@ -473,37 +430,12 @@ int librados::IoCtxImpl::create(const object_t& oid, bool exclusive,
 int librados::IoCtxImpl::write(const object_t& oid, bufferlist& bl,
                               size_t len, uint64_t off)
 {
-  utime_t ut = ceph_clock_now(client->cct);
-
-  /* can't write to a snapshot */
-  if (snap_seq != CEPH_NOSNAP)
-    return -EROFS;
-
-  Mutex mylock("IoCtxImpl::write::mylock");
-  Cond cond;
-  bool done;
-  int r;
-
-  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
-  eversion_t ver;
-
-  // extra ops?
   ::ObjectOperation op;
-  ::ObjectOperation *pop = prepare_assert_ops(&op);
-
-  lock->Lock();
-  objecter->write(oid, oloc,
-                 off, len, snapc, bl, ut, 0,
-                 onack, NULL, &ver, pop);
-  lock->Unlock();
-
-  mylock.Lock();
-  while (!done)
-    cond.Wait(mylock);
-  mylock.Unlock();
-
-  set_sync_op_version(ver);
-
+  prepare_assert_ops(&op);
+  bufferlist mybl;
+  mybl.substr_of(bl, 0, len);
+  op.write(off, mybl);
+  int r =  operate(oid, &op, NULL);
   if (r < 0)
     return r;
 
@@ -512,36 +444,12 @@ int librados::IoCtxImpl::write(const object_t& oid, bufferlist& bl,
 
 int librados::IoCtxImpl::append(const object_t& oid, bufferlist& bl, size_t len)
 {
-  utime_t ut = ceph_clock_now(client->cct);
-
-  /* can't write to a snapshot */
-  if (snap_seq != CEPH_NOSNAP)
-    return -EROFS;
-
-  Mutex mylock("IoCtxImpl::append::mylock");
-  Cond cond;
-  bool done;
-  int r;
-
-  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
-  eversion_t ver;
-
   ::ObjectOperation op;
-  ::ObjectOperation *pop = prepare_assert_ops(&op);
-
-  lock->Lock();
-  objecter->append(oid, oloc,
-                  len, snapc, bl, ut, 0,
-                  onack, NULL, &ver, pop);
-  lock->Unlock();
-
-  mylock.Lock();
-  while (!done)
-    cond.Wait(mylock);
-  mylock.Unlock();
-
-  set_sync_op_version(ver);
-
+  prepare_assert_ops(&op);
+  bufferlist mybl;
+  mybl.substr_of(bl, 0, len);
+  op.append(mybl);
+  int r = operate(oid, &op, NULL);
   if (r < 0)
     return r;
 
@@ -550,38 +458,10 @@ int librados::IoCtxImpl::append(const object_t& oid, bufferlist& bl, size_t len)
 
 int librados::IoCtxImpl::write_full(const object_t& oid, bufferlist& bl)
 {
-  utime_t ut = ceph_clock_now(client->cct);
-
-  /* can't write to a snapshot */
-  if (snap_seq != CEPH_NOSNAP)
-    return -EROFS;
-
-  Mutex mylock("IoCtxImpl::write_full::mylock");
-  Cond cond;
-  bool done;
-  int r;
-
-  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
-
-  eversion_t ver;
-
   ::ObjectOperation op;
-  ::ObjectOperation *pop = prepare_assert_ops(&op);
-
-  lock->Lock();
-  objecter->write_full(oid, oloc,
-                      snapc, bl, ut, 0,
-                      onack, NULL, &ver, pop);
-  lock->Unlock();
-
-  mylock.Lock();
-  while (!done)
-    cond.Wait(mylock);
-  mylock.Unlock();
-
-  set_sync_op_version(ver);
-
-  return r;
+  prepare_assert_ops(&op);
+  op.write_full(bl);
+  return operate(oid, &op, NULL);
 }
 
 int librados::IoCtxImpl::clone_range(const object_t& dst_oid,
@@ -590,36 +470,10 @@ int librados::IoCtxImpl::clone_range(const object_t& dst_oid,
                                     uint64_t src_offset,
                                     uint64_t len)
 {
-  utime_t ut = ceph_clock_now(client->cct);
-
-  /* can't write to a snapshot */
-  if (snap_seq != CEPH_NOSNAP)
-    return -EROFS;
-
-  Mutex mylock("IoCtxImpl::clone_range::mylock");
-  Cond cond;
-  bool done;
-  int r;
-  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
-  eversion_t ver;
-
-  bufferlist outbl;
-
-  lock->Lock();
   ::ObjectOperation wr;
   prepare_assert_ops(&wr);
   wr.clone_range(src_oid, src_offset, len, dst_offset);
-  objecter->mutate(dst_oid, oloc, wr, snapc, ut, 0, onack, NULL, &ver);
-  lock->Unlock();
-
-  mylock.Lock();
-  while (!done)
-    cond.Wait(mylock);
-  mylock.Unlock();
-
-  set_sync_op_version(ver);
-
-  return r;
+  return operate(dst_oid, &wr, NULL);
 }
 
 int librados::IoCtxImpl::operate(const object_t& oid, ::ObjectOperation *o,
@@ -924,172 +778,42 @@ int librados::IoCtxImpl::aio_stat(const object_t& oid, AioCompletionImpl *c,
 
 int librados::IoCtxImpl::remove(const object_t& oid)
 {
-  utime_t ut = ceph_clock_now(client->cct);
-
-  /* can't write to a snapshot */
-  if (snap_seq != CEPH_NOSNAP)
-    return -EROFS;
-
-  Mutex mylock("IoCtxImpl::remove::mylock");
-  Cond cond;
-  bool done;
-  int r;
-  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
-  eversion_t ver;
-
   ::ObjectOperation op;
-  ::ObjectOperation *pop = prepare_assert_ops(&op);
-
-  lock->Lock();
-  objecter->remove(oid, oloc,
-                  snapc, ut, 0,
-                  onack, NULL, &ver, pop);
-  lock->Unlock();
-
-  mylock.Lock();
-  while (!done)
-    cond.Wait(mylock);
-  mylock.Unlock();
-
-  set_sync_op_version(ver);
-
-  return r;
+  prepare_assert_ops(&op);
+  op.remove();
+  return operate(oid, &op, NULL);
 }
 
 int librados::IoCtxImpl::trunc(const object_t& oid, uint64_t size)
 {
-  utime_t ut = ceph_clock_now(client->cct);
-
-  /* can't write to a snapshot */
-  if (snap_seq != CEPH_NOSNAP)
-    return -EROFS;
-
-  Mutex mylock("IoCtxImpl::trunc::mylock");
-  Cond cond;
-  bool done;
-  int r;
-
-  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
-  eversion_t ver;
-
   ::ObjectOperation op;
-  ::ObjectOperation *pop = prepare_assert_ops(&op);
-
-  lock->Lock();
-  objecter->trunc(oid, oloc,
-                 snapc, ut, 0,
-                 size, 0,
-                 onack, NULL, &ver, pop);
-  lock->Unlock();
-
-  mylock.Lock();
-  while (!done)
-    cond.Wait(mylock);
-  mylock.Unlock();
-
-  set_sync_op_version(ver);
-
-  return r;
+  prepare_assert_ops(&op);
+  op.truncate(size);
+  return operate(oid, &op, NULL);
 }
 
 int librados::IoCtxImpl::tmap_update(const object_t& oid, bufferlist& cmdbl)
 {
-  utime_t ut = ceph_clock_now(client->cct);
-
-  /* can't write to a snapshot */
-  if (snap_seq != CEPH_NOSNAP)
-    return -EROFS;
-
-  Mutex mylock("IoCtxImpl::tmap_update::mylock");
-  Cond cond;
-  bool done;
-  int r;
-  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
-  eversion_t ver;
-
-  bufferlist outbl;
-
-  lock->Lock();
   ::ObjectOperation wr;
   prepare_assert_ops(&wr);
   wr.tmap_update(cmdbl);
-  objecter->mutate(oid, oloc, wr, snapc, ut, 0, onack, NULL, &ver);
-  lock->Unlock();
-
-  mylock.Lock();
-  while (!done)
-    cond.Wait(mylock);
-  mylock.Unlock();
-
-  set_sync_op_version(ver);
-
-  return r;
+  return operate(oid, &wr, NULL);
 }
 
 int librados::IoCtxImpl::tmap_put(const object_t& oid, bufferlist& bl)
 {
-  utime_t ut = ceph_clock_now(client->cct);
-
-  /* can't write to a snapshot */
-  if (snap_seq != CEPH_NOSNAP)
-    return -EROFS;
-
-  Mutex mylock("IoCtxImpl::tmap_put::mylock");
-  Cond cond;
-  bool done;
-  int r;
-  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
-  eversion_t ver;
-
-  bufferlist outbl;
-
-  lock->Lock();
   ::ObjectOperation wr;
   prepare_assert_ops(&wr);
   wr.tmap_put(bl);
-  objecter->mutate(oid, oloc, wr, snapc, ut, 0, onack, NULL, &ver);
-  lock->Unlock();
-
-  mylock.Lock();
-  while (!done)
-    cond.Wait(mylock);
-  mylock.Unlock();
-
-  set_sync_op_version(ver);
-
-  return r;
+  return operate(oid, &wr, NULL);
 }
 
 int librados::IoCtxImpl::tmap_get(const object_t& oid, bufferlist& bl)
 {
-  /* can't write to a snapshot */
-  if (snap_seq != CEPH_NOSNAP)
-    return -EROFS;
-
-  Mutex mylock("IoCtxImpl::tmap_put::mylock");
-  Cond cond;
-  bool done;
-  int r = 0;
-  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
-  eversion_t ver;
-
-  bufferlist outbl;
-
-  lock->Lock();
   ::ObjectOperation rd;
   prepare_assert_ops(&rd);
   rd.tmap_get(&bl, NULL);
-  objecter->read(oid, oloc, rd, snap_seq, 0, 0, onack, &ver);
-  lock->Unlock();
-
-  mylock.Lock();
-  while (!done)
-    cond.Wait(mylock);
-  mylock.Unlock();
-
-  set_sync_op_version(ver);
-
-  return r;
+  return operate_read(oid, &rd, NULL);
 }
 
 
@@ -1097,29 +821,10 @@ int librados::IoCtxImpl::exec(const object_t& oid,
                              const char *cls, const char *method,
                              bufferlist& inbl, bufferlist& outbl)
 {
-  Mutex mylock("IoCtxImpl::exec::mylock");
-  Cond cond;
-  bool done;
-  int r;
-  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
-  eversion_t ver;
-
-
-  lock->Lock();
   ::ObjectOperation rd;
   prepare_assert_ops(&rd);
   rd.call(cls, method, inbl);
-  objecter->read(oid, oloc, rd, snap_seq, &outbl, 0, onack, &ver);
-  lock->Unlock();
-
-  mylock.Lock();
-  while (!done)
-    cond.Wait(mylock);
-  mylock.Unlock();
-
-  set_sync_op_version(ver);
-
-  return r;
+  return operate_read(oid, &rd, &outbl);
 }
 
 int librados::IoCtxImpl::aio_exec(const object_t& oid, AioCompletionImpl *c,
@@ -1146,31 +851,10 @@ int librados::IoCtxImpl::read(const object_t& oid,
   if (len > (size_t) INT_MAX)
     return -EDOM;
 
-  Mutex mylock("IoCtxImpl::read::mylock");
-  Cond cond;
-  bool done;
-  int r;
-  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
-  eversion_t ver;
-
-
-  ::ObjectOperation op;
-  ::ObjectOperation *pop = prepare_assert_ops(&op);
-
-  lock->Lock();
-  objecter->read(oid, oloc,
-                off, len, snap_seq, &bl, 0,
-                onack, &ver, pop);
-  lock->Unlock();
-
-  mylock.Lock();
-  while (!done)
-    cond.Wait(mylock);
-  mylock.Unlock();
-  ldout(client->cct, 10) << "Objecter returned from read r=" << r << dendl;
-
-  set_sync_op_version(ver);
-
+  ::ObjectOperation rd;
+  prepare_assert_ops(&rd);
+  rd.read(off, len, &bl, NULL);
+  int r = operate_read(oid, &rd, &bl);
   if (r < 0)
     return r;
 
@@ -1223,12 +907,11 @@ int librados::IoCtxImpl::sparse_read(const object_t& oid,
   if (len > (size_t) INT_MAX)
     return -EDOM;
 
-  int r;
   ::ObjectOperation rd;
   prepare_assert_ops(&rd);
   rd.sparse_read(off, len, &m, &data_bl, NULL);
 
-  r = operate_read(oid, &rd, NULL);
+  int r = operate_read(oid, &rd, NULL);
   if (r < 0)
     return r;
 
@@ -1237,69 +920,31 @@ int librados::IoCtxImpl::sparse_read(const object_t& oid,
 
 int librados::IoCtxImpl::stat(const object_t& oid, uint64_t *psize, time_t *pmtime)
 {
-  Mutex mylock("IoCtxImpl::stat::mylock");
-  Cond cond;
-  bool done;
-  int r;
-  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
   uint64_t size;
   utime_t mtime;
-  eversion_t ver;
 
   if (!psize)
     psize = &size;
 
-  ::ObjectOperation op;
-  ::ObjectOperation *pop = prepare_assert_ops(&op);
-
-  lock->Lock();
-  objecter->stat(oid, oloc,
-                snap_seq, psize, &mtime, 0,
-                onack, &ver, pop);
-  lock->Unlock();
-
-  mylock.Lock();
-  while (!done)
-    cond.Wait(mylock);
-  mylock.Unlock();
-  ldout(client->cct, 10) << "Objecter returned from stat" << dendl;
+  ::ObjectOperation rd;
+  prepare_assert_ops(&rd);
+  rd.stat(psize, &mtime, NULL);
+  int r = operate_read(oid, &rd, NULL);
 
   if (r >= 0 && pmtime) {
     *pmtime = mtime.sec();
   }
 
-  set_sync_op_version(ver);
-
   return r;
 }
 
 int librados::IoCtxImpl::getxattr(const object_t& oid,
                                    const char *name, bufferlist& bl)
 {
-  Mutex mylock("IoCtxImpl::getxattr::mylock");
-  Cond cond;
-  bool done;
-  int r;
-  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
-  eversion_t ver;
-
-  ::ObjectOperation op;
-  ::ObjectOperation *pop = prepare_assert_ops(&op);
-
-  lock->Lock();
-  objecter->getxattr(oid, oloc,
-                    name, snap_seq, &bl, 0,
-                    onack, &ver, pop);
-  lock->Unlock();
-
-  mylock.Lock();
-  while (!done)
-    cond.Wait(mylock);
-  mylock.Unlock();
-  ldout(client->cct, 10) << "Objecter returned from getxattr" << dendl;
-
-  set_sync_op_version(ver);
-
+  ::ObjectOperation rd;
+  prepare_assert_ops(&rd);
+  rd.getxattr(name, &bl, NULL);
+  int r = operate_read(oid, &rd, NULL);
   if (r < 0)
     return r;
 
@@ -1308,117 +953,39 @@ int librados::IoCtxImpl::getxattr(const object_t& oid,
 
 int librados::IoCtxImpl::rmxattr(const object_t& oid, const char *name)
 {
-  utime_t ut = ceph_clock_now(client->cct);
-
-  /* can't write to a snapshot */
-  if (snap_seq != CEPH_NOSNAP)
-    return -EROFS;
-
-  Mutex mylock("IoCtxImpl::rmxattr::mylock");
-  Cond cond;
-  bool done;
-  int r;
-
-  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
-  eversion_t ver;
-
   ::ObjectOperation op;
-  ::ObjectOperation *pop = prepare_assert_ops(&op);
-
-  lock->Lock();
-  objecter->removexattr(oid, oloc, name,
-                       snapc, ut, 0,
-                       onack, NULL, &ver, pop);
-  lock->Unlock();
-
-  mylock.Lock();
-  while (!done)
-    cond.Wait(mylock);
-  mylock.Unlock();
-
-  set_sync_op_version(ver);
-
-  if (r < 0)
-    return r;
-
-  return 0;
+  prepare_assert_ops(&op);
+  op.rmxattr(name);
+  return operate(oid, &op, NULL);
 }
 
 int librados::IoCtxImpl::setxattr(const object_t& oid,
                                    const char *name, bufferlist& bl)
 {
-  utime_t ut = ceph_clock_now(client->cct);
-
-  /* can't write to a snapshot */
-  if (snap_seq != CEPH_NOSNAP)
-    return -EROFS;
-
-  Mutex mylock("IoCtxImpl::setxattr::mylock");
-  Cond cond;
-  bool done;
-  int r;
-
-  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
-  eversion_t ver;
-
   ::ObjectOperation op;
-  ::ObjectOperation *pop = prepare_assert_ops(&op);
-
-  lock->Lock();
-  objecter->setxattr(oid, oloc, name,
-                    snapc, bl, ut, 0,
-                    onack, NULL, &ver, pop);
-  lock->Unlock();
-
-  mylock.Lock();
-  while (!done)
-    cond.Wait(mylock);
-  mylock.Unlock();
-
-  set_sync_op_version(ver);
-
-  if (r < 0)
-    return r;
-
-  return 0;
+  prepare_assert_ops(&op);
+  op.setxattr(name, bl);
+  return operate(oid, &op, NULL);
 }
 
 int librados::IoCtxImpl::getxattrs(const object_t& oid,
                                     map<std::string, bufferlist>& attrset)
 {
-  Mutex mylock("IoCtxImpl::getexattrs::mylock");
-  Cond cond;
-  bool done;
-  int r;
-  eversion_t ver;
-
-  ::ObjectOperation op;
-  ::ObjectOperation *pop = prepare_assert_ops(&op);
-
-  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
-
-  lock->Lock();
   map<string, bufferlist> aset;
-  objecter->getxattrs(oid, oloc, snap_seq,
-                     aset,
-                     0, onack, &ver, pop);
-  lock->Unlock();
 
-  attrset.clear();
-
-
-  mylock.Lock();
-  while (!done)
-    cond.Wait(mylock);
-  mylock.Unlock();
+  ::ObjectOperation rd;
+  prepare_assert_ops(&rd);
+  rd.getxattrs(&aset, NULL);
+  int r = operate_read(oid, &rd, NULL);
 
-  for (map<string,bufferlist>::iterator p = aset.begin(); p != aset.end(); ++p) {
-    ldout(client->cct, 10) << "IoCtxImpl::getxattrs: xattr=" << p->first << dendl;
-    attrset[p->first.c_str()] = p->second;
+  attrset.clear();
+  if (r >= 0) {
+    for (map<string,bufferlist>::iterator p = aset.begin(); p != aset.end(); ++p) {
+      ldout(client->cct, 10) << "IoCtxImpl::getxattrs: xattr=" << p->first << dendl;
+      attrset[p->first.c_str()] = p->second;
+    }
   }
 
-  set_sync_op_version(ver);
-
   return r;
 }