ceph/src/test/librbd/test_ObjectMap.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 #include "test/librbd/test_fixture.h"
   4 #include "test/librbd/test_support.h"
   5 #include "librbd/ExclusiveLock.h"
   6 #include "librbd/ImageCtx.h"
   7 #include "librbd/ImageState.h"
   8 #include "librbd/ImageWatcher.h"
   9 #include "librbd/internal.h"
  10 #include "librbd/ObjectMap.h"
  11 #include "common/Cond.h"
  12 #include "common/Throttle.h"
  13 #include "cls/rbd/cls_rbd_client.h"
  14 #include "cls/rbd/cls_rbd_types.h"
  15 #include <list>
  16 #include <boost/accumulators/accumulators.hpp>
  17 #include <boost/accumulators/statistics/stats.hpp>
  18 #include <boost/accumulators/statistics/rolling_sum.hpp>
  19
  20 void register_test_object_map() {
  21 }
  22
  23 class TestObjectMap : public TestFixture {
  24 public:
  25
  26   int when_open_object_map(librbd::ImageCtx *ictx) {
  27     C_SaferCond ctx;
  28     librbd::ObjectMap<> *object_map = new librbd::ObjectMap<>(*ictx, ictx->snap_id);
  29     object_map->open(&ctx);
  30     int r = ctx.wait();
  31     object_map->put();
  32
  33     return r;
  34   }
  35 };
  36
  37 TEST_F(TestObjectMap, RefreshInvalidatesWhenCorrupt) {
  38   REQUIRE_FEATURE(RBD_FEATURE_OBJECT_MAP);
  39
  40   librbd::ImageCtx *ictx;
  41   ASSERT_EQ(0, open_image(m_image_name, &ictx));
  42   bool flags_set;
  43   ASSERT_EQ(0, ictx->test_flags(CEPH_NOSNAP, RBD_FLAG_OBJECT_MAP_INVALID,
  44                                 &flags_set));
  45   ASSERT_FALSE(flags_set);
  46
  47   C_SaferCond lock_ctx;
  48   {
  49     std::unique_lock owner_locker{ictx->owner_lock};
  50     ictx->exclusive_lock->try_acquire_lock(&lock_ctx);
  51   }
  52   ASSERT_EQ(0, lock_ctx.wait());
  53
  54   std::string oid = librbd::ObjectMap<>::object_map_name(ictx->id, CEPH_NOSNAP);
  55   bufferlist bl;
  56   bl.append("corrupt");
  57   ASSERT_EQ(0, ictx->md_ctx.write_full(oid, bl));
  58
  59   ASSERT_EQ(0, when_open_object_map(ictx));
  60   ASSERT_EQ(0, ictx->test_flags(CEPH_NOSNAP, RBD_FLAG_OBJECT_MAP_INVALID,
  61                                 &flags_set));
  62   ASSERT_TRUE(flags_set);
  63 }
  64
  65 TEST_F(TestObjectMap, RefreshInvalidatesWhenTooSmall) {
  66   REQUIRE_FEATURE(RBD_FEATURE_OBJECT_MAP);
  67
  68   librbd::ImageCtx *ictx;
  69   ASSERT_EQ(0, open_image(m_image_name, &ictx));
  70   bool flags_set;
  71   ASSERT_EQ(0, ictx->test_flags(CEPH_NOSNAP, RBD_FLAG_OBJECT_MAP_INVALID,
  72                                 &flags_set));
  73   ASSERT_FALSE(flags_set);
  74
  75   C_SaferCond lock_ctx;
  76   {
  77     std::unique_lock owner_locker{ictx->owner_lock};
  78     ictx->exclusive_lock->try_acquire_lock(&lock_ctx);
  79   }
  80   ASSERT_EQ(0, lock_ctx.wait());
  81
  82   librados::ObjectWriteOperation op;
  83   librbd::cls_client::object_map_resize(&op, 0, OBJECT_NONEXISTENT);
  84
  85   std::string oid = librbd::ObjectMap<>::object_map_name(ictx->id, CEPH_NOSNAP);
  86   ASSERT_EQ(0, ictx->md_ctx.operate(oid, &op));
  87
  88   ASSERT_EQ(0, when_open_object_map(ictx));
  89   ASSERT_EQ(0, ictx->test_flags(CEPH_NOSNAP, RBD_FLAG_OBJECT_MAP_INVALID,
  90                                 &flags_set));
  91   ASSERT_TRUE(flags_set);
  92 }
  93
  94 TEST_F(TestObjectMap, InvalidateFlagOnDisk) {
  95   REQUIRE_FEATURE(RBD_FEATURE_OBJECT_MAP);
  96
  97   librbd::ImageCtx *ictx;
  98   ASSERT_EQ(0, open_image(m_image_name, &ictx));
  99   bool flags_set;
 100   ASSERT_EQ(0, ictx->test_flags(CEPH_NOSNAP, RBD_FLAG_OBJECT_MAP_INVALID,
 101                                 &flags_set));
 102   ASSERT_FALSE(flags_set);
 103
 104   C_SaferCond lock_ctx;
 105   {
 106     std::unique_lock owner_locker{ictx->owner_lock};
 107     ictx->exclusive_lock->try_acquire_lock(&lock_ctx);
 108   }
 109   ASSERT_EQ(0, lock_ctx.wait());
 110
 111   std::string oid = librbd::ObjectMap<>::object_map_name(ictx->id, CEPH_NOSNAP);
 112   bufferlist bl;
 113   bl.append("corrupt");
 114   ASSERT_EQ(0, ictx->md_ctx.write_full(oid, bl));
 115
 116   ASSERT_EQ(0, when_open_object_map(ictx));
 117   ASSERT_EQ(0, ictx->test_flags(CEPH_NOSNAP, RBD_FLAG_OBJECT_MAP_INVALID,
 118                                 &flags_set));
 119   ASSERT_TRUE(flags_set);
 120
 121   ASSERT_EQ(0, open_image(m_image_name, &ictx));
 122   ASSERT_EQ(0, ictx->test_flags(CEPH_NOSNAP, RBD_FLAG_OBJECT_MAP_INVALID,
 123                                 &flags_set));
 124   ASSERT_TRUE(flags_set);
 125 }
 126
 127 TEST_F(TestObjectMap, AcquireLockInvalidatesWhenTooSmall) {
 128   REQUIRE_FEATURE(RBD_FEATURE_OBJECT_MAP);
 129
 130   librbd::ImageCtx *ictx;
 131   ASSERT_EQ(0, open_image(m_image_name, &ictx));
 132   bool flags_set;
 133   ASSERT_EQ(0, ictx->test_flags(CEPH_NOSNAP, RBD_FLAG_OBJECT_MAP_INVALID,
 134                                 &flags_set));
 135   ASSERT_FALSE(flags_set);
 136
 137   librados::ObjectWriteOperation op;
 138   librbd::cls_client::object_map_resize(&op, 0, OBJECT_NONEXISTENT);
 139
 140   std::string oid = librbd::ObjectMap<>::object_map_name(ictx->id, CEPH_NOSNAP);
 141   ASSERT_EQ(0, ictx->md_ctx.operate(oid, &op));
 142
 143   C_SaferCond lock_ctx;
 144   {
 145     std::unique_lock owner_locker{ictx->owner_lock};
 146     ictx->exclusive_lock->try_acquire_lock(&lock_ctx);
 147   }
 148   ASSERT_EQ(0, lock_ctx.wait());
 149
 150   ASSERT_EQ(0, ictx->test_flags(CEPH_NOSNAP, RBD_FLAG_OBJECT_MAP_INVALID,
 151                                 &flags_set));
 152   ASSERT_TRUE(flags_set);
 153
 154   // Test the flag is stored on disk
 155   ASSERT_EQ(0, ictx->state->refresh());
 156   ASSERT_EQ(0, ictx->test_flags(CEPH_NOSNAP, RBD_FLAG_OBJECT_MAP_INVALID,
 157                                 &flags_set));
 158   ASSERT_TRUE(flags_set);
 159 }
 160
 161 namespace chrono = std::chrono;
 162
 163 TEST_F(TestObjectMap, DISABLED_StressTest) {
 164   REQUIRE_FEATURE(RBD_FEATURE_OBJECT_MAP);
 165
 166   uint64_t object_count = cls::rbd::MAX_OBJECT_MAP_OBJECT_COUNT;
 167   librbd::ImageCtx *ictx;
 168   ASSERT_EQ(0, open_image(m_image_name, &ictx));
 169   ASSERT_EQ(0, resize(ictx, ictx->layout.object_size * object_count));
 170
 171   bool flags_set;
 172   ASSERT_EQ(0, ictx->test_flags(CEPH_NOSNAP, RBD_FLAG_OBJECT_MAP_INVALID,
 173                                 &flags_set));
 174   ASSERT_FALSE(flags_set);
 175
 176   srand(time(NULL) % (unsigned long) -1);
 177
 178   coarse_mono_time start = coarse_mono_clock::now();
 179   chrono::duration<double> last = chrono::duration<double>::zero();
 180
 181   const int WINDOW_SIZE = 5;
 182   typedef boost::accumulators::accumulator_set<
 183     double, boost::accumulators::stats<
 184       boost::accumulators::tag::rolling_sum> > RollingSum;
 185
 186   RollingSum time_acc(
 187     boost::accumulators::tag::rolling_window::window_size = WINDOW_SIZE);
 188   RollingSum ios_acc(
 189     boost::accumulators::tag::rolling_window::window_size = WINDOW_SIZE);
 190
 191   uint32_t io_threads = 16;
 192   uint64_t cur_ios = 0;
 193   SimpleThrottle throttle(io_threads, false);
 194   for (uint64_t ios = 0; ios < 100000;) {
 195     if (throttle.pending_error()) {
 196       break;
 197     }
 198
 199     throttle.start_op();
 200     uint64_t object_no = (rand() % object_count);
 201     auto ctx = new LambdaContext([&throttle, object_no](int r) {
 202         ASSERT_EQ(0, r) << "object_no=" << object_no;
 203         throttle.end_op(r);
 204       });
 205
 206     std::shared_lock owner_locker{ictx->owner_lock};
 207     std::shared_lock image_locker{ictx->image_lock};
 208     ASSERT_TRUE(ictx->object_map != nullptr);
 209
 210     if (!ictx->object_map->aio_update<
 211           Context, &Context::complete>(CEPH_NOSNAP, object_no,
 212                                        OBJECT_EXISTS, {}, {}, true,
 213                                        ctx)) {
 214       ctx->complete(0);
 215     } else {
 216       ++cur_ios;
 217       ++ios;
 218     }
 219
 220     coarse_mono_time now = coarse_mono_clock::now();
 221     chrono::duration<double> elapsed = now - start;
 222     if (last == chrono::duration<double>::zero()) {
 223       last = elapsed;
 224     } else if ((int)elapsed.count() != (int)last.count()) {
 225       time_acc((elapsed - last).count());
 226       ios_acc(static_cast<double>(cur_ios));
 227       cur_ios = 0;
 228
 229       double time_sum = boost::accumulators::rolling_sum(time_acc);
 230       std::cerr << std::setw(5) << (int)elapsed.count() << "\t"
 231                 << std::setw(8) << (int)ios << "\t"
 232                 << std::fixed << std::setw(8) << std::setprecision(2)
 233                 << boost::accumulators::rolling_sum(ios_acc) / time_sum
 234                 << std::endl;
 235       last = elapsed;
 236     }
 237   }
 238
 239   ASSERT_EQ(0, throttle.wait_for_ret());
 240 }