src/ceph/src/test/librados_test_stub/TestIoCtxImpl.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #include "test/librados_test_stub/TestIoCtxImpl.h"
   5 #include "test/librados_test_stub/TestClassHandler.h"
   6 #include "test/librados_test_stub/TestRadosClient.h"
   7 #include "test/librados_test_stub/TestWatchNotify.h"
   8 #include "librados/AioCompletionImpl.h"
   9 #include "include/assert.h"
  10 #include "common/Finisher.h"
  11 #include "common/valgrind.h"
  12 #include "objclass/objclass.h"
  13 #include <boost/bind.hpp>
  14 #include <errno.h>
  15
  16 namespace librados {
  17
  18 TestIoCtxImpl::TestIoCtxImpl() : m_client(NULL) {
  19   get();
  20 }
  21
  22 TestIoCtxImpl::TestIoCtxImpl(TestRadosClient *client, int64_t pool_id,
  23                              const std::string& pool_name)
  24   : m_client(client), m_pool_id(pool_id), m_pool_name(pool_name),
  25     m_snap_seq(CEPH_NOSNAP)
  26 {
  27   m_client->get();
  28   get();
  29 }
  30
  31 TestIoCtxImpl::TestIoCtxImpl(const TestIoCtxImpl& rhs)
  32   : m_client(rhs.m_client),
  33     m_pool_id(rhs.m_pool_id),
  34     m_pool_name(rhs.m_pool_name),
  35     m_snap_seq(rhs.m_snap_seq)
  36 {
  37   m_client->get();
  38   get();
  39 }
  40
  41 TestIoCtxImpl::~TestIoCtxImpl() {
  42   assert(m_pending_ops == 0);
  43 }
  44
  45 void TestObjectOperationImpl::get() {
  46   m_refcount++;
  47 }
  48
  49 void TestObjectOperationImpl::put() {
  50   if (--m_refcount == 0) {
  51     ANNOTATE_HAPPENS_AFTER(&m_refcount);
  52     ANNOTATE_HAPPENS_BEFORE_FORGET_ALL(&m_refcount);
  53     delete this;
  54   } else {
  55     ANNOTATE_HAPPENS_BEFORE(&m_refcount);
  56   }
  57 }
  58
  59 void TestIoCtxImpl::get() {
  60   m_refcount++;
  61 }
  62
  63 void TestIoCtxImpl::put() {
  64   if (--m_refcount == 0) {
  65     m_client->put();
  66     delete this;
  67   }
  68 }
  69
  70 uint64_t TestIoCtxImpl::get_instance_id() const {
  71   return m_client->get_instance_id();
  72 }
  73
  74 int64_t TestIoCtxImpl::get_id() {
  75   return m_pool_id;
  76 }
  77
  78 uint64_t TestIoCtxImpl::get_last_version() {
  79   return 0;
  80 }
  81
  82 std::string TestIoCtxImpl::get_pool_name() {
  83   return m_pool_name;
  84 }
  85
  86 int TestIoCtxImpl::aio_flush() {
  87   m_client->flush_aio_operations();
  88   return 0;
  89 }
  90
  91 void TestIoCtxImpl::aio_flush_async(AioCompletionImpl *c) {
  92   m_client->flush_aio_operations(c);
  93 }
  94
  95 void TestIoCtxImpl::aio_notify(const std::string& oid, AioCompletionImpl *c,
  96                                bufferlist& bl, uint64_t timeout_ms,
  97                                bufferlist *pbl) {
  98   m_pending_ops++;
  99   c->get();
 100   C_AioNotify *ctx = new C_AioNotify(this, c);
 101   m_client->get_watch_notify()->aio_notify(m_client, oid, bl, timeout_ms, pbl,
 102                                            ctx);
 103 }
 104
 105 int TestIoCtxImpl::aio_operate(const std::string& oid, TestObjectOperationImpl &ops,
 106                                AioCompletionImpl *c, SnapContext *snap_context,
 107                                int flags) {
 108   // TODO flags for now
 109   ops.get();
 110   m_pending_ops++;
 111   m_client->add_aio_operation(oid, true, boost::bind(
 112     &TestIoCtxImpl::execute_aio_operations, this, oid, &ops,
 113     reinterpret_cast<bufferlist*>(0),
 114     snap_context != NULL ? *snap_context : m_snapc), c);
 115   return 0;
 116 }
 117
 118 int TestIoCtxImpl::aio_operate_read(const std::string& oid,
 119                                     TestObjectOperationImpl &ops,
 120                                     AioCompletionImpl *c, int flags,
 121                                     bufferlist *pbl) {
 122   // TODO ignoring flags for now
 123   ops.get();
 124   m_pending_ops++;
 125   m_client->add_aio_operation(oid, true, boost::bind(
 126     &TestIoCtxImpl::execute_aio_operations, this, oid, &ops, pbl, m_snapc), c);
 127   return 0;
 128 }
 129
 130 int TestIoCtxImpl::aio_watch(const std::string& o, AioCompletionImpl *c,
 131                              uint64_t *handle, librados::WatchCtx2 *watch_ctx) {
 132   m_pending_ops++;
 133   c->get();
 134   C_AioNotify *ctx = new C_AioNotify(this, c);
 135   if (m_client->is_blacklisted()) {
 136     m_client->get_aio_finisher()->queue(ctx, -EBLACKLISTED);
 137   } else {
 138     m_client->get_watch_notify()->aio_watch(m_client, o, get_instance_id(),
 139                                             handle, watch_ctx, ctx);
 140   }
 141   return 0;
 142 }
 143
 144 int TestIoCtxImpl::aio_unwatch(uint64_t handle, AioCompletionImpl *c) {
 145   m_pending_ops++;
 146   c->get();
 147   C_AioNotify *ctx = new C_AioNotify(this, c);
 148   if (m_client->is_blacklisted()) {
 149     m_client->get_aio_finisher()->queue(ctx, -EBLACKLISTED);
 150   } else {
 151     m_client->get_watch_notify()->aio_unwatch(m_client, handle, ctx);
 152   }
 153   return 0;
 154 }
 155
 156 int TestIoCtxImpl::exec(const std::string& oid, TestClassHandler *handler,
 157                         const char *cls, const char *method,
 158                         bufferlist& inbl, bufferlist* outbl,
 159                         const SnapContext &snapc) {
 160   if (m_client->is_blacklisted()) {
 161     return -EBLACKLISTED;
 162   }
 163
 164   cls_method_cxx_call_t call = handler->get_method(cls, method);
 165   if (call == NULL) {
 166     return -ENOSYS;
 167   }
 168
 169   return (*call)(reinterpret_cast<cls_method_context_t>(
 170     handler->get_method_context(this, oid, snapc).get()), &inbl, outbl);
 171 }
 172
 173 int TestIoCtxImpl::list_watchers(const std::string& o,
 174                                  std::list<obj_watch_t> *out_watchers) {
 175   if (m_client->is_blacklisted()) {
 176     return -EBLACKLISTED;
 177   }
 178
 179   return m_client->get_watch_notify()->list_watchers(o, out_watchers);
 180 }
 181
 182 int TestIoCtxImpl::notify(const std::string& o, bufferlist& bl,
 183                           uint64_t timeout_ms, bufferlist *pbl) {
 184   if (m_client->is_blacklisted()) {
 185     return -EBLACKLISTED;
 186   }
 187
 188   return m_client->get_watch_notify()->notify(m_client, o, bl, timeout_ms, pbl);
 189 }
 190
 191 void TestIoCtxImpl::notify_ack(const std::string& o, uint64_t notify_id,
 192                                uint64_t handle, bufferlist& bl) {
 193   m_client->get_watch_notify()->notify_ack(m_client, o, notify_id, handle,
 194                                            m_client->get_instance_id(), bl);
 195 }
 196
 197 int TestIoCtxImpl::operate(const std::string& oid, TestObjectOperationImpl &ops) {
 198   AioCompletionImpl *comp = new AioCompletionImpl();
 199
 200   ops.get();
 201   m_pending_ops++;
 202   m_client->add_aio_operation(oid, false, boost::bind(
 203     &TestIoCtxImpl::execute_aio_operations, this, oid, &ops,
 204     reinterpret_cast<bufferlist*>(0), m_snapc), comp);
 205
 206   comp->wait_for_safe();
 207   int ret = comp->get_return_value();
 208   comp->put();
 209   return ret;
 210 }
 211
 212 int TestIoCtxImpl::operate_read(const std::string& oid, TestObjectOperationImpl &ops,
 213                                 bufferlist *pbl) {
 214   AioCompletionImpl *comp = new AioCompletionImpl();
 215
 216   ops.get();
 217   m_pending_ops++;
 218   m_client->add_aio_operation(oid, false, boost::bind(
 219     &TestIoCtxImpl::execute_aio_operations, this, oid, &ops, pbl,
 220     m_snapc), comp);
 221
 222   comp->wait_for_complete();
 223   int ret = comp->get_return_value();
 224   comp->put();
 225   return ret;
 226 }
 227
 228 void TestIoCtxImpl::aio_selfmanaged_snap_create(uint64_t *snapid,
 229                                                 AioCompletionImpl *c) {
 230   m_client->add_aio_operation(
 231     "", true,
 232     boost::bind(&TestIoCtxImpl::selfmanaged_snap_create, this, snapid), c);
 233 }
 234
 235 void TestIoCtxImpl::aio_selfmanaged_snap_remove(uint64_t snapid,
 236                                                 AioCompletionImpl *c) {
 237   m_client->add_aio_operation(
 238     "", true,
 239     boost::bind(&TestIoCtxImpl::selfmanaged_snap_remove, this, snapid), c);
 240 }
 241
 242 int TestIoCtxImpl::selfmanaged_snap_set_write_ctx(snap_t seq,
 243                                                   std::vector<snap_t>& snaps) {
 244   std::vector<snapid_t> snap_ids(snaps.begin(), snaps.end());
 245   m_snapc = SnapContext(seq, snap_ids);
 246   return 0;
 247 }
 248
 249 int TestIoCtxImpl::set_alloc_hint(const std::string& oid,
 250                                   uint64_t expected_object_size,
 251                                   uint64_t expected_write_size) {
 252   return 0;
 253 }
 254
 255 void TestIoCtxImpl::set_snap_read(snap_t seq) {
 256   if (seq == 0) {
 257     seq = CEPH_NOSNAP;
 258   }
 259   m_snap_seq = seq;
 260 }
 261
 262 int TestIoCtxImpl::tmap_update(const std::string& oid, bufferlist& cmdbl) {
 263   if (m_client->is_blacklisted()) {
 264     return -EBLACKLISTED;
 265   }
 266
 267   // TODO: protect against concurrent tmap updates
 268   bufferlist tmap_header;
 269   std::map<string,bufferlist> tmap;
 270   uint64_t size = 0;
 271   int r = stat(oid, &size, NULL);
 272   if (r == -ENOENT) {
 273     r = create(oid, false);
 274   }
 275   if (r < 0) {
 276     return r;
 277   }
 278
 279   if (size > 0) {
 280     bufferlist inbl;
 281     r = read(oid, size, 0, &inbl);
 282     if (r < 0) {
 283       return r;
 284     }
 285     bufferlist::iterator iter = inbl.begin();
 286     ::decode(tmap_header, iter);
 287     ::decode(tmap, iter);
 288   }
 289
 290   __u8 c;
 291   std::string key;
 292   bufferlist value;
 293   bufferlist::iterator iter = cmdbl.begin();
 294   ::decode(c, iter);
 295   ::decode(key, iter);
 296
 297   switch (c) {
 298     case CEPH_OSD_TMAP_SET:
 299       ::decode(value, iter);
 300       tmap[key] = value;
 301       break;
 302     case CEPH_OSD_TMAP_RM:
 303       r = tmap.erase(key);
 304       if (r == 0) {
 305         return -ENOENT;
 306       }
 307       break;
 308     default:
 309       return -EINVAL;
 310   }
 311
 312   bufferlist out;
 313   ::encode(tmap_header, out);
 314   ::encode(tmap, out);
 315   r = write_full(oid, out, m_snapc);
 316   return r;
 317 }
 318
 319 int TestIoCtxImpl::unwatch(uint64_t handle) {
 320   if (m_client->is_blacklisted()) {
 321     return -EBLACKLISTED;
 322   }
 323
 324   return m_client->get_watch_notify()->unwatch(m_client, handle);
 325 }
 326
 327 int TestIoCtxImpl::watch(const std::string& o, uint64_t *handle,
 328                          librados::WatchCtx *ctx, librados::WatchCtx2 *ctx2) {
 329   if (m_client->is_blacklisted()) {
 330     return -EBLACKLISTED;
 331   }
 332
 333   return m_client->get_watch_notify()->watch(m_client, o, get_instance_id(),
 334                                              handle, ctx, ctx2);
 335 }
 336
 337 int TestIoCtxImpl::execute_operation(const std::string& oid,
 338                                      const Operation &operation) {
 339   if (m_client->is_blacklisted()) {
 340     return -EBLACKLISTED;
 341   }
 342
 343   TestRadosClient::Transaction transaction(m_client, oid);
 344   return operation(this, oid);
 345 }
 346
 347 int TestIoCtxImpl::execute_aio_operations(const std::string& oid,
 348                                           TestObjectOperationImpl *ops,
 349                                           bufferlist *pbl,
 350                                           const SnapContext &snapc) {
 351   int ret = 0;
 352   if (m_client->is_blacklisted()) {
 353     ret = -EBLACKLISTED;
 354   } else {
 355     TestRadosClient::Transaction transaction(m_client, oid);
 356     for (ObjectOperations::iterator it = ops->ops.begin();
 357          it != ops->ops.end(); ++it) {
 358       ret = (*it)(this, oid, pbl, snapc);
 359       if (ret < 0) {
 360         break;
 361       }
 362     }
 363   }
 364   m_pending_ops--;
 365   ops->put();
 366   return ret;
 367 }
 368
 369 void TestIoCtxImpl::handle_aio_notify_complete(AioCompletionImpl *c, int r) {
 370   m_pending_ops--;
 371
 372   m_client->finish_aio_completion(c, r);
 373 }
 374
 375 } // namespace librados