src/ceph/src/test/librbd/test_librbd.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2011 New Dream Network
   7  *
   8  * This is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU General Public
  10  * License version 2, as published by the Free Software
  11  * Foundation.  See file COPYING.
  12  *
  13  */
  14
  15 #include "include/int_types.h"
  16 #include "include/rados/librados.h"
  17 #include "include/rbd_types.h"
  18 #include "include/rbd/librbd.h"
  19 #include "include/rbd/librbd.hpp"
  20 #include "include/event_type.h"
  21 #include "include/err.h"
  22
  23 #include "gtest/gtest.h"
  24
  25 #include <errno.h>
  26 #include <stdarg.h>
  27 #include <stdio.h>
  28 #include <stdlib.h>
  29 #include <sys/types.h>
  30 #include <poll.h>
  31 #include <time.h>
  32 #include <unistd.h>
  33 #include <algorithm>
  34 #include <chrono>
  35 #include <condition_variable>
  36 #include <iostream>
  37 #include <sstream>
  38 #include <list>
  39 #include <set>
  40 #include <thread>
  41 #include <vector>
  42
  43 #include "test/librados/test.h"
  44 #include "test/librbd/test_support.h"
  45 #include "common/event_socket.h"
  46 #include "include/interval_set.h"
  47 #include "include/stringify.h"
  48
  49 #include <boost/assign/list_of.hpp>
  50 #include <boost/scope_exit.hpp>
  51
  52 #ifdef HAVE_EVENTFD
  53 #include <sys/eventfd.h>
  54 #endif
  55
  56 using namespace std;
  57
  58 using std::chrono::seconds;
  59
  60 #define ASSERT_PASSED(x, args...) \
  61   do {                            \
  62     bool passed = false;          \
  63     x(args, &passed);             \
  64     ASSERT_TRUE(passed);          \
  65   } while(0)
  66
  67 void register_test_librbd() {
  68 }
  69
  70 static int get_features(bool *old_format, uint64_t *features)
  71 {
  72   const char *c = getenv("RBD_FEATURES");
  73   if (c) {
  74     stringstream ss;
  75     ss << c;
  76     ss >> *features;
  77     if (ss.fail())
  78       return -EINVAL;
  79     *old_format = false;
  80     cout << "using new format!" << std::endl;
  81   } else {
  82     *old_format = true;
  83     *features = 0;
  84     cout << "using old format" << std::endl;
  85   }
  86
  87   return 0;
  88 }
  89
  90 static int create_image_full(rados_ioctx_t ioctx, const char *name,
  91                              uint64_t size, int *order, int old_format,
  92                              uint64_t features)
  93 {
  94   if (old_format) {
  95     // ensure old-format tests actually use the old format
  96     int r = rados_conf_set(rados_ioctx_get_cluster(ioctx),
  97                            "rbd_default_format", "1");
  98     if (r < 0) {
  99       return r;
 100     }
 101     return rbd_create(ioctx, name, size, order);
 102   } else if ((features & RBD_FEATURE_STRIPINGV2) != 0) {
 103     uint64_t stripe_unit = IMAGE_STRIPE_UNIT;
 104     if (*order) {
 105       // use a conservative stripe_unit for non default order
 106       stripe_unit = (1ull << (*order-1));
 107     }
 108
 109     printf("creating image with stripe unit: %" PRIu64 ", "
 110            "stripe count: %" PRIu64 "\n",
 111            stripe_unit, IMAGE_STRIPE_COUNT);
 112     return rbd_create3(ioctx, name, size, features, order,
 113                        stripe_unit, IMAGE_STRIPE_COUNT);
 114   } else {
 115     return rbd_create2(ioctx, name, size, features, order);
 116   }
 117 }
 118
 119 static int clone_image(rados_ioctx_t p_ioctx,
 120                        rbd_image_t p_image, const char *p_name,
 121                        const char *p_snap_name, rados_ioctx_t c_ioctx,
 122                        const char *c_name, uint64_t features, int *c_order)
 123 {
 124   uint64_t stripe_unit, stripe_count;
 125
 126   int r;
 127   r = rbd_get_stripe_unit(p_image, &stripe_unit);
 128   if (r != 0) {
 129     return r;
 130   }
 131
 132   r = rbd_get_stripe_count(p_image, &stripe_count);
 133   if (r != 0) {
 134     return r;
 135   }
 136
 137   return rbd_clone2(p_ioctx, p_name, p_snap_name, c_ioctx,
 138                     c_name, features, c_order, stripe_unit, stripe_count);
 139 }
 140
 141
 142 static int create_image(rados_ioctx_t ioctx, const char *name,
 143                         uint64_t size, int *order)
 144 {
 145   bool old_format;
 146   uint64_t features;
 147
 148   int r = get_features(&old_format, &features);
 149   if (r < 0)
 150     return r;
 151   return create_image_full(ioctx, name, size, order, old_format, features);
 152 }
 153
 154 static int create_image_pp(librbd::RBD &rbd,
 155                            librados::IoCtx &ioctx,
 156                            const char *name,
 157                            uint64_t size, int *order) {
 158   bool old_format;
 159   uint64_t features;
 160   int r = get_features(&old_format, &features);
 161   if (r < 0)
 162     return r;
 163   if (old_format) {
 164     librados::Rados rados(ioctx);
 165     int r = rados.conf_set("rbd_default_format", "1");
 166     if (r < 0) {
 167       return r;
 168     }
 169     return rbd.create(ioctx, name, size, order);
 170   } else {
 171     return rbd.create2(ioctx, name, size, features, order);
 172   }
 173 }
 174
 175 class TestLibRBD : public ::testing::Test {
 176 public:
 177
 178   TestLibRBD() : m_pool_number() {
 179   }
 180
 181   static void SetUpTestCase() {
 182     static bool seeded = false;
 183     if (!seeded) {
 184       seeded = true;
 185       int seed = getpid();
 186       cout << "seed " << seed << std::endl;
 187       srand(seed);
 188     }
 189
 190     _pool_names.clear();
 191     _unique_pool_names.clear();
 192     _image_number = 0;
 193     ASSERT_EQ("", connect_cluster(&_cluster));
 194     ASSERT_EQ("", connect_cluster_pp(_rados));
 195
 196     create_optional_data_pool();
 197   }
 198
 199   static void TearDownTestCase() {
 200     rados_shutdown(_cluster);
 201     _rados.wait_for_latest_osdmap();
 202     _pool_names.insert(_pool_names.end(), _unique_pool_names.begin(),
 203                        _unique_pool_names.end());
 204     for (size_t i = 1; i < _pool_names.size(); ++i) {
 205       ASSERT_EQ(0, _rados.pool_delete(_pool_names[i].c_str()));
 206     }
 207     if (!_pool_names.empty()) {
 208       ASSERT_EQ(0, destroy_one_pool_pp(_pool_names[0], _rados));
 209     }
 210   }
 211
 212   void SetUp() override {
 213     ASSERT_NE("", m_pool_name = create_pool());
 214   }
 215
 216   bool is_skip_partial_discard_enabled() {
 217     std::string value;
 218     EXPECT_EQ(0, _rados.conf_get("rbd_skip_partial_discard", value));
 219     return value == "true";
 220   }
 221
 222   void validate_object_map(rbd_image_t image, bool *passed) {
 223     uint64_t flags;
 224     ASSERT_EQ(0, rbd_get_flags(image, &flags));
 225     *passed = ((flags & RBD_FLAG_OBJECT_MAP_INVALID) == 0);
 226   }
 227
 228   void validate_object_map(librbd::Image &image, bool *passed) {
 229     uint64_t flags;
 230     ASSERT_EQ(0, image.get_flags(&flags));
 231     *passed = ((flags & RBD_FLAG_OBJECT_MAP_INVALID) == 0);
 232   }
 233
 234   std::string get_temp_image_name() {
 235     ++_image_number;
 236     return "image" + stringify(_image_number);
 237   }
 238
 239   static void create_optional_data_pool() {
 240     bool created = false;
 241     std::string data_pool;
 242     ASSERT_EQ(0, create_image_data_pool(_rados, data_pool, &created));
 243     if (!data_pool.empty()) {
 244       printf("using image data pool: %s\n", data_pool.c_str());
 245       if (created) {
 246         _unique_pool_names.push_back(data_pool);
 247       }
 248     }
 249   }
 250
 251   std::string create_pool(bool unique = false) {
 252     librados::Rados rados;
 253     std::string pool_name;
 254     if (unique) {
 255       pool_name = get_temp_pool_name("test-librbd-");
 256       EXPECT_EQ("", create_one_pool_pp(pool_name, rados));
 257       _unique_pool_names.push_back(pool_name);
 258     } else if (m_pool_number < _pool_names.size()) {
 259       pool_name = _pool_names[m_pool_number];
 260     } else {
 261       pool_name = get_temp_pool_name("test-librbd-");
 262       EXPECT_EQ("", create_one_pool_pp(pool_name, rados));
 263       _pool_names.push_back(pool_name);
 264     }
 265     ++m_pool_number;
 266     return pool_name;
 267   }
 268
 269   static std::vector<std::string> _pool_names;
 270   static std::vector<std::string> _unique_pool_names;
 271   static rados_t _cluster;
 272   static librados::Rados _rados;
 273   static uint64_t _image_number;
 274
 275   std::string m_pool_name;
 276   uint32_t m_pool_number;
 277
 278 };
 279
 280 std::vector<std::string> TestLibRBD::_pool_names;
 281 std::vector<std::string> TestLibRBD::_unique_pool_names;
 282 rados_t TestLibRBD::_cluster;
 283 librados::Rados TestLibRBD::_rados;
 284 uint64_t TestLibRBD::_image_number = 0;
 285
 286 TEST_F(TestLibRBD, CreateAndStat)
 287 {
 288   rados_ioctx_t ioctx;
 289   ASSERT_EQ(0, rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx));
 290
 291   rbd_image_info_t info;
 292   rbd_image_t image;
 293   int order = 0;
 294   std::string name = get_temp_image_name();
 295   uint64_t size = 2 << 20;
 296
 297   ASSERT_EQ(0, create_image(ioctx, name.c_str(), size, &order));
 298   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
 299   ASSERT_EQ(0, rbd_stat(image, &info, sizeof(info)));
 300   printf("image has size %llu and order %d\n", (unsigned long long) info.size, info.order);
 301   ASSERT_EQ(info.size, size);
 302   ASSERT_EQ(info.order, order);
 303   ASSERT_EQ(0, rbd_close(image));
 304
 305   rados_ioctx_destroy(ioctx);
 306 }
 307
 308 TEST_F(TestLibRBD, CreateWithSameDataPool)
 309 {
 310   REQUIRE_FORMAT_V2();
 311
 312   rados_ioctx_t ioctx;
 313   ASSERT_EQ(0, rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx));
 314
 315   rbd_image_t image;
 316   std::string name = get_temp_image_name();
 317   uint64_t size = 2 << 20;
 318
 319   bool old_format;
 320   uint64_t features;
 321   ASSERT_EQ(0, get_features(&old_format, &features));
 322   ASSERT_FALSE(old_format);
 323
 324   rbd_image_options_t image_options;
 325   rbd_image_options_create(&image_options);
 326   BOOST_SCOPE_EXIT( (&image_options) ) {
 327     rbd_image_options_destroy(image_options);
 328   } BOOST_SCOPE_EXIT_END;
 329
 330   ASSERT_EQ(0, rbd_image_options_set_uint64(image_options,
 331                                             RBD_IMAGE_OPTION_FEATURES,
 332                                             features));
 333   ASSERT_EQ(0, rbd_image_options_set_string(image_options,
 334                                             RBD_IMAGE_OPTION_DATA_POOL,
 335                                             m_pool_name.c_str()));
 336
 337   ASSERT_EQ(0, rbd_create4(ioctx, name.c_str(), size, image_options));
 338   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
 339
 340   ASSERT_EQ(0, rbd_close(image));
 341
 342   rados_ioctx_destroy(ioctx);
 343 }
 344
 345 TEST_F(TestLibRBD, CreateAndStatPP)
 346 {
 347   librados::IoCtx ioctx;
 348   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
 349
 350   {
 351     librbd::RBD rbd;
 352     librbd::image_info_t info;
 353     librbd::Image image;
 354     int order = 0;
 355     std::string name = get_temp_image_name();
 356     uint64_t size = 2 << 20;
 357
 358     ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
 359     ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
 360     ASSERT_EQ(0, image.stat(info, sizeof(info)));
 361     ASSERT_EQ(info.size, size);
 362     ASSERT_EQ(info.order, order);
 363   }
 364
 365   ioctx.close();
 366 }
 367
 368 TEST_F(TestLibRBD, GetId)
 369 {
 370   rados_ioctx_t ioctx;
 371   ASSERT_EQ(0, rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx));
 372
 373   rbd_image_t image;
 374   int order = 0;
 375   std::string name = get_temp_image_name();
 376
 377   ASSERT_EQ(0, create_image(ioctx, name.c_str(), 0, &order));
 378   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
 379
 380   char id[4096];
 381   if (!is_feature_enabled(0)) {
 382     // V1 image
 383     ASSERT_EQ(-EINVAL, rbd_get_id(image, id, sizeof(id)));
 384   } else {
 385     ASSERT_EQ(-ERANGE, rbd_get_id(image, id, 0));
 386     ASSERT_EQ(0, rbd_get_id(image, id, sizeof(id)));
 387     ASSERT_LT(0U, strlen(id));
 388   }
 389
 390   ASSERT_EQ(0, rbd_close(image));
 391   rados_ioctx_destroy(ioctx);
 392 }
 393
 394 TEST_F(TestLibRBD, GetIdPP)
 395 {
 396   librados::IoCtx ioctx;
 397   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
 398
 399   librbd::RBD rbd;
 400   librbd::Image image;
 401   int order = 0;
 402   std::string name = get_temp_image_name();
 403
 404   std::string id;
 405   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), 0, &order));
 406   ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
 407   if (!is_feature_enabled(0)) {
 408     // V1 image
 409     ASSERT_EQ(-EINVAL, image.get_id(&id));
 410   } else {
 411     ASSERT_EQ(0, image.get_id(&id));
 412     ASSERT_LT(0U, id.size());
 413   }
 414 }
 415
 416 TEST_F(TestLibRBD, GetBlockNamePrefix)
 417 {
 418   rados_ioctx_t ioctx;
 419   ASSERT_EQ(0, rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx));
 420
 421   rbd_image_t image;
 422   int order = 0;
 423   std::string name = get_temp_image_name();
 424
 425   ASSERT_EQ(0, create_image(ioctx, name.c_str(), 0, &order));
 426   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
 427
 428   char prefix[4096];
 429   ASSERT_EQ(-ERANGE, rbd_get_block_name_prefix(image, prefix, 0));
 430   ASSERT_EQ(0, rbd_get_block_name_prefix(image, prefix, sizeof(prefix)));
 431   ASSERT_LT(0U, strlen(prefix));
 432
 433   ASSERT_EQ(0, rbd_close(image));
 434   rados_ioctx_destroy(ioctx);
 435 }
 436
 437 TEST_F(TestLibRBD, GetBlockNamePrefixPP)
 438 {
 439   librados::IoCtx ioctx;
 440   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
 441
 442   librbd::RBD rbd;
 443   librbd::Image image;
 444   int order = 0;
 445   std::string name = get_temp_image_name();
 446
 447   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), 0, &order));
 448   ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
 449   ASSERT_LT(0U, image.get_block_name_prefix().size());
 450 }
 451
 452 TEST_F(TestLibRBD, TestGetCreateTimestamp)
 453 {
 454   REQUIRE_FORMAT_V2();
 455
 456   rados_ioctx_t ioctx;
 457   ASSERT_EQ(0, rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx));
 458
 459   rbd_image_t image;
 460   int order = 0;
 461   std::string name = get_temp_image_name();
 462
 463   ASSERT_EQ(0, create_image(ioctx, name.c_str(), 0, &order));
 464   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
 465
 466   struct timespec timestamp;
 467   ASSERT_EQ(0, rbd_get_create_timestamp(image, &timestamp));
 468   ASSERT_LT(0, timestamp.tv_sec);
 469
 470   ASSERT_EQ(0, rbd_close(image));
 471
 472   rados_ioctx_destroy(ioctx);
 473 }
 474
 475 TEST_F(TestLibRBD, GetCreateTimestampPP)
 476 {
 477   REQUIRE_FORMAT_V2();
 478
 479   librados::IoCtx ioctx;
 480   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
 481
 482   librbd::RBD rbd;
 483   librbd::Image image;
 484   int order = 0;
 485   std::string name = get_temp_image_name();
 486
 487   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), 0, &order));
 488   ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
 489
 490   struct timespec timestamp;
 491   ASSERT_EQ(0, image.get_create_timestamp(&timestamp));
 492   ASSERT_LT(0, timestamp.tv_sec);
 493 }
 494
 495 TEST_F(TestLibRBD, OpenAio)
 496 {
 497   rados_ioctx_t ioctx;
 498   ASSERT_EQ(0, rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx));
 499
 500   rbd_image_info_t info;
 501   rbd_image_t image;
 502   int order = 0;
 503   std::string name = get_temp_image_name();
 504   uint64_t size = 2 << 20;
 505
 506   ASSERT_EQ(0, create_image(ioctx, name.c_str(), size, &order));
 507
 508   rbd_completion_t open_comp;
 509   ASSERT_EQ(0, rbd_aio_create_completion(NULL, NULL, &open_comp));
 510   ASSERT_EQ(0, rbd_aio_open(ioctx, name.c_str(), &image, NULL, open_comp));
 511   ASSERT_EQ(0, rbd_aio_wait_for_complete(open_comp));
 512   ASSERT_EQ(1, rbd_aio_is_complete(open_comp));
 513   ASSERT_EQ(0, rbd_aio_get_return_value(open_comp));
 514   rbd_aio_release(open_comp);
 515
 516   ASSERT_EQ(0, rbd_stat(image, &info, sizeof(info)));
 517   printf("image has size %llu and order %d\n", (unsigned long long) info.size, info.order);
 518   ASSERT_EQ(info.size, size);
 519   ASSERT_EQ(info.order, order);
 520
 521   rbd_completion_t close_comp;
 522   ASSERT_EQ(0, rbd_aio_create_completion(NULL, NULL, &close_comp));
 523   ASSERT_EQ(0, rbd_aio_close(image, close_comp));
 524   ASSERT_EQ(0, rbd_aio_wait_for_complete(close_comp));
 525   ASSERT_EQ(1, rbd_aio_is_complete(close_comp));
 526   ASSERT_EQ(0, rbd_aio_get_return_value(close_comp));
 527   rbd_aio_release(close_comp);
 528
 529   rados_ioctx_destroy(ioctx);
 530 }
 531
 532 TEST_F(TestLibRBD, OpenAioFail)
 533 {
 534   rados_ioctx_t ioctx;
 535   ASSERT_EQ(0, rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx));
 536
 537   std::string name = get_temp_image_name();
 538   rbd_image_t image;
 539   rbd_completion_t open_comp;
 540   ASSERT_EQ(0, rbd_aio_create_completion(NULL, NULL, &open_comp));
 541   ASSERT_EQ(0, rbd_aio_open(ioctx, name.c_str(), &image, NULL, open_comp));
 542   ASSERT_EQ(0, rbd_aio_wait_for_complete(open_comp));
 543   ASSERT_EQ(1, rbd_aio_is_complete(open_comp));
 544   ASSERT_EQ(-ENOENT, rbd_aio_get_return_value(open_comp));
 545   rbd_aio_release(open_comp);
 546
 547   rados_ioctx_destroy(ioctx);
 548 }
 549
 550 TEST_F(TestLibRBD, OpenAioPP)
 551 {
 552   librados::IoCtx ioctx;
 553   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
 554
 555   librbd::RBD rbd;
 556   librbd::image_info_t info;
 557   librbd::Image image;
 558   int order = 0;
 559   std::string name = get_temp_image_name();
 560   uint64_t size = 2 << 20;
 561
 562   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
 563
 564   librbd::RBD::AioCompletion *open_comp =
 565       new librbd::RBD::AioCompletion(NULL, NULL);
 566   ASSERT_EQ(0, rbd.aio_open(ioctx, image, name.c_str(), NULL, open_comp));
 567   ASSERT_EQ(0, open_comp->wait_for_complete());
 568   ASSERT_EQ(1, open_comp->is_complete());
 569   ASSERT_EQ(0, open_comp->get_return_value());
 570   open_comp->release();
 571
 572   ASSERT_EQ(0, image.stat(info, sizeof(info)));
 573   ASSERT_EQ(info.size, size);
 574   ASSERT_EQ(info.order, order);
 575
 576   // reopen
 577   open_comp = new librbd::RBD::AioCompletion(NULL, NULL);
 578   ASSERT_EQ(0, rbd.aio_open(ioctx, image, name.c_str(), NULL, open_comp));
 579   ASSERT_EQ(0, open_comp->wait_for_complete());
 580   ASSERT_EQ(1, open_comp->is_complete());
 581   ASSERT_EQ(0, open_comp->get_return_value());
 582   open_comp->release();
 583
 584   // close
 585   librbd::RBD::AioCompletion *close_comp =
 586       new librbd::RBD::AioCompletion(NULL, NULL);
 587   ASSERT_EQ(0, image.aio_close(close_comp));
 588   ASSERT_EQ(0, close_comp->wait_for_complete());
 589   ASSERT_EQ(1, close_comp->is_complete());
 590   ASSERT_EQ(0, close_comp->get_return_value());
 591   close_comp->release();
 592
 593   // close closed image
 594   close_comp = new librbd::RBD::AioCompletion(NULL, NULL);
 595   ASSERT_EQ(-EINVAL, image.aio_close(close_comp));
 596   close_comp->release();
 597
 598   ioctx.close();
 599 }
 600
 601 TEST_F(TestLibRBD, OpenAioFailPP)
 602 {
 603   librados::IoCtx ioctx;
 604   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
 605
 606   {
 607     librbd::RBD rbd;
 608     librbd::Image image;
 609     std::string name = get_temp_image_name();
 610
 611     librbd::RBD::AioCompletion *open_comp =
 612       new librbd::RBD::AioCompletion(NULL, NULL);
 613     ASSERT_EQ(0, rbd.aio_open(ioctx, image, name.c_str(), NULL, open_comp));
 614     ASSERT_EQ(0, open_comp->wait_for_complete());
 615     ASSERT_EQ(1, open_comp->is_complete());
 616     ASSERT_EQ(-ENOENT, open_comp->get_return_value());
 617     open_comp->release();
 618   }
 619
 620   ioctx.close();
 621 }
 622
 623 TEST_F(TestLibRBD, ResizeAndStat)
 624 {
 625   rados_ioctx_t ioctx;
 626   rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx);
 627
 628   rbd_image_info_t info;
 629   rbd_image_t image;
 630   int order = 0;
 631   std::string name = get_temp_image_name();
 632   uint64_t size = 2 << 20;
 633
 634   ASSERT_EQ(0, create_image(ioctx, name.c_str(), size, &order));
 635   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
 636
 637   ASSERT_EQ(0, rbd_resize(image, size * 4));
 638   ASSERT_EQ(0, rbd_stat(image, &info, sizeof(info)));
 639   ASSERT_EQ(info.size, size * 4);
 640
 641   ASSERT_EQ(0, rbd_resize(image, size / 2));
 642   ASSERT_EQ(0, rbd_stat(image, &info, sizeof(info)));
 643   ASSERT_EQ(info.size, size / 2);
 644
 645   // downsizing without allowing shrink should fail
 646   // and image size should not change
 647   ASSERT_EQ(-EINVAL, rbd_resize2(image, size / 4, false, NULL, NULL));
 648   ASSERT_EQ(0, rbd_stat(image, &info, sizeof(info)));
 649   ASSERT_EQ(info.size, size / 2);
 650
 651   ASSERT_EQ(0, rbd_resize2(image, size / 4, true, NULL, NULL));
 652   ASSERT_EQ(0, rbd_stat(image, &info, sizeof(info)));
 653   ASSERT_EQ(info.size, size / 4);
 654
 655   ASSERT_PASSED(validate_object_map, image);
 656   ASSERT_EQ(0, rbd_close(image));
 657
 658   rados_ioctx_destroy(ioctx);
 659 }
 660
 661 TEST_F(TestLibRBD, ResizeAndStatPP)
 662 {
 663   librados::IoCtx ioctx;
 664   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
 665
 666   {
 667     librbd::RBD rbd;
 668     librbd::image_info_t info;
 669     librbd::Image image;
 670     int order = 0;
 671     std::string name = get_temp_image_name();
 672     uint64_t size = 2 << 20;
 673
 674     ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
 675     ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
 676
 677     ASSERT_EQ(0, image.resize(size * 4));
 678     ASSERT_EQ(0, image.stat(info, sizeof(info)));
 679     ASSERT_EQ(info.size, size * 4);
 680
 681     ASSERT_EQ(0, image.resize(size / 2));
 682     ASSERT_EQ(0, image.stat(info, sizeof(info)));
 683     ASSERT_EQ(info.size, size / 2);
 684     ASSERT_PASSED(validate_object_map, image);
 685   }
 686
 687   ioctx.close();
 688 }
 689
 690 TEST_F(TestLibRBD, UpdateWatchAndResize)
 691 {
 692   rados_ioctx_t ioctx;
 693   rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx);
 694
 695   rbd_image_t image;
 696   int order = 0;
 697   std::string name = get_temp_image_name();
 698   uint64_t size = 2 << 20;
 699   struct Watcher {
 700     rbd_image_t &m_image;
 701     mutex m_lock;
 702     condition_variable m_cond;
 703     size_t m_size = 0;
 704     static void cb(void *arg) {
 705       Watcher *watcher = static_cast<Watcher *>(arg);
 706       watcher->handle_notify();
 707     }
 708     Watcher(rbd_image_t &image) : m_image(image) {}
 709     void handle_notify() {
 710       rbd_image_info_t info;
 711       ASSERT_EQ(0, rbd_stat(m_image, &info, sizeof(info)));
 712       lock_guard<mutex> locker(m_lock);
 713       m_size = info.size;
 714       m_cond.notify_one();
 715     }
 716     void wait_for_size(size_t size) {
 717       unique_lock<mutex> locker(m_lock);
 718       ASSERT_TRUE(m_cond.wait_for(locker, seconds(5),
 719                                   [size, this] {
 720                                     return this->m_size == size;}));
 721     }
 722   } watcher(image);
 723   uint64_t handle;
 724
 725   ASSERT_EQ(0, create_image(ioctx, name.c_str(), size, &order));
 726   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
 727
 728   ASSERT_EQ(0, rbd_update_watch(image, &handle, Watcher::cb, &watcher));
 729
 730   ASSERT_EQ(0, rbd_resize(image, size * 4));
 731   watcher.wait_for_size(size * 4);
 732
 733   ASSERT_EQ(0, rbd_resize(image, size / 2));
 734   watcher.wait_for_size(size / 2);
 735
 736   ASSERT_EQ(0, rbd_update_unwatch(image, handle));
 737
 738   ASSERT_EQ(0, rbd_close(image));
 739   rados_ioctx_destroy(ioctx);
 740 }
 741
 742 TEST_F(TestLibRBD, UpdateWatchAndResizePP)
 743 {
 744   librados::IoCtx ioctx;
 745   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
 746
 747   {
 748     librbd::RBD rbd;
 749     librbd::Image image;
 750     int order = 0;
 751     std::string name = get_temp_image_name();
 752     uint64_t size = 2 << 20;
 753     struct Watcher : public librbd::UpdateWatchCtx {
 754       Watcher(librbd::Image &image) : m_image(image) {
 755       }
 756       void handle_notify() override {
 757         librbd::image_info_t info;
 758         ASSERT_EQ(0, m_image.stat(info, sizeof(info)));
 759         lock_guard<mutex> locker(m_lock);
 760         m_size = info.size;
 761         m_cond.notify_one();
 762       }
 763       void wait_for_size(size_t size) {
 764         unique_lock<mutex> locker(m_lock);
 765         ASSERT_TRUE(m_cond.wait_for(locker, seconds(5),
 766                                     [size, this] {
 767                                       return this->m_size == size;}));
 768       }
 769       librbd::Image &m_image;
 770       mutex m_lock;
 771       condition_variable m_cond;
 772       size_t m_size = 0;
 773     } watcher(image);
 774     uint64_t handle;
 775
 776     ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
 777     ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
 778
 779     ASSERT_EQ(0, image.update_watch(&watcher, &handle));
 780
 781     ASSERT_EQ(0, image.resize(size * 4));
 782     watcher.wait_for_size(size * 4);
 783
 784     ASSERT_EQ(0, image.resize(size / 2));
 785     watcher.wait_for_size(size / 2);
 786
 787     ASSERT_EQ(0, image.update_unwatch(handle));
 788   }
 789
 790   ioctx.close();
 791 }
 792
 793 int test_ls(rados_ioctx_t io_ctx, size_t num_expected, ...)
 794 {
 795   int num_images, i;
 796   char *names, *cur_name;
 797   va_list ap;
 798   size_t max_size = 1024;
 799
 800   names = (char *) malloc(sizeof(char) * 1024);
 801   int len = rbd_list(io_ctx, names, &max_size);
 802
 803   std::set<std::string> image_names;
 804   for (i = 0, num_images = 0, cur_name = names; cur_name < names + len; i++) {
 805     printf("image: %s\n", cur_name);
 806     image_names.insert(cur_name);
 807     cur_name += strlen(cur_name) + 1;
 808     num_images++;
 809   }
 810   free(names);
 811
 812   va_start(ap, num_expected);
 813   for (i = num_expected; i > 0; i--) {
 814     char *expected = va_arg(ap, char *);
 815     printf("expected = %s\n", expected);
 816     std::set<std::string>::iterator it = image_names.find(expected);
 817     if (it != image_names.end()) {
 818       printf("found %s\n", expected);
 819       image_names.erase(it);
 820       printf("erased %s\n", expected);
 821     } else {
 822       ADD_FAILURE() << "Unable to find image " << expected;
 823       va_end(ap);
 824       return -ENOENT;
 825     }
 826   }
 827   va_end(ap);
 828
 829   if (!image_names.empty()) {
 830     ADD_FAILURE() << "Unexpected images discovered";
 831     return -EINVAL;
 832   }
 833   return num_images;
 834 }
 835
 836 TEST_F(TestLibRBD, TestCreateLsDelete)
 837 {
 838   rados_ioctx_t ioctx;
 839   rados_ioctx_create(_cluster, create_pool(true).c_str(), &ioctx);
 840
 841   int order = 0;
 842   std::string name = get_temp_image_name();
 843   std::string name2 = get_temp_image_name();
 844   uint64_t size = 2 << 20;
 845
 846   ASSERT_EQ(0, create_image(ioctx, name.c_str(), size, &order));
 847   ASSERT_EQ(1, test_ls(ioctx, 1, name.c_str()));
 848   ASSERT_EQ(0, create_image(ioctx, name2.c_str(), size, &order));
 849   ASSERT_EQ(2, test_ls(ioctx, 2, name.c_str(), name2.c_str()));
 850   ASSERT_EQ(0, rbd_remove(ioctx, name.c_str()));
 851   ASSERT_EQ(1, test_ls(ioctx, 1, name2.c_str()));
 852
 853   ASSERT_EQ(-ENOENT, rbd_remove(ioctx, name.c_str()));
 854
 855   rados_ioctx_destroy(ioctx);
 856 }
 857
 858 int test_ls_pp(librbd::RBD& rbd, librados::IoCtx& io_ctx, size_t num_expected, ...)
 859 {
 860   int r;
 861   size_t i;
 862   va_list ap;
 863   vector<string> names;
 864   r = rbd.list(io_ctx, names);
 865   if (r == -ENOENT)
 866     r = 0;
 867   EXPECT_TRUE(r >= 0);
 868   cout << "num images is: " << names.size() << std::endl
 869             << "expected: " << num_expected << std::endl;
 870   int num = names.size();
 871
 872   for (i = 0; i < names.size(); i++) {
 873     cout << "image: " << names[i] << std::endl;
 874   }
 875
 876   va_start(ap, num_expected);
 877   for (i = num_expected; i > 0; i--) {
 878     char *expected = va_arg(ap, char *);
 879     cout << "expected = " << expected << std::endl;
 880     vector<string>::iterator listed_name = find(names.begin(), names.end(), string(expected));
 881     if (listed_name == names.end()) {
 882       ADD_FAILURE() << "Unable to find image " << expected;
 883       va_end(ap);
 884       return -ENOENT;
 885     }
 886     names.erase(listed_name);
 887   }
 888   va_end(ap);
 889
 890   if (!names.empty()) {
 891     ADD_FAILURE() << "Unexpected images discovered";
 892     return -EINVAL;
 893   }
 894   return num;
 895 }
 896
 897 TEST_F(TestLibRBD, TestCreateLsDeletePP)
 898 {
 899   librados::IoCtx ioctx;
 900   ASSERT_EQ(0, _rados.ioctx_create(create_pool(true).c_str(), ioctx));
 901
 902   {
 903     librbd::RBD rbd;
 904     librbd::Image image;
 905     int order = 0;
 906     std::string name = get_temp_image_name();
 907     std::string name2 = get_temp_image_name();
 908     uint64_t size = 2 << 20;
 909
 910     ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
 911     ASSERT_EQ(1, test_ls_pp(rbd, ioctx, 1, name.c_str()));
 912     ASSERT_EQ(0, create_image_pp(rbd, ioctx, name2.c_str(), size, &order));
 913     ASSERT_EQ(2, test_ls_pp(rbd, ioctx, 2, name.c_str(), name2.c_str()));
 914     ASSERT_EQ(0, rbd.remove(ioctx, name.c_str()));
 915     ASSERT_EQ(1, test_ls_pp(rbd, ioctx, 1, name2.c_str()));
 916   }
 917
 918   ioctx.close();
 919 }
 920
 921
 922 static int print_progress_percent(uint64_t offset, uint64_t src_size,
 923                                      void *data)
 924 {
 925   float percent = ((float)offset * 100) / src_size;
 926   printf("%3.2f%% done\n", percent);
 927   return 0;
 928 }
 929
 930 TEST_F(TestLibRBD, TestCopy)
 931 {
 932   rados_ioctx_t ioctx;
 933   rados_ioctx_create(_cluster, create_pool(true).c_str(), &ioctx);
 934
 935   rbd_image_t image;
 936   int order = 0;
 937   std::string name = get_temp_image_name();
 938   std::string name2 = get_temp_image_name();
 939   std::string name3 = get_temp_image_name();
 940
 941   uint64_t size = 2 << 20;
 942
 943   ASSERT_EQ(0, create_image(ioctx, name.c_str(), size, &order));
 944   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
 945   ASSERT_EQ(1, test_ls(ioctx, 1, name.c_str()));
 946   ASSERT_EQ(0, rbd_copy(image, ioctx, name2.c_str()));
 947   ASSERT_EQ(2, test_ls(ioctx, 2, name.c_str(), name2.c_str()));
 948   ASSERT_EQ(0, rbd_copy_with_progress(image, ioctx, name3.c_str(),
 949                                       print_progress_percent, NULL));
 950   ASSERT_EQ(3, test_ls(ioctx, 3, name.c_str(), name2.c_str(), name3.c_str()));
 951
 952   ASSERT_EQ(0, rbd_close(image));
 953   rados_ioctx_destroy(ioctx);
 954 }
 955
 956 class PrintProgress : public librbd::ProgressContext
 957 {
 958 public:
 959   int update_progress(uint64_t offset, uint64_t src_size) override
 960   {
 961     float percent = ((float)offset * 100) / src_size;
 962     printf("%3.2f%% done\n", percent);
 963     return 0;
 964   }
 965 };
 966
 967 TEST_F(TestLibRBD, TestCopyPP)
 968 {
 969   librados::IoCtx ioctx;
 970   ASSERT_EQ(0, _rados.ioctx_create(create_pool(true).c_str(), ioctx));
 971
 972   {
 973     librbd::RBD rbd;
 974     librbd::Image image;
 975     int order = 0;
 976     std::string name = get_temp_image_name();
 977     std::string name2 = get_temp_image_name();
 978     std::string name3 = get_temp_image_name();
 979     uint64_t size = 2 << 20;
 980     PrintProgress pp;
 981
 982     ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
 983     ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
 984     ASSERT_EQ(1, test_ls_pp(rbd, ioctx, 1, name.c_str()));
 985     ASSERT_EQ(0, image.copy(ioctx, name2.c_str()));
 986     ASSERT_EQ(2, test_ls_pp(rbd, ioctx, 2, name.c_str(), name2.c_str()));
 987     ASSERT_EQ(0, image.copy_with_progress(ioctx, name3.c_str(), pp));
 988     ASSERT_EQ(3, test_ls_pp(rbd, ioctx, 3, name.c_str(), name2.c_str(),
 989                             name3.c_str()));
 990   }
 991
 992   ioctx.close();
 993 }
 994
 995 int test_ls_snaps(rbd_image_t image, int num_expected, ...)
 996 {
 997   int num_snaps, i, j, max_size = 10;
 998   va_list ap;
 999   rbd_snap_info_t snaps[max_size];
1000   num_snaps = rbd_snap_list(image, snaps, &max_size);
1001   printf("num snaps is: %d\nexpected: %d\n", num_snaps, num_expected);
1002
1003   for (i = 0; i < num_snaps; i++) {
1004     printf("snap: %s\n", snaps[i].name);
1005   }
1006
1007   va_start(ap, num_expected);
1008   for (i = num_expected; i > 0; i--) {
1009     char *expected = va_arg(ap, char *);
1010     uint64_t expected_size = va_arg(ap, uint64_t);
1011     bool found = false;
1012     for (j = 0; j < num_snaps; j++) {
1013       if (snaps[j].name == NULL)
1014         continue;
1015       if (strcmp(snaps[j].name, expected) == 0) {
1016         printf("found %s with size %llu\n", snaps[j].name, (unsigned long long) snaps[j].size);
1017         EXPECT_EQ(expected_size, snaps[j].size);
1018         free((void *) snaps[j].name);
1019         snaps[j].name = NULL;
1020         found = true;
1021         break;
1022       }
1023     }
1024     EXPECT_TRUE(found);
1025   }
1026   va_end(ap);
1027
1028   for (i = 0; i < num_snaps; i++) {
1029     EXPECT_EQ((const char *)0, snaps[i].name);
1030   }
1031
1032   return num_snaps;
1033 }
1034
1035 TEST_F(TestLibRBD, TestCreateLsDeleteSnap)
1036 {
1037   rados_ioctx_t ioctx;
1038   rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx);
1039
1040   rbd_image_t image;
1041   int order = 0;
1042   std::string name = get_temp_image_name();
1043   uint64_t size = 2 << 20;
1044   uint64_t size2 = 4 << 20;
1045
1046   ASSERT_EQ(0, create_image(ioctx, name.c_str(), size, &order));
1047   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
1048
1049   ASSERT_EQ(0, rbd_snap_create(image, "snap1"));
1050   ASSERT_EQ(1, test_ls_snaps(image, 1, "snap1", size));
1051   ASSERT_EQ(0, rbd_resize(image, size2));
1052   ASSERT_EQ(0, rbd_snap_create(image, "snap2"));
1053   ASSERT_EQ(2, test_ls_snaps(image, 2, "snap1", size, "snap2", size2));
1054   ASSERT_EQ(0, rbd_snap_remove(image, "snap1"));
1055   ASSERT_EQ(1, test_ls_snaps(image, 1, "snap2", size2));
1056   ASSERT_EQ(0, rbd_snap_remove(image, "snap2"));
1057   ASSERT_EQ(0, test_ls_snaps(image, 0));
1058
1059   ASSERT_EQ(0, rbd_close(image));
1060
1061   rados_ioctx_destroy(ioctx);
1062 }
1063
1064 int test_get_snapshot_timestamp(rbd_image_t image, uint64_t snap_id)
1065 {
1066   struct timespec timestamp;
1067   EXPECT_EQ(0, rbd_snap_get_timestamp(image, snap_id, &timestamp));
1068   EXPECT_LT(0, timestamp.tv_sec);
1069   return 0;
1070 }
1071
1072 TEST_F(TestLibRBD, TestGetSnapShotTimeStamp)
1073 {
1074   REQUIRE_FORMAT_V2();
1075
1076   rados_ioctx_t ioctx;
1077   rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx);
1078
1079   rbd_image_t image;
1080   int order = 0;
1081   std::string name = get_temp_image_name();
1082   uint64_t size = 2 << 20;
1083   int num_snaps, max_size = 10;
1084   rbd_snap_info_t snaps[max_size];
1085
1086   ASSERT_EQ(0, create_image(ioctx, name.c_str(), size, &order));
1087   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
1088
1089   ASSERT_EQ(0, rbd_snap_create(image, "snap1"));
1090   num_snaps = rbd_snap_list(image, snaps, &max_size);
1091   ASSERT_EQ(1, num_snaps);
1092   ASSERT_EQ(0, test_get_snapshot_timestamp(image, snaps[0].id));
1093   free((void *)snaps[0].name);
1094
1095   ASSERT_EQ(0, rbd_snap_create(image, "snap2"));
1096   num_snaps = rbd_snap_list(image, snaps, &max_size);
1097   ASSERT_EQ(2, num_snaps);
1098   ASSERT_EQ(0, test_get_snapshot_timestamp(image, snaps[0].id));
1099   ASSERT_EQ(0, test_get_snapshot_timestamp(image, snaps[1].id));
1100   free((void *)snaps[0].name);
1101   free((void *)snaps[1].name);
1102
1103   ASSERT_EQ(0, rbd_close(image));
1104
1105   rados_ioctx_destroy(ioctx);
1106 }
1107
1108
1109 int test_ls_snaps(librbd::Image& image, size_t num_expected, ...)
1110 {
1111   int r;
1112   size_t i, j;
1113   va_list ap;
1114   vector<librbd::snap_info_t> snaps;
1115   r = image.snap_list(snaps);
1116   EXPECT_TRUE(r >= 0);
1117   cout << "num snaps is: " << snaps.size() << std::endl
1118             << "expected: " << num_expected << std::endl;
1119
1120   for (i = 0; i < snaps.size(); i++) {
1121     cout << "snap: " << snaps[i].name << std::endl;
1122   }
1123
1124   va_start(ap, num_expected);
1125   for (i = num_expected; i > 0; i--) {
1126     char *expected = va_arg(ap, char *);
1127     uint64_t expected_size = va_arg(ap, uint64_t);
1128     int found = 0;
1129     for (j = 0; j < snaps.size(); j++) {
1130       if (snaps[j].name == "")
1131         continue;
1132       if (strcmp(snaps[j].name.c_str(), expected) == 0) {
1133         cout << "found " << snaps[j].name << " with size " << snaps[j].size
1134              << std::endl;
1135         EXPECT_EQ(expected_size, snaps[j].size);
1136         snaps[j].name = "";
1137         found = 1;
1138         break;
1139       }
1140     }
1141     EXPECT_TRUE(found);
1142   }
1143   va_end(ap);
1144
1145   for (i = 0; i < snaps.size(); i++) {
1146     EXPECT_EQ("", snaps[i].name);
1147   }
1148
1149   return snaps.size();
1150 }
1151
1152 TEST_F(TestLibRBD, TestCreateLsDeleteSnapPP)
1153 {
1154   librados::IoCtx ioctx;
1155   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
1156
1157   {
1158     librbd::RBD rbd;
1159     librbd::Image image;
1160     int order = 0;
1161     std::string name = get_temp_image_name();
1162     uint64_t size = 2 << 20;
1163     uint64_t size2 = 4 << 20;
1164
1165     ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
1166     ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
1167
1168     bool exists;
1169     ASSERT_EQ(0, image.snap_exists2("snap1", &exists));
1170     ASSERT_FALSE(exists);
1171     ASSERT_EQ(0, image.snap_create("snap1"));
1172     ASSERT_EQ(0, image.snap_exists2("snap1", &exists));
1173     ASSERT_TRUE(exists);
1174     ASSERT_EQ(1, test_ls_snaps(image, 1, "snap1", size));
1175     ASSERT_EQ(0, image.resize(size2));
1176     ASSERT_EQ(0, image.snap_exists2("snap2", &exists));
1177     ASSERT_FALSE(exists);
1178     ASSERT_EQ(0, image.snap_create("snap2"));
1179     ASSERT_EQ(0, image.snap_exists2("snap2", &exists));
1180     ASSERT_TRUE(exists);
1181     ASSERT_EQ(2, test_ls_snaps(image, 2, "snap1", size, "snap2", size2));
1182     ASSERT_EQ(0, image.snap_remove("snap1"));
1183     ASSERT_EQ(0, image.snap_exists2("snap1", &exists));
1184     ASSERT_FALSE(exists);
1185     ASSERT_EQ(1, test_ls_snaps(image, 1, "snap2", size2));
1186     ASSERT_EQ(0, image.snap_remove("snap2"));
1187     ASSERT_EQ(0, image.snap_exists2("snap2", &exists));
1188     ASSERT_FALSE(exists);
1189     ASSERT_EQ(0, test_ls_snaps(image, 0));
1190   }
1191
1192   ioctx.close();
1193 }
1194
1195 TEST_F(TestLibRBD, TestCreateLsRenameSnapPP)
1196 {
1197   librados::IoCtx ioctx;
1198   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
1199
1200   {
1201     librbd::RBD rbd;
1202     librbd::Image image;
1203     int order = 0;
1204     std::string name = get_temp_image_name();
1205     uint64_t size = 2 << 20;
1206     uint64_t size2 = 4 << 20;
1207
1208     ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
1209     ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
1210
1211     bool exists;
1212     ASSERT_EQ(0, image.snap_exists2("snap1", &exists));
1213     ASSERT_FALSE(exists);
1214     ASSERT_EQ(0, image.snap_create("snap1"));
1215     ASSERT_EQ(0, image.snap_exists2("snap1", &exists));
1216     ASSERT_TRUE(exists);
1217     ASSERT_EQ(1, test_ls_snaps(image, 1, "snap1", size));
1218     ASSERT_EQ(0, image.resize(size2));
1219     ASSERT_EQ(0, image.snap_exists2("snap2", &exists));
1220     ASSERT_FALSE(exists);
1221     ASSERT_EQ(0, image.snap_create("snap2"));
1222     ASSERT_EQ(0, image.snap_exists2("snap2", &exists));
1223     ASSERT_TRUE(exists);
1224     ASSERT_EQ(2, test_ls_snaps(image, 2, "snap1", size, "snap2", size2));
1225     ASSERT_EQ(0, image.snap_rename("snap1","snap1-rename"));
1226     ASSERT_EQ(2, test_ls_snaps(image, 2, "snap1-rename", size, "snap2", size2));
1227     ASSERT_EQ(0, image.snap_exists2("snap1", &exists));
1228     ASSERT_FALSE(exists);
1229     ASSERT_EQ(0, image.snap_exists2("snap1-rename", &exists));
1230     ASSERT_TRUE(exists);
1231     ASSERT_EQ(0, image.snap_remove("snap1-rename"));
1232     ASSERT_EQ(0, image.snap_rename("snap2","snap2-rename"));
1233     ASSERT_EQ(1, test_ls_snaps(image, 1, "snap2-rename", size2));
1234     ASSERT_EQ(0, image.snap_exists2("snap2", &exists));
1235     ASSERT_FALSE(exists);
1236     ASSERT_EQ(0, image.snap_exists2("snap2-rename", &exists));
1237     ASSERT_TRUE(exists);
1238     ASSERT_EQ(0, image.snap_remove("snap2-rename"));
1239     ASSERT_EQ(0, test_ls_snaps(image, 0));
1240   }
1241
1242   ioctx.close();
1243 }
1244
1245 void simple_write_cb(rbd_completion_t cb, void *arg)
1246 {
1247   printf("write completion cb called!\n");
1248 }
1249
1250 void simple_read_cb(rbd_completion_t cb, void *arg)
1251 {
1252   printf("read completion cb called!\n");
1253 }
1254
1255 void aio_write_test_data_and_poll(rbd_image_t image, int fd, const char *test_data,
1256                                   uint64_t off, size_t len, uint32_t iohint, bool *passed)
1257 {
1258   rbd_completion_t comp;
1259   uint64_t data = 0x123;
1260   rbd_aio_create_completion((void*)&data, (rbd_callback_t) simple_write_cb, &comp);
1261   printf("created completion\n");
1262   printf("started write\n");
1263   if (iohint)
1264     rbd_aio_write2(image, off, len, test_data, comp, iohint);
1265   else
1266     rbd_aio_write(image, off, len, test_data, comp);
1267
1268   struct pollfd pfd;
1269   pfd.fd = fd;
1270   pfd.events = POLLIN;
1271
1272   ASSERT_EQ(1, poll(&pfd, 1, -1));
1273   ASSERT_TRUE(pfd.revents & POLLIN);
1274
1275   rbd_completion_t comps[1];
1276   ASSERT_EQ(1, rbd_poll_io_events(image, comps, 1));
1277   uint64_t count;
1278   ASSERT_EQ(static_cast<ssize_t>(sizeof(count)),
1279             read(fd, &count, sizeof(count)));
1280   int r = rbd_aio_get_return_value(comps[0]);
1281   ASSERT_TRUE(rbd_aio_is_complete(comps[0]));
1282   ASSERT_TRUE(*(uint64_t*)rbd_aio_get_arg(comps[0]) == data);
1283   printf("return value is: %d\n", r);
1284   ASSERT_EQ(0, r);
1285   printf("finished write\n");
1286   rbd_aio_release(comps[0]);
1287   *passed = true;
1288 }
1289
1290 void aio_write_test_data(rbd_image_t image, const char *test_data, uint64_t off, size_t len, uint32_t iohint, bool *passed)
1291 {
1292   rbd_completion_t comp;
1293   rbd_aio_create_completion(NULL, (rbd_callback_t) simple_write_cb, &comp);
1294   printf("created completion\n");
1295   if (iohint)
1296     rbd_aio_write2(image, off, len, test_data, comp, iohint);
1297   else
1298     rbd_aio_write(image, off, len, test_data, comp);
1299   printf("started write\n");
1300   rbd_aio_wait_for_complete(comp);
1301   int r = rbd_aio_get_return_value(comp);
1302   printf("return value is: %d\n", r);
1303   ASSERT_EQ(0, r);
1304   printf("finished write\n");
1305   rbd_aio_release(comp);
1306   *passed = true;
1307 }
1308
1309 void write_test_data(rbd_image_t image, const char *test_data, uint64_t off, size_t len, uint32_t iohint, bool *passed)
1310 {
1311   ssize_t written;
1312   if (iohint)
1313     written = rbd_write2(image, off, len, test_data, iohint);
1314   else
1315     written = rbd_write(image, off, len, test_data);
1316   printf("wrote: %d\n", (int) written);
1317   ASSERT_EQ(len, static_cast<size_t>(written));
1318   *passed = true;
1319 }
1320
1321 void aio_discard_test_data(rbd_image_t image, uint64_t off, uint64_t len, bool *passed)
1322 {
1323   rbd_completion_t comp;
1324   rbd_aio_create_completion(NULL, (rbd_callback_t) simple_write_cb, &comp);
1325   rbd_aio_discard(image, off, len, comp);
1326   rbd_aio_wait_for_complete(comp);
1327   int r = rbd_aio_get_return_value(comp);
1328   ASSERT_EQ(0, r);
1329   printf("aio discard: %d~%d = %d\n", (int)off, (int)len, (int)r);
1330   rbd_aio_release(comp);
1331   *passed = true;
1332 }
1333
1334 void discard_test_data(rbd_image_t image, uint64_t off, size_t len, bool *passed)
1335 {
1336   ssize_t written;
1337   written = rbd_discard(image, off, len);
1338   printf("discard: %d~%d = %d\n", (int)off, (int)len, (int)written);
1339   ASSERT_EQ(len, static_cast<size_t>(written));
1340   *passed = true;
1341 }
1342
1343 void aio_read_test_data_and_poll(rbd_image_t image, int fd, const char *expected,
1344                                  uint64_t off, size_t len, uint32_t iohint, bool *passed)
1345 {
1346   rbd_completion_t comp;
1347   char *result = (char *)malloc(len + 1);
1348
1349   ASSERT_NE(static_cast<char *>(NULL), result);
1350   rbd_aio_create_completion(NULL, (rbd_callback_t) simple_read_cb, &comp);
1351   printf("created completion\n");
1352   printf("started read\n");
1353   if (iohint)
1354     rbd_aio_read2(image, off, len, result, comp, iohint);
1355   else
1356     rbd_aio_read(image, off, len, result, comp);
1357
1358   struct pollfd pfd;
1359   pfd.fd = fd;
1360   pfd.events = POLLIN;
1361
1362   ASSERT_EQ(1, poll(&pfd, 1, -1));
1363   ASSERT_TRUE(pfd.revents & POLLIN);
1364
1365   rbd_completion_t comps[1];
1366   ASSERT_EQ(1, rbd_poll_io_events(image, comps, 1));
1367   uint64_t count;
1368   ASSERT_EQ(static_cast<ssize_t>(sizeof(count)),
1369             read(fd, &count, sizeof(count)));
1370
1371   int r = rbd_aio_get_return_value(comps[0]);
1372   ASSERT_TRUE(rbd_aio_is_complete(comps[0]));
1373   printf("return value is: %d\n", r);
1374   ASSERT_EQ(len, static_cast<size_t>(r));
1375   rbd_aio_release(comps[0]);
1376   if (memcmp(result, expected, len)) {
1377     printf("read: %s\nexpected: %s\n", result, expected);
1378     ASSERT_EQ(0, memcmp(result, expected, len));
1379   }
1380   free(result);
1381   *passed = true;
1382 }
1383
1384 void aio_read_test_data(rbd_image_t image, const char *expected, uint64_t off, size_t len, uint32_t iohint, bool *passed)
1385 {
1386   rbd_completion_t comp;
1387   char *result = (char *)malloc(len + 1);
1388
1389   ASSERT_NE(static_cast<char *>(NULL), result);
1390   rbd_aio_create_completion(NULL, (rbd_callback_t) simple_read_cb, &comp);
1391   printf("created completion\n");
1392   if (iohint)
1393     rbd_aio_read2(image, off, len, result, comp, iohint);
1394   else
1395     rbd_aio_read(image, off, len, result, comp);
1396   printf("started read\n");
1397   rbd_aio_wait_for_complete(comp);
1398   int r = rbd_aio_get_return_value(comp);
1399   printf("return value is: %d\n", r);
1400   ASSERT_EQ(len, static_cast<size_t>(r));
1401   rbd_aio_release(comp);
1402   if (memcmp(result, expected, len)) {
1403     printf("read: %s\nexpected: %s\n", result, expected);
1404     ASSERT_EQ(0, memcmp(result, expected, len));
1405   }
1406   free(result);
1407   *passed = true;
1408 }
1409
1410 void read_test_data(rbd_image_t image, const char *expected, uint64_t off, size_t len, uint32_t iohint, bool *passed)
1411 {
1412   ssize_t read;
1413   char *result = (char *)malloc(len + 1);
1414
1415   ASSERT_NE(static_cast<char *>(NULL), result);
1416   if (iohint)
1417     read = rbd_read2(image, off, len, result, iohint);
1418   else
1419     read = rbd_read(image, off, len, result);
1420   printf("read: %d\n", (int) read);
1421   ASSERT_EQ(len, static_cast<size_t>(read));
1422   result[len] = '\0';
1423   if (memcmp(result, expected, len)) {
1424     printf("read: %s\nexpected: %s\n", result, expected);
1425     ASSERT_EQ(0, memcmp(result, expected, len));
1426   }
1427   free(result);
1428   *passed = true;
1429 }
1430
1431 void aio_writesame_test_data(rbd_image_t image, const char *test_data, uint64_t off, uint64_t len,
1432                              uint64_t data_len, uint32_t iohint, bool *passed)
1433 {
1434   rbd_completion_t comp;
1435   rbd_aio_create_completion(NULL, (rbd_callback_t) simple_write_cb, &comp);
1436   printf("created completion\n");
1437   int r;
1438   r = rbd_aio_writesame(image, off, len, test_data, data_len, comp, iohint);
1439   printf("started writesame\n");
1440   if (len % data_len) {
1441     ASSERT_EQ(-EINVAL, r);
1442     printf("expected fail, finished writesame\n");
1443     rbd_aio_release(comp);
1444     *passed = true;
1445     return;
1446   }
1447
1448   rbd_aio_wait_for_complete(comp);
1449   r = rbd_aio_get_return_value(comp);
1450   printf("return value is: %d\n", r);
1451   ASSERT_EQ(0, r);
1452   printf("finished writesame\n");
1453   rbd_aio_release(comp);
1454
1455   //verify data
1456   printf("to verify the data\n");
1457   ssize_t read;
1458   char *result = (char *)malloc(data_len+ 1);
1459   ASSERT_NE(static_cast<char *>(NULL), result);
1460   uint64_t left = len;
1461   while (left > 0) {
1462     read = rbd_read(image, off, data_len, result);
1463     ASSERT_EQ(data_len, static_cast<size_t>(read));
1464     result[data_len] = '\0';
1465     if (memcmp(result, test_data, data_len)) {
1466       printf("read: %d ~ %d\n", (int) off, (int) read);
1467       printf("read: %s\nexpected: %s\n", result, test_data);
1468       ASSERT_EQ(0, memcmp(result, test_data, data_len));
1469     }
1470     off += data_len;
1471     left -= data_len;
1472   }
1473   ASSERT_EQ(0U, left);
1474   free(result);
1475   printf("verified\n");
1476
1477   *passed = true;
1478 }
1479
1480 void writesame_test_data(rbd_image_t image, const char *test_data, uint64_t off, uint64_t len,
1481                          uint64_t data_len, uint32_t iohint, bool *passed)
1482 {
1483   ssize_t written;
1484   written = rbd_writesame(image, off, len, test_data, data_len, iohint);
1485   if (len % data_len) {
1486     ASSERT_EQ(-EINVAL, written);
1487     printf("expected fail, finished writesame\n");
1488     *passed = true;
1489     return;
1490   }
1491   ASSERT_EQ(len, static_cast<size_t>(written));
1492   printf("wrote: %d\n", (int) written);
1493
1494   //verify data
1495   printf("to verify the data\n");
1496   ssize_t read;
1497   char *result = (char *)malloc(data_len+ 1);
1498   ASSERT_NE(static_cast<char *>(NULL), result);
1499   uint64_t left = len;
1500   while (left > 0) {
1501     read = rbd_read(image, off, data_len, result);
1502     ASSERT_EQ(data_len, static_cast<size_t>(read));
1503     result[data_len] = '\0';
1504     if (memcmp(result, test_data, data_len)) {
1505       printf("read: %d ~ %d\n", (int) off, (int) read);
1506       printf("read: %s\nexpected: %s\n", result, test_data);
1507       ASSERT_EQ(0, memcmp(result, test_data, data_len));
1508     }
1509     off += data_len;
1510     left -= data_len;
1511   }
1512   ASSERT_EQ(0U, left);
1513   free(result);
1514   printf("verified\n");
1515
1516   *passed = true;
1517 }
1518
1519 void aio_compare_and_write_test_data(rbd_image_t image, const char *cmp_data,
1520                                      const char *test_data, uint64_t off,
1521                                      size_t len, uint32_t iohint, bool *passed)
1522 {
1523   rbd_completion_t comp;
1524   rbd_aio_create_completion(NULL, (rbd_callback_t) simple_write_cb, &comp);
1525   printf("created completion\n");
1526
1527   uint64_t mismatch_offset;
1528   rbd_aio_compare_and_write(image, off, len, cmp_data, test_data, comp, &mismatch_offset, iohint);
1529   printf("started aio compare and write\n");
1530   rbd_aio_wait_for_complete(comp);
1531   int r = rbd_aio_get_return_value(comp);
1532   printf("return value is: %d\n", r);
1533   ASSERT_EQ(0, r);
1534   printf("finished aio compare and write\n");
1535   rbd_aio_release(comp);
1536   *passed = true;
1537 }
1538
1539 void compare_and_write_test_data(rbd_image_t image, const char *cmp_data,
1540                                  const char *test_data, uint64_t off, size_t len,
1541                                  uint64_t *mismatch_off, uint32_t iohint, bool *passed)
1542 {
1543   printf("start compare and write\n");
1544   ssize_t written;
1545   written = rbd_compare_and_write(image, off, len, cmp_data, test_data, mismatch_off, iohint);
1546   printf("compare and  wrote: %d\n", (int) written);
1547   ASSERT_EQ(len, static_cast<size_t>(written));
1548   *passed = true;
1549 }
1550
1551
1552 TEST_F(TestLibRBD, TestIO)
1553 {
1554   rados_ioctx_t ioctx;
1555   rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx);
1556
1557   bool skip_discard = is_skip_partial_discard_enabled();
1558
1559   rbd_image_t image;
1560   int order = 0;
1561   std::string name = get_temp_image_name();
1562   uint64_t size = 2 << 20;
1563
1564   ASSERT_EQ(0, create_image(ioctx, name.c_str(), size, &order));
1565   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
1566
1567   char test_data[TEST_IO_SIZE + 1];
1568   char zero_data[TEST_IO_SIZE + 1];
1569   char mismatch_data[TEST_IO_SIZE + 1];
1570   int i;
1571   uint64_t mismatch_offset;
1572
1573   for (i = 0; i < TEST_IO_SIZE; ++i) {
1574     test_data[i] = (char) (rand() % (126 - 33) + 33);
1575   }
1576   test_data[TEST_IO_SIZE] = '\0';
1577   memset(zero_data, 0, sizeof(zero_data));
1578   memset(mismatch_data, 9, sizeof(mismatch_data));
1579
1580   for (i = 0; i < 5; ++i)
1581     ASSERT_PASSED(write_test_data, image, test_data, TEST_IO_SIZE * i, TEST_IO_SIZE, 0);
1582
1583   for (i = 5; i < 10; ++i)
1584     ASSERT_PASSED(aio_write_test_data, image, test_data, TEST_IO_SIZE * i, TEST_IO_SIZE, 0);
1585
1586   for (i = 0; i < 5; ++i)
1587     ASSERT_PASSED(compare_and_write_test_data, image, test_data, test_data, TEST_IO_SIZE * i, TEST_IO_SIZE, &mismatch_offset, 0);
1588
1589   for (i = 5; i < 10; ++i)
1590     ASSERT_PASSED(aio_compare_and_write_test_data, image, test_data, test_data, TEST_IO_SIZE * i, TEST_IO_SIZE, 0);
1591
1592   for (i = 0; i < 5; ++i)
1593     ASSERT_PASSED(read_test_data, image, test_data, TEST_IO_SIZE * i, TEST_IO_SIZE, 0);
1594
1595   for (i = 5; i < 10; ++i)
1596     ASSERT_PASSED(aio_read_test_data, image, test_data, TEST_IO_SIZE * i, TEST_IO_SIZE, 0);
1597
1598   // discard 2nd, 4th sections.
1599   ASSERT_PASSED(discard_test_data, image, TEST_IO_SIZE, TEST_IO_SIZE);
1600   ASSERT_PASSED(aio_discard_test_data, image, TEST_IO_SIZE*3, TEST_IO_SIZE);
1601
1602   ASSERT_PASSED(read_test_data, image, test_data,  0, TEST_IO_SIZE, 0);
1603   ASSERT_PASSED(read_test_data, image, skip_discard ? test_data : zero_data,
1604                 TEST_IO_SIZE, TEST_IO_SIZE, 0);
1605   ASSERT_PASSED(read_test_data, image, test_data,  TEST_IO_SIZE*2, TEST_IO_SIZE, 0);
1606   ASSERT_PASSED(read_test_data, image, skip_discard ? test_data : zero_data,
1607                 TEST_IO_SIZE*3, TEST_IO_SIZE, 0);
1608   ASSERT_PASSED(read_test_data, image, test_data,  TEST_IO_SIZE*4, TEST_IO_SIZE, 0);
1609
1610   for (i = 0; i < 15; ++i) {
1611     if (i % 3 == 2) {
1612       ASSERT_PASSED(writesame_test_data, image, test_data, TEST_IO_SIZE * i, TEST_IO_SIZE * i * 32 + i, TEST_IO_SIZE, 0);
1613       ASSERT_PASSED(writesame_test_data, image, zero_data, TEST_IO_SIZE * i, TEST_IO_SIZE * i * 32 + i, TEST_IO_SIZE, 0);
1614     } else if (i % 3 == 1) {
1615       ASSERT_PASSED(writesame_test_data, image, test_data, TEST_IO_SIZE + i, TEST_IO_SIZE * i * 32, TEST_IO_SIZE, 0);
1616       ASSERT_PASSED(writesame_test_data, image, zero_data, TEST_IO_SIZE + i, TEST_IO_SIZE * i * 32, TEST_IO_SIZE, 0);
1617     } else {
1618       ASSERT_PASSED(writesame_test_data, image, test_data, TEST_IO_SIZE * i, TEST_IO_SIZE * i * 32, TEST_IO_SIZE, 0);
1619       ASSERT_PASSED(writesame_test_data, image, zero_data, TEST_IO_SIZE * i, TEST_IO_SIZE * i * 32, TEST_IO_SIZE, 0);
1620     }
1621   }
1622   for (i = 0; i < 15; ++i) {
1623     if (i % 3 == 2) {
1624       ASSERT_PASSED(aio_writesame_test_data, image, test_data, TEST_IO_SIZE * i, TEST_IO_SIZE * i * 32 + i, TEST_IO_SIZE, 0);
1625       ASSERT_PASSED(aio_writesame_test_data, image, zero_data, TEST_IO_SIZE * i, TEST_IO_SIZE * i * 32 + i, TEST_IO_SIZE, 0);
1626     } else if (i % 3 == 1) {
1627       ASSERT_PASSED(aio_writesame_test_data, image, test_data, TEST_IO_SIZE + i, TEST_IO_SIZE * i * 32, TEST_IO_SIZE, 0);
1628       ASSERT_PASSED(aio_writesame_test_data, image, zero_data, TEST_IO_SIZE + i, TEST_IO_SIZE * i * 32, TEST_IO_SIZE, 0);
1629     } else {
1630       ASSERT_PASSED(aio_writesame_test_data, image, test_data, TEST_IO_SIZE * i, TEST_IO_SIZE * i * 32, TEST_IO_SIZE, 0);
1631       ASSERT_PASSED(aio_writesame_test_data, image, zero_data, TEST_IO_SIZE * i, TEST_IO_SIZE * i * 32, TEST_IO_SIZE, 0);
1632     }
1633   }
1634
1635   rbd_image_info_t info;
1636   rbd_completion_t comp;
1637   ASSERT_EQ(0, rbd_stat(image, &info, sizeof(info)));
1638   // can't read or write starting past end
1639   ASSERT_EQ(-EINVAL, rbd_write(image, info.size, 1, test_data));
1640   ASSERT_EQ(-EINVAL, rbd_read(image, info.size, 1, test_data));
1641   // reading through end returns amount up to end
1642   ASSERT_EQ(10, rbd_read(image, info.size - 10, 100, test_data));
1643   // writing through end returns amount up to end
1644   ASSERT_EQ(10, rbd_write(image, info.size - 10, 100, test_data));
1645
1646   rbd_aio_create_completion(NULL, (rbd_callback_t) simple_read_cb, &comp);
1647   ASSERT_EQ(0, rbd_aio_write(image, info.size, 1, test_data, comp));
1648   ASSERT_EQ(0, rbd_aio_wait_for_complete(comp));
1649   ASSERT_EQ(-EINVAL, rbd_aio_get_return_value(comp));
1650   rbd_aio_release(comp);
1651
1652   rbd_aio_create_completion(NULL, (rbd_callback_t) simple_read_cb, &comp);
1653   ASSERT_EQ(0, rbd_aio_read(image, info.size, 1, test_data, comp));
1654   ASSERT_EQ(0, rbd_aio_wait_for_complete(comp));
1655   ASSERT_EQ(-EINVAL, rbd_aio_get_return_value(comp));
1656   rbd_aio_release(comp);
1657
1658   ASSERT_PASSED(write_test_data, image, zero_data, 0, TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_NOCACHE);
1659   ASSERT_EQ(-EILSEQ, rbd_compare_and_write(image, 0, TEST_IO_SIZE, mismatch_data, mismatch_data, &mismatch_offset, 0));
1660   ASSERT_EQ(0U, mismatch_offset);
1661   rbd_aio_create_completion(NULL, (rbd_callback_t) simple_read_cb, &comp);
1662   ASSERT_EQ(0, rbd_aio_compare_and_write(image, 0, TEST_IO_SIZE, mismatch_data, mismatch_data, comp, &mismatch_offset, 0));
1663   ASSERT_EQ(0, rbd_aio_wait_for_complete(comp));
1664   ASSERT_EQ(0U, mismatch_offset);
1665   rbd_aio_release(comp);
1666
1667   ASSERT_PASSED(validate_object_map, image);
1668   ASSERT_EQ(0, rbd_close(image));
1669
1670   rados_ioctx_destroy(ioctx);
1671 }
1672
1673 TEST_F(TestLibRBD, TestIOWithIOHint)
1674 {
1675   rados_ioctx_t ioctx;
1676   rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx);
1677
1678   bool skip_discard = is_skip_partial_discard_enabled();
1679
1680   rbd_image_t image;
1681   int order = 0;
1682   std::string name = get_temp_image_name();
1683   uint64_t size = 2 << 20;
1684
1685   ASSERT_EQ(0, create_image(ioctx, name.c_str(), size, &order));
1686   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
1687
1688   char test_data[TEST_IO_SIZE + 1];
1689   char zero_data[TEST_IO_SIZE + 1];
1690   char mismatch_data[TEST_IO_SIZE + 1];
1691   int i;
1692   uint64_t mismatch_offset;
1693
1694   for (i = 0; i < TEST_IO_SIZE; ++i) {
1695     test_data[i] = (char) (rand() % (126 - 33) + 33);
1696   }
1697   test_data[TEST_IO_SIZE] = '\0';
1698   memset(zero_data, 0, sizeof(zero_data));
1699   memset(mismatch_data, 9, sizeof(mismatch_data));
1700
1701   for (i = 0; i < 5; ++i)
1702     ASSERT_PASSED(write_test_data, image, test_data, TEST_IO_SIZE * i,
1703                   TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_NOCACHE);
1704
1705   for (i = 5; i < 10; ++i)
1706     ASSERT_PASSED(aio_write_test_data, image, test_data, TEST_IO_SIZE * i,
1707                   TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_DONTNEED);
1708
1709   for (i = 0; i < 5; ++i)
1710     ASSERT_PASSED(compare_and_write_test_data, image, test_data, test_data,
1711       TEST_IO_SIZE * i, TEST_IO_SIZE, &mismatch_offset, LIBRADOS_OP_FLAG_FADVISE_DONTNEED);
1712
1713   for (i = 5; i < 10; ++i)
1714     ASSERT_PASSED(aio_compare_and_write_test_data, image, test_data, test_data,
1715       TEST_IO_SIZE * i, TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_DONTNEED);
1716
1717   for (i = 0; i < 5; ++i)
1718     ASSERT_PASSED(read_test_data, image, test_data, TEST_IO_SIZE * i, TEST_IO_SIZE,
1719                   LIBRADOS_OP_FLAG_FADVISE_SEQUENTIAL);
1720
1721   for (i = 5; i < 10; ++i)
1722     ASSERT_PASSED(aio_read_test_data, image, test_data, TEST_IO_SIZE * i,
1723                   TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_SEQUENTIAL|LIBRADOS_OP_FLAG_FADVISE_DONTNEED);
1724
1725   // discard 2nd, 4th sections.
1726   ASSERT_PASSED(discard_test_data, image, TEST_IO_SIZE, TEST_IO_SIZE);
1727   ASSERT_PASSED(aio_discard_test_data, image, TEST_IO_SIZE*3, TEST_IO_SIZE);
1728
1729   ASSERT_PASSED(read_test_data, image, test_data,  0, TEST_IO_SIZE,
1730                 LIBRADOS_OP_FLAG_FADVISE_SEQUENTIAL);
1731   ASSERT_PASSED(read_test_data, image, skip_discard ? test_data : zero_data,
1732                 TEST_IO_SIZE, TEST_IO_SIZE,
1733                 LIBRADOS_OP_FLAG_FADVISE_SEQUENTIAL);
1734   ASSERT_PASSED(read_test_data, image, test_data,  TEST_IO_SIZE*2, TEST_IO_SIZE,
1735                 LIBRADOS_OP_FLAG_FADVISE_SEQUENTIAL);
1736   ASSERT_PASSED(read_test_data, image, skip_discard ? test_data : zero_data,
1737                 TEST_IO_SIZE*3, TEST_IO_SIZE,
1738                 LIBRADOS_OP_FLAG_FADVISE_SEQUENTIAL);
1739   ASSERT_PASSED(read_test_data, image, test_data,  TEST_IO_SIZE*4, TEST_IO_SIZE, 0);
1740
1741   for (i = 0; i < 15; ++i) {
1742     if (i % 3 == 2) {
1743       ASSERT_PASSED(writesame_test_data, image, test_data, TEST_IO_SIZE * i, TEST_IO_SIZE * i * 32 + i,
1744                     TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_NOCACHE);
1745       ASSERT_PASSED(writesame_test_data, image, zero_data, TEST_IO_SIZE * i, TEST_IO_SIZE * i * 32 + i,
1746                     TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_NOCACHE);
1747     } else if (i % 3 == 1) {
1748       ASSERT_PASSED(writesame_test_data, image, test_data, TEST_IO_SIZE + i, TEST_IO_SIZE * i * 32,
1749                     TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_NOCACHE);
1750       ASSERT_PASSED(writesame_test_data, image, zero_data, TEST_IO_SIZE + i, TEST_IO_SIZE * i * 32,
1751                     TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_NOCACHE);
1752     } else {
1753       ASSERT_PASSED(writesame_test_data, image, test_data, TEST_IO_SIZE * i, TEST_IO_SIZE * i * 32,
1754                     TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_NOCACHE);
1755       ASSERT_PASSED(writesame_test_data, image, zero_data, TEST_IO_SIZE * i, TEST_IO_SIZE * i * 32,
1756                     TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_NOCACHE);
1757     }
1758   }
1759   for (i = 0; i < 15; ++i) {
1760     if (i % 3 == 2) {
1761       ASSERT_PASSED(aio_writesame_test_data, image, test_data, TEST_IO_SIZE * i, TEST_IO_SIZE * i * 32 + i,
1762                     TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_DONTNEED);
1763       ASSERT_PASSED(aio_writesame_test_data, image, zero_data, TEST_IO_SIZE * i, TEST_IO_SIZE * i * 32 + i,
1764                     TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_DONTNEED);
1765     } else if (i % 3 == 1) {
1766       ASSERT_PASSED(aio_writesame_test_data, image, test_data, TEST_IO_SIZE + i, TEST_IO_SIZE * i * 32,
1767                     TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_DONTNEED);
1768       ASSERT_PASSED(aio_writesame_test_data, image, zero_data, TEST_IO_SIZE + i, TEST_IO_SIZE * i * 32,
1769                     TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_DONTNEED);
1770     } else {
1771       ASSERT_PASSED(aio_writesame_test_data, image, test_data, TEST_IO_SIZE * i, TEST_IO_SIZE * i * 32,
1772                     TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_DONTNEED);
1773       ASSERT_PASSED(aio_writesame_test_data, image, zero_data, TEST_IO_SIZE * i, TEST_IO_SIZE * i * 32,
1774                     TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_DONTNEED);
1775     }
1776   }
1777
1778   rbd_image_info_t info;
1779   rbd_completion_t comp;
1780   ASSERT_EQ(0, rbd_stat(image, &info, sizeof(info)));
1781   // can't read or write starting past end
1782   ASSERT_EQ(-EINVAL, rbd_write(image, info.size, 1, test_data));
1783   ASSERT_EQ(-EINVAL, rbd_read(image, info.size, 1, test_data));
1784   // reading through end returns amount up to end
1785   ASSERT_EQ(10, rbd_read2(image, info.size - 10, 100, test_data,
1786                           LIBRADOS_OP_FLAG_FADVISE_NOCACHE));
1787   // writing through end returns amount up to end
1788   ASSERT_EQ(10, rbd_write2(image, info.size - 10, 100, test_data,
1789                             LIBRADOS_OP_FLAG_FADVISE_DONTNEED));
1790
1791   rbd_aio_create_completion(NULL, (rbd_callback_t) simple_read_cb, &comp);
1792   ASSERT_EQ(0, rbd_aio_read2(image, info.size, 1, test_data, comp,
1793                              LIBRADOS_OP_FLAG_FADVISE_DONTNEED));
1794   ASSERT_EQ(0, rbd_aio_wait_for_complete(comp));
1795   ASSERT_EQ(-EINVAL, rbd_aio_get_return_value(comp));
1796   rbd_aio_release(comp);
1797
1798   ASSERT_PASSED(write_test_data, image, zero_data, 0, TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_NOCACHE);
1799   ASSERT_EQ(-EILSEQ, rbd_compare_and_write(image, 0, TEST_IO_SIZE, mismatch_data, mismatch_data,
1800                                            &mismatch_offset, LIBRADOS_OP_FLAG_FADVISE_DONTNEED));
1801   ASSERT_EQ(0U, mismatch_offset);
1802   rbd_aio_create_completion(NULL, (rbd_callback_t) simple_read_cb, &comp);
1803   ASSERT_EQ(0, rbd_aio_compare_and_write(image, 0, TEST_IO_SIZE, mismatch_data, mismatch_data,
1804                                          comp, &mismatch_offset, LIBRADOS_OP_FLAG_FADVISE_DONTNEED));
1805   ASSERT_EQ(0, rbd_aio_wait_for_complete(comp));
1806   ASSERT_EQ(0U, mismatch_offset);
1807   rbd_aio_release(comp);
1808
1809   ASSERT_PASSED(validate_object_map, image);
1810   ASSERT_EQ(0, rbd_close(image));
1811
1812   rados_ioctx_destroy(ioctx);
1813 }
1814
1815 TEST_F(TestLibRBD, TestDataPoolIO)
1816 {
1817   REQUIRE_FORMAT_V2();
1818
1819   rados_ioctx_t ioctx;
1820   rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx);
1821
1822   std::string data_pool_name = create_pool(true);
1823
1824   bool skip_discard = is_skip_partial_discard_enabled();
1825
1826   rbd_image_t image;
1827   std::string name = get_temp_image_name();
1828   uint64_t size = 2 << 20;
1829
1830   bool old_format;
1831   uint64_t features;
1832   ASSERT_EQ(0, get_features(&old_format, &features));
1833   ASSERT_FALSE(old_format);
1834
1835   rbd_image_options_t image_options;
1836   rbd_image_options_create(&image_options);
1837   BOOST_SCOPE_EXIT( (&image_options) ) {
1838     rbd_image_options_destroy(image_options);
1839   } BOOST_SCOPE_EXIT_END;
1840
1841   ASSERT_EQ(0, rbd_image_options_set_uint64(image_options,
1842                                             RBD_IMAGE_OPTION_FEATURES,
1843                                             features));
1844   ASSERT_EQ(0, rbd_image_options_set_string(image_options,
1845                                             RBD_IMAGE_OPTION_DATA_POOL,
1846                                             data_pool_name.c_str()));
1847
1848   ASSERT_EQ(0, rbd_create4(ioctx, name.c_str(), size, image_options));
1849   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
1850   ASSERT_NE(-1, rbd_get_data_pool_id(image));
1851
1852   char test_data[TEST_IO_SIZE + 1];
1853   char zero_data[TEST_IO_SIZE + 1];
1854   int i;
1855
1856   for (i = 0; i < TEST_IO_SIZE; ++i) {
1857     test_data[i] = (char) (rand() % (126 - 33) + 33);
1858   }
1859   test_data[TEST_IO_SIZE] = '\0';
1860   memset(zero_data, 0, sizeof(zero_data));
1861
1862   for (i = 0; i < 5; ++i)
1863     ASSERT_PASSED(write_test_data, image, test_data, TEST_IO_SIZE * i, TEST_IO_SIZE, 0);
1864
1865   for (i = 5; i < 10; ++i)
1866     ASSERT_PASSED(aio_write_test_data, image, test_data, TEST_IO_SIZE * i, TEST_IO_SIZE, 0);
1867
1868   for (i = 0; i < 5; ++i)
1869     ASSERT_PASSED(read_test_data, image, test_data, TEST_IO_SIZE * i, TEST_IO_SIZE, 0);
1870
1871   for (i = 5; i < 10; ++i)
1872     ASSERT_PASSED(aio_read_test_data, image, test_data, TEST_IO_SIZE * i, TEST_IO_SIZE, 0);
1873
1874   // discard 2nd, 4th sections.
1875   ASSERT_PASSED(discard_test_data, image, TEST_IO_SIZE, TEST_IO_SIZE);
1876   ASSERT_PASSED(aio_discard_test_data, image, TEST_IO_SIZE*3, TEST_IO_SIZE);
1877
1878   ASSERT_PASSED(read_test_data, image, test_data,  0, TEST_IO_SIZE, 0);
1879   ASSERT_PASSED(read_test_data, image, skip_discard ? test_data : zero_data,
1880                 TEST_IO_SIZE, TEST_IO_SIZE, 0);
1881   ASSERT_PASSED(read_test_data, image, test_data,  TEST_IO_SIZE*2, TEST_IO_SIZE, 0);
1882   ASSERT_PASSED(read_test_data, image, skip_discard ? test_data : zero_data,
1883                 TEST_IO_SIZE*3, TEST_IO_SIZE, 0);
1884   ASSERT_PASSED(read_test_data, image, test_data,  TEST_IO_SIZE*4, TEST_IO_SIZE, 0);
1885
1886   rbd_image_info_t info;
1887   rbd_completion_t comp;
1888   ASSERT_EQ(0, rbd_stat(image, &info, sizeof(info)));
1889   // can't read or write starting past end
1890   ASSERT_EQ(-EINVAL, rbd_write(image, info.size, 1, test_data));
1891   ASSERT_EQ(-EINVAL, rbd_read(image, info.size, 1, test_data));
1892   // reading through end returns amount up to end
1893   ASSERT_EQ(10, rbd_read(image, info.size - 10, 100, test_data));
1894   // writing through end returns amount up to end
1895   ASSERT_EQ(10, rbd_write(image, info.size - 10, 100, test_data));
1896
1897   rbd_aio_create_completion(NULL, (rbd_callback_t) simple_read_cb, &comp);
1898   ASSERT_EQ(0, rbd_aio_write(image, info.size, 1, test_data, comp));
1899   ASSERT_EQ(0, rbd_aio_wait_for_complete(comp));
1900   ASSERT_EQ(-EINVAL, rbd_aio_get_return_value(comp));
1901   rbd_aio_release(comp);
1902
1903   rbd_aio_create_completion(NULL, (rbd_callback_t) simple_read_cb, &comp);
1904   ASSERT_EQ(0, rbd_aio_read(image, info.size, 1, test_data, comp));
1905   ASSERT_EQ(0, rbd_aio_wait_for_complete(comp));
1906   ASSERT_EQ(-EINVAL, rbd_aio_get_return_value(comp));
1907   rbd_aio_release(comp);
1908
1909   ASSERT_PASSED(validate_object_map, image);
1910   ASSERT_EQ(0, rbd_close(image));
1911
1912   rados_ioctx_destroy(ioctx);
1913 }
1914
1915 TEST_F(TestLibRBD, TestScatterGatherIO)
1916 {
1917   rados_ioctx_t ioctx;
1918   rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx);
1919
1920   rbd_image_t image;
1921   int order = 0;
1922   std::string name = get_temp_image_name();
1923   uint64_t size = 20 << 20;
1924
1925   ASSERT_EQ(0, create_image(ioctx, name.c_str(), size, &order));
1926   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
1927
1928   std::string write_buffer("This is a test");
1929   struct iovec bad_iovs[] = {
1930     {.iov_base = NULL, .iov_len = static_cast<size_t>(-1)}
1931   };
1932   struct iovec write_iovs[] = {
1933     {.iov_base = &write_buffer[0],  .iov_len = 5},
1934     {.iov_base = &write_buffer[5],  .iov_len = 3},
1935     {.iov_base = &write_buffer[8],  .iov_len = 2},
1936     {.iov_base = &write_buffer[10], .iov_len = 4}
1937   };
1938
1939   rbd_completion_t comp;
1940   rbd_aio_create_completion(NULL, NULL, &comp);
1941   ASSERT_EQ(-EINVAL, rbd_aio_writev(image, write_iovs, 0, 0, comp));
1942   ASSERT_EQ(-EINVAL, rbd_aio_writev(image, bad_iovs, 1, 0, comp));
1943   ASSERT_EQ(0, rbd_aio_writev(image, write_iovs,
1944                               sizeof(write_iovs) / sizeof(struct iovec),
1945                               1<<order, comp));
1946   ASSERT_EQ(0, rbd_aio_wait_for_complete(comp));
1947   ASSERT_EQ(0, rbd_aio_get_return_value(comp));
1948   rbd_aio_release(comp);
1949
1950   std::string read_buffer(write_buffer.size(), '1');
1951   struct iovec read_iovs[] = {
1952     {.iov_base = &read_buffer[0],  .iov_len = 4},
1953     {.iov_base = &read_buffer[8],  .iov_len = 4},
1954     {.iov_base = &read_buffer[12], .iov_len = 2}
1955   };
1956
1957   rbd_aio_create_completion(NULL, NULL, &comp);
1958   ASSERT_EQ(-EINVAL, rbd_aio_readv(image, read_iovs, 0, 0, comp));
1959   ASSERT_EQ(-EINVAL, rbd_aio_readv(image, bad_iovs, 1, 0, comp));
1960   ASSERT_EQ(0, rbd_aio_readv(image, read_iovs,
1961                              sizeof(read_iovs) / sizeof(struct iovec),
1962                              1<<order, comp));
1963   ASSERT_EQ(0, rbd_aio_wait_for_complete(comp));
1964   ASSERT_EQ(10, rbd_aio_get_return_value(comp));
1965   rbd_aio_release(comp);
1966   ASSERT_EQ("This1111 is a ", read_buffer);
1967
1968   std::string linear_buffer(write_buffer.size(), '1');
1969   struct iovec linear_iovs[] = {
1970     {.iov_base = &linear_buffer[4], .iov_len = 4}
1971   };
1972   rbd_aio_create_completion(NULL, NULL, &comp);
1973   ASSERT_EQ(0, rbd_aio_readv(image, linear_iovs,
1974                              sizeof(linear_iovs) / sizeof(struct iovec),
1975                              1<<order, comp));
1976   ASSERT_EQ(0, rbd_aio_wait_for_complete(comp));
1977   ASSERT_EQ(4, rbd_aio_get_return_value(comp));
1978   rbd_aio_release(comp);
1979   ASSERT_EQ("1111This111111", linear_buffer);
1980
1981   ASSERT_PASSED(validate_object_map, image);
1982   ASSERT_EQ(0, rbd_close(image));
1983
1984   rados_ioctx_destroy(ioctx);
1985 }
1986
1987 TEST_F(TestLibRBD, TestEmptyDiscard)
1988 {
1989   rados_ioctx_t ioctx;
1990   rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx);
1991
1992   rbd_image_t image;
1993   int order = 0;
1994   std::string name = get_temp_image_name();
1995   uint64_t size = 20 << 20;
1996
1997   ASSERT_EQ(0, create_image(ioctx, name.c_str(), size, &order));
1998   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
1999
2000   ASSERT_PASSED(aio_discard_test_data, image, 0, 1*1024*1024);
2001   ASSERT_PASSED(aio_discard_test_data, image, 0, 4*1024*1024);
2002   ASSERT_PASSED(aio_discard_test_data, image, 3*1024*1024, 1*1024*1024);
2003
2004   ASSERT_PASSED(validate_object_map, image);
2005   ASSERT_EQ(0, rbd_close(image));
2006
2007   rados_ioctx_destroy(ioctx);
2008 }
2009
2010
2011 void simple_write_cb_pp(librbd::completion_t cb, void *arg)
2012 {
2013   cout << "write completion cb called!" << std::endl;
2014 }
2015
2016 void simple_read_cb_pp(librbd::completion_t cb, void *arg)
2017 {
2018   cout << "read completion cb called!" << std::endl;
2019 }
2020
2021 void aio_write_test_data(librbd::Image& image, const char *test_data,
2022                          off_t off, uint32_t iohint, bool *passed)
2023 {
2024   ceph::bufferlist bl;
2025   bl.append(test_data, strlen(test_data));
2026   librbd::RBD::AioCompletion *comp = new librbd::RBD::AioCompletion(NULL, (librbd::callback_t) simple_write_cb_pp);
2027   printf("created completion\n");
2028   if (iohint)
2029     image.aio_write2(off, strlen(test_data), bl, comp, iohint);
2030   else
2031     image.aio_write(off, strlen(test_data), bl, comp);
2032   printf("started write\n");
2033   comp->wait_for_complete();
2034   int r = comp->get_return_value();
2035   printf("return value is: %d\n", r);
2036   ASSERT_EQ(0, r);
2037   printf("finished write\n");
2038   comp->release();
2039   *passed = true;
2040 }
2041
2042 void aio_discard_test_data(librbd::Image& image, off_t off, size_t len, bool *passed)
2043 {
2044   librbd::RBD::AioCompletion *comp = new librbd::RBD::AioCompletion(NULL, (librbd::callback_t) simple_write_cb_pp);
2045   image.aio_discard(off, len, comp);
2046   comp->wait_for_complete();
2047   int r = comp->get_return_value();
2048   ASSERT_EQ(0, r);
2049   comp->release();
2050   *passed = true;
2051 }
2052
2053 void write_test_data(librbd::Image& image, const char *test_data, off_t off, uint32_t iohint, bool *passed)
2054 {
2055   size_t written;
2056   size_t len = strlen(test_data);
2057   ceph::bufferlist bl;
2058   bl.append(test_data, len);
2059   if (iohint)
2060     written = image.write2(off, len, bl, iohint);
2061   else
2062     written = image.write(off, len, bl);
2063   printf("wrote: %u\n", (unsigned int) written);
2064   ASSERT_EQ(bl.length(), written);
2065   *passed = true;
2066 }
2067
2068 void discard_test_data(librbd::Image& image, off_t off, size_t len, bool *passed)
2069 {
2070   size_t written;
2071   written = image.discard(off, len);
2072   printf("discard: %u~%u\n", (unsigned)off, (unsigned)len);
2073   ASSERT_EQ(len, written);
2074   *passed = true;
2075 }
2076
2077 void aio_read_test_data(librbd::Image& image, const char *expected, off_t off, size_t expected_len, uint32_t iohint, bool *passed)
2078 {
2079   librbd::RBD::AioCompletion *comp = new librbd::RBD::AioCompletion(NULL, (librbd::callback_t) simple_read_cb_pp);
2080   ceph::bufferlist bl;
2081   printf("created completion\n");
2082   if (iohint)
2083     image.aio_read2(off, expected_len, bl, comp, iohint);
2084   else
2085     image.aio_read(off, expected_len, bl, comp);
2086   printf("started read\n");
2087   comp->wait_for_complete();
2088   int r = comp->get_return_value();
2089   printf("return value is: %d\n", r);
2090   ASSERT_EQ(TEST_IO_SIZE, r);
2091   ASSERT_EQ(0, memcmp(expected, bl.c_str(), TEST_IO_SIZE));
2092   printf("finished read\n");
2093   comp->release();
2094   *passed = true;
2095 }
2096
2097 void read_test_data(librbd::Image& image, const char *expected, off_t off, size_t expected_len, uint32_t iohint, bool *passed)
2098 {
2099   int read;
2100   size_t len = expected_len;
2101   ceph::bufferlist bl;
2102   if (iohint)
2103     read = image.read2(off, len, bl, iohint);
2104   else
2105     read = image.read(off, len, bl);
2106   ASSERT_TRUE(read >= 0);
2107   std::string bl_str(bl.c_str(), read);
2108
2109   printf("read: %u\n", (unsigned int) read);
2110   int result = memcmp(bl_str.c_str(), expected, expected_len);
2111   if (result != 0) {
2112     printf("read: %s\nexpected: %s\n", bl_str.c_str(), expected);
2113     ASSERT_EQ(0, result);
2114   }
2115   *passed = true;
2116 }
2117
2118 void aio_writesame_test_data(librbd::Image& image, const char *test_data, off_t off,
2119                              size_t len, size_t data_len, uint32_t iohint, bool *passed)
2120 {
2121   ceph::bufferlist bl;
2122   bl.append(test_data, data_len);
2123   librbd::RBD::AioCompletion *comp = new librbd::RBD::AioCompletion(NULL, (librbd::callback_t) simple_write_cb_pp);
2124   printf("created completion\n");
2125   int r;
2126   r = image.aio_writesame(off, len, bl, comp, iohint);
2127   printf("started writesame\n");
2128   if (len % data_len) {
2129     ASSERT_EQ(-EINVAL, r);
2130     printf("expected fail, finished writesame\n");
2131     comp->release();
2132     *passed = true;
2133     return;
2134   }
2135
2136   comp->wait_for_complete();
2137   r = comp->get_return_value();
2138   printf("return value is: %d\n", r);
2139   ASSERT_EQ(0, r);
2140   printf("finished writesame\n");
2141   comp->release();
2142
2143   //verify data
2144   printf("to verify the data\n");
2145   int read;
2146   uint64_t left = len;
2147   while (left > 0) {
2148     ceph::bufferlist bl;
2149     read = image.read(off, data_len, bl);
2150     ASSERT_EQ(data_len, static_cast<size_t>(read));
2151     std::string bl_str(bl.c_str(), read);
2152     int result = memcmp(bl_str.c_str(), test_data, data_len);
2153     if (result !=0 ) {
2154       printf("read: %u ~ %u\n", (unsigned int) off, (unsigned int) read);
2155       printf("read: %s\nexpected: %s\n", bl_str.c_str(), test_data);
2156       ASSERT_EQ(0, result);
2157     }
2158     off += data_len;
2159     left -= data_len;
2160   }
2161   ASSERT_EQ(0U, left);
2162   printf("verified\n");
2163
2164   *passed = true;
2165 }
2166
2167 void writesame_test_data(librbd::Image& image, const char *test_data, off_t off,
2168                          ssize_t len, size_t data_len, uint32_t iohint,
2169                          bool *passed)
2170 {
2171   ssize_t written;
2172   ceph::bufferlist bl;
2173   bl.append(test_data, data_len);
2174   written = image.writesame(off, len, bl, iohint);
2175   if (len % data_len) {
2176     ASSERT_EQ(-EINVAL, written);
2177     printf("expected fail, finished writesame\n");
2178     *passed = true;
2179     return;
2180   }
2181   ASSERT_EQ(len, written);
2182   printf("wrote: %u\n", (unsigned int) written);
2183   *passed = true;
2184
2185   //verify data
2186   printf("to verify the data\n");
2187   int read;
2188   uint64_t left = len;
2189   while (left > 0) {
2190     ceph::bufferlist bl;
2191     read = image.read(off, data_len, bl);
2192     ASSERT_EQ(data_len, static_cast<size_t>(read));
2193     std::string bl_str(bl.c_str(), read);
2194     int result = memcmp(bl_str.c_str(), test_data, data_len);
2195     if (result !=0 ) {
2196       printf("read: %u ~ %u\n", (unsigned int) off, (unsigned int) read);
2197       printf("read: %s\nexpected: %s\n", bl_str.c_str(), test_data);
2198       ASSERT_EQ(0, result);
2199     }
2200     off += data_len;
2201     left -= data_len;
2202   }
2203   ASSERT_EQ(0U, left);
2204   printf("verified\n");
2205
2206   *passed = true;
2207 }
2208
2209 void aio_compare_and_write_test_data(librbd::Image& image, const char *cmp_data,
2210                                      const char *test_data, off_t off, ssize_t len,
2211                                      uint32_t iohint, bool *passed)
2212 {
2213   ceph::bufferlist cmp_bl;
2214   cmp_bl.append(cmp_data, strlen(cmp_data));
2215   ceph::bufferlist test_bl;
2216   test_bl.append(test_data, strlen(test_data));
2217   librbd::RBD::AioCompletion *comp = new librbd::RBD::AioCompletion(NULL, (librbd::callback_t) simple_write_cb_pp);
2218   printf("created completion\n");
2219
2220   uint64_t mismatch_offset;
2221   image.aio_compare_and_write(off, len, cmp_bl, test_bl, comp, &mismatch_offset, iohint);
2222   printf("started aio compare and write\n");
2223   comp->wait_for_complete();
2224   int r = comp->get_return_value();
2225   printf("return value is: %d\n", r);
2226   ASSERT_EQ(0, r);
2227   printf("finished aio compare and write\n");
2228   comp->release();
2229   *passed = true;
2230 }
2231
2232 void compare_and_write_test_data(librbd::Image& image, const char *cmp_data, const char *test_data,
2233                                  off_t off, ssize_t len, uint64_t *mismatch_off, uint32_t iohint, bool *passed)
2234 {
2235   size_t written;
2236   ceph::bufferlist cmp_bl;
2237   cmp_bl.append(cmp_data, strlen(cmp_data));
2238   ceph::bufferlist test_bl;
2239   test_bl.append(test_data, strlen(test_data));
2240   printf("start compare and write\n");
2241   written = image.compare_and_write(off, len, cmp_bl, test_bl, mismatch_off, iohint);
2242   printf("compare and  wrote: %d\n", (int) written);
2243   ASSERT_EQ(len, static_cast<ssize_t>(written));
2244   *passed = true;
2245 }
2246
2247 TEST_F(TestLibRBD, TestIOPP)
2248 {
2249   librados::IoCtx ioctx;
2250   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
2251
2252   bool skip_discard = is_skip_partial_discard_enabled();
2253
2254   {
2255     librbd::RBD rbd;
2256     librbd::Image image;
2257     int order = 0;
2258     std::string name = get_temp_image_name();
2259     uint64_t size = 2 << 20;
2260
2261     ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
2262     ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
2263
2264     char test_data[TEST_IO_SIZE + 1];
2265     char zero_data[TEST_IO_SIZE + 1];
2266     int i;
2267     uint64_t mismatch_offset;
2268
2269     for (i = 0; i < TEST_IO_SIZE; ++i) {
2270       test_data[i] = (char) (rand() % (126 - 33) + 33);
2271     }
2272     test_data[TEST_IO_SIZE] = '\0';
2273     memset(zero_data, 0, sizeof(zero_data));
2274
2275     for (i = 0; i < 5; ++i)
2276       ASSERT_PASSED(write_test_data, image, test_data, strlen(test_data) * i, 0);
2277
2278     for (i = 5; i < 10; ++i)
2279       ASSERT_PASSED(aio_write_test_data, image, test_data, strlen(test_data) * i, 0);
2280
2281     for (i = 0; i < 5; ++i)
2282       ASSERT_PASSED(compare_and_write_test_data, image, test_data, test_data, TEST_IO_SIZE * i,
2283         TEST_IO_SIZE, &mismatch_offset, 0);
2284
2285     for (i = 5; i < 10; ++i)
2286       ASSERT_PASSED(aio_compare_and_write_test_data, image, test_data, test_data, TEST_IO_SIZE * i,
2287         TEST_IO_SIZE, 0);
2288
2289     for (i = 0; i < 5; ++i)
2290       ASSERT_PASSED(read_test_data, image, test_data, strlen(test_data) * i, TEST_IO_SIZE, 0);
2291
2292     for (i = 5; i < 10; ++i)
2293       ASSERT_PASSED(aio_read_test_data, image, test_data, strlen(test_data) * i, TEST_IO_SIZE, 0);
2294
2295     // discard 2nd, 4th sections.
2296     ASSERT_PASSED(discard_test_data, image, TEST_IO_SIZE, TEST_IO_SIZE);
2297     ASSERT_PASSED(aio_discard_test_data, image, TEST_IO_SIZE*3, TEST_IO_SIZE);
2298
2299     ASSERT_PASSED(read_test_data, image, test_data,  0, TEST_IO_SIZE, 0);
2300     ASSERT_PASSED(read_test_data, image, skip_discard ? test_data : zero_data,
2301                   TEST_IO_SIZE, TEST_IO_SIZE, 0);
2302     ASSERT_PASSED(read_test_data, image, test_data,  TEST_IO_SIZE*2, TEST_IO_SIZE, 0);
2303     ASSERT_PASSED(read_test_data, image, skip_discard ? test_data : zero_data,
2304                   TEST_IO_SIZE*3, TEST_IO_SIZE, 0);
2305     ASSERT_PASSED(read_test_data, image, test_data,  TEST_IO_SIZE*4, TEST_IO_SIZE, 0);
2306
2307     for (i = 0; i < 15; ++i) {
2308       if (i % 3 == 2) {
2309         ASSERT_PASSED(writesame_test_data, image, test_data, TEST_IO_SIZE * i, TEST_IO_SIZE * i * 32 + i, TEST_IO_SIZE, 0);
2310         ASSERT_PASSED(writesame_test_data, image, zero_data, TEST_IO_SIZE * i, TEST_IO_SIZE * i * 32 + i, TEST_IO_SIZE, 0);
2311       } else if (i % 3 == 1) {
2312         ASSERT_PASSED(writesame_test_data, image, test_data, TEST_IO_SIZE + i, TEST_IO_SIZE * i * 32, TEST_IO_SIZE, 0);
2313         ASSERT_PASSED(writesame_test_data, image, zero_data, TEST_IO_SIZE + i, TEST_IO_SIZE * i * 32, TEST_IO_SIZE, 0);
2314       } else {
2315         ASSERT_PASSED(writesame_test_data, image, test_data, TEST_IO_SIZE * i, TEST_IO_SIZE * i * 32, TEST_IO_SIZE, 0);
2316         ASSERT_PASSED(writesame_test_data, image, zero_data, TEST_IO_SIZE * i, TEST_IO_SIZE * i * 32, TEST_IO_SIZE, 0);
2317       }
2318     }
2319     for (i = 0; i < 15; ++i) {
2320       if (i % 3 == 2) {
2321         ASSERT_PASSED(aio_writesame_test_data, image, test_data, TEST_IO_SIZE * i, TEST_IO_SIZE * i * 32 + i, TEST_IO_SIZE, 0);
2322         ASSERT_PASSED(aio_writesame_test_data, image, zero_data, TEST_IO_SIZE * i, TEST_IO_SIZE * i * 32 + i, TEST_IO_SIZE, 0);
2323       } else if (i % 3 == 1) {
2324         ASSERT_PASSED(aio_writesame_test_data, image, test_data, TEST_IO_SIZE + i, TEST_IO_SIZE * i * 32, TEST_IO_SIZE, 0);
2325         ASSERT_PASSED(aio_writesame_test_data, image, zero_data, TEST_IO_SIZE + i, TEST_IO_SIZE * i * 32, TEST_IO_SIZE, 0);
2326       } else {
2327         ASSERT_PASSED(aio_writesame_test_data, image, test_data, TEST_IO_SIZE * i, TEST_IO_SIZE * i * 32, TEST_IO_SIZE, 0);
2328         ASSERT_PASSED(aio_writesame_test_data, image, zero_data, TEST_IO_SIZE * i, TEST_IO_SIZE * i * 32, TEST_IO_SIZE, 0);
2329       }
2330     }
2331
2332     ASSERT_PASSED(validate_object_map, image);
2333   }
2334
2335   ioctx.close();
2336 }
2337
2338 TEST_F(TestLibRBD, TestIOPPWithIOHint)
2339 {
2340   librados::IoCtx ioctx;
2341   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
2342
2343   {
2344     librbd::RBD rbd;
2345     librbd::Image image;
2346     int order = 0;
2347     std::string name = get_temp_image_name();
2348     uint64_t size = 2 << 20;
2349
2350     ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
2351     ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
2352
2353     char test_data[TEST_IO_SIZE + 1];
2354     char zero_data[TEST_IO_SIZE + 1];
2355     test_data[TEST_IO_SIZE] = '\0';
2356     int i;
2357
2358     for (i = 0; i < TEST_IO_SIZE; ++i) {
2359       test_data[i] = (char) (rand() % (126 - 33) + 33);
2360     }
2361     memset(zero_data, 0, sizeof(zero_data));
2362
2363     for (i = 0; i < 5; ++i)
2364       ASSERT_PASSED(write_test_data, image, test_data, strlen(test_data) * i,
2365                     LIBRADOS_OP_FLAG_FADVISE_NOCACHE);
2366
2367     for (i = 5; i < 10; ++i)
2368       ASSERT_PASSED(aio_write_test_data, image, test_data, strlen(test_data) * i,
2369                     LIBRADOS_OP_FLAG_FADVISE_DONTNEED);
2370
2371     ASSERT_PASSED(read_test_data, image, test_data, strlen(test_data),
2372                   TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_RANDOM);
2373
2374     for (i = 5; i < 10; ++i)
2375       ASSERT_PASSED(aio_read_test_data, image, test_data, strlen(test_data) * i,
2376                     TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_SEQUENTIAL|LIBRADOS_OP_FLAG_FADVISE_DONTNEED);
2377
2378     for (i = 0; i < 15; ++i) {
2379       if (i % 3 == 2) {
2380         ASSERT_PASSED(writesame_test_data, image, test_data, TEST_IO_SIZE * i,
2381                       TEST_IO_SIZE * i * 32 + i, TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_NOCACHE);
2382         ASSERT_PASSED(writesame_test_data, image, zero_data, TEST_IO_SIZE * i,
2383                       TEST_IO_SIZE * i * 32 + i, TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_NOCACHE);
2384       } else if (i % 3 == 1) {
2385         ASSERT_PASSED(writesame_test_data, image, test_data, TEST_IO_SIZE + i,
2386                       TEST_IO_SIZE * i * 32, TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_NOCACHE);
2387         ASSERT_PASSED(writesame_test_data, image, zero_data, TEST_IO_SIZE + i,
2388                       TEST_IO_SIZE * i * 32, TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_NOCACHE);
2389       } else {
2390         ASSERT_PASSED(writesame_test_data, image, test_data, TEST_IO_SIZE * i,
2391                       TEST_IO_SIZE * i * 32, TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_NOCACHE);
2392         ASSERT_PASSED(writesame_test_data, image, zero_data, TEST_IO_SIZE * i,
2393                       TEST_IO_SIZE * i * 32, TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_NOCACHE);
2394       }
2395     }
2396     for (i = 0; i < 15; ++i) {
2397       if (i % 3 == 2) {
2398         ASSERT_PASSED(aio_writesame_test_data, image, test_data, TEST_IO_SIZE * i,
2399                       TEST_IO_SIZE * i * 32 + i, TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_DONTNEED);
2400         ASSERT_PASSED(aio_writesame_test_data, image, zero_data, TEST_IO_SIZE * i,
2401                       TEST_IO_SIZE * i * 32 + i, TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_DONTNEED);
2402       } else if (i % 3 == 1) {
2403         ASSERT_PASSED(aio_writesame_test_data, image, test_data, TEST_IO_SIZE + i,
2404                       TEST_IO_SIZE * i * 32, TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_DONTNEED);
2405         ASSERT_PASSED(aio_writesame_test_data, image, zero_data, TEST_IO_SIZE + i,
2406                       TEST_IO_SIZE * i * 32, TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_DONTNEED);
2407       } else {
2408         ASSERT_PASSED(aio_writesame_test_data, image, test_data, TEST_IO_SIZE * i,
2409                       TEST_IO_SIZE * i * 32, TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_DONTNEED);
2410         ASSERT_PASSED(aio_writesame_test_data, image, zero_data, TEST_IO_SIZE * i,
2411                       TEST_IO_SIZE * i * 32, TEST_IO_SIZE, LIBRADOS_OP_FLAG_FADVISE_DONTNEED);
2412       }
2413     }
2414
2415     ASSERT_PASSED(validate_object_map, image);
2416   }
2417
2418   ioctx.close();
2419 }
2420
2421
2422
2423 TEST_F(TestLibRBD, TestIOToSnapshot)
2424 {
2425   rados_ioctx_t ioctx;
2426   rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx);
2427
2428   rbd_image_t image;
2429   int order = 0;
2430   std::string name = get_temp_image_name();
2431   uint64_t isize = 2 << 20;
2432
2433   ASSERT_EQ(0, create_image(ioctx, name.c_str(), isize, &order));
2434   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
2435
2436   int i, r;
2437   rbd_image_t image_at_snap;
2438   char orig_data[TEST_IO_TO_SNAP_SIZE + 1];
2439   char test_data[TEST_IO_TO_SNAP_SIZE + 1];
2440
2441   for (i = 0; i < TEST_IO_TO_SNAP_SIZE; ++i)
2442     test_data[i] = (char) (i + 48);
2443   test_data[TEST_IO_TO_SNAP_SIZE] = '\0';
2444   orig_data[TEST_IO_TO_SNAP_SIZE] = '\0';
2445
2446   r = rbd_read(image, 0, TEST_IO_TO_SNAP_SIZE, orig_data);
2447   ASSERT_EQ(r, TEST_IO_TO_SNAP_SIZE);
2448
2449   ASSERT_EQ(0, test_ls_snaps(image, 0));
2450   ASSERT_EQ(0, rbd_snap_create(image, "orig"));
2451   ASSERT_EQ(1, test_ls_snaps(image, 1, "orig", isize));
2452   ASSERT_PASSED(read_test_data, image, orig_data, 0, TEST_IO_TO_SNAP_SIZE, 0);
2453
2454   printf("write test data!\n");
2455   ASSERT_PASSED(write_test_data, image, test_data, 0, TEST_IO_TO_SNAP_SIZE, 0);
2456   ASSERT_EQ(0, rbd_snap_create(image, "written"));
2457   ASSERT_EQ(2, test_ls_snaps(image, 2, "orig", isize, "written", isize));
2458
2459   ASSERT_PASSED(read_test_data, image, test_data, 0, TEST_IO_TO_SNAP_SIZE, 0);
2460
2461   rbd_snap_set(image, "orig");
2462   ASSERT_PASSED(read_test_data, image, orig_data, 0, TEST_IO_TO_SNAP_SIZE, 0);
2463
2464   rbd_snap_set(image, "written");
2465   ASSERT_PASSED(read_test_data, image, test_data, 0, TEST_IO_TO_SNAP_SIZE, 0);
2466
2467   rbd_snap_set(image, "orig");
2468
2469   r = rbd_write(image, 0, TEST_IO_TO_SNAP_SIZE, test_data);
2470   printf("write to snapshot returned %d\n", r);
2471   ASSERT_LT(r, 0);
2472   cout << strerror(-r) << std::endl;
2473
2474   ASSERT_PASSED(read_test_data, image, orig_data, 0, TEST_IO_TO_SNAP_SIZE, 0);
2475   rbd_snap_set(image, "written");
2476   ASSERT_PASSED(read_test_data, image, test_data, 0, TEST_IO_TO_SNAP_SIZE, 0);
2477
2478   r = rbd_snap_rollback(image, "orig");
2479   ASSERT_EQ(r, -EROFS);
2480
2481   r = rbd_snap_set(image, NULL);
2482   ASSERT_EQ(r, 0);
2483   r = rbd_snap_rollback(image, "orig");
2484   ASSERT_EQ(r, 0);
2485
2486   ASSERT_PASSED(write_test_data, image, test_data, 0, TEST_IO_TO_SNAP_SIZE, 0);
2487
2488   rbd_flush(image);
2489
2490   printf("opening testimg@orig\n");
2491   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image_at_snap, "orig"));
2492   ASSERT_PASSED(read_test_data, image_at_snap, orig_data, 0, TEST_IO_TO_SNAP_SIZE, 0);
2493   r = rbd_write(image_at_snap, 0, TEST_IO_TO_SNAP_SIZE, test_data);
2494   printf("write to snapshot returned %d\n", r);
2495   ASSERT_LT(r, 0);
2496   cout << strerror(-r) << std::endl;
2497   ASSERT_EQ(0, rbd_close(image_at_snap));
2498
2499   ASSERT_EQ(2, test_ls_snaps(image, 2, "orig", isize, "written", isize));
2500   ASSERT_EQ(0, rbd_snap_remove(image, "written"));
2501   ASSERT_EQ(1, test_ls_snaps(image, 1, "orig", isize));
2502   ASSERT_EQ(0, rbd_snap_remove(image, "orig"));
2503   ASSERT_EQ(0, test_ls_snaps(image, 0));
2504
2505   ASSERT_PASSED(validate_object_map, image);
2506   ASSERT_EQ(0, rbd_close(image));
2507
2508   rados_ioctx_destroy(ioctx);
2509 }
2510
2511 TEST_F(TestLibRBD, TestClone)
2512 {
2513   REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
2514
2515   rados_ioctx_t ioctx;
2516   rbd_image_info_t pinfo, cinfo;
2517   rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx);
2518
2519   bool old_format;
2520   uint64_t features;
2521   rbd_image_t parent, child;
2522   int order = 0;
2523
2524   ASSERT_EQ(0, get_features(&old_format, &features));
2525   ASSERT_FALSE(old_format);
2526
2527   std::string parent_name = get_temp_image_name();
2528   std::string child_name = get_temp_image_name();
2529
2530   // make a parent to clone from
2531   ASSERT_EQ(0, create_image_full(ioctx, parent_name.c_str(), 4<<20, &order,
2532                                  false, features));
2533   ASSERT_EQ(0, rbd_open(ioctx, parent_name.c_str(), &parent, NULL));
2534   printf("made parent image \"parent\"\n");
2535
2536   char *data = (char *)"testdata";
2537   ASSERT_EQ((ssize_t)strlen(data), rbd_write(parent, 0, strlen(data), data));
2538
2539   // can't clone a non-snapshot, expect failure
2540   EXPECT_NE(0, clone_image(ioctx, parent, parent_name.c_str(), NULL, ioctx,
2541                            child_name.c_str(), features, &order));
2542
2543   // verify that there is no parent info on "parent"
2544   ASSERT_EQ(-ENOENT, rbd_get_parent_info(parent, NULL, 0, NULL, 0, NULL, 0));
2545   printf("parent has no parent info\n");
2546
2547   // create a snapshot, reopen as the parent we're interested in
2548   ASSERT_EQ(0, rbd_snap_create(parent, "parent_snap"));
2549   printf("made snapshot \"parent@parent_snap\"\n");
2550   ASSERT_EQ(0, rbd_close(parent));
2551   ASSERT_EQ(0, rbd_open(ioctx, parent_name.c_str(), &parent, "parent_snap"));
2552
2553   ASSERT_EQ(-EINVAL, clone_image(ioctx, parent, parent_name.c_str(), "parent_snap",
2554                                  ioctx, child_name.c_str(), features, &order));
2555
2556   // unprotected image should fail unprotect
2557   ASSERT_EQ(-EINVAL, rbd_snap_unprotect(parent, "parent_snap"));
2558   printf("can't unprotect an unprotected snap\n");
2559
2560   ASSERT_EQ(0, rbd_snap_protect(parent, "parent_snap"));
2561   // protecting again should fail
2562   ASSERT_EQ(-EBUSY, rbd_snap_protect(parent, "parent_snap"));
2563   printf("can't protect a protected snap\n");
2564
2565   // This clone and open should work
2566   ASSERT_EQ(0, clone_image(ioctx, parent, parent_name.c_str(), "parent_snap",
2567                            ioctx, child_name.c_str(), features, &order));
2568   ASSERT_EQ(0, rbd_open(ioctx, child_name.c_str(), &child, NULL));
2569   printf("made and opened clone \"child\"\n");
2570
2571   // check read
2572   ASSERT_PASSED(read_test_data, child, data, 0, strlen(data), 0);
2573
2574   // check write
2575   ASSERT_EQ((ssize_t)strlen(data), rbd_write(child, 20, strlen(data), data));
2576   ASSERT_PASSED(read_test_data, child, data, 20, strlen(data), 0);
2577   ASSERT_PASSED(read_test_data, child, data, 0, strlen(data), 0);
2578
2579   // check attributes
2580   ASSERT_EQ(0, rbd_stat(parent, &pinfo, sizeof(pinfo)));
2581   ASSERT_EQ(0, rbd_stat(child, &cinfo, sizeof(cinfo)));
2582   EXPECT_EQ(cinfo.size, pinfo.size);
2583   uint64_t overlap;
2584   rbd_get_overlap(child, &overlap);
2585   EXPECT_EQ(overlap, pinfo.size);
2586   EXPECT_EQ(cinfo.obj_size, pinfo.obj_size);
2587   EXPECT_EQ(cinfo.order, pinfo.order);
2588   printf("sizes and overlaps are good between parent and child\n");
2589
2590   // sizing down child results in changing overlap and size, not parent size
2591   ASSERT_EQ(0, rbd_resize(child, 2UL<<20));
2592   ASSERT_EQ(0, rbd_stat(child, &cinfo, sizeof(cinfo)));
2593   rbd_get_overlap(child, &overlap);
2594   ASSERT_EQ(overlap, 2UL<<20);
2595   ASSERT_EQ(cinfo.size, 2UL<<20);
2596   ASSERT_EQ(0, rbd_resize(child, 4UL<<20));
2597   ASSERT_EQ(0, rbd_stat(child, &cinfo, sizeof(cinfo)));
2598   rbd_get_overlap(child, &overlap);
2599   ASSERT_EQ(overlap, 2UL<<20);
2600   ASSERT_EQ(cinfo.size, 4UL<<20);
2601   printf("sized down clone, changed overlap\n");
2602
2603   // sizing back up doesn't change that
2604   ASSERT_EQ(0, rbd_resize(child, 5UL<<20));
2605   ASSERT_EQ(0, rbd_stat(child, &cinfo, sizeof(cinfo)));
2606   rbd_get_overlap(child, &overlap);
2607   ASSERT_EQ(overlap, 2UL<<20);
2608   ASSERT_EQ(cinfo.size, 5UL<<20);
2609   ASSERT_EQ(0, rbd_stat(parent, &pinfo, sizeof(pinfo)));
2610   printf("parent info: size %lld obj_size %lld parent_pool %lld\n",
2611          (unsigned long long)pinfo.size, (unsigned long long)pinfo.obj_size,
2612          (unsigned long long)pinfo.parent_pool);
2613   ASSERT_EQ(pinfo.size, 4UL<<20);
2614   printf("sized up clone, changed size but not overlap or parent's size\n");
2615
2616   ASSERT_PASSED(validate_object_map, child);
2617   ASSERT_EQ(0, rbd_close(child));
2618
2619   ASSERT_PASSED(validate_object_map, parent);
2620   ASSERT_EQ(-EBUSY, rbd_snap_remove(parent, "parent_snap"));
2621   printf("can't remove parent while child still exists\n");
2622   ASSERT_EQ(0, rbd_remove(ioctx, child_name.c_str()));
2623   ASSERT_EQ(-EBUSY, rbd_snap_remove(parent, "parent_snap"));
2624   printf("can't remove parent while still protected\n");
2625   ASSERT_EQ(0, rbd_snap_unprotect(parent, "parent_snap"));
2626   ASSERT_EQ(0, rbd_snap_remove(parent, "parent_snap"));
2627   printf("removed parent snap after unprotecting\n");
2628
2629   ASSERT_EQ(0, rbd_close(parent));
2630   rados_ioctx_destroy(ioctx);
2631 }
2632
2633 TEST_F(TestLibRBD, TestClone2)
2634 {
2635   REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
2636
2637   rados_ioctx_t ioctx;
2638   rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx);
2639
2640   bool old_format;
2641   uint64_t features;
2642   rbd_image_t parent, child;
2643   int order = 0;
2644
2645   ASSERT_EQ(0, get_features(&old_format, &features));
2646   ASSERT_FALSE(old_format);
2647
2648   std::string parent_name = get_temp_image_name();
2649   std::string child_name = get_temp_image_name();
2650
2651   // make a parent to clone from
2652   ASSERT_EQ(0, create_image_full(ioctx, parent_name.c_str(), 4<<20, &order,
2653                                  false, features));
2654   ASSERT_EQ(0, rbd_open(ioctx, parent_name.c_str(), &parent, NULL));
2655   printf("made parent image \"parent\"\n");
2656
2657   char *data = (char *)"testdata";
2658   char *childata = (char *)"childata";
2659   ASSERT_EQ((ssize_t)strlen(data), rbd_write(parent, 0, strlen(data), data));
2660   ASSERT_EQ((ssize_t)strlen(data), rbd_write(parent, 12, strlen(data), data));
2661
2662   // can't clone a non-snapshot, expect failure
2663   EXPECT_NE(0, clone_image(ioctx, parent, parent_name.c_str(), NULL, ioctx,
2664                            child_name.c_str(), features, &order));
2665
2666   // verify that there is no parent info on "parent"
2667   ASSERT_EQ(-ENOENT, rbd_get_parent_info(parent, NULL, 0, NULL, 0, NULL, 0));
2668   printf("parent has no parent info\n");
2669
2670   // create a snapshot, reopen as the parent we're interested in
2671   ASSERT_EQ(0, rbd_snap_create(parent, "parent_snap"));
2672   printf("made snapshot \"parent@parent_snap\"\n");
2673   ASSERT_EQ(0, rbd_close(parent));
2674   ASSERT_EQ(0, rbd_open(ioctx, parent_name.c_str(), &parent, "parent_snap"));
2675
2676   ASSERT_EQ(-EINVAL, clone_image(ioctx, parent, parent_name.c_str(), "parent_snap",
2677                                  ioctx, child_name.c_str(), features, &order));
2678
2679   // unprotected image should fail unprotect
2680   ASSERT_EQ(-EINVAL, rbd_snap_unprotect(parent, "parent_snap"));
2681   printf("can't unprotect an unprotected snap\n");
2682
2683   ASSERT_EQ(0, rbd_snap_protect(parent, "parent_snap"));
2684   // protecting again should fail
2685   ASSERT_EQ(-EBUSY, rbd_snap_protect(parent, "parent_snap"));
2686   printf("can't protect a protected snap\n");
2687
2688   // This clone and open should work
2689   ASSERT_EQ(0, clone_image(ioctx, parent, parent_name.c_str(), "parent_snap",
2690                            ioctx, child_name.c_str(), features, &order));
2691   ASSERT_EQ(0, rbd_open(ioctx, child_name.c_str(), &child, NULL));
2692   printf("made and opened clone \"child\"\n");
2693
2694   // write something in
2695   ASSERT_EQ((ssize_t)strlen(childata), rbd_write(child, 20, strlen(childata), childata));
2696
2697   char test[strlen(data) * 2];
2698   ASSERT_EQ((ssize_t)strlen(data), rbd_read(child, 20, strlen(data), test));
2699   ASSERT_EQ(0, memcmp(test, childata, strlen(childata)));
2700
2701   // overlap
2702   ASSERT_EQ((ssize_t)sizeof(test), rbd_read(child, 20 - strlen(data), sizeof(test), test));
2703   ASSERT_EQ(0, memcmp(test, data, strlen(data)));
2704   ASSERT_EQ(0, memcmp(test + strlen(data), childata, strlen(childata)));
2705
2706   // all parent
2707   ASSERT_EQ((ssize_t)sizeof(test), rbd_read(child, 0, sizeof(test), test));
2708   ASSERT_EQ(0, memcmp(test, data, strlen(data)));
2709
2710   ASSERT_PASSED(validate_object_map, child);
2711   ASSERT_PASSED(validate_object_map, parent);
2712
2713   ASSERT_EQ(0, rbd_close(child));
2714   ASSERT_EQ(0, rbd_close(parent));
2715   rados_ioctx_destroy(ioctx);
2716 }
2717
2718 static void test_list_children(rbd_image_t image, ssize_t num_expected, ...)
2719 {
2720   va_list ap;
2721   va_start(ap, num_expected);
2722   size_t pools_len = 100;
2723   size_t children_len = 100;
2724   char *pools = NULL;
2725   char *children = NULL;
2726   ssize_t num_children;
2727
2728   do {
2729     free(pools);
2730     free(children);
2731     pools = (char *) malloc(pools_len);
2732     children = (char *) malloc(children_len);
2733     num_children = rbd_list_children(image, pools, &pools_len,
2734                                      children, &children_len);
2735   } while (num_children == -ERANGE);
2736
2737   ASSERT_EQ(num_expected, num_children);
2738   for (ssize_t i = num_expected; i > 0; --i) {
2739     char *expected_pool = va_arg(ap, char *);
2740     char *expected_image = va_arg(ap, char *);
2741     char *pool = pools;
2742     char *image = children;
2743     bool found = 0;
2744     printf("\ntrying to find %s/%s\n", expected_pool, expected_image);
2745     for (ssize_t j = 0; j < num_children; ++j) {
2746       printf("checking %s/%s\n", pool, image);
2747       if (strcmp(expected_pool, pool) == 0 &&
2748           strcmp(expected_image, image) == 0) {
2749         printf("found child %s/%s\n\n", pool, image);
2750         found = 1;
2751         break;
2752       }
2753       pool += strlen(pool) + 1;
2754       image += strlen(image) + 1;
2755       if (j == num_children - 1) {
2756         ASSERT_EQ(pool - pools - 1, (ssize_t) pools_len);
2757         ASSERT_EQ(image - children - 1, (ssize_t) children_len);
2758       }
2759     }
2760     ASSERT_TRUE(found);
2761   }
2762   va_end(ap);
2763
2764   if (pools)
2765     free(pools);
2766   if (children)
2767     free(children);
2768 }
2769
2770 TEST_F(TestLibRBD, ListChildren)
2771 {
2772   REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
2773
2774   rados_ioctx_t ioctx1, ioctx2;
2775   string pool_name1 = create_pool(true);
2776   string pool_name2 = create_pool(true);
2777   ASSERT_NE("", pool_name2);
2778
2779   rados_ioctx_create(_cluster, pool_name1.c_str(), &ioctx1);
2780   rados_ioctx_create(_cluster, pool_name2.c_str(), &ioctx2);
2781
2782   bool old_format;
2783   uint64_t features;
2784   rbd_image_t parent;
2785   int order = 0;
2786
2787   ASSERT_EQ(0, get_features(&old_format, &features));
2788   ASSERT_FALSE(old_format);
2789
2790   std::string parent_name = get_temp_image_name();
2791   std::string child_name1 = get_temp_image_name();
2792   std::string child_name2 = get_temp_image_name();
2793   std::string child_name3 = get_temp_image_name();
2794   std::string child_name4 = get_temp_image_name();
2795
2796   // make a parent to clone from
2797   ASSERT_EQ(0, create_image_full(ioctx1, parent_name.c_str(), 4<<20, &order,
2798                                  false, features));
2799   ASSERT_EQ(0, rbd_open(ioctx1, parent_name.c_str(), &parent, NULL));
2800   // create a snapshot, reopen as the parent we're interested in
2801   ASSERT_EQ(0, rbd_snap_create(parent, "parent_snap"));
2802   ASSERT_EQ(0, rbd_snap_set(parent, "parent_snap"));
2803   ASSERT_EQ(0, rbd_snap_protect(parent, "parent_snap"));
2804
2805   ASSERT_EQ(0, rbd_close(parent));
2806   ASSERT_EQ(0, rbd_open(ioctx1, parent_name.c_str(), &parent, "parent_snap"));
2807
2808   ASSERT_EQ(0, clone_image(ioctx1, parent, parent_name.c_str(), "parent_snap",
2809                            ioctx2, child_name1.c_str(), features, &order));
2810   test_list_children(parent, 1, pool_name2.c_str(), child_name1.c_str());
2811
2812   ASSERT_EQ(0, clone_image(ioctx1, parent, parent_name.c_str(), "parent_snap",
2813                            ioctx1, child_name2.c_str(), features, &order));
2814   test_list_children(parent, 2, pool_name2.c_str(), child_name1.c_str(),
2815                      pool_name1.c_str(), child_name2.c_str());
2816
2817   ASSERT_EQ(0, clone_image(ioctx1, parent, parent_name.c_str(), "parent_snap",
2818                            ioctx2, child_name3.c_str(), features, &order));
2819   test_list_children(parent, 3, pool_name2.c_str(), child_name1.c_str(),
2820                      pool_name1.c_str(), child_name2.c_str(),
2821                      pool_name2.c_str(), child_name3.c_str());
2822
2823   ASSERT_EQ(0, clone_image(ioctx1, parent, parent_name.c_str(), "parent_snap",
2824                            ioctx2, child_name4.c_str(), features, &order));
2825   test_list_children(parent, 4, pool_name2.c_str(), child_name1.c_str(),
2826                      pool_name1.c_str(), child_name2.c_str(),
2827                      pool_name2.c_str(), child_name3.c_str(),
2828                      pool_name2.c_str(), child_name4.c_str());
2829
2830   ASSERT_EQ(0, rbd_remove(ioctx2, child_name1.c_str()));
2831   test_list_children(parent, 3,
2832                      pool_name1.c_str(), child_name2.c_str(),
2833                      pool_name2.c_str(), child_name3.c_str(),
2834                      pool_name2.c_str(), child_name4.c_str());
2835
2836   ASSERT_EQ(0, rbd_remove(ioctx2, child_name3.c_str()));
2837   test_list_children(parent, 2,
2838                      pool_name1.c_str(), child_name2.c_str(),
2839                      pool_name2.c_str(), child_name4.c_str());
2840
2841   ASSERT_EQ(0, rbd_remove(ioctx2, child_name4.c_str()));
2842   test_list_children(parent, 1,
2843                      pool_name1.c_str(), child_name2.c_str());
2844
2845   ASSERT_EQ(0, rbd_remove(ioctx1, child_name2.c_str()));
2846   test_list_children(parent, 0);
2847
2848   ASSERT_EQ(0, rbd_snap_unprotect(parent, "parent_snap"));
2849   ASSERT_EQ(0, rbd_snap_remove(parent, "parent_snap"));
2850   ASSERT_EQ(0, rbd_close(parent));
2851   ASSERT_EQ(0, rbd_remove(ioctx1, parent_name.c_str()));
2852   rados_ioctx_destroy(ioctx1);
2853   rados_ioctx_destroy(ioctx2);
2854 }
2855
2856 TEST_F(TestLibRBD, ListChildrenTiered)
2857 {
2858   REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
2859
2860   string pool_name1 = m_pool_name;
2861   string pool_name2 = create_pool(true);
2862   string pool_name3 = create_pool(true);
2863   ASSERT_NE("", pool_name2);
2864   ASSERT_NE("", pool_name3);
2865
2866   std::string cmdstr = "{\"prefix\": \"osd tier add\", \"pool\": \"" +
2867      pool_name1 + "\", \"tierpool\":\"" + pool_name3 + "\", \"force_nonempty\":\"\"}";
2868   char *cmd[1];
2869   cmd[0] = (char *)cmdstr.c_str();
2870   ASSERT_EQ(0, rados_mon_command(_cluster, (const char **)cmd, 1, "", 0, NULL, 0, NULL, 0));
2871
2872   cmdstr = "{\"prefix\": \"osd tier cache-mode\", \"pool\": \"" +
2873      pool_name3 + "\", \"mode\":\"writeback\"}";
2874   cmd[0] = (char *)cmdstr.c_str();
2875   ASSERT_EQ(0, rados_mon_command(_cluster, (const char **)cmd, 1, "", 0, NULL, 0, NULL, 0));
2876
2877   cmdstr = "{\"prefix\": \"osd tier set-overlay\", \"pool\": \"" +
2878      pool_name1 + "\", \"overlaypool\":\"" + pool_name3 + "\"}";
2879   cmd[0] = (char *)cmdstr.c_str();
2880   ASSERT_EQ(0, rados_mon_command(_cluster, (const char **)cmd, 1, "", 0, NULL, 0, NULL, 0));
2881
2882   EXPECT_EQ(0, rados_wait_for_latest_osdmap(_cluster));
2883
2884   string parent_name = get_temp_image_name();
2885   string child_name1 = get_temp_image_name();
2886   string child_name2 = get_temp_image_name();
2887   string child_name3 = get_temp_image_name();
2888   string child_name4 = get_temp_image_name();
2889
2890   rados_ioctx_t ioctx1, ioctx2;
2891   rados_ioctx_create(_cluster, pool_name1.c_str(), &ioctx1);
2892   rados_ioctx_create(_cluster, pool_name2.c_str(), &ioctx2);
2893
2894   bool old_format;
2895   uint64_t features;
2896   rbd_image_t parent;
2897   int order = 0;
2898
2899   ASSERT_EQ(0, get_features(&old_format, &features));
2900   ASSERT_FALSE(old_format);
2901
2902   // make a parent to clone from
2903   ASSERT_EQ(0, create_image_full(ioctx1, parent_name.c_str(), 4<<20, &order,
2904                                  false, features));
2905   ASSERT_EQ(0, rbd_open(ioctx1, parent_name.c_str(), &parent, NULL));
2906   // create a snapshot, reopen as the parent we're interested in
2907   ASSERT_EQ(0, rbd_snap_create(parent, "parent_snap"));
2908   ASSERT_EQ(0, rbd_snap_set(parent, "parent_snap"));
2909   ASSERT_EQ(0, rbd_snap_protect(parent, "parent_snap"));
2910
2911   ASSERT_EQ(0, rbd_close(parent));
2912   ASSERT_EQ(0, rbd_open(ioctx1, parent_name.c_str(), &parent, "parent_snap"));
2913
2914   ASSERT_EQ(0, clone_image(ioctx1, parent, parent_name.c_str(), "parent_snap",
2915                            ioctx2, child_name1.c_str(), features, &order));
2916   test_list_children(parent, 1, pool_name2.c_str(), child_name1.c_str());
2917
2918   ASSERT_EQ(0, clone_image(ioctx1, parent, parent_name.c_str(), "parent_snap",
2919                            ioctx1, child_name2.c_str(), features, &order));
2920   test_list_children(parent, 2, pool_name2.c_str(), child_name1.c_str(),
2921                      pool_name1.c_str(), child_name2.c_str());
2922
2923   // read from the cache to populate it
2924   rbd_image_t tier_image;
2925   ASSERT_EQ(0, rbd_open(ioctx1, child_name2.c_str(), &tier_image, NULL));
2926   size_t len = 4 * 1024 * 1024;
2927   char* buf = (char*)malloc(len);
2928   ssize_t size = rbd_read(tier_image, 0, len, buf);
2929   ASSERT_GT(size, 0);
2930   free(buf);
2931   ASSERT_EQ(0, rbd_close(tier_image));
2932
2933   ASSERT_EQ(0, clone_image(ioctx1, parent, parent_name.c_str(), "parent_snap",
2934                            ioctx2, child_name3.c_str(), features, &order));
2935   test_list_children(parent, 3, pool_name2.c_str(), child_name1.c_str(),
2936                      pool_name1.c_str(), child_name2.c_str(),
2937                      pool_name2.c_str(), child_name3.c_str());
2938
2939   ASSERT_EQ(0, clone_image(ioctx1, parent, parent_name.c_str(), "parent_snap",
2940                            ioctx2, child_name4.c_str(), features, &order));
2941   test_list_children(parent, 4, pool_name2.c_str(), child_name1.c_str(),
2942                      pool_name1.c_str(), child_name2.c_str(),
2943                      pool_name2.c_str(), child_name3.c_str(),
2944                      pool_name2.c_str(), child_name4.c_str());
2945
2946   ASSERT_EQ(0, rbd_remove(ioctx2, child_name1.c_str()));
2947   test_list_children(parent, 3,
2948                      pool_name1.c_str(), child_name2.c_str(),
2949                      pool_name2.c_str(), child_name3.c_str(),
2950                      pool_name2.c_str(), child_name4.c_str());
2951
2952   ASSERT_EQ(0, rbd_remove(ioctx2, child_name3.c_str()));
2953   test_list_children(parent, 2,
2954                      pool_name1.c_str(), child_name2.c_str(),
2955                      pool_name2.c_str(), child_name4.c_str());
2956
2957   ASSERT_EQ(0, rbd_remove(ioctx2, child_name4.c_str()));
2958   test_list_children(parent, 1,
2959                      pool_name1.c_str(), child_name2.c_str());
2960
2961   ASSERT_EQ(0, rbd_remove(ioctx1, child_name2.c_str()));
2962   test_list_children(parent, 0);
2963
2964   ASSERT_EQ(0, rbd_snap_unprotect(parent, "parent_snap"));
2965   ASSERT_EQ(0, rbd_snap_remove(parent, "parent_snap"));
2966   ASSERT_EQ(0, rbd_close(parent));
2967   ASSERT_EQ(0, rbd_remove(ioctx1, parent_name.c_str()));
2968   rados_ioctx_destroy(ioctx1);
2969   rados_ioctx_destroy(ioctx2);
2970   cmdstr = "{\"prefix\": \"osd tier remove-overlay\", \"pool\": \"" +
2971      pool_name1 + "\"}";
2972   cmd[0] = (char *)cmdstr.c_str();
2973   ASSERT_EQ(0, rados_mon_command(_cluster, (const char **)cmd, 1, "", 0, NULL, 0, NULL, 0));
2974   cmdstr = "{\"prefix\": \"osd tier remove\", \"pool\": \"" +
2975      pool_name1 + "\", \"tierpool\":\"" + pool_name3 + "\"}";
2976   cmd[0] = (char *)cmdstr.c_str();
2977   ASSERT_EQ(0, rados_mon_command(_cluster, (const char **)cmd, 1, "", 0, NULL, 0, NULL, 0));
2978 }
2979
2980 TEST_F(TestLibRBD, LockingPP)
2981 {
2982   librados::IoCtx ioctx;
2983   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
2984
2985   {
2986     librbd::RBD rbd;
2987     librbd::Image image;
2988     int order = 0;
2989     std::string name = get_temp_image_name();
2990     uint64_t size = 2 << 20;
2991     std::string cookie1 = "foo";
2992     std::string cookie2 = "bar";
2993
2994     ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
2995     ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
2996
2997     // no lockers initially
2998     std::list<librbd::locker_t> lockers;
2999     std::string tag;
3000     bool exclusive;
3001     ASSERT_EQ(0, image.list_lockers(&lockers, &exclusive, &tag));
3002     ASSERT_EQ(0u, lockers.size());
3003     ASSERT_EQ("", tag);
3004
3005     // exclusive lock is exclusive
3006     ASSERT_EQ(0, image.lock_exclusive(cookie1));
3007     ASSERT_EQ(-EEXIST, image.lock_exclusive(cookie1));
3008     ASSERT_EQ(-EBUSY, image.lock_exclusive(""));
3009     ASSERT_EQ(-EEXIST, image.lock_shared(cookie1, ""));
3010     ASSERT_EQ(-EBUSY, image.lock_shared(cookie1, "test"));
3011     ASSERT_EQ(-EBUSY, image.lock_shared("", "test"));
3012     ASSERT_EQ(-EBUSY, image.lock_shared("", ""));
3013
3014     // list exclusive
3015     ASSERT_EQ(0, image.list_lockers(&lockers, &exclusive, &tag));
3016     ASSERT_TRUE(exclusive);
3017     ASSERT_EQ("", tag);
3018     ASSERT_EQ(1u, lockers.size());
3019     ASSERT_EQ(cookie1, lockers.front().cookie);
3020
3021     // unlock
3022     ASSERT_EQ(-ENOENT, image.unlock(""));
3023     ASSERT_EQ(-ENOENT, image.unlock(cookie2));
3024     ASSERT_EQ(0, image.unlock(cookie1));
3025     ASSERT_EQ(-ENOENT, image.unlock(cookie1));
3026     ASSERT_EQ(0, image.list_lockers(&lockers, &exclusive, &tag));
3027     ASSERT_EQ(0u, lockers.size());
3028
3029     ASSERT_EQ(0, image.lock_shared(cookie1, ""));
3030     ASSERT_EQ(-EEXIST, image.lock_shared(cookie1, ""));
3031     ASSERT_EQ(0, image.lock_shared(cookie2, ""));
3032     ASSERT_EQ(-EEXIST, image.lock_shared(cookie2, ""));
3033     ASSERT_EQ(-EEXIST, image.lock_exclusive(cookie1));
3034     ASSERT_EQ(-EEXIST, image.lock_exclusive(cookie2));
3035     ASSERT_EQ(-EBUSY, image.lock_exclusive(""));
3036     ASSERT_EQ(-EBUSY, image.lock_exclusive("test"));
3037
3038     // list shared
3039     ASSERT_EQ(0, image.list_lockers(&lockers, &exclusive, &tag));
3040     ASSERT_EQ(2u, lockers.size());
3041   }
3042
3043   ioctx.close();
3044 }
3045
3046 TEST_F(TestLibRBD, FlushAio)
3047 {
3048   rados_ioctx_t ioctx;
3049   rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx);
3050
3051   rbd_image_t image;
3052   int order = 0;
3053   std::string name = get_temp_image_name();
3054   uint64_t size = 2 << 20;
3055   size_t num_aios = 256;
3056
3057   ASSERT_EQ(0, create_image(ioctx, name.c_str(), size, &order));
3058   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
3059
3060   char test_data[TEST_IO_SIZE + 1];
3061   size_t i;
3062   for (i = 0; i < TEST_IO_SIZE; ++i) {
3063     test_data[i] = (char) (rand() % (126 - 33) + 33);
3064   }
3065
3066   rbd_completion_t write_comps[num_aios];
3067   for (i = 0; i < num_aios; ++i) {
3068     ASSERT_EQ(0, rbd_aio_create_completion(NULL, NULL, &write_comps[i]));
3069     uint64_t offset = rand() % (size - TEST_IO_SIZE);
3070     ASSERT_EQ(0, rbd_aio_write(image, offset, TEST_IO_SIZE, test_data,
3071                                write_comps[i]));
3072   }
3073
3074   rbd_completion_t flush_comp;
3075   ASSERT_EQ(0, rbd_aio_create_completion(NULL, NULL, &flush_comp));
3076   ASSERT_EQ(0, rbd_aio_flush(image, flush_comp));
3077   ASSERT_EQ(0, rbd_aio_wait_for_complete(flush_comp));
3078   ASSERT_EQ(1, rbd_aio_is_complete(flush_comp));
3079   rbd_aio_release(flush_comp);
3080
3081   for (i = 0; i < num_aios; ++i) {
3082     ASSERT_EQ(1, rbd_aio_is_complete(write_comps[i]));
3083     rbd_aio_release(write_comps[i]);
3084   }
3085
3086   ASSERT_PASSED(validate_object_map, image);
3087   ASSERT_EQ(0, rbd_close(image));
3088   ASSERT_EQ(0, rbd_remove(ioctx, name.c_str()));
3089   rados_ioctx_destroy(ioctx);
3090 }
3091
3092 TEST_F(TestLibRBD, FlushAioPP)
3093 {
3094   librados::IoCtx ioctx;
3095   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
3096
3097   {
3098     librbd::RBD rbd;
3099     librbd::Image image;
3100     int order = 0;
3101     std::string name = get_temp_image_name();
3102     uint64_t size = 2 << 20;
3103     const size_t num_aios = 256;
3104
3105     ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
3106     ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
3107
3108     char test_data[TEST_IO_SIZE + 1];
3109     size_t i;
3110     for (i = 0; i < TEST_IO_SIZE; ++i) {
3111       test_data[i] = (char) (rand() % (126 - 33) + 33);
3112     }
3113     test_data[TEST_IO_SIZE] = '\0';
3114
3115     librbd::RBD::AioCompletion *write_comps[num_aios];
3116     ceph::bufferlist bls[num_aios];
3117     for (i = 0; i < num_aios; ++i) {
3118       bls[i].append(test_data, strlen(test_data));
3119       write_comps[i] = new librbd::RBD::AioCompletion(NULL, NULL);
3120       uint64_t offset = rand() % (size - TEST_IO_SIZE);
3121       ASSERT_EQ(0, image.aio_write(offset, TEST_IO_SIZE, bls[i],
3122                                    write_comps[i]));
3123     }
3124
3125     librbd::RBD::AioCompletion *flush_comp =
3126       new librbd::RBD::AioCompletion(NULL, NULL);
3127     ASSERT_EQ(0, image.aio_flush(flush_comp));
3128     ASSERT_EQ(0, flush_comp->wait_for_complete());
3129     ASSERT_EQ(1, flush_comp->is_complete());
3130     flush_comp->release();
3131
3132     for (i = 0; i < num_aios; ++i) {
3133       librbd::RBD::AioCompletion *comp = write_comps[i];
3134       ASSERT_EQ(1, comp->is_complete());
3135       comp->release();
3136     }
3137     ASSERT_PASSED(validate_object_map, image);
3138   }
3139
3140   ioctx.close();
3141 }
3142
3143
3144 int iterate_cb(uint64_t off, size_t len, int exists, void *arg)
3145 {
3146   //cout << "iterate_cb " << off << "~" << len << std::endl;
3147   interval_set<uint64_t> *diff = static_cast<interval_set<uint64_t> *>(arg);
3148   diff->insert(off, len);
3149   return 0;
3150 }
3151
3152 static int iterate_error_cb(uint64_t off, size_t len, int exists, void *arg)
3153 {
3154   return -EINVAL;
3155 }
3156
3157 void scribble(librbd::Image& image, int n, int max, bool skip_discard,
3158               interval_set<uint64_t> *exists,
3159               interval_set<uint64_t> *what)
3160 {
3161   uint64_t size;
3162   image.size(&size);
3163   interval_set<uint64_t> exists_at_start = *exists;
3164
3165   for (int i=0; i<n; i++) {
3166     uint64_t off = rand() % (size - max + 1);
3167     uint64_t len = 1 + rand() % max;
3168     if (!skip_discard && rand() % 4 == 0) {
3169       ASSERT_EQ((int)len, image.discard(off, len));
3170       interval_set<uint64_t> w;
3171       w.insert(off, len);
3172
3173       // the zeroed bit no longer exists...
3174       w.intersection_of(*exists);
3175       exists->subtract(w);
3176
3177       // the bits we discarded are no long written...
3178       interval_set<uint64_t> w2 = w;
3179       w2.intersection_of(*what);
3180       what->subtract(w2);
3181
3182       // except for the extents that existed at the start that we overwrote.
3183       interval_set<uint64_t> w3;
3184       w3.insert(off, len);
3185       w3.intersection_of(exists_at_start);
3186       what->union_of(w3);
3187
3188     } else {
3189       bufferlist bl;
3190       bl.append(buffer::create(len));
3191       bl.zero();
3192       ASSERT_EQ((int)len, image.write(off, len, bl));
3193       interval_set<uint64_t> w;
3194       w.insert(off, len);
3195       what->union_of(w);
3196       exists->union_of(w);
3197     }
3198   }
3199 }
3200
3201 interval_set<uint64_t> round_diff_interval(const interval_set<uint64_t>& diff,
3202                                            uint64_t object_size)
3203 {
3204   if (object_size == 0) {
3205     return diff;
3206   }
3207
3208   interval_set<uint64_t> rounded_diff;
3209   for (interval_set<uint64_t>::const_iterator it = diff.begin();
3210        it != diff.end(); ++it) {
3211     uint64_t off = it.get_start();
3212     uint64_t len = it.get_len();
3213     off -= off % object_size;
3214     len += (object_size - (len % object_size));
3215     interval_set<uint64_t> interval;
3216     interval.insert(off, len);
3217     rounded_diff.union_of(interval);
3218   }
3219   return rounded_diff;
3220 }
3221
3222 template <typename T>
3223 class DiffIterateTest : public TestLibRBD {
3224 public:
3225   static const uint8_t whole_object = T::whole_object;
3226 };
3227
3228 template <bool _whole_object>
3229 class DiffIterateParams {
3230 public:
3231   static const uint8_t whole_object = _whole_object;
3232 };
3233
3234 typedef ::testing::Types<DiffIterateParams<false>,
3235                          DiffIterateParams<true> > DiffIterateTypes;
3236 TYPED_TEST_CASE(DiffIterateTest, DiffIterateTypes);
3237
3238 TYPED_TEST(DiffIterateTest, DiffIterate)
3239 {
3240   librados::IoCtx ioctx;
3241   ASSERT_EQ(0, this->_rados.ioctx_create(this->m_pool_name.c_str(), ioctx));
3242
3243   bool skip_discard = this->is_skip_partial_discard_enabled();
3244
3245   {
3246     librbd::RBD rbd;
3247     librbd::Image image;
3248     int order = 0;
3249     std::string name = this->get_temp_image_name();
3250     uint64_t size = 20 << 20;
3251
3252     ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
3253     ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
3254
3255     uint64_t object_size = 0;
3256     if (this->whole_object) {
3257       object_size = 1 << order;
3258     }
3259
3260     interval_set<uint64_t> exists;
3261     interval_set<uint64_t> one, two;
3262     scribble(image, 10, 102400, skip_discard, &exists, &one);
3263     cout << " wrote " << one << std::endl;
3264     ASSERT_EQ(0, image.snap_create("one"));
3265     scribble(image, 10, 102400, skip_discard, &exists, &two);
3266
3267     two = round_diff_interval(two, object_size);
3268     cout << " wrote " << two << std::endl;
3269
3270     interval_set<uint64_t> diff;
3271     ASSERT_EQ(0, image.diff_iterate2("one", 0, size, true, this->whole_object,
3272                                      iterate_cb, (void *)&diff));
3273     cout << " diff was " << diff << std::endl;
3274     if (!two.subset_of(diff)) {
3275       interval_set<uint64_t> i;
3276       i.intersection_of(two, diff);
3277       interval_set<uint64_t> l = two;
3278       l.subtract(i);
3279       cout << " ... two - (two*diff) = " << l << std::endl;
3280     }
3281     ASSERT_TRUE(two.subset_of(diff));
3282   }
3283   ioctx.close();
3284 }
3285
3286 struct diff_extent {
3287   diff_extent(uint64_t _offset, uint64_t _length, bool _exists,
3288               uint64_t object_size) :
3289     offset(_offset), length(_length), exists(_exists)
3290   {
3291     if (object_size != 0) {
3292       offset -= offset % object_size;
3293       length = object_size;
3294     }
3295   }
3296   uint64_t offset;
3297   uint64_t length;
3298   bool exists;
3299   bool operator==(const diff_extent& o) const {
3300     return offset == o.offset && length == o.length && exists == o.exists;
3301   }
3302 };
3303
3304 ostream& operator<<(ostream & o, const diff_extent& e) {
3305   return o << '(' << e.offset << '~' << e.length << ' ' << (e.exists ? "true" : "false") << ')';
3306 }
3307
3308 int vector_iterate_cb(uint64_t off, size_t len, int exists, void *arg)
3309 {
3310   cout << "iterate_cb " << off << "~" << len << std::endl;
3311   vector<diff_extent> *diff = static_cast<vector<diff_extent> *>(arg);
3312   diff->push_back(diff_extent(off, len, exists, 0));
3313   return 0;
3314 }
3315
3316 TYPED_TEST(DiffIterateTest, DiffIterateDiscard)
3317 {
3318   librados::IoCtx ioctx;
3319   ASSERT_EQ(0, this->_rados.ioctx_create(this->m_pool_name.c_str(), ioctx));
3320
3321   librbd::RBD rbd;
3322   librbd::Image image;
3323   int order = 0;
3324   std::string name = this->get_temp_image_name();
3325   uint64_t size = 20 << 20;
3326
3327   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
3328   ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
3329
3330   uint64_t object_size = 0;
3331   if (this->whole_object) {
3332     object_size = 1 << order;
3333   }
3334   vector<diff_extent> extents;
3335   ceph::bufferlist bl;
3336
3337   ASSERT_EQ(0, image.diff_iterate2(NULL, 0, size, true, this->whole_object,
3338                                    vector_iterate_cb, (void *) &extents));
3339   ASSERT_EQ(0u, extents.size());
3340
3341   char data[256];
3342   memset(data, 1, sizeof(data));
3343   bl.append(data, 256);
3344   ASSERT_EQ(256, image.write(0, 256, bl));
3345   ASSERT_EQ(0, image.diff_iterate2(NULL, 0, size, true, this->whole_object,
3346                                    vector_iterate_cb, (void *) &extents));
3347   ASSERT_EQ(1u, extents.size());
3348   ASSERT_EQ(diff_extent(0, 256, true, object_size), extents[0]);
3349
3350   int obj_ofs = 256;
3351   ASSERT_EQ(1 << order, image.discard(0, 1 << order));
3352
3353   extents.clear();
3354   ASSERT_EQ(0, image.diff_iterate2(NULL, 0, size, true, this->whole_object,
3355                                    vector_iterate_cb, (void *) &extents));
3356   ASSERT_EQ(0u, extents.size());
3357
3358   ASSERT_EQ(0, image.snap_create("snap1"));
3359   ASSERT_EQ(256, image.write(0, 256, bl));
3360   ASSERT_EQ(0, image.diff_iterate2(NULL, 0, size, true, this->whole_object,
3361                                    vector_iterate_cb, (void *) &extents));
3362   ASSERT_EQ(1u, extents.size());
3363   ASSERT_EQ(diff_extent(0, 256, true, object_size), extents[0]);
3364   ASSERT_EQ(0, image.snap_create("snap2"));
3365
3366   ASSERT_EQ(obj_ofs, image.discard(0, obj_ofs));
3367
3368   extents.clear();
3369   ASSERT_EQ(0, image.snap_set("snap2"));
3370   ASSERT_EQ(0, image.diff_iterate2("snap1", 0, size, true, this->whole_object,
3371                                    vector_iterate_cb, (void *) &extents));
3372   ASSERT_EQ(1u, extents.size());
3373   ASSERT_EQ(diff_extent(0, 256, true, object_size), extents[0]);
3374
3375   ASSERT_EQ(0, image.snap_set(NULL));
3376   ASSERT_EQ(1 << order, image.discard(0, 1 << order));
3377   ASSERT_EQ(0, image.snap_create("snap3"));
3378   ASSERT_EQ(0, image.snap_set("snap3"));
3379
3380   extents.clear();
3381   ASSERT_EQ(0, image.diff_iterate2("snap1", 0, size, true, this->whole_object,
3382                                    vector_iterate_cb, (void *) &extents));
3383   ASSERT_EQ(1u, extents.size());
3384   ASSERT_EQ(diff_extent(0, 256, false, object_size), extents[0]);
3385   ASSERT_PASSED(this->validate_object_map, image);
3386 }
3387
3388 TYPED_TEST(DiffIterateTest, DiffIterateStress)
3389 {
3390   librados::IoCtx ioctx;
3391   ASSERT_EQ(0, this->_rados.ioctx_create(this->m_pool_name.c_str(), ioctx));
3392
3393   bool skip_discard = this->is_skip_partial_discard_enabled();
3394
3395   librbd::RBD rbd;
3396   librbd::Image image;
3397   int order = 0;
3398   std::string name = this->get_temp_image_name();
3399   uint64_t size = 400 << 20;
3400
3401   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
3402   ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
3403
3404   uint64_t object_size = 0;
3405   if (this->whole_object) {
3406     object_size = 1 << order;
3407   }
3408
3409   interval_set<uint64_t> curexists;
3410   vector<interval_set<uint64_t> > wrote;
3411   vector<interval_set<uint64_t> > exists;
3412   vector<string> snap;
3413   int n = 20;
3414   for (int i=0; i<n; i++) {
3415     interval_set<uint64_t> w;
3416     scribble(image, 10, 8192000, skip_discard, &curexists, &w);
3417     cout << " i=" << i << " exists " << curexists << " wrote " << w << std::endl;
3418     string s = "snap" + stringify(i);
3419     ASSERT_EQ(0, image.snap_create(s.c_str()));
3420     wrote.push_back(w);
3421     exists.push_back(curexists);
3422     snap.push_back(s);
3423   }
3424
3425   for (int h=0; h<n-1; h++) {
3426     for (int i=0; i<n-h-1; i++) {
3427       for (int j=(h==0 ? i+1 : n-1); j<n; j++) {
3428         interval_set<uint64_t> diff, actual, uex;
3429         for (int k=i+1; k<=j; k++)
3430           diff.union_of(wrote[k]);
3431         cout << "from " << i << " to "
3432              << (h != 0 ? string("HEAD") : stringify(j)) << " diff "
3433              << round_diff_interval(diff, object_size) << std::endl;
3434
3435         // limit to extents that exists both at the beginning and at the end
3436         uex.union_of(exists[i], exists[j]);
3437         diff.intersection_of(uex);
3438         diff = round_diff_interval(diff, object_size);
3439         cout << " limited diff " << diff << std::endl;
3440
3441         ASSERT_EQ(0, image.snap_set(h==0 ? snap[j].c_str() : NULL));
3442         ASSERT_EQ(0, image.diff_iterate2(snap[i].c_str(), 0, size, true,
3443                                          this->whole_object, iterate_cb,
3444                                          (void *)&actual));
3445         cout << " actual was " << actual << std::endl;
3446         if (!diff.subset_of(actual)) {
3447           interval_set<uint64_t> i;
3448           i.intersection_of(diff, actual);
3449           interval_set<uint64_t> l = diff;
3450           l.subtract(i);
3451           cout << " ... diff - (actual*diff) = " << l << std::endl;
3452         }
3453         ASSERT_TRUE(diff.subset_of(actual));
3454       }
3455     }
3456     ASSERT_EQ(0, image.snap_set(NULL));
3457     ASSERT_EQ(0, image.snap_remove(snap[n-h-1].c_str()));
3458   }
3459
3460   ASSERT_PASSED(this->validate_object_map, image);
3461 }
3462
3463 TYPED_TEST(DiffIterateTest, DiffIterateRegression6926)
3464 {
3465   librados::IoCtx ioctx;
3466   ASSERT_EQ(0, this->_rados.ioctx_create(this->m_pool_name.c_str(), ioctx));
3467
3468   librbd::RBD rbd;
3469   librbd::Image image;
3470   int order = 0;
3471   std::string name = this->get_temp_image_name();
3472   uint64_t size = 20 << 20;
3473
3474   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
3475   ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
3476
3477   uint64_t object_size = 0;
3478   if (this->whole_object) {
3479     object_size = 1 << order;
3480   }
3481   vector<diff_extent> extents;
3482   ceph::bufferlist bl;
3483
3484   ASSERT_EQ(0, image.diff_iterate2(NULL, 0, size, true, this->whole_object,
3485                                    vector_iterate_cb, (void *) &extents));
3486   ASSERT_EQ(0u, extents.size());
3487
3488   ASSERT_EQ(0, image.snap_create("snap1"));
3489   char data[256];
3490   memset(data, 1, sizeof(data));
3491   bl.append(data, 256);
3492   ASSERT_EQ(256, image.write(0, 256, bl));
3493
3494   extents.clear();
3495   ASSERT_EQ(0, image.diff_iterate2(NULL, 0, size, true, this->whole_object,
3496                                    vector_iterate_cb, (void *) &extents));
3497   ASSERT_EQ(1u, extents.size());
3498   ASSERT_EQ(diff_extent(0, 256, true, object_size), extents[0]);
3499
3500   ASSERT_EQ(0, image.snap_set("snap1"));
3501   extents.clear();
3502   ASSERT_EQ(0, image.diff_iterate2(NULL, 0, size, true, this->whole_object,
3503                                    vector_iterate_cb, (void *) &extents));
3504   ASSERT_EQ(static_cast<size_t>(0), extents.size());
3505 }
3506
3507 TYPED_TEST(DiffIterateTest, DiffIterateIgnoreParent)
3508 {
3509   REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
3510
3511   librados::IoCtx ioctx;
3512   ASSERT_EQ(0, this->_rados.ioctx_create(this->m_pool_name.c_str(), ioctx));
3513
3514   bool skip_discard = this->is_skip_partial_discard_enabled();
3515
3516   librbd::RBD rbd;
3517   librbd::Image image;
3518   std::string name = this->get_temp_image_name();
3519   uint64_t size = 20 << 20;
3520   int order = 0;
3521
3522   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
3523   ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
3524
3525   uint64_t object_size = 0;
3526   if (this->whole_object) {
3527     object_size = 1 << order;
3528   }
3529
3530   bufferlist bl;
3531   bl.append(buffer::create(size));
3532   bl.zero();
3533   interval_set<uint64_t> one;
3534   one.insert(0, size);
3535   ASSERT_EQ((int)size, image.write(0, size, bl));
3536   ASSERT_EQ(0, image.snap_create("one"));
3537   ASSERT_EQ(0, image.snap_protect("one"));
3538
3539   std::string clone_name = this->get_temp_image_name();
3540   ASSERT_EQ(0, rbd.clone(ioctx, name.c_str(), "one", ioctx, clone_name.c_str(),
3541                          RBD_FEATURE_LAYERING, &order));
3542   ASSERT_EQ(0, rbd.open(ioctx, image, clone_name.c_str(), NULL));
3543
3544   interval_set<uint64_t> exists;
3545   interval_set<uint64_t> two;
3546   scribble(image, 10, 102400, skip_discard, &exists, &two);
3547   two = round_diff_interval(two, object_size);
3548   cout << " wrote " << two << " to clone" << std::endl;
3549
3550   interval_set<uint64_t> diff;
3551   ASSERT_EQ(0, image.diff_iterate2(NULL, 0, size, false, this->whole_object,
3552                                    iterate_cb, (void *)&diff));
3553   cout << " diff was " << diff << std::endl;
3554   if (!this->whole_object) {
3555     ASSERT_FALSE(one.subset_of(diff));
3556   }
3557   ASSERT_TRUE(two.subset_of(diff));
3558 }
3559
3560 TYPED_TEST(DiffIterateTest, DiffIterateCallbackError)
3561 {
3562   librados::IoCtx ioctx;
3563   ASSERT_EQ(0, this->_rados.ioctx_create(this->m_pool_name.c_str(), ioctx));
3564
3565   bool skip_discard = this->is_skip_partial_discard_enabled();
3566
3567   {
3568     librbd::RBD rbd;
3569     librbd::Image image;
3570     int order = 0;
3571     std::string name = this->get_temp_image_name();
3572     uint64_t size = 20 << 20;
3573
3574     ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
3575     ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
3576
3577     interval_set<uint64_t> exists;
3578     interval_set<uint64_t> one;
3579     scribble(image, 10, 102400, skip_discard, &exists, &one);
3580     cout << " wrote " << one << std::endl;
3581
3582     interval_set<uint64_t> diff;
3583     ASSERT_EQ(-EINVAL, image.diff_iterate2(NULL, 0, size, true,
3584                                            this->whole_object,
3585                                            iterate_error_cb, NULL));
3586   }
3587   ioctx.close();
3588 }
3589
3590 TYPED_TEST(DiffIterateTest, DiffIterateParentDiscard)
3591 {
3592   REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
3593
3594   librados::IoCtx ioctx;
3595   ASSERT_EQ(0, this->_rados.ioctx_create(this->m_pool_name.c_str(), ioctx));
3596
3597   bool skip_discard = this->is_skip_partial_discard_enabled();
3598
3599   librbd::RBD rbd;
3600   librbd::Image image;
3601   std::string name = this->get_temp_image_name();
3602   uint64_t size = 20 << 20;
3603   int order = 0;
3604
3605   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
3606   ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
3607
3608   uint64_t object_size = 0;
3609   if (this->whole_object) {
3610     object_size = 1 << order;
3611   }
3612
3613   interval_set<uint64_t> exists;
3614   interval_set<uint64_t> one;
3615   scribble(image, 10, 102400, skip_discard, &exists, &one);
3616   ASSERT_EQ(0, image.snap_create("one"));
3617
3618   ASSERT_EQ(1 << order, image.discard(0, 1 << order));
3619   ASSERT_EQ(0, image.snap_create("two"));
3620   ASSERT_EQ(0, image.snap_protect("two"));
3621   exists.clear();
3622   one.clear();
3623
3624   std::string clone_name = this->get_temp_image_name();
3625   ASSERT_EQ(0, rbd.clone(ioctx, name.c_str(), "two", ioctx,
3626                          clone_name.c_str(), RBD_FEATURE_LAYERING, &order));
3627   ASSERT_EQ(0, rbd.open(ioctx, image, clone_name.c_str(), NULL));
3628
3629   interval_set<uint64_t> two;
3630   scribble(image, 10, 102400, skip_discard, &exists, &two);
3631   two = round_diff_interval(two, object_size);
3632
3633   interval_set<uint64_t> diff;
3634   ASSERT_EQ(0, image.diff_iterate2(NULL, 0, size, true, this->whole_object,
3635                                    iterate_cb, (void *)&diff));
3636   ASSERT_TRUE(two.subset_of(diff));
3637 }
3638
3639 TEST_F(TestLibRBD, ZeroLengthWrite)
3640 {
3641   rados_ioctx_t ioctx;
3642   rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx);
3643
3644   rbd_image_t image;
3645   int order = 0;
3646   std::string name = get_temp_image_name();
3647   uint64_t size = 2 << 20;
3648
3649   ASSERT_EQ(0, create_image(ioctx, name.c_str(), size, &order));
3650   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
3651
3652   char read_data[1];
3653   ASSERT_EQ(0, rbd_write(image, 0, 0, NULL));
3654   ASSERT_EQ(1, rbd_read(image, 0, 1, read_data));
3655   ASSERT_EQ('\0', read_data[0]);
3656
3657   ASSERT_PASSED(validate_object_map, image);
3658   ASSERT_EQ(0, rbd_close(image));
3659
3660   rados_ioctx_destroy(ioctx);
3661 }
3662
3663
3664 TEST_F(TestLibRBD, ZeroLengthDiscard)
3665 {
3666   rados_ioctx_t ioctx;
3667   rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx);
3668
3669   rbd_image_t image;
3670   int order = 0;
3671   std::string name = get_temp_image_name();
3672   uint64_t size = 2 << 20;
3673
3674   ASSERT_EQ(0, create_image(ioctx, name.c_str(), size, &order));
3675   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
3676
3677   const char data[] = "blah";
3678   char read_data[sizeof(data)];
3679   ASSERT_EQ((int)strlen(data), rbd_write(image, 0, strlen(data), data));
3680   ASSERT_EQ(0, rbd_discard(image, 0, 0));
3681   ASSERT_EQ((int)strlen(data), rbd_read(image, 0, strlen(data), read_data));
3682   ASSERT_EQ(0, memcmp(data, read_data, strlen(data)));
3683
3684   ASSERT_PASSED(validate_object_map, image);
3685   ASSERT_EQ(0, rbd_close(image));
3686
3687   rados_ioctx_destroy(ioctx);
3688 }
3689
3690 TEST_F(TestLibRBD, ZeroLengthRead)
3691 {
3692   rados_ioctx_t ioctx;
3693   rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx);
3694
3695   rbd_image_t image;
3696   int order = 0;
3697   std::string name = get_temp_image_name();
3698   uint64_t size = 2 << 20;
3699
3700   ASSERT_EQ(0, create_image(ioctx, name.c_str(), size, &order));
3701   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
3702
3703   char read_data[1];
3704   ASSERT_EQ(0, rbd_read(image, 0, 0, read_data));
3705
3706   ASSERT_EQ(0, rbd_close(image));
3707
3708   rados_ioctx_destroy(ioctx);
3709 }
3710
3711 TEST_F(TestLibRBD, LargeCacheRead)
3712 {
3713   std::string config_value;
3714   ASSERT_EQ(0, _rados.conf_get("rbd_cache", config_value));
3715   if (config_value == "false") {
3716     std::cout << "SKIPPING due to disabled cache" << std::endl;
3717     return;
3718   }
3719
3720   rados_ioctx_t ioctx;
3721   rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx);
3722
3723   uint32_t new_cache_size = 1 << 20;
3724   std::string orig_cache_size;
3725   ASSERT_EQ(0, _rados.conf_get("rbd_cache_size", orig_cache_size));
3726   ASSERT_EQ(0, _rados.conf_set("rbd_cache_size",
3727                                stringify(new_cache_size).c_str()));
3728   ASSERT_EQ(0, _rados.conf_get("rbd_cache_size", config_value));
3729   ASSERT_EQ(stringify(new_cache_size), config_value);
3730   BOOST_SCOPE_EXIT( (orig_cache_size) ) {
3731     ASSERT_EQ(0, _rados.conf_set("rbd_cache_size", orig_cache_size.c_str()));
3732   } BOOST_SCOPE_EXIT_END;
3733
3734   rbd_image_t image;
3735   int order = 21;
3736   std::string name = get_temp_image_name();
3737   uint64_t size = 1 << order;
3738
3739   ASSERT_EQ(0, create_image(ioctx, name.c_str(), size, &order));
3740   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
3741
3742   std::string buffer(1 << order, '1');
3743
3744   ASSERT_EQ(static_cast<ssize_t>(buffer.size()),
3745             rbd_write(image, 0, buffer.size(), buffer.c_str()));
3746
3747   ASSERT_EQ(0, rbd_invalidate_cache(image));
3748
3749   ASSERT_EQ(static_cast<ssize_t>(buffer.size()),
3750             rbd_read(image, 0, buffer.size(), &buffer[0]));
3751
3752   ASSERT_EQ(0, rbd_close(image));
3753
3754   rados_ioctx_destroy(ioctx);
3755 }
3756
3757 TEST_F(TestLibRBD, TestPendingAio)
3758 {
3759   REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
3760
3761   rados_ioctx_t ioctx;
3762   rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx);
3763
3764   bool old_format;
3765   uint64_t features;
3766   rbd_image_t image;
3767   int order = 0;
3768
3769   ASSERT_EQ(0, get_features(&old_format, &features));
3770   ASSERT_FALSE(old_format);
3771
3772   std::string name = get_temp_image_name();
3773
3774   uint64_t size = 4 << 20;
3775   ASSERT_EQ(0, create_image_full(ioctx, name.c_str(), size, &order,
3776                                  false, features));
3777   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
3778
3779   char test_data[TEST_IO_SIZE];
3780   for (size_t i = 0; i < TEST_IO_SIZE; ++i) {
3781     test_data[i] = (char) (rand() % (126 - 33) + 33);
3782   }
3783
3784   size_t num_aios = 256;
3785   rbd_completion_t comps[num_aios];
3786   for (size_t i = 0; i < num_aios; ++i) {
3787     ASSERT_EQ(0, rbd_aio_create_completion(NULL, NULL, &comps[i]));
3788     uint64_t offset = rand() % (size - TEST_IO_SIZE);
3789     ASSERT_EQ(0, rbd_aio_write(image, offset, TEST_IO_SIZE, test_data,
3790                                comps[i]));
3791   }
3792   for (size_t i = 0; i < num_aios; ++i) {
3793     ASSERT_EQ(0, rbd_aio_wait_for_complete(comps[i]));
3794     rbd_aio_release(comps[i]);
3795   }
3796   ASSERT_EQ(0, rbd_invalidate_cache(image));
3797
3798   for (size_t i = 0; i < num_aios; ++i) {
3799     ASSERT_EQ(0, rbd_aio_create_completion(NULL, NULL, &comps[i]));
3800     uint64_t offset = rand() % (size - TEST_IO_SIZE);
3801     ASSERT_LE(0, rbd_aio_read(image, offset, TEST_IO_SIZE, test_data,
3802                               comps[i]));
3803   }
3804
3805   ASSERT_PASSED(validate_object_map, image);
3806   ASSERT_EQ(0, rbd_close(image));
3807   for (size_t i = 0; i < num_aios; ++i) {
3808     ASSERT_EQ(1, rbd_aio_is_complete(comps[i]));
3809     rbd_aio_release(comps[i]);
3810   }
3811
3812   rados_ioctx_destroy(ioctx);
3813 }
3814
3815 TEST_F(TestLibRBD, Flatten)
3816 {
3817   REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
3818
3819   librados::IoCtx ioctx;
3820   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
3821
3822   librbd::RBD rbd;
3823   std::string parent_name = get_temp_image_name();
3824   uint64_t size = 2 << 20;
3825   int order = 0;
3826   ASSERT_EQ(0, create_image_pp(rbd, ioctx, parent_name.c_str(), size, &order));
3827
3828   librbd::Image parent_image;
3829   ASSERT_EQ(0, rbd.open(ioctx, parent_image, parent_name.c_str(), NULL));
3830
3831   bufferlist bl;
3832   bl.append(std::string(4096, '1'));
3833   ASSERT_EQ((ssize_t)bl.length(), parent_image.write(0, bl.length(), bl));
3834
3835   ASSERT_EQ(0, parent_image.snap_create("snap1"));
3836   ASSERT_EQ(0, parent_image.snap_protect("snap1"));
3837
3838   uint64_t features;
3839   ASSERT_EQ(0, parent_image.features(&features));
3840
3841   std::string clone_name = get_temp_image_name();
3842   EXPECT_EQ(0, rbd.clone(ioctx, parent_name.c_str(), "snap1", ioctx,
3843        clone_name.c_str(), features, &order));
3844
3845   librbd::Image clone_image;
3846   ASSERT_EQ(0, rbd.open(ioctx, clone_image, clone_name.c_str(), NULL));
3847   ASSERT_EQ(0, clone_image.flatten());
3848
3849   librbd::RBD::AioCompletion *read_comp =
3850     new librbd::RBD::AioCompletion(NULL, NULL);
3851   bufferlist read_bl;
3852   clone_image.aio_read(0, bl.length(), read_bl, read_comp);
3853   ASSERT_EQ(0, read_comp->wait_for_complete());
3854   ASSERT_EQ((ssize_t)bl.length(), read_comp->get_return_value());
3855   read_comp->release();
3856   ASSERT_TRUE(bl.contents_equal(read_bl));
3857
3858   ASSERT_PASSED(validate_object_map, clone_image);
3859 }
3860
3861 TEST_F(TestLibRBD, SnapshotLimit)
3862 {
3863   rados_ioctx_t ioctx;
3864   rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx);
3865
3866   rbd_image_t image;
3867   int order = 0;
3868   std::string name = get_temp_image_name();
3869   uint64_t size = 2 << 20;
3870   uint64_t limit;
3871
3872   ASSERT_EQ(0, create_image(ioctx, name.c_str(), size, &order));
3873   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
3874
3875   ASSERT_EQ(0, rbd_snap_get_limit(image, &limit));
3876   ASSERT_EQ(UINT64_MAX, limit);
3877   ASSERT_EQ(0, rbd_snap_set_limit(image, 2));
3878   ASSERT_EQ(0, rbd_snap_get_limit(image, &limit));
3879   ASSERT_EQ(2U, limit);
3880
3881   ASSERT_EQ(0, rbd_snap_create(image, "snap1"));
3882   ASSERT_EQ(0, rbd_snap_create(image, "snap2"));
3883   ASSERT_EQ(-EDQUOT, rbd_snap_create(image, "snap3"));
3884   ASSERT_EQ(0, rbd_snap_set_limit(image, UINT64_MAX));
3885   ASSERT_EQ(0, rbd_snap_create(image, "snap3"));
3886   ASSERT_EQ(0, rbd_close(image));
3887
3888   rados_ioctx_destroy(ioctx);
3889 }
3890
3891
3892 TEST_F(TestLibRBD, SnapshotLimitPP)
3893 {
3894   librados::IoCtx ioctx;
3895   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
3896
3897   {
3898     librbd::RBD rbd;
3899     librbd::Image image;
3900     std::string name = get_temp_image_name();
3901     uint64_t size = 2 << 20;
3902     int order = 0;
3903     uint64_t limit;
3904
3905     ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
3906     ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
3907
3908     ASSERT_EQ(0, image.snap_get_limit(&limit));
3909     ASSERT_EQ(UINT64_MAX, limit);
3910     ASSERT_EQ(0, image.snap_set_limit(2));
3911     ASSERT_EQ(0, image.snap_get_limit(&limit));
3912     ASSERT_EQ(2U, limit);
3913
3914     ASSERT_EQ(0, image.snap_create("snap1"));
3915     ASSERT_EQ(0, image.snap_create("snap2"));
3916     ASSERT_EQ(-EDQUOT, image.snap_create("snap3"));
3917     ASSERT_EQ(0, image.snap_set_limit(UINT64_MAX));
3918     ASSERT_EQ(0, image.snap_create("snap3"));
3919   }
3920
3921   ioctx.close();
3922 }
3923
3924 TEST_F(TestLibRBD, RebuildObjectMapViaLockOwner)
3925 {
3926   REQUIRE_FEATURE(RBD_FEATURE_EXCLUSIVE_LOCK | RBD_FEATURE_OBJECT_MAP);
3927
3928   librados::IoCtx ioctx;
3929   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
3930
3931   librbd::RBD rbd;
3932   std::string name = get_temp_image_name();
3933   uint64_t size = 2 << 20;
3934   int order = 0;
3935   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
3936
3937   std::string object_map_oid;
3938   {
3939     librbd::Image image;
3940     ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
3941
3942     std::string image_id;
3943     ASSERT_EQ(0, get_image_id(image, &image_id));
3944     object_map_oid = RBD_OBJECT_MAP_PREFIX + image_id;
3945   }
3946
3947   // corrupt the object map
3948   bufferlist bl;
3949   bl.append("foo");
3950   ASSERT_EQ(0, ioctx.write(object_map_oid, bl, bl.length(), 0));
3951
3952   librbd::Image image1;
3953   ASSERT_EQ(0, rbd.open(ioctx, image1, name.c_str(), NULL));
3954
3955   bool lock_owner;
3956   bl.clear();
3957   ASSERT_EQ(0, image1.write(0, 0, bl));
3958   ASSERT_EQ(0, image1.is_exclusive_lock_owner(&lock_owner));
3959   ASSERT_TRUE(lock_owner);
3960
3961   uint64_t flags;
3962   ASSERT_EQ(0, image1.get_flags(&flags));
3963   ASSERT_TRUE((flags & RBD_FLAG_OBJECT_MAP_INVALID) != 0);
3964
3965   librbd::Image image2;
3966   ASSERT_EQ(0, rbd.open(ioctx, image2, name.c_str(), NULL));
3967   ASSERT_EQ(0, image2.is_exclusive_lock_owner(&lock_owner));
3968   ASSERT_FALSE(lock_owner);
3969
3970   PrintProgress prog_ctx;
3971   ASSERT_EQ(0, image2.rebuild_object_map(prog_ctx));
3972   ASSERT_PASSED(validate_object_map, image1);
3973   ASSERT_PASSED(validate_object_map, image2);
3974 }
3975
3976 TEST_F(TestLibRBD, RenameViaLockOwner)
3977 {
3978   REQUIRE_FEATURE(RBD_FEATURE_LAYERING | RBD_FEATURE_EXCLUSIVE_LOCK);
3979
3980   librados::IoCtx ioctx;
3981   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
3982
3983   librbd::RBD rbd;
3984   std::string name = get_temp_image_name();
3985   uint64_t size = 2 << 20;
3986   int order = 0;
3987   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
3988
3989   librbd::Image image1;
3990   ASSERT_EQ(0, rbd.open(ioctx, image1, name.c_str(), NULL));
3991
3992   bufferlist bl;
3993   ASSERT_EQ(0, image1.write(0, 0, bl));
3994
3995   bool lock_owner;
3996   ASSERT_EQ(0, image1.is_exclusive_lock_owner(&lock_owner));
3997   ASSERT_TRUE(lock_owner);
3998
3999   std::string new_name = get_temp_image_name();
4000   ASSERT_EQ(0, rbd.rename(ioctx, name.c_str(), new_name.c_str()));
4001   ASSERT_EQ(0, image1.is_exclusive_lock_owner(&lock_owner));
4002   ASSERT_TRUE(lock_owner);
4003
4004   librbd::Image image2;
4005   ASSERT_EQ(0, rbd.open(ioctx, image2, new_name.c_str(), NULL));
4006 }
4007
4008 TEST_F(TestLibRBD, SnapCreateViaLockOwner)
4009 {
4010   REQUIRE_FEATURE(RBD_FEATURE_LAYERING | RBD_FEATURE_EXCLUSIVE_LOCK);
4011
4012   librados::IoCtx ioctx;
4013   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
4014
4015   librbd::RBD rbd;
4016   std::string name = get_temp_image_name();
4017   uint64_t size = 2 << 20;
4018   int order = 0;
4019   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
4020
4021   librbd::Image image1;
4022   ASSERT_EQ(0, rbd.open(ioctx, image1, name.c_str(), NULL));
4023
4024   // switch to writeback cache
4025   ASSERT_EQ(0, image1.flush());
4026
4027   bufferlist bl;
4028   bl.append(std::string(4096, '1'));
4029   ASSERT_EQ((ssize_t)bl.length(), image1.write(0, bl.length(), bl));
4030
4031   bool lock_owner;
4032   ASSERT_EQ(0, image1.is_exclusive_lock_owner(&lock_owner));
4033   ASSERT_TRUE(lock_owner);
4034
4035   librbd::Image image2;
4036   ASSERT_EQ(0, rbd.open(ioctx, image2, name.c_str(), NULL));
4037
4038   ASSERT_EQ(0, image2.is_exclusive_lock_owner(&lock_owner));
4039   ASSERT_FALSE(lock_owner);
4040
4041   ASSERT_EQ(0, image2.snap_create("snap1"));
4042   bool exists;
4043   ASSERT_EQ(0, image1.snap_exists2("snap1", &exists));
4044   ASSERT_TRUE(exists);
4045   ASSERT_EQ(0, image2.snap_exists2("snap1", &exists));
4046   ASSERT_TRUE(exists);
4047
4048   ASSERT_EQ(0, image1.is_exclusive_lock_owner(&lock_owner));
4049   ASSERT_TRUE(lock_owner);
4050 }
4051
4052 TEST_F(TestLibRBD, SnapRemoveViaLockOwner)
4053 {
4054   REQUIRE_FEATURE(RBD_FEATURE_LAYERING | RBD_FEATURE_EXCLUSIVE_LOCK);
4055
4056   librados::IoCtx ioctx;
4057   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
4058
4059   librbd::RBD rbd;
4060   std::string name = get_temp_image_name();
4061   uint64_t size = 2 << 20;
4062   int order = 0;
4063   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
4064
4065   librbd::Image image1;
4066   ASSERT_EQ(0, rbd.open(ioctx, image1, name.c_str(), NULL));
4067
4068   bufferlist bl;
4069   ASSERT_EQ(0, image1.write(0, 0, bl));
4070   ASSERT_EQ(0, image1.snap_create("snap1"));
4071
4072   bool lock_owner;
4073   ASSERT_EQ(0, image1.is_exclusive_lock_owner(&lock_owner));
4074   ASSERT_TRUE(lock_owner);
4075
4076   librbd::Image image2;
4077   ASSERT_EQ(0, rbd.open(ioctx, image2, name.c_str(), NULL));
4078
4079   ASSERT_EQ(0, image2.is_exclusive_lock_owner(&lock_owner));
4080   ASSERT_FALSE(lock_owner);
4081
4082   ASSERT_EQ(0, image2.snap_remove("snap1"));
4083   bool exists;
4084   ASSERT_EQ(0, image1.snap_exists2("snap1", &exists));
4085   ASSERT_FALSE(exists);
4086   ASSERT_EQ(0, image2.snap_exists2("snap1", &exists));
4087   ASSERT_FALSE(exists);
4088
4089   ASSERT_EQ(0, image1.is_exclusive_lock_owner(&lock_owner));
4090   ASSERT_TRUE(lock_owner);
4091 }
4092
4093 TEST_F(TestLibRBD, SnapRemove2)
4094 {
4095   REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
4096
4097   librados::IoCtx ioctx;
4098   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
4099
4100   librbd::RBD rbd;
4101   std::string name = get_temp_image_name();
4102   uint64_t size = 2 << 20;
4103   int order = 0;
4104   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
4105
4106   librbd::Image image1;
4107   ASSERT_EQ(0, rbd.open(ioctx, image1, name.c_str(), NULL));
4108
4109   bufferlist bl;
4110   ASSERT_EQ(0, image1.write(0, 0, bl));
4111   ASSERT_EQ(0, image1.snap_create("snap1"));
4112   bool exists;
4113   ASSERT_EQ(0, image1.snap_exists2("snap1", &exists));
4114   ASSERT_TRUE(exists);
4115   ASSERT_EQ(0, image1.snap_protect("snap1"));
4116   bool is_protected;
4117   ASSERT_EQ(0, image1.snap_is_protected("snap1", &is_protected));
4118   ASSERT_TRUE(is_protected);
4119
4120   uint64_t features;
4121   ASSERT_EQ(0, image1.features(&features));
4122
4123   std::string child_name = get_temp_image_name();
4124   EXPECT_EQ(0, rbd.clone(ioctx, name.c_str(), "snap1", ioctx,
4125                          child_name.c_str(), features, &order));
4126
4127   ASSERT_EQ(0, image1.snap_exists2("snap1", &exists));
4128   ASSERT_TRUE(exists);
4129   ASSERT_EQ(0, image1.snap_is_protected("snap1", &is_protected));
4130   ASSERT_TRUE(is_protected);
4131
4132   ASSERT_EQ(-EBUSY, image1.snap_remove("snap1"));
4133   PrintProgress pp;
4134   ASSERT_EQ(0, image1.snap_remove2("snap1", RBD_SNAP_REMOVE_FORCE, pp));
4135   ASSERT_EQ(0, image1.snap_exists2("snap1", &exists));
4136   ASSERT_FALSE(exists);
4137 }
4138
4139 TEST_F(TestLibRBD, SnapRenameViaLockOwner)
4140 {
4141   REQUIRE_FEATURE(RBD_FEATURE_LAYERING | RBD_FEATURE_EXCLUSIVE_LOCK);
4142
4143   librados::IoCtx ioctx;
4144   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
4145
4146   librbd::RBD rbd;
4147   std::string name = get_temp_image_name();
4148   uint64_t size = 2 << 20;
4149   int order = 0;
4150   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
4151
4152   librbd::Image image1;
4153   ASSERT_EQ(0, rbd.open(ioctx, image1, name.c_str(), NULL));
4154
4155   bufferlist bl;
4156   ASSERT_EQ(0, image1.write(0, 0, bl));
4157   ASSERT_EQ(0, image1.snap_create("snap1"));
4158
4159   bool lock_owner;
4160   ASSERT_EQ(0, image1.is_exclusive_lock_owner(&lock_owner));
4161   ASSERT_TRUE(lock_owner);
4162
4163   librbd::Image image2;
4164   ASSERT_EQ(0, rbd.open(ioctx, image2, name.c_str(), NULL));
4165
4166   ASSERT_EQ(0, image2.is_exclusive_lock_owner(&lock_owner));
4167   ASSERT_FALSE(lock_owner);
4168
4169   ASSERT_EQ(0, image2.snap_rename("snap1", "snap1-rename"));
4170   bool exists;
4171   ASSERT_EQ(0, image1.snap_exists2("snap1-rename", &exists));
4172   ASSERT_TRUE(exists);
4173   ASSERT_EQ(0, image2.snap_exists2("snap1-rename", &exists));
4174   ASSERT_TRUE(exists);
4175
4176   ASSERT_EQ(0, image1.is_exclusive_lock_owner(&lock_owner));
4177   ASSERT_TRUE(lock_owner);
4178 }
4179
4180 TEST_F(TestLibRBD, SnapProtectViaLockOwner)
4181 {
4182   REQUIRE_FEATURE(RBD_FEATURE_LAYERING | RBD_FEATURE_EXCLUSIVE_LOCK);
4183
4184   librados::IoCtx ioctx;
4185   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
4186
4187   librbd::RBD rbd;
4188   std::string name = get_temp_image_name();
4189   uint64_t size = 2 << 20;
4190   int order = 0;
4191   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
4192
4193   librbd::Image image1;
4194   ASSERT_EQ(0, rbd.open(ioctx, image1, name.c_str(), NULL));
4195
4196   bufferlist bl;
4197   ASSERT_EQ(0, image1.write(0, 0, bl));
4198
4199   bool lock_owner;
4200   ASSERT_EQ(0, image1.is_exclusive_lock_owner(&lock_owner));
4201   ASSERT_TRUE(lock_owner);
4202   ASSERT_EQ(0, image1.snap_create("snap1"));
4203
4204   librbd::Image image2;
4205   ASSERT_EQ(0, rbd.open(ioctx, image2, name.c_str(), NULL));
4206
4207   ASSERT_EQ(0, image2.is_exclusive_lock_owner(&lock_owner));
4208   ASSERT_FALSE(lock_owner);
4209
4210   ASSERT_EQ(0, image2.snap_protect("snap1"));
4211   bool is_protected;
4212   ASSERT_EQ(0, image2.snap_is_protected("snap1", &is_protected));
4213   ASSERT_TRUE(is_protected);
4214   ASSERT_EQ(0, image1.snap_is_protected("snap1", &is_protected));
4215   ASSERT_TRUE(is_protected);
4216
4217   ASSERT_EQ(0, image1.is_exclusive_lock_owner(&lock_owner));
4218   ASSERT_TRUE(lock_owner);
4219 }
4220
4221 TEST_F(TestLibRBD, SnapUnprotectViaLockOwner)
4222 {
4223   REQUIRE_FEATURE(RBD_FEATURE_LAYERING | RBD_FEATURE_EXCLUSIVE_LOCK);
4224
4225   librados::IoCtx ioctx;
4226   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
4227
4228   librbd::RBD rbd;
4229   std::string name = get_temp_image_name();
4230   uint64_t size = 2 << 20;
4231   int order = 0;
4232   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
4233
4234   librbd::Image image1;
4235   ASSERT_EQ(0, rbd.open(ioctx, image1, name.c_str(), NULL));
4236
4237   bufferlist bl;
4238   ASSERT_EQ(0, image1.write(0, 0, bl));
4239
4240   bool lock_owner;
4241   ASSERT_EQ(0, image1.is_exclusive_lock_owner(&lock_owner));
4242   ASSERT_TRUE(lock_owner);
4243   ASSERT_EQ(0, image1.snap_create("snap1"));
4244   ASSERT_EQ(0, image1.snap_protect("snap1"));
4245   bool is_protected;
4246   ASSERT_EQ(0, image1.snap_is_protected("snap1", &is_protected));
4247   ASSERT_TRUE(is_protected);
4248
4249   librbd::Image image2;
4250   ASSERT_EQ(0, rbd.open(ioctx, image2, name.c_str(), NULL));
4251
4252   ASSERT_EQ(0, image2.is_exclusive_lock_owner(&lock_owner));
4253   ASSERT_FALSE(lock_owner);
4254
4255   ASSERT_EQ(0, image2.snap_unprotect("snap1"));
4256   ASSERT_EQ(0, image2.snap_is_protected("snap1", &is_protected));
4257   ASSERT_FALSE(is_protected);
4258   ASSERT_EQ(0, image1.snap_is_protected("snap1", &is_protected));
4259   ASSERT_FALSE(is_protected);
4260
4261   ASSERT_EQ(0, image1.is_exclusive_lock_owner(&lock_owner));
4262   ASSERT_TRUE(lock_owner);
4263 }
4264
4265 TEST_F(TestLibRBD, FlattenViaLockOwner)
4266 {
4267   REQUIRE_FEATURE(RBD_FEATURE_EXCLUSIVE_LOCK);
4268
4269   librados::IoCtx ioctx;
4270   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
4271
4272   librbd::RBD rbd;
4273   std::string parent_name = get_temp_image_name();
4274   uint64_t size = 2 << 20;
4275   int order = 0;
4276   ASSERT_EQ(0, create_image_pp(rbd, ioctx, parent_name.c_str(), size, &order));
4277
4278   librbd::Image parent_image;
4279   ASSERT_EQ(0, rbd.open(ioctx, parent_image, parent_name.c_str(), NULL));
4280   ASSERT_EQ(0, parent_image.snap_create("snap1"));
4281   ASSERT_EQ(0, parent_image.snap_protect("snap1"));
4282
4283   uint64_t features;
4284   ASSERT_EQ(0, parent_image.features(&features));
4285
4286   std::string name = get_temp_image_name();
4287   EXPECT_EQ(0, rbd.clone(ioctx, parent_name.c_str(), "snap1", ioctx,
4288                          name.c_str(), features, &order));
4289
4290   librbd::Image image1;
4291   ASSERT_EQ(0, rbd.open(ioctx, image1, name.c_str(), NULL));
4292
4293   bufferlist bl;
4294   ASSERT_EQ(0, image1.write(0, 0, bl));
4295
4296   bool lock_owner;
4297   ASSERT_EQ(0, image1.is_exclusive_lock_owner(&lock_owner));
4298   ASSERT_TRUE(lock_owner);
4299
4300   librbd::Image image2;
4301   ASSERT_EQ(0, rbd.open(ioctx, image2, name.c_str(), NULL));
4302
4303   ASSERT_EQ(0, image2.is_exclusive_lock_owner(&lock_owner));
4304   ASSERT_FALSE(lock_owner);
4305
4306   ASSERT_EQ(0, image2.flatten());
4307
4308   ASSERT_EQ(0, image1.is_exclusive_lock_owner(&lock_owner));
4309   ASSERT_TRUE(lock_owner);
4310   ASSERT_PASSED(validate_object_map, image1);
4311 }
4312
4313 TEST_F(TestLibRBD, ResizeViaLockOwner)
4314 {
4315   REQUIRE_FEATURE(RBD_FEATURE_EXCLUSIVE_LOCK);
4316
4317   librados::IoCtx ioctx;
4318   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
4319
4320   librbd::RBD rbd;
4321   std::string name = get_temp_image_name();
4322   uint64_t size = 2 << 20;
4323   int order = 0;
4324   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
4325
4326   librbd::Image image1;
4327   ASSERT_EQ(0, rbd.open(ioctx, image1, name.c_str(), NULL));
4328
4329   bufferlist bl;
4330   ASSERT_EQ(0, image1.write(0, 0, bl));
4331
4332   bool lock_owner;
4333   ASSERT_EQ(0, image1.is_exclusive_lock_owner(&lock_owner));
4334   ASSERT_TRUE(lock_owner);
4335
4336   librbd::Image image2;
4337   ASSERT_EQ(0, rbd.open(ioctx, image2, name.c_str(), NULL));
4338
4339   ASSERT_EQ(0, image2.is_exclusive_lock_owner(&lock_owner));
4340   ASSERT_FALSE(lock_owner);
4341
4342   ASSERT_EQ(0, image2.resize(0));
4343
4344   ASSERT_EQ(0, image1.is_exclusive_lock_owner(&lock_owner));
4345   ASSERT_TRUE(lock_owner);
4346   ASSERT_PASSED(validate_object_map, image1);
4347 }
4348
4349 TEST_F(TestLibRBD, ObjectMapConsistentSnap)
4350 {
4351   REQUIRE_FEATURE(RBD_FEATURE_OBJECT_MAP);
4352
4353   librados::IoCtx ioctx;
4354   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
4355
4356   librbd::RBD rbd;
4357   std::string name = get_temp_image_name();
4358   uint64_t size = 1 << 20;
4359   int order = 12;
4360   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
4361
4362   librbd::Image image1;
4363   ASSERT_EQ(0, rbd.open(ioctx, image1, name.c_str(), NULL));
4364
4365   int num_snaps = 10;
4366   for (int i = 0; i < num_snaps; ++i) {
4367     std::string snap_name = "snap" + stringify(i);
4368     ASSERT_EQ(0, image1.snap_create(snap_name.c_str()));
4369   }
4370
4371
4372   thread writer([&image1](){
4373       librbd::image_info_t info;
4374       int r = image1.stat(info, sizeof(info));
4375       assert(r == 0);
4376       bufferlist bl;
4377       bl.append("foo");
4378       for (unsigned i = 0; i < info.num_objs; ++i) {
4379         r = image1.write((1 << info.order) * i, bl.length(), bl);
4380         assert(r == (int) bl.length());
4381       }
4382     });
4383   writer.join();
4384
4385   for (int i = 0; i < num_snaps; ++i) {
4386     std::string snap_name = "snap" + stringify(i);
4387     ASSERT_EQ(0, image1.snap_set(snap_name.c_str()));
4388     ASSERT_PASSED(validate_object_map, image1);
4389   }
4390
4391   ASSERT_EQ(0, image1.snap_set(NULL));
4392   ASSERT_PASSED(validate_object_map, image1);
4393 }
4394
4395 void memset_rand(char *buf, size_t len) {
4396   for (size_t i = 0; i < len; ++i) {
4397     buf[i] = (char) (rand() % (126 - 33) + 33);
4398   }
4399 }
4400
4401 TEST_F(TestLibRBD, Metadata)
4402 {
4403   REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
4404
4405   rados_ioctx_t ioctx;
4406   rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx);
4407
4408   std::string name = get_temp_image_name();
4409   uint64_t size = 2 << 20;
4410   int order = 0;
4411   ASSERT_EQ(0, create_image(ioctx, name.c_str(), size, &order));
4412
4413   rbd_image_t image;
4414   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
4415
4416   rbd_image_t image1;
4417   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image1, NULL));
4418
4419   char keys[1024];
4420   char vals[1024];
4421   size_t keys_len = sizeof(keys);
4422   size_t vals_len = sizeof(vals);
4423
4424   memset_rand(keys, keys_len);
4425   memset_rand(vals, vals_len);
4426
4427   ASSERT_EQ(0, rbd_metadata_list(image, "", 0, keys, &keys_len, vals,
4428                                  &vals_len));
4429   ASSERT_EQ(0U, keys_len);
4430   ASSERT_EQ(0U, vals_len);
4431
4432   char value[1024];
4433   size_t value_len = sizeof(value);
4434   memset_rand(value, value_len);
4435
4436   ASSERT_EQ(0, rbd_metadata_set(image1, "key1", "value1"));
4437   ASSERT_EQ(0, rbd_metadata_set(image1, "key2", "value2"));
4438   ASSERT_EQ(0, rbd_metadata_get(image1, "key1", value, &value_len));
4439   ASSERT_STREQ(value, "value1");
4440   value_len = 1;
4441   ASSERT_EQ(-ERANGE, rbd_metadata_get(image1, "key1", value, &value_len));
4442   ASSERT_EQ(value_len, strlen("value1") + 1);
4443
4444   ASSERT_EQ(-ERANGE, rbd_metadata_list(image1, "", 0, keys, &keys_len, vals,
4445                                        &vals_len));
4446   keys_len = sizeof(keys);
4447   vals_len = sizeof(vals);
4448   memset_rand(keys, keys_len);
4449   memset_rand(vals, vals_len);
4450   ASSERT_EQ(0, rbd_metadata_list(image1, "", 0, keys, &keys_len, vals,
4451                                  &vals_len));
4452   ASSERT_EQ(keys_len, strlen("key1") + 1 + strlen("key2") + 1);
4453   ASSERT_EQ(vals_len, strlen("value1") + 1 + strlen("value2") + 1);
4454   ASSERT_STREQ(keys, "key1");
4455   ASSERT_STREQ(keys + strlen(keys) + 1, "key2");
4456   ASSERT_STREQ(vals, "value1");
4457   ASSERT_STREQ(vals + strlen(vals) + 1, "value2");
4458
4459   ASSERT_EQ(0, rbd_metadata_remove(image1, "key1"));
4460   ASSERT_EQ(-ENOENT, rbd_metadata_remove(image1, "key3"));
4461   value_len = sizeof(value);
4462   ASSERT_EQ(-ENOENT, rbd_metadata_get(image1, "key3", value, &value_len));
4463   ASSERT_EQ(0, rbd_metadata_list(image1, "", 0, keys, &keys_len, vals,
4464                                  &vals_len));
4465   ASSERT_EQ(keys_len, strlen("key2") + 1);
4466   ASSERT_EQ(vals_len, strlen("value2") + 1);
4467   ASSERT_STREQ(keys, "key2");
4468   ASSERT_STREQ(vals, "value2");
4469
4470   // test config setting
4471   ASSERT_EQ(0, rbd_metadata_set(image1, "conf_rbd_cache", "false"));
4472   ASSERT_EQ(-EINVAL, rbd_metadata_set(image1, "conf_rbd_cache", "INVALID_VAL"));
4473   ASSERT_EQ(0, rbd_metadata_remove(image1, "conf_rbd_cache"));
4474
4475   // test metadata with snapshot adding
4476   ASSERT_EQ(0, rbd_snap_create(image1, "snap1"));
4477   ASSERT_EQ(0, rbd_snap_protect(image1, "snap1"));
4478   ASSERT_EQ(0, rbd_snap_set(image1, "snap1"));
4479
4480   ASSERT_EQ(0, rbd_metadata_set(image1, "key1", "value1"));
4481   ASSERT_EQ(0, rbd_metadata_set(image1, "key3", "value3"));
4482
4483   keys_len = sizeof(keys);
4484   vals_len = sizeof(vals);
4485   memset_rand(keys, keys_len);
4486   memset_rand(vals, vals_len);
4487   ASSERT_EQ(0, rbd_metadata_list(image1, "", 0, keys, &keys_len, vals,
4488                                  &vals_len));
4489   ASSERT_EQ(keys_len,
4490             strlen("key1") + 1 + strlen("key2") + 1 + strlen("key3") + 1);
4491   ASSERT_EQ(vals_len,
4492             strlen("value1") + 1 + strlen("value2") + 1 + strlen("value3") + 1);
4493   ASSERT_STREQ(keys, "key1");
4494   ASSERT_STREQ(keys + strlen("key1") + 1, "key2");
4495   ASSERT_STREQ(keys + strlen("key1") + 1 + strlen("key2") + 1, "key3");
4496   ASSERT_STREQ(vals, "value1");
4497   ASSERT_STREQ(vals + strlen("value1") + 1, "value2");
4498   ASSERT_STREQ(vals + strlen("value1") + 1 + strlen("value2") + 1, "value3");
4499
4500   ASSERT_EQ(0, rbd_snap_set(image1, NULL));
4501   keys_len = sizeof(keys);
4502   vals_len = sizeof(vals);
4503   memset_rand(keys, keys_len);
4504   memset_rand(vals, vals_len);
4505   ASSERT_EQ(0, rbd_metadata_list(image1, "", 0, keys, &keys_len, vals,
4506                                  &vals_len));
4507   ASSERT_EQ(keys_len,
4508             strlen("key1") + 1 + strlen("key2") + 1 + strlen("key3") + 1);
4509   ASSERT_EQ(vals_len,
4510             strlen("value1") + 1 + strlen("value2") + 1 + strlen("value3") + 1);
4511   ASSERT_STREQ(keys, "key1");
4512   ASSERT_STREQ(keys + strlen("key1") + 1, "key2");
4513   ASSERT_STREQ(keys + strlen("key1") + 1 + strlen("key2") + 1, "key3");
4514   ASSERT_STREQ(vals, "value1");
4515   ASSERT_STREQ(vals + strlen("value1") + 1, "value2");
4516   ASSERT_STREQ(vals + strlen("value1") + 1 + strlen("value2") + 1, "value3");
4517
4518   // test metadata with cloning
4519   uint64_t features;
4520   ASSERT_EQ(0, rbd_get_features(image1, &features));
4521
4522   string cname = get_temp_image_name();
4523   EXPECT_EQ(0, rbd_clone(ioctx, name.c_str(), "snap1", ioctx,
4524                          cname.c_str(), features, &order));
4525   rbd_image_t image2;
4526   ASSERT_EQ(0, rbd_open(ioctx, cname.c_str(), &image2, NULL));
4527   ASSERT_EQ(0, rbd_metadata_set(image2, "key4", "value4"));
4528
4529   keys_len = sizeof(keys);
4530   vals_len = sizeof(vals);
4531   memset_rand(keys, keys_len);
4532   memset_rand(vals, vals_len);
4533   ASSERT_EQ(0, rbd_metadata_list(image2, "", 0, keys, &keys_len, vals,
4534                                  &vals_len));
4535   ASSERT_EQ(keys_len, strlen("key1") + 1 + strlen("key2") + 1 + strlen("key3") +
4536             1 + strlen("key4") + 1);
4537   ASSERT_EQ(vals_len, strlen("value1") + 1 + strlen("value2") + 1 +
4538             strlen("value3") + 1 + strlen("value4") + 1);
4539   ASSERT_STREQ(keys + strlen("key1") + 1 + strlen("key2") + 1 + strlen("key3") +
4540                1, "key4");
4541   ASSERT_STREQ(vals + strlen("value1") + 1 + strlen("value2") + 1 +
4542                strlen("value3") + 1, "value4");
4543
4544   ASSERT_EQ(0, rbd_metadata_list(image1, "", 0, keys, &keys_len, vals,
4545                                  &vals_len));
4546   ASSERT_EQ(keys_len,
4547             strlen("key1") + 1 + strlen("key2") + 1 + strlen("key3") + 1);
4548   ASSERT_EQ(vals_len,
4549             strlen("value1") + 1 + strlen("value2") + 1 + strlen("value3") + 1);
4550   ASSERT_EQ(-ENOENT, rbd_metadata_get(image1, "key4", value, &value_len));
4551
4552   // test short buffer cases
4553   keys_len = strlen("key1") + 1;
4554   vals_len = strlen("value1") + 1;
4555   memset_rand(keys, keys_len);
4556   memset_rand(vals, vals_len);
4557   ASSERT_EQ(0, rbd_metadata_list(image2, "", 1, keys, &keys_len, vals,
4558                                  &vals_len));
4559   ASSERT_EQ(keys_len, strlen("key1") + 1);
4560   ASSERT_EQ(vals_len, strlen("value1") + 1);
4561   ASSERT_STREQ(keys, "key1");
4562   ASSERT_STREQ(vals, "value1");
4563
4564   ASSERT_EQ(-ERANGE, rbd_metadata_list(image2, "", 2, keys, &keys_len, vals,
4565                                        &vals_len));
4566   ASSERT_EQ(keys_len, strlen("key1") + 1 + strlen("key2") + 1);
4567   ASSERT_EQ(vals_len, strlen("value1") + 1 + strlen("value2") + 1);
4568
4569   ASSERT_EQ(-ERANGE, rbd_metadata_list(image2, "", 0, keys, &keys_len, vals,
4570                                        &vals_len));
4571   ASSERT_EQ(keys_len, strlen("key1") + 1 + strlen("key2") + 1 + strlen("key3") +
4572             1 + strlen("key4") + 1);
4573   ASSERT_EQ(vals_len, strlen("value1") + 1 + strlen("value2") + 1 +
4574             strlen("value3") + 1 + strlen("value4") + 1);
4575
4576   // test `start` param
4577   keys_len = sizeof(keys);
4578   vals_len = sizeof(vals);
4579   memset_rand(keys, keys_len);
4580   memset_rand(vals, vals_len);
4581   ASSERT_EQ(0, rbd_metadata_list(image2, "key2", 0, keys, &keys_len, vals,
4582                                  &vals_len));
4583   ASSERT_EQ(keys_len, strlen("key3") + 1 + strlen("key4") + 1);
4584   ASSERT_EQ(vals_len, strlen("value3") + 1 + strlen("value4") + 1);
4585   ASSERT_STREQ(keys, "key3");
4586   ASSERT_STREQ(vals, "value3");
4587
4588   ASSERT_EQ(0, rbd_close(image));
4589   ASSERT_EQ(0, rbd_close(image1));
4590   ASSERT_EQ(0, rbd_close(image2));
4591   rados_ioctx_destroy(ioctx);
4592 }
4593
4594 TEST_F(TestLibRBD, MetadataPP)
4595 {
4596   REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
4597
4598   librados::IoCtx ioctx;
4599   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
4600
4601   librbd::RBD rbd;
4602   string name = get_temp_image_name();
4603   uint64_t size = 2 << 20;
4604   int order = 0;
4605   uint64_t features;
4606   string value;
4607   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
4608
4609   librbd::Image image1;
4610   ASSERT_EQ(0, rbd.open(ioctx, image1, name.c_str(), NULL));
4611   map<string, bufferlist> pairs;
4612   ASSERT_EQ(0, image1.metadata_list("", 0, &pairs));
4613   ASSERT_TRUE(pairs.empty());
4614
4615   ASSERT_EQ(0, image1.metadata_set("key1", "value1"));
4616   ASSERT_EQ(0, image1.metadata_set("key2", "value2"));
4617   ASSERT_EQ(0, image1.metadata_get("key1", &value));
4618   ASSERT_EQ(0, strcmp("value1", value.c_str()));
4619   ASSERT_EQ(0, image1.metadata_list("", 0, &pairs));
4620   ASSERT_EQ(2U, pairs.size());
4621   ASSERT_EQ(0, strncmp("value1", pairs["key1"].c_str(), 6));
4622   ASSERT_EQ(0, strncmp("value2", pairs["key2"].c_str(), 6));
4623
4624   pairs.clear();
4625   ASSERT_EQ(0, image1.metadata_remove("key1"));
4626   ASSERT_EQ(-ENOENT, image1.metadata_remove("key3"));
4627   ASSERT_TRUE(image1.metadata_get("key3", &value) < 0);
4628   ASSERT_EQ(0, image1.metadata_list("", 0, &pairs));
4629   ASSERT_EQ(1U, pairs.size());
4630   ASSERT_EQ(0, strncmp("value2", pairs["key2"].c_str(), 6));
4631
4632   // test config setting
4633   ASSERT_EQ(0, image1.metadata_set("conf_rbd_cache", "false"));
4634   ASSERT_EQ(-EINVAL, image1.metadata_set("conf_rbd_cache", "INVALID_VALUE"));
4635   ASSERT_EQ(0, image1.metadata_remove("conf_rbd_cache"));
4636
4637   // test metadata with snapshot adding
4638   ASSERT_EQ(0, image1.snap_create("snap1"));
4639   ASSERT_EQ(0, image1.snap_protect("snap1"));
4640   ASSERT_EQ(0, image1.snap_set("snap1"));
4641
4642   pairs.clear();
4643   ASSERT_EQ(0, image1.metadata_set("key1", "value1"));
4644   ASSERT_EQ(0, image1.metadata_set("key3", "value3"));
4645   ASSERT_EQ(0, image1.metadata_list("", 0, &pairs));
4646   ASSERT_EQ(3U, pairs.size());
4647   ASSERT_EQ(0, strncmp("value1", pairs["key1"].c_str(), 6));
4648   ASSERT_EQ(0, strncmp("value2", pairs["key2"].c_str(), 6));
4649   ASSERT_EQ(0, strncmp("value3", pairs["key3"].c_str(), 6));
4650
4651   ASSERT_EQ(0, image1.snap_set(NULL));
4652   ASSERT_EQ(0, image1.metadata_list("", 0, &pairs));
4653   ASSERT_EQ(3U, pairs.size());
4654   ASSERT_EQ(0, strncmp("value1", pairs["key1"].c_str(), 6));
4655   ASSERT_EQ(0, strncmp("value2", pairs["key2"].c_str(), 6));
4656   ASSERT_EQ(0, strncmp("value3", pairs["key3"].c_str(), 6));
4657
4658   // test metadata with cloning
4659   string cname = get_temp_image_name();
4660   librbd::Image image2;
4661   ASSERT_EQ(0, image1.features(&features));
4662   EXPECT_EQ(0, rbd.clone(ioctx, name.c_str(), "snap1", ioctx,
4663                          cname.c_str(), features, &order));
4664   ASSERT_EQ(0, rbd.open(ioctx, image2, cname.c_str(), NULL));
4665   ASSERT_EQ(0, image2.metadata_set("key4", "value4"));
4666   pairs.clear();
4667   ASSERT_EQ(0, image2.metadata_list("", 0, &pairs));
4668   ASSERT_EQ(4U, pairs.size());
4669   pairs.clear();
4670   ASSERT_EQ(0, image1.metadata_list("", 0, &pairs));
4671   ASSERT_EQ(3U, pairs.size());
4672   ASSERT_EQ(-ENOENT, image1.metadata_get("key4", &value));
4673 }
4674
4675 TEST_F(TestLibRBD, UpdateFeatures)
4676 {
4677   REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
4678
4679   librados::IoCtx ioctx;
4680   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
4681
4682   librbd::RBD rbd;
4683   std::string name = get_temp_image_name();
4684   uint64_t size = 1 << 20;
4685   int order = 0;
4686   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
4687
4688   librbd::Image image;
4689   ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
4690
4691   uint8_t old_format;
4692   ASSERT_EQ(0, image.old_format(&old_format));
4693   if (old_format) {
4694     ASSERT_EQ(-EINVAL, image.update_features(RBD_FEATURE_EXCLUSIVE_LOCK, true));
4695     return;
4696   }
4697
4698   uint64_t features;
4699   ASSERT_EQ(0, image.features(&features));
4700
4701   // must provide a single feature
4702   ASSERT_EQ(-EINVAL, image.update_features(0, true));
4703
4704   uint64_t disable_features;
4705   disable_features = features & (RBD_FEATURE_EXCLUSIVE_LOCK |
4706                                  RBD_FEATURE_OBJECT_MAP |
4707                                  RBD_FEATURE_FAST_DIFF |
4708                                  RBD_FEATURE_JOURNALING);
4709   if (disable_features != 0) {
4710     ASSERT_EQ(0, image.update_features(disable_features, false));
4711   }
4712
4713   ASSERT_EQ(0, image.features(&features));
4714   ASSERT_EQ(0U, features & disable_features);
4715
4716   // cannot enable object map nor journaling w/o exclusive lock
4717   ASSERT_EQ(-EINVAL, image.update_features(RBD_FEATURE_OBJECT_MAP, true));
4718   ASSERT_EQ(-EINVAL, image.update_features(RBD_FEATURE_JOURNALING, true));
4719   ASSERT_EQ(0, image.update_features(RBD_FEATURE_EXCLUSIVE_LOCK, true));
4720
4721   ASSERT_EQ(0, image.features(&features));
4722   ASSERT_NE(0U, features & RBD_FEATURE_EXCLUSIVE_LOCK);
4723
4724   // cannot enable fast diff w/o object map
4725   ASSERT_EQ(-EINVAL, image.update_features(RBD_FEATURE_FAST_DIFF, true));
4726   ASSERT_EQ(0, image.update_features(RBD_FEATURE_OBJECT_MAP, true));
4727   ASSERT_EQ(0, image.features(&features));
4728   ASSERT_NE(0U, features & RBD_FEATURE_OBJECT_MAP);
4729
4730   uint64_t expected_flags = RBD_FLAG_OBJECT_MAP_INVALID;
4731   uint64_t flags;
4732   ASSERT_EQ(0, image.get_flags(&flags));
4733   ASSERT_EQ(expected_flags, flags);
4734
4735   ASSERT_EQ(0, image.update_features(RBD_FEATURE_OBJECT_MAP, false));
4736   ASSERT_EQ(0, image.features(&features));
4737   ASSERT_EQ(0U, features & RBD_FEATURE_OBJECT_MAP);
4738
4739   ASSERT_EQ(0, image.update_features(RBD_FEATURE_OBJECT_MAP |
4740                                      RBD_FEATURE_FAST_DIFF |
4741                                      RBD_FEATURE_JOURNALING, true));
4742
4743   expected_flags = RBD_FLAG_OBJECT_MAP_INVALID | RBD_FLAG_FAST_DIFF_INVALID;
4744   ASSERT_EQ(0, image.get_flags(&flags));
4745   ASSERT_EQ(expected_flags, flags);
4746
4747   // cannot disable object map w/ fast diff
4748   ASSERT_EQ(-EINVAL, image.update_features(RBD_FEATURE_OBJECT_MAP, false));
4749   ASSERT_EQ(0, image.update_features(RBD_FEATURE_FAST_DIFF, false));
4750   ASSERT_EQ(0, image.features(&features));
4751   ASSERT_EQ(0U, features & RBD_FEATURE_FAST_DIFF);
4752
4753   expected_flags = RBD_FLAG_OBJECT_MAP_INVALID;
4754   ASSERT_EQ(0, image.get_flags(&flags));
4755   ASSERT_EQ(expected_flags, flags);
4756
4757   // cannot disable exclusive lock w/ object map
4758   ASSERT_EQ(-EINVAL, image.update_features(RBD_FEATURE_EXCLUSIVE_LOCK, false));
4759   ASSERT_EQ(0, image.update_features(RBD_FEATURE_OBJECT_MAP, false));
4760
4761   // cannot disable exclusive lock w/ journaling
4762   ASSERT_EQ(-EINVAL, image.update_features(RBD_FEATURE_EXCLUSIVE_LOCK, false));
4763   ASSERT_EQ(0, image.update_features(RBD_FEATURE_JOURNALING, false));
4764
4765   ASSERT_EQ(0, image.get_flags(&flags));
4766   ASSERT_EQ(0U, flags);
4767
4768   ASSERT_EQ(0, image.update_features(RBD_FEATURE_EXCLUSIVE_LOCK, false));
4769
4770   ASSERT_EQ(0, image.features(&features));
4771   if ((features & RBD_FEATURE_DEEP_FLATTEN) != 0) {
4772     ASSERT_EQ(0, image.update_features(RBD_FEATURE_DEEP_FLATTEN, false));
4773   }
4774   ASSERT_EQ(-EINVAL, image.update_features(RBD_FEATURE_DEEP_FLATTEN, true));
4775 }
4776
4777 TEST_F(TestLibRBD, RebuildObjectMap)
4778 {
4779   librados::IoCtx ioctx;
4780   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
4781
4782   librbd::RBD rbd;
4783   std::string name = get_temp_image_name();
4784   uint64_t size = 1 << 20;
4785   int order = 18;
4786   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
4787
4788   PrintProgress prog_ctx;
4789   std::string object_map_oid;
4790   bufferlist bl;
4791   bl.append("foo");
4792   {
4793     librbd::Image image;
4794     ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
4795
4796     uint64_t features;
4797     ASSERT_EQ(0, image.features(&features));
4798     if ((features & RBD_FEATURE_OBJECT_MAP) == 0) {
4799       ASSERT_EQ(-EINVAL, image.rebuild_object_map(prog_ctx));
4800       return;
4801     }
4802
4803     ASSERT_EQ((ssize_t)bl.length(), image.write(0, bl.length(), bl));
4804
4805     ASSERT_EQ(0, image.snap_create("snap1"));
4806     ASSERT_EQ((ssize_t)bl.length(), image.write(1<<order, bl.length(), bl));
4807
4808     std::string image_id;
4809     ASSERT_EQ(0, get_image_id(image, &image_id));
4810     object_map_oid = RBD_OBJECT_MAP_PREFIX + image_id;
4811   }
4812
4813   // corrupt the object map
4814   ASSERT_EQ(0, ioctx.write(object_map_oid, bl, bl.length(), 0));
4815
4816   librbd::Image image1;
4817   ASSERT_EQ(0, rbd.open(ioctx, image1, name.c_str(), NULL));
4818
4819   bool lock_owner;
4820   bl.clear();
4821   ASSERT_EQ(0, image1.write(0, 0, bl));
4822   ASSERT_EQ(0, image1.is_exclusive_lock_owner(&lock_owner));
4823   ASSERT_TRUE(lock_owner);
4824
4825   uint64_t flags;
4826   ASSERT_EQ(0, image1.get_flags(&flags));
4827   ASSERT_TRUE((flags & RBD_FLAG_OBJECT_MAP_INVALID) != 0);
4828
4829   ASSERT_EQ(0, image1.rebuild_object_map(prog_ctx));
4830
4831   librbd::Image image2;
4832   ASSERT_EQ(0, rbd.open(ioctx, image2, name.c_str(), NULL));
4833
4834   bufferlist read_bl;
4835   ASSERT_EQ((ssize_t)bl.length(), image2.read(0, bl.length(), read_bl));
4836   ASSERT_TRUE(bl.contents_equal(read_bl));
4837
4838   read_bl.clear();
4839   ASSERT_EQ((ssize_t)bl.length(), image2.read(1<<order, bl.length(), read_bl));
4840   ASSERT_TRUE(bl.contents_equal(read_bl));
4841
4842   ASSERT_PASSED(validate_object_map, image1);
4843   ASSERT_PASSED(validate_object_map, image2);
4844 }
4845
4846 TEST_F(TestLibRBD, RebuildNewObjectMap)
4847 {
4848   REQUIRE_FEATURE(RBD_FEATURE_OBJECT_MAP);
4849
4850   rados_ioctx_t ioctx;
4851   rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx);
4852
4853   std::string name = get_temp_image_name();
4854   uint64_t size = 1 << 20;
4855   int order = 18;
4856   uint64_t features = RBD_FEATURE_EXCLUSIVE_LOCK;
4857   ASSERT_EQ(0, create_image_full(ioctx, name.c_str(), size, &order,
4858                                  false, features));
4859
4860   rbd_image_t image;
4861   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
4862   ASSERT_EQ(0, rbd_update_features(image, RBD_FEATURE_OBJECT_MAP, true));
4863   ASSERT_EQ(0, rbd_rebuild_object_map(image, print_progress_percent, NULL));
4864
4865   ASSERT_PASSED(validate_object_map, image);
4866
4867   ASSERT_EQ(0, rbd_close(image));
4868   rados_ioctx_destroy(ioctx);
4869 }
4870
4871 TEST_F(TestLibRBD, CheckObjectMap)
4872 {
4873   REQUIRE_FEATURE(RBD_FEATURE_OBJECT_MAP);
4874
4875   librados::IoCtx ioctx;
4876   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
4877
4878   librbd::RBD rbd;
4879   std::string name = get_temp_image_name();
4880   uint64_t size = 1 << 20;
4881   int order = 18;
4882   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
4883
4884   PrintProgress prog_ctx;
4885   bufferlist bl1;
4886   bufferlist bl2;
4887   bl1.append("foo");
4888   {
4889     librbd::Image image;
4890     ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
4891
4892     uint64_t features;
4893     ASSERT_EQ(0, image.features(&features));
4894
4895     ASSERT_EQ((ssize_t)bl1.length(), image.write(0, bl1.length(), bl1));
4896
4897     ASSERT_EQ(0, image.snap_create("snap1"));
4898     ASSERT_EQ((ssize_t)bl1.length(), image.write(1<<order, bl1.length(), bl1));
4899   }
4900
4901   librbd::Image image1;
4902   ASSERT_EQ(0, rbd.open(ioctx, image1, name.c_str(), NULL));
4903
4904   std::string image_id;
4905   ASSERT_EQ(0, get_image_id(image1, &image_id));
4906
4907   std::string object_map_oid = RBD_OBJECT_MAP_PREFIX + image_id;
4908
4909   ASSERT_LT(0, ioctx.read(object_map_oid, bl2, 1024, 0));
4910
4911   bool lock_owner;
4912   ASSERT_EQ((ssize_t)bl1.length(), image1.write(3 * (1 << 18), bl1.length(), bl1));
4913   ASSERT_EQ(0, image1.is_exclusive_lock_owner(&lock_owner));
4914   ASSERT_TRUE(lock_owner);
4915
4916   //reopen image to reread now corrupt object map from disk
4917   image1.close();
4918
4919   bl1.clear();
4920   ASSERT_LT(0, ioctx.read(object_map_oid, bl1, 1024, 0));
4921   ASSERT_FALSE(bl1.contents_equal(bl2));
4922
4923   ASSERT_EQ(0, ioctx.write_full(object_map_oid, bl2));
4924   ASSERT_EQ(0, rbd.open(ioctx, image1, name.c_str(), NULL));
4925
4926   uint64_t flags;
4927   ASSERT_EQ(0, image1.get_flags(&flags));
4928   ASSERT_TRUE((flags & RBD_FLAG_OBJECT_MAP_INVALID) == 0);
4929
4930   ASSERT_EQ(0, image1.check_object_map(prog_ctx));
4931
4932   ASSERT_EQ(0, image1.get_flags(&flags));
4933   ASSERT_TRUE((flags & RBD_FLAG_OBJECT_MAP_INVALID) != 0);
4934 }
4935
4936 TEST_F(TestLibRBD, BlockingAIO)
4937 {
4938   librados::IoCtx ioctx;
4939   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
4940
4941   bool skip_discard = is_skip_partial_discard_enabled();
4942
4943   librbd::RBD rbd;
4944   std::string name = get_temp_image_name();
4945   uint64_t size = 1 << 20;
4946   int order = 18;
4947   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
4948
4949   std::string non_blocking_aio;
4950   ASSERT_EQ(0, _rados.conf_get("rbd_non_blocking_aio", non_blocking_aio));
4951   ASSERT_EQ(0, _rados.conf_set("rbd_non_blocking_aio", "0"));
4952   BOOST_SCOPE_EXIT( (non_blocking_aio) ) {
4953     ASSERT_EQ(0, _rados.conf_set("rbd_non_blocking_aio",
4954                                  non_blocking_aio.c_str()));
4955   } BOOST_SCOPE_EXIT_END;
4956
4957   librbd::Image image;
4958   ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
4959
4960   bufferlist bl;
4961   ASSERT_EQ(0, image.write(0, bl.length(), bl));
4962
4963   bl.append(std::string(256, '1'));
4964   librbd::RBD::AioCompletion *write_comp =
4965     new librbd::RBD::AioCompletion(NULL, NULL);
4966   ASSERT_EQ(0, image.aio_write(0, bl.length(), bl, write_comp));
4967
4968   librbd::RBD::AioCompletion *flush_comp =
4969     new librbd::RBD::AioCompletion(NULL, NULL);
4970   ASSERT_EQ(0, image.aio_flush(flush_comp));
4971   ASSERT_EQ(0, flush_comp->wait_for_complete());
4972   ASSERT_EQ(0, flush_comp->get_return_value());
4973   flush_comp->release();
4974
4975   ASSERT_EQ(1, write_comp->is_complete());
4976   ASSERT_EQ(0, write_comp->get_return_value());
4977   write_comp->release();
4978
4979   librbd::RBD::AioCompletion *discard_comp =
4980     new librbd::RBD::AioCompletion(NULL, NULL);
4981   ASSERT_EQ(0, image.aio_discard(128, 128, discard_comp));
4982   ASSERT_EQ(0, discard_comp->wait_for_complete());
4983   discard_comp->release();
4984
4985   librbd::RBD::AioCompletion *read_comp =
4986     new librbd::RBD::AioCompletion(NULL, NULL);
4987   bufferlist read_bl;
4988   image.aio_read(0, bl.length(), read_bl, read_comp);
4989   ASSERT_EQ(0, read_comp->wait_for_complete());
4990   ASSERT_EQ((ssize_t)bl.length(), read_comp->get_return_value());
4991   read_comp->release();
4992
4993   bufferlist expected_bl;
4994   expected_bl.append(std::string(128, '1'));
4995   expected_bl.append(std::string(128, skip_discard ? '1' : '\0'));
4996   ASSERT_TRUE(expected_bl.contents_equal(read_bl));
4997 }
4998
4999 TEST_F(TestLibRBD, ExclusiveLockTransition)
5000 {
5001   REQUIRE_FEATURE(RBD_FEATURE_EXCLUSIVE_LOCK);
5002
5003   librados::IoCtx ioctx;
5004   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
5005
5006   librbd::RBD rbd;
5007   std::string name = get_temp_image_name();
5008
5009   uint64_t size = 1 << 18;
5010   int order = 12;
5011   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
5012
5013   librbd::Image image1;
5014   ASSERT_EQ(0, rbd.open(ioctx, image1, name.c_str(), NULL));
5015
5016   librbd::Image image2;
5017   ASSERT_EQ(0, rbd.open(ioctx, image2, name.c_str(), NULL));
5018
5019   std::list<librbd::RBD::AioCompletion *> comps;
5020   ceph::bufferlist bl;
5021   bl.append(std::string(1 << order, '1'));
5022   for (size_t object_no = 0; object_no < (size >> 12); ++object_no) {
5023     librbd::RBD::AioCompletion *comp = new librbd::RBD::AioCompletion(NULL,
5024                                                                       NULL);
5025     comps.push_back(comp);
5026     if (object_no % 2 == 0) {
5027       ASSERT_EQ(0, image1.aio_write(object_no << order, bl.length(), bl, comp));
5028     } else {
5029       ASSERT_EQ(0, image2.aio_write(object_no << order, bl.length(), bl, comp));
5030     }
5031   }
5032
5033   while (!comps.empty()) {
5034     librbd::RBD::AioCompletion *comp = comps.front();
5035     comps.pop_front();
5036     ASSERT_EQ(0, comp->wait_for_complete());
5037     ASSERT_EQ(1, comp->is_complete());
5038     comp->release();
5039   }
5040
5041   librbd::Image image3;
5042   ASSERT_EQ(0, rbd.open(ioctx, image3, name.c_str(), NULL));
5043   for (size_t object_no = 0; object_no < (size >> 12); ++object_no) {
5044     bufferlist read_bl;
5045     ASSERT_EQ((ssize_t)bl.length(), image3.read(object_no << order, bl.length(),
5046                                                 read_bl));
5047     ASSERT_TRUE(bl.contents_equal(read_bl));
5048   }
5049
5050   ASSERT_PASSED(validate_object_map, image1);
5051   ASSERT_PASSED(validate_object_map, image2);
5052   ASSERT_PASSED(validate_object_map, image3);
5053 }
5054
5055 TEST_F(TestLibRBD, ExclusiveLockReadTransition)
5056 {
5057   REQUIRE_FEATURE(RBD_FEATURE_JOURNALING);
5058
5059   librados::IoCtx ioctx;
5060   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
5061
5062   librbd::RBD rbd;
5063   std::string name = get_temp_image_name();
5064
5065   uint64_t size = 1 << 18;
5066   int order = 12;
5067   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
5068
5069   librbd::Image image1;
5070   ASSERT_EQ(0, rbd.open(ioctx, image1, name.c_str(), NULL));
5071
5072   bool lock_owner;
5073   ASSERT_EQ(0, image1.is_exclusive_lock_owner(&lock_owner));
5074   ASSERT_FALSE(lock_owner);
5075
5076   // journaling should force read ops to acquire the lock
5077   bufferlist read_bl;
5078   ASSERT_EQ(0, image1.read(0, 0, read_bl));
5079
5080   ASSERT_EQ(0, image1.is_exclusive_lock_owner(&lock_owner));
5081   ASSERT_TRUE(lock_owner);
5082
5083   librbd::Image image2;
5084   ASSERT_EQ(0, rbd.open(ioctx, image2, name.c_str(), NULL));
5085
5086   std::list<librbd::RBD::AioCompletion *> comps;
5087   std::list<bufferlist> read_bls;
5088   for (size_t object_no = 0; object_no < (size >> 12); ++object_no) {
5089     librbd::RBD::AioCompletion *comp = new librbd::RBD::AioCompletion(NULL,
5090                                                                       NULL);
5091     comps.push_back(comp);
5092     read_bls.emplace_back();
5093     if (object_no % 2 == 0) {
5094       ASSERT_EQ(0, image1.aio_read(object_no << order, 1 << order, read_bls.back(), comp));
5095     } else {
5096       ASSERT_EQ(0, image2.aio_read(object_no << order, 1 << order, read_bls.back(), comp));
5097     }
5098   }
5099
5100   while (!comps.empty()) {
5101     librbd::RBD::AioCompletion *comp = comps.front();
5102     comps.pop_front();
5103     ASSERT_EQ(0, comp->wait_for_complete());
5104     ASSERT_EQ(1, comp->is_complete());
5105     comp->release();
5106   }
5107 }
5108
5109 TEST_F(TestLibRBD, CacheMayCopyOnWrite) {
5110   REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
5111
5112   librados::IoCtx ioctx;
5113   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
5114
5115   librbd::RBD rbd;
5116   std::string name = get_temp_image_name();
5117
5118   uint64_t size = 1 << 18;
5119   int order = 12;
5120   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
5121
5122   librbd::Image image;
5123   ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
5124   ASSERT_EQ(0, image.snap_create("one"));
5125   ASSERT_EQ(0, image.snap_protect("one"));
5126
5127   std::string clone_name = this->get_temp_image_name();
5128   ASSERT_EQ(0, rbd.clone(ioctx, name.c_str(), "one", ioctx, clone_name.c_str(),
5129                          RBD_FEATURE_LAYERING, &order));
5130
5131   librbd::Image clone;
5132   ASSERT_EQ(0, rbd.open(ioctx, clone, clone_name.c_str(), NULL));
5133   ASSERT_EQ(0, clone.flush());
5134
5135   bufferlist expect_bl;
5136   expect_bl.append(std::string(1024, '\0'));
5137
5138   // test double read path
5139   bufferlist read_bl;
5140   uint64_t offset = 0;
5141   ASSERT_EQ(1024, clone.read(offset + 2048, 1024, read_bl));
5142   ASSERT_TRUE(expect_bl.contents_equal(read_bl));
5143
5144   bufferlist write_bl;
5145   write_bl.append(std::string(1024, '1'));
5146   ASSERT_EQ(1024, clone.write(offset, write_bl.length(), write_bl));
5147
5148   read_bl.clear();
5149   ASSERT_EQ(1024, clone.read(offset + 2048, 1024, read_bl));
5150   ASSERT_TRUE(expect_bl.contents_equal(read_bl));
5151
5152   // test read retry path
5153   offset = 1 << order;
5154   ASSERT_EQ(1024, clone.write(offset, write_bl.length(), write_bl));
5155
5156   read_bl.clear();
5157   ASSERT_EQ(1024, clone.read(offset + 2048, 1024, read_bl));
5158   ASSERT_TRUE(expect_bl.contents_equal(read_bl));
5159 }
5160
5161 TEST_F(TestLibRBD, FlushEmptyOpsOnExternalSnapshot) {
5162   std::string cache_enabled;
5163   ASSERT_EQ(0, _rados.conf_get("rbd_cache", cache_enabled));
5164   ASSERT_EQ(0, _rados.conf_set("rbd_cache", "false"));
5165   BOOST_SCOPE_EXIT( (cache_enabled) ) {
5166     ASSERT_EQ(0, _rados.conf_set("rbd_cache", cache_enabled.c_str()));
5167   } BOOST_SCOPE_EXIT_END;
5168
5169   librados::IoCtx ioctx;
5170   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
5171
5172   librbd::RBD rbd;
5173   std::string name = get_temp_image_name();
5174   uint64_t size = 1 << 18;
5175   int order = 0;
5176   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
5177
5178   librbd::Image image1;
5179   librbd::Image image2;
5180   ASSERT_EQ(0, rbd.open(ioctx, image1, name.c_str(), NULL));
5181   ASSERT_EQ(0, rbd.open(ioctx, image2, name.c_str(), NULL));
5182   ASSERT_EQ(0, image1.snap_create("snap1"));
5183
5184   librbd::RBD::AioCompletion *read_comp =
5185     new librbd::RBD::AioCompletion(NULL, NULL);
5186   bufferlist read_bl;
5187   image2.aio_read(0, 1024, read_bl, read_comp);
5188   ASSERT_EQ(0, read_comp->wait_for_complete());
5189   read_comp->release();
5190 }
5191
5192 TEST_F(TestLibRBD, TestImageOptions)
5193 {
5194   rados_ioctx_t ioctx;
5195   rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx);
5196
5197   //make create image options
5198   uint64_t features = RBD_FEATURE_LAYERING | RBD_FEATURE_STRIPINGV2 ;
5199   uint64_t order = 0;
5200   uint64_t stripe_unit = IMAGE_STRIPE_UNIT;
5201   uint64_t stripe_count = IMAGE_STRIPE_COUNT;
5202   rbd_image_options_t opts;
5203   rbd_image_options_create(&opts);
5204
5205   bool is_set;
5206   ASSERT_EQ(-EINVAL, rbd_image_options_is_set(opts, 12345, &is_set));
5207   ASSERT_EQ(0, rbd_image_options_is_set(opts, RBD_IMAGE_OPTION_FORMAT,
5208                                         &is_set));
5209   ASSERT_FALSE(is_set);
5210
5211   ASSERT_EQ(0, rbd_image_options_set_uint64(opts, RBD_IMAGE_OPTION_FORMAT,
5212           2));
5213   ASSERT_EQ(0, rbd_image_options_set_uint64(opts, RBD_IMAGE_OPTION_FEATURES,
5214           features));
5215   ASSERT_EQ(0, rbd_image_options_set_uint64(opts, RBD_IMAGE_OPTION_ORDER,
5216           order));
5217   ASSERT_EQ(0, rbd_image_options_set_uint64(opts, RBD_IMAGE_OPTION_STRIPE_UNIT,
5218           stripe_unit));
5219   ASSERT_EQ(0, rbd_image_options_set_uint64(opts, RBD_IMAGE_OPTION_STRIPE_COUNT,
5220           stripe_count));
5221
5222   ASSERT_EQ(0, rbd_image_options_is_set(opts, RBD_IMAGE_OPTION_FORMAT,
5223                                         &is_set));
5224   ASSERT_TRUE(is_set);
5225
5226   std::string parent_name = get_temp_image_name();
5227
5228   // make parent
5229   ASSERT_EQ(0, rbd_create4(ioctx, parent_name.c_str(), 4<<20, opts));
5230
5231   // check order is returned in opts
5232   ASSERT_EQ(0, rbd_image_options_get_uint64(opts, RBD_IMAGE_OPTION_ORDER,
5233           &order));
5234   ASSERT_NE((uint64_t)0, order);
5235
5236   // write some data to parent
5237   rbd_image_t parent;
5238   ASSERT_EQ(0, rbd_open(ioctx, parent_name.c_str(), &parent, NULL));
5239   char *data = (char *)"testdata";
5240   ASSERT_EQ((ssize_t)strlen(data), rbd_write(parent, 0, strlen(data), data));
5241   ASSERT_EQ((ssize_t)strlen(data), rbd_write(parent, 12, strlen(data), data));
5242
5243   // create a snapshot, reopen as the parent we're interested in
5244   ASSERT_EQ(0, rbd_snap_create(parent, "parent_snap"));
5245   ASSERT_EQ(0, rbd_close(parent));
5246   ASSERT_EQ(0, rbd_open(ioctx, parent_name.c_str(), &parent, "parent_snap"));
5247
5248   // clone
5249   std::string child_name = get_temp_image_name();
5250   ASSERT_EQ(0, rbd_snap_protect(parent, "parent_snap"));
5251   ASSERT_EQ(0, rbd_clone3(ioctx, parent_name.c_str(), "parent_snap", ioctx,
5252           child_name.c_str(), opts));
5253
5254   // copy
5255   std::string copy1_name = get_temp_image_name();
5256   ASSERT_EQ(0, rbd_copy3(parent, ioctx, copy1_name.c_str(), opts));
5257   std::string copy2_name = get_temp_image_name();
5258   ASSERT_EQ(0, rbd_copy_with_progress3(parent, ioctx, copy2_name.c_str(), opts,
5259           print_progress_percent, NULL));
5260
5261   ASSERT_EQ(0, rbd_close(parent));
5262
5263   rbd_image_options_destroy(opts);
5264
5265   rados_ioctx_destroy(ioctx);
5266 }
5267
5268 TEST_F(TestLibRBD, TestImageOptionsPP)
5269 {
5270   librados::IoCtx ioctx;
5271   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
5272
5273   //make create image options
5274   uint64_t features = RBD_FEATURE_LAYERING | RBD_FEATURE_STRIPINGV2 ;
5275   uint64_t order = 0;
5276   uint64_t stripe_unit = IMAGE_STRIPE_UNIT;
5277   uint64_t stripe_count = IMAGE_STRIPE_COUNT;
5278   librbd::ImageOptions opts;
5279   ASSERT_EQ(0, opts.set(RBD_IMAGE_OPTION_FORMAT, static_cast<uint64_t>(2)));
5280   ASSERT_EQ(0, opts.set(RBD_IMAGE_OPTION_FEATURES, features));
5281   ASSERT_EQ(0, opts.set(RBD_IMAGE_OPTION_ORDER, order));
5282   ASSERT_EQ(0, opts.set(RBD_IMAGE_OPTION_STRIPE_UNIT, stripe_unit));
5283   ASSERT_EQ(0, opts.set(RBD_IMAGE_OPTION_STRIPE_COUNT, stripe_count));
5284
5285   librbd::RBD rbd;
5286   std::string parent_name = get_temp_image_name();
5287
5288   // make parent
5289   ASSERT_EQ(0, rbd.create4(ioctx, parent_name.c_str(), 4<<20, opts));
5290
5291   // check order is returned in opts
5292   ASSERT_EQ(0, opts.get(RBD_IMAGE_OPTION_ORDER, &order));
5293   ASSERT_NE((uint64_t)0, order);
5294
5295   // write some data to parent
5296   librbd::Image parent;
5297   ASSERT_EQ(0, rbd.open(ioctx, parent, parent_name.c_str(), NULL));
5298
5299   ssize_t len = 1024;
5300   bufferlist bl;
5301   bl.append(buffer::create(len));
5302   bl.zero();
5303   ASSERT_EQ(len, parent.write(0, len, bl));
5304   ASSERT_EQ(len, parent.write(len, len, bl));
5305
5306   // create a snapshot, reopen as the parent we're interested in
5307   ASSERT_EQ(0, parent.snap_create("parent_snap"));
5308   ASSERT_EQ(0, parent.close());
5309   ASSERT_EQ(0, rbd.open(ioctx, parent, parent_name.c_str(), "parent_snap"));
5310
5311   // clone
5312   std::string child_name = get_temp_image_name();
5313   ASSERT_EQ(0, parent.snap_protect("parent_snap"));
5314   ASSERT_EQ(0, rbd.clone3(ioctx, parent_name.c_str(), "parent_snap", ioctx,
5315           child_name.c_str(), opts));
5316
5317   // copy
5318   std::string copy1_name = get_temp_image_name();
5319   ASSERT_EQ(0, parent.copy3(ioctx, copy1_name.c_str(), opts));
5320   std::string copy2_name = get_temp_image_name();
5321   PrintProgress pp;
5322   ASSERT_EQ(0, parent.copy_with_progress3(ioctx, copy2_name.c_str(), opts, pp));
5323
5324   ASSERT_EQ(0, parent.close());
5325 }
5326
5327 TEST_F(TestLibRBD, EventSocketPipe)
5328 {
5329   EventSocket event_sock;
5330   int pipe_fd[2]; // read and write fd
5331   char buf[32];
5332
5333   ASSERT_EQ(0, pipe(pipe_fd));
5334
5335   ASSERT_FALSE(event_sock.is_valid());
5336
5337   ASSERT_EQ(-EINVAL, event_sock.init(pipe_fd[1], EVENT_SOCKET_TYPE_NONE));
5338   ASSERT_FALSE(event_sock.is_valid());
5339
5340   ASSERT_EQ(-EINVAL, event_sock.init(pipe_fd[1], 44));
5341   ASSERT_FALSE(event_sock.is_valid());
5342
5343 #ifndef HAVE_EVENTFD
5344   ASSERT_EQ(-EINVAL, event_sock.init(pipe_fd[1], EVENT_SOCKET_TYPE_EVENTFD));
5345   ASSERT_FALSE(event_sock.is_valid());
5346 #endif
5347
5348   ASSERT_EQ(0, event_sock.init(pipe_fd[1], EVENT_SOCKET_TYPE_PIPE));
5349   ASSERT_TRUE(event_sock.is_valid());
5350   ASSERT_EQ(0, event_sock.notify());
5351   ASSERT_EQ(1, read(pipe_fd[0], buf, 32));
5352   ASSERT_EQ('i', buf[0]);
5353
5354   close(pipe_fd[0]);
5355   close(pipe_fd[1]);
5356 }
5357
5358 TEST_F(TestLibRBD, EventSocketEventfd)
5359 {
5360 #ifdef HAVE_EVENTFD
5361   EventSocket event_sock;
5362   int event_fd;
5363   struct pollfd poll_fd;
5364   char buf[32];
5365
5366   event_fd = eventfd(0, EFD_NONBLOCK);
5367   ASSERT_NE(-1, event_fd);
5368
5369   ASSERT_FALSE(event_sock.is_valid());
5370
5371   ASSERT_EQ(-EINVAL, event_sock.init(event_fd, EVENT_SOCKET_TYPE_NONE));
5372   ASSERT_FALSE(event_sock.is_valid());
5373
5374   ASSERT_EQ(-EINVAL, event_sock.init(event_fd, 44));
5375   ASSERT_FALSE(event_sock.is_valid());
5376
5377   ASSERT_EQ(0, event_sock.init(event_fd, EVENT_SOCKET_TYPE_EVENTFD));
5378   ASSERT_TRUE(event_sock.is_valid());
5379   ASSERT_EQ(0, event_sock.notify());
5380
5381   poll_fd.fd = event_fd;
5382   poll_fd.events = POLLIN;
5383   ASSERT_EQ(1, poll(&poll_fd, 1, -1));
5384   ASSERT_TRUE(poll_fd.revents & POLLIN);
5385
5386   ASSERT_EQ(static_cast<ssize_t>(sizeof(uint64_t)), read(event_fd, buf, 32));
5387   ASSERT_EQ(1U, *reinterpret_cast<uint64_t *>(buf));
5388
5389   close(event_fd);
5390 #endif
5391 }
5392
5393 TEST_F(TestLibRBD, ImagePollIO)
5394 {
5395 #ifdef HAVE_EVENTFD
5396   rados_ioctx_t ioctx;
5397   rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx);
5398
5399   rbd_image_t image;
5400   int order = 0;
5401   std::string name = get_temp_image_name();
5402   uint64_t size = 2 << 20;
5403   int fd = eventfd(0, EFD_NONBLOCK);
5404
5405   ASSERT_EQ(0, create_image(ioctx, name.c_str(), size, &order));
5406   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
5407
5408   ASSERT_EQ(0, rbd_set_image_notification(image, fd, EVENT_SOCKET_TYPE_EVENTFD));
5409
5410   char test_data[TEST_IO_SIZE + 1];
5411   char zero_data[TEST_IO_SIZE + 1];
5412   int i;
5413
5414   for (i = 0; i < TEST_IO_SIZE; ++i)
5415     test_data[i] = (char) (rand() % (126 - 33) + 33);
5416   test_data[TEST_IO_SIZE] = '\0';
5417   memset(zero_data, 0, sizeof(zero_data));
5418
5419   for (i = 0; i < 5; ++i)
5420     ASSERT_PASSED(write_test_data, image, test_data, TEST_IO_SIZE * i, TEST_IO_SIZE, 0);
5421
5422   for (i = 5; i < 10; ++i)
5423     ASSERT_PASSED(aio_write_test_data_and_poll, image, fd, test_data, TEST_IO_SIZE * i, TEST_IO_SIZE, 0);
5424
5425   for (i = 5; i < 10; ++i)
5426     ASSERT_PASSED(aio_read_test_data_and_poll, image, fd, test_data, TEST_IO_SIZE * i, TEST_IO_SIZE, 0);
5427
5428   ASSERT_EQ(0, rbd_close(image));
5429   rados_ioctx_destroy(ioctx);
5430 #endif
5431 }
5432
5433 namespace librbd {
5434
5435 static bool operator==(const mirror_peer_t &lhs, const mirror_peer_t &rhs) {
5436   return (lhs.uuid == rhs.uuid &&
5437           lhs.cluster_name == rhs.cluster_name &&
5438           lhs.client_name == rhs.client_name);
5439 }
5440
5441 static std::ostream& operator<<(std::ostream &os, const mirror_peer_t &peer) {
5442   os << "uuid=" << peer.uuid << ", "
5443      << "cluster=" << peer.cluster_name << ", "
5444      << "client=" << peer.client_name;
5445   return os;
5446 }
5447
5448 } // namespace librbd
5449
5450 TEST_F(TestLibRBD, Mirror) {
5451   librados::IoCtx ioctx;
5452   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
5453
5454   librbd::RBD rbd;
5455
5456   std::vector<librbd::mirror_peer_t> expected_peers;
5457   std::vector<librbd::mirror_peer_t> peers;
5458   ASSERT_EQ(0, rbd.mirror_peer_list(ioctx, &peers));
5459   ASSERT_EQ(expected_peers, peers);
5460
5461   std::string uuid1;
5462   ASSERT_EQ(-EINVAL, rbd.mirror_peer_add(ioctx, &uuid1, "cluster1", "client"));
5463
5464   rbd_mirror_mode_t mirror_mode;
5465   ASSERT_EQ(0, rbd.mirror_mode_get(ioctx, &mirror_mode));
5466   ASSERT_EQ(RBD_MIRROR_MODE_DISABLED, mirror_mode);
5467
5468   ASSERT_EQ(0, rbd.mirror_mode_set(ioctx, RBD_MIRROR_MODE_IMAGE));
5469   ASSERT_EQ(0, rbd.mirror_mode_get(ioctx, &mirror_mode));
5470
5471   // Add some images to the pool
5472   int order = 0;
5473   std::string parent_name = get_temp_image_name();
5474   std::string child_name = get_temp_image_name();
5475   ASSERT_EQ(0, create_image_pp(rbd, ioctx, parent_name.c_str(), 2 << 20,
5476                                &order));
5477   bool old_format;
5478   uint64_t features;
5479   ASSERT_EQ(0, get_features(&old_format, &features));
5480   if ((features & RBD_FEATURE_LAYERING) != 0) {
5481     librbd::Image parent;
5482     ASSERT_EQ(0, rbd.open(ioctx, parent, parent_name.c_str(), NULL));
5483     ASSERT_EQ(0, parent.snap_create("parent_snap"));
5484     ASSERT_EQ(0, parent.close());
5485     ASSERT_EQ(0, rbd.open(ioctx, parent, parent_name.c_str(), "parent_snap"));
5486     ASSERT_EQ(0, parent.snap_protect("parent_snap"));
5487     ASSERT_EQ(0, parent.close());
5488     ASSERT_EQ(0, rbd.clone(ioctx, parent_name.c_str(), "parent_snap", ioctx,
5489                            child_name.c_str(), features, &order));
5490   }
5491
5492   ASSERT_EQ(RBD_MIRROR_MODE_IMAGE, mirror_mode);
5493
5494   ASSERT_EQ(0, rbd.mirror_mode_set(ioctx, RBD_MIRROR_MODE_POOL));
5495   ASSERT_EQ(0, rbd.mirror_mode_get(ioctx, &mirror_mode));
5496   ASSERT_EQ(RBD_MIRROR_MODE_POOL, mirror_mode);
5497
5498   std::string uuid2;
5499   std::string uuid3;
5500   ASSERT_EQ(0, rbd.mirror_peer_add(ioctx, &uuid1, "cluster1", "client"));
5501   ASSERT_EQ(0, rbd.mirror_peer_add(ioctx, &uuid2, "cluster2", "admin"));
5502   ASSERT_EQ(-EEXIST, rbd.mirror_peer_add(ioctx, &uuid3, "cluster1", "foo"));
5503   ASSERT_EQ(0, rbd.mirror_peer_add(ioctx, &uuid3, "cluster3", "admin"));
5504
5505   ASSERT_EQ(0, rbd.mirror_peer_list(ioctx, &peers));
5506   auto sort_peers = [](const librbd::mirror_peer_t &lhs,
5507                          const librbd::mirror_peer_t &rhs) {
5508       return lhs.uuid < rhs.uuid;
5509     };
5510   expected_peers = {
5511     {uuid1, "cluster1", "client"},
5512     {uuid2, "cluster2", "admin"},
5513     {uuid3, "cluster3", "admin"}};
5514   std::sort(expected_peers.begin(), expected_peers.end(), sort_peers);
5515   ASSERT_EQ(expected_peers, peers);
5516
5517   ASSERT_EQ(0, rbd.mirror_peer_remove(ioctx, "uuid4"));
5518   ASSERT_EQ(0, rbd.mirror_peer_remove(ioctx, uuid2));
5519
5520   ASSERT_EQ(-ENOENT, rbd.mirror_peer_set_client(ioctx, "uuid4", "new client"));
5521   ASSERT_EQ(0, rbd.mirror_peer_set_client(ioctx, uuid1, "new client"));
5522
5523   ASSERT_EQ(-ENOENT, rbd.mirror_peer_set_cluster(ioctx, "uuid4",
5524                                                  "new cluster"));
5525   ASSERT_EQ(0, rbd.mirror_peer_set_cluster(ioctx, uuid3, "new cluster"));
5526
5527   ASSERT_EQ(0, rbd.mirror_peer_list(ioctx, &peers));
5528   expected_peers = {
5529     {uuid1, "cluster1", "new client"},
5530     {uuid3, "new cluster", "admin"}};
5531   std::sort(expected_peers.begin(), expected_peers.end(), sort_peers);
5532   ASSERT_EQ(expected_peers, peers);
5533
5534   ASSERT_EQ(-EBUSY, rbd.mirror_mode_set(ioctx, RBD_MIRROR_MODE_DISABLED));
5535 }
5536
5537 TEST_F(TestLibRBD, FlushCacheWithCopyupOnExternalSnapshot) {
5538   REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
5539
5540   librados::IoCtx ioctx;
5541   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
5542
5543   librbd::RBD rbd;
5544   librbd::Image image;
5545   std::string name = get_temp_image_name();
5546
5547   uint64_t size = 1 << 18;
5548   int order = 0;
5549
5550   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
5551   ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
5552
5553   bufferlist bl;
5554   bl.append(std::string(size, '1'));
5555   ASSERT_EQ((int)size, image.write(0, size, bl));
5556   ASSERT_EQ(0, image.snap_create("one"));
5557   ASSERT_EQ(0, image.snap_protect("one"));
5558
5559   std::string clone_name = this->get_temp_image_name();
5560   ASSERT_EQ(0, rbd.clone(ioctx, name.c_str(), "one", ioctx, clone_name.c_str(),
5561                          RBD_FEATURE_LAYERING, &order));
5562   ASSERT_EQ(0, rbd.open(ioctx, image, clone_name.c_str(), NULL));
5563
5564   librbd::Image image2;
5565   ASSERT_EQ(0, rbd.open(ioctx, image2, clone_name.c_str(), NULL));
5566
5567   // prepare CoW writeback that will be flushed on next op
5568   bl.clear();
5569   bl.append(std::string(1, '1'));
5570   ASSERT_EQ(0, image.flush());
5571   ASSERT_EQ(1, image.write(0, 1, bl));
5572   ASSERT_EQ(0, image2.snap_create("snap1"));
5573
5574   librbd::RBD::AioCompletion *read_comp =
5575     new librbd::RBD::AioCompletion(NULL, NULL);
5576   bufferlist read_bl;
5577   image.aio_read(0, 1024, read_bl, read_comp);
5578   ASSERT_EQ(0, read_comp->wait_for_complete());
5579   read_comp->release();
5580 }
5581
5582 TEST_F(TestLibRBD, ExclusiveLock)
5583 {
5584   REQUIRE_FEATURE(RBD_FEATURE_EXCLUSIVE_LOCK);
5585
5586   static char buf[10];
5587
5588   rados_ioctx_t ioctx;
5589   rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx);
5590
5591   std::string name = get_temp_image_name();
5592   uint64_t size = 2 << 20;
5593   int order = 0;
5594   ASSERT_EQ(0, create_image(ioctx, name.c_str(), size, &order));
5595
5596   rbd_image_t image1;
5597   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image1, NULL));
5598
5599   int lock_owner;
5600   ASSERT_EQ(0, rbd_lock_acquire(image1, RBD_LOCK_MODE_EXCLUSIVE));
5601   ASSERT_EQ(0, rbd_is_exclusive_lock_owner(image1, &lock_owner));
5602   ASSERT_TRUE(lock_owner);
5603
5604   rbd_lock_mode_t lock_mode;
5605   char *lock_owners[1];
5606   size_t max_lock_owners = 0;
5607   ASSERT_EQ(-ERANGE, rbd_lock_get_owners(image1, &lock_mode, lock_owners,
5608                                          &max_lock_owners));
5609   ASSERT_EQ(1U, max_lock_owners);
5610
5611   max_lock_owners = 2;
5612   ASSERT_EQ(0, rbd_lock_get_owners(image1, &lock_mode, lock_owners,
5613                                    &max_lock_owners));
5614   ASSERT_EQ(RBD_LOCK_MODE_EXCLUSIVE, lock_mode);
5615   ASSERT_STRNE("", lock_owners[0]);
5616   ASSERT_EQ(1U, max_lock_owners);
5617
5618   rbd_image_t image2;
5619   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image2, NULL));
5620
5621   ASSERT_EQ(0, rbd_is_exclusive_lock_owner(image2, &lock_owner));
5622   ASSERT_FALSE(lock_owner);
5623
5624   ASSERT_EQ(-EOPNOTSUPP, rbd_lock_break(image1, RBD_LOCK_MODE_SHARED, ""));
5625   ASSERT_EQ(-EBUSY, rbd_lock_break(image1, RBD_LOCK_MODE_EXCLUSIVE,
5626                                    "not the owner"));
5627
5628   ASSERT_EQ(0, rbd_lock_release(image1));
5629   ASSERT_EQ(0, rbd_is_exclusive_lock_owner(image1, &lock_owner));
5630   ASSERT_FALSE(lock_owner);
5631
5632   ASSERT_EQ(-ENOENT, rbd_lock_break(image1, RBD_LOCK_MODE_EXCLUSIVE,
5633                                     lock_owners[0]));
5634   rbd_lock_get_owners_cleanup(lock_owners, max_lock_owners);
5635
5636   ASSERT_EQ(-EROFS, rbd_write(image1, 0, sizeof(buf), buf));
5637   ASSERT_EQ((ssize_t)sizeof(buf), rbd_write(image2, 0, sizeof(buf), buf));
5638
5639   ASSERT_EQ(0, rbd_lock_acquire(image2, RBD_LOCK_MODE_EXCLUSIVE));
5640   ASSERT_EQ(0, rbd_is_exclusive_lock_owner(image2, &lock_owner));
5641   ASSERT_TRUE(lock_owner);
5642
5643   ASSERT_EQ(0, rbd_lock_release(image2));
5644   ASSERT_EQ(0, rbd_is_exclusive_lock_owner(image2, &lock_owner));
5645   ASSERT_FALSE(lock_owner);
5646
5647   ASSERT_EQ(0, rbd_lock_acquire(image1, RBD_LOCK_MODE_EXCLUSIVE));
5648   ASSERT_EQ(0, rbd_is_exclusive_lock_owner(image1, &lock_owner));
5649   ASSERT_TRUE(lock_owner);
5650
5651   ASSERT_EQ((ssize_t)sizeof(buf), rbd_write(image1, 0, sizeof(buf), buf));
5652   ASSERT_EQ(-EROFS, rbd_write(image2, 0, sizeof(buf), buf));
5653
5654   ASSERT_EQ(0, rbd_lock_release(image1));
5655   ASSERT_EQ(0, rbd_is_exclusive_lock_owner(image1, &lock_owner));
5656   ASSERT_FALSE(lock_owner);
5657
5658   int owner_id = -1;
5659   mutex lock;
5660   const auto pingpong = [&,this](int m_id, rbd_image_t &m_image) {
5661       for (int i = 0; i < 10; i++) {
5662         {
5663           lock_guard<mutex> locker(lock);
5664           if (owner_id == m_id) {
5665             std::cout << m_id << ": releasing exclusive lock" << std::endl;
5666             EXPECT_EQ(0, rbd_lock_release(m_image));
5667             int lock_owner;
5668             EXPECT_EQ(0, rbd_is_exclusive_lock_owner(m_image, &lock_owner));
5669             EXPECT_FALSE(lock_owner);
5670             owner_id = -1;
5671             std::cout << m_id << ": exclusive lock released" << std::endl;
5672             continue;
5673           }
5674         }
5675
5676         std::cout << m_id << ": acquiring exclusive lock" << std::endl;
5677         int r;
5678         do {
5679           r = rbd_lock_acquire(m_image, RBD_LOCK_MODE_EXCLUSIVE);
5680           if (r == -EROFS) {
5681             usleep(1000);
5682           }
5683         } while (r == -EROFS);
5684         EXPECT_EQ(0, r);
5685
5686         int lock_owner;
5687         EXPECT_EQ(0, rbd_is_exclusive_lock_owner(m_image, &lock_owner));
5688         EXPECT_TRUE(lock_owner);
5689         std::cout << m_id << ": exclusive lock acquired" << std::endl;
5690         {
5691           lock_guard<mutex> locker(lock);
5692           owner_id = m_id;
5693         }
5694         usleep(rand() % 50000);
5695       }
5696
5697       lock_guard<mutex> locker(lock);
5698       if (owner_id == m_id) {
5699         EXPECT_EQ(0, rbd_lock_release(m_image));
5700         int lock_owner;
5701         EXPECT_EQ(0, rbd_is_exclusive_lock_owner(m_image, &lock_owner));
5702         EXPECT_FALSE(lock_owner);
5703         owner_id = -1;
5704       }
5705   };
5706   thread ping(bind(pingpong, 1, ref(image1)));
5707   thread pong(bind(pingpong, 2, ref(image2)));
5708
5709   ping.join();
5710   pong.join();
5711
5712   ASSERT_EQ(0, rbd_lock_acquire(image2, RBD_LOCK_MODE_EXCLUSIVE));
5713   ASSERT_EQ(0, rbd_is_exclusive_lock_owner(image2, &lock_owner));
5714   ASSERT_TRUE(lock_owner);
5715
5716   ASSERT_EQ(0, rbd_close(image2));
5717
5718   ASSERT_EQ(0, rbd_lock_acquire(image1, RBD_LOCK_MODE_EXCLUSIVE));
5719   ASSERT_EQ(0, rbd_is_exclusive_lock_owner(image1, &lock_owner));
5720   ASSERT_TRUE(lock_owner);
5721
5722   ASSERT_EQ(0, rbd_close(image1));
5723   rados_ioctx_destroy(ioctx);
5724 }
5725
5726 TEST_F(TestLibRBD, BreakLock)
5727 {
5728   REQUIRE_FEATURE(RBD_FEATURE_EXCLUSIVE_LOCK);
5729
5730   static char buf[10];
5731
5732   rados_t blacklist_cluster;
5733   ASSERT_EQ("", connect_cluster(&blacklist_cluster));
5734
5735   rados_ioctx_t ioctx, blacklist_ioctx;
5736   ASSERT_EQ(0, rados_ioctx_create(_cluster, m_pool_name.c_str(), &ioctx));
5737   ASSERT_EQ(0, rados_ioctx_create(blacklist_cluster, m_pool_name.c_str(),
5738                                   &blacklist_ioctx));
5739
5740   std::string name = get_temp_image_name();
5741   uint64_t size = 2 << 20;
5742   int order = 0;
5743   ASSERT_EQ(0, create_image(ioctx, name.c_str(), size, &order));
5744
5745   rbd_image_t image, blacklist_image;
5746   ASSERT_EQ(0, rbd_open(ioctx, name.c_str(), &image, NULL));
5747   ASSERT_EQ(0, rbd_open(blacklist_ioctx, name.c_str(), &blacklist_image, NULL));
5748
5749   ASSERT_EQ(0, rbd_metadata_set(image, "rbd_blacklist_on_break_lock", "true"));
5750   ASSERT_EQ(0, rbd_lock_acquire(blacklist_image, RBD_LOCK_MODE_EXCLUSIVE));
5751
5752   rbd_lock_mode_t lock_mode;
5753   char *lock_owners[1];
5754   size_t max_lock_owners = 1;
5755   ASSERT_EQ(0, rbd_lock_get_owners(image, &lock_mode, lock_owners,
5756                                    &max_lock_owners));
5757   ASSERT_EQ(RBD_LOCK_MODE_EXCLUSIVE, lock_mode);
5758   ASSERT_STRNE("", lock_owners[0]);
5759   ASSERT_EQ(1U, max_lock_owners);
5760
5761   ASSERT_EQ(0, rbd_lock_break(image, RBD_LOCK_MODE_EXCLUSIVE, lock_owners[0]));
5762   ASSERT_EQ(0, rbd_lock_acquire(image, RBD_LOCK_MODE_EXCLUSIVE));
5763   EXPECT_EQ(0, rados_wait_for_latest_osdmap(blacklist_cluster));
5764
5765   ASSERT_EQ((ssize_t)sizeof(buf), rbd_write(image, 0, sizeof(buf), buf));
5766   ASSERT_EQ(-EBLACKLISTED, rbd_write(blacklist_image, 0, sizeof(buf), buf));
5767
5768   ASSERT_EQ(0, rbd_close(image));
5769   ASSERT_EQ(0, rbd_close(blacklist_image));
5770
5771   rbd_lock_get_owners_cleanup(lock_owners, max_lock_owners);
5772
5773   rados_ioctx_destroy(ioctx);
5774   rados_ioctx_destroy(blacklist_ioctx);
5775   rados_shutdown(blacklist_cluster);
5776 }
5777
5778 TEST_F(TestLibRBD, DiscardAfterWrite)
5779 {
5780   REQUIRE(!is_skip_partial_discard_enabled());
5781
5782   librados::IoCtx ioctx;
5783   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
5784
5785   librbd::RBD rbd;
5786   std::string name = get_temp_image_name();
5787   uint64_t size = 1 << 20;
5788   int order = 18;
5789   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
5790
5791   librbd::Image image;
5792   ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
5793
5794   // enable writeback cache
5795   ASSERT_EQ(0, image.flush());
5796
5797   bufferlist bl;
5798   bl.append(std::string(256, '1'));
5799
5800   librbd::RBD::AioCompletion *write_comp =
5801     new librbd::RBD::AioCompletion(NULL, NULL);
5802   ASSERT_EQ(0, image.aio_write(0, bl.length(), bl, write_comp));
5803   ASSERT_EQ(0, write_comp->wait_for_complete());
5804   write_comp->release();
5805
5806   librbd::RBD::AioCompletion *discard_comp =
5807     new librbd::RBD::AioCompletion(NULL, NULL);
5808   ASSERT_EQ(0, image.aio_discard(0, 256, discard_comp));
5809   ASSERT_EQ(0, discard_comp->wait_for_complete());
5810   discard_comp->release();
5811
5812   librbd::RBD::AioCompletion *read_comp =
5813     new librbd::RBD::AioCompletion(NULL, NULL);
5814   bufferlist read_bl;
5815   image.aio_read(0, bl.length(), read_bl, read_comp);
5816   ASSERT_EQ(0, read_comp->wait_for_complete());
5817   ASSERT_EQ(bl.length(), read_comp->get_return_value());
5818   ASSERT_TRUE(read_bl.is_zero());
5819   read_comp->release();
5820 }
5821
5822 TEST_F(TestLibRBD, DefaultFeatures) {
5823   std::string orig_default_features;
5824   ASSERT_EQ(0, _rados.conf_get("rbd_default_features", orig_default_features));
5825   BOOST_SCOPE_EXIT_ALL(orig_default_features) {
5826     ASSERT_EQ(0, _rados.conf_set("rbd_default_features",
5827                                  orig_default_features.c_str()));
5828   };
5829
5830   std::list<std::pair<std::string, std::string> > feature_names_to_bitmask = {
5831     {"", orig_default_features},
5832     {"layering", "1"},
5833     {"layering, exclusive-lock", "5"},
5834     {"exclusive-lock,journaling", "68"},
5835     {"125", "125"}
5836   };
5837
5838   for (auto &pair : feature_names_to_bitmask) {
5839     ASSERT_EQ(0, _rados.conf_set("rbd_default_features", pair.first.c_str()));
5840     std::string features;
5841     ASSERT_EQ(0, _rados.conf_get("rbd_default_features", features));
5842     ASSERT_EQ(pair.second, features);
5843   }
5844 }
5845
5846 TEST_F(TestLibRBD, TestTrashMoveAndPurge) {
5847   librados::IoCtx ioctx;
5848   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
5849
5850   librbd::RBD rbd;
5851   std::string name = get_temp_image_name();
5852
5853   uint64_t size = 1 << 18;
5854   int order = 12;
5855   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
5856
5857   librbd::Image image;
5858   ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), nullptr));
5859   uint8_t old_format;
5860   ASSERT_EQ(0, image.old_format(&old_format));
5861
5862   if (old_format) {
5863     ASSERT_EQ(-EOPNOTSUPP, rbd.trash_move(ioctx, name.c_str(), 0));
5864     image.close();
5865     return;
5866   }
5867   std::string image_id;
5868   ASSERT_EQ(0, image.get_id(&image_id));
5869   image.close();
5870
5871   ASSERT_EQ(0, rbd.trash_move(ioctx, name.c_str(), 0));
5872
5873   std::vector<std::string> images;
5874   ASSERT_EQ(0, rbd.list(ioctx, images));
5875   for (const auto& image : images) {
5876     ASSERT_TRUE(image != name);
5877   }
5878
5879   librbd::trash_image_info_t info;
5880   ASSERT_EQ(-ENOENT, rbd.trash_get(ioctx, "dummy image id", &info));
5881   ASSERT_EQ(0, rbd.trash_get(ioctx, image_id.c_str(), &info));
5882   ASSERT_EQ(image_id, info.id);
5883
5884   std::vector<librbd::trash_image_info_t> entries;
5885   ASSERT_EQ(0, rbd.trash_list(ioctx, entries));
5886   ASSERT_FALSE(entries.empty());
5887   ASSERT_EQ(entries.begin()->id, image_id);
5888
5889   entries.clear();
5890   PrintProgress pp;
5891   ASSERT_EQ(0, rbd.trash_remove_with_progress(ioctx, image_id.c_str(),
5892                                               false, pp));
5893   ASSERT_EQ(0, rbd.trash_list(ioctx, entries));
5894   ASSERT_TRUE(entries.empty());
5895 }
5896
5897 TEST_F(TestLibRBD, TestTrashMoveAndPurgeNonExpiredDelay) {
5898   librados::IoCtx ioctx;
5899   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
5900
5901   librbd::RBD rbd;
5902   std::string name = get_temp_image_name();
5903
5904   uint64_t size = 1 << 18;
5905   int order = 12;
5906   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
5907
5908   librbd::Image image;
5909   ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), nullptr));
5910   uint8_t old_format;
5911   ASSERT_EQ(0, image.old_format(&old_format));
5912
5913   if (old_format) {
5914     ASSERT_EQ(-EOPNOTSUPP, rbd.trash_move(ioctx, name.c_str(), 0));
5915     image.close();
5916     return;
5917   }
5918   std::string image_id;
5919   ASSERT_EQ(0, image.get_id(&image_id));
5920   image.close();
5921
5922   ASSERT_EQ(0, rbd.trash_move(ioctx, name.c_str(), 100));
5923
5924   PrintProgress pp;
5925   ASSERT_EQ(-EPERM, rbd.trash_remove_with_progress(ioctx, image_id.c_str(),
5926                                                    false, pp));
5927
5928   PrintProgress pp2;
5929   ASSERT_EQ(0, rbd.trash_remove_with_progress(ioctx, image_id.c_str(),
5930                                               true, pp2));
5931 }
5932
5933 TEST_F(TestLibRBD, TestTrashMoveAndRestore) {
5934   librados::IoCtx ioctx;
5935   ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
5936
5937   librbd::RBD rbd;
5938   std::string name = get_temp_image_name();
5939
5940   uint64_t size = 1 << 18;
5941   int order = 12;
5942   ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
5943
5944   librbd::Image image;
5945   ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), nullptr));
5946   uint8_t old_format;
5947   ASSERT_EQ(0, image.old_format(&old_format));
5948
5949   if (old_format) {
5950     ASSERT_EQ(-EOPNOTSUPP, rbd.trash_move(ioctx, name.c_str(), 0));
5951     image.close();
5952     return;
5953   }
5954   std::string image_id;
5955   ASSERT_EQ(0, image.get_id(&image_id));
5956   image.close();
5957
5958   ASSERT_EQ(0, rbd.trash_move(ioctx, name.c_str(), 10));
5959
5960   std::vector<std::string> images;
5961   ASSERT_EQ(0, rbd.list(ioctx, images));
5962   for (const auto& image : images) {
5963     ASSERT_TRUE(image != name);
5964   }
5965
5966   std::vector<librbd::trash_image_info_t> entries;
5967   ASSERT_EQ(0, rbd.trash_list(ioctx, entries));
5968   ASSERT_FALSE(entries.empty());
5969   ASSERT_EQ(entries.begin()->id, image_id);
5970
5971   images.clear();
5972   ASSERT_EQ(0, rbd.trash_restore(ioctx, image_id.c_str(), ""));
5973   ASSERT_EQ(0, rbd.list(ioctx, images));
5974   ASSERT_FALSE(images.empty());
5975   bool found = false;
5976   for (const auto& image : images) {
5977     if (image == name) {
5978       found = true;
5979       break;
5980     }
5981   }
5982   ASSERT_TRUE(found);
5983 }
5984
5985 // poorman's assert()
5986 namespace ceph {
5987   void __ceph_assert_fail(const char *assertion, const char *file, int line,
5988                           const char *func) {
5989     assert(false);
5990   }
5991 }