ceph/src/rocksdb/utilities/env_librados_test.cc

   1 //  Copyright (c) 2016, Red Hat, Inc.  All rights reserved.
   2 //  This source code is licensed under the BSD-style license found in the
   3 //  LICENSE file in the root directory of this source tree. An additional grant
   4 //  of patent rights can be found in the PATENTS file in the same directory.
   5
   6 #ifndef ROCKSDB_LITE
   7
   8 #include "rocksdb/utilities/env_librados.h"
   9 #include <rados/librados.hpp>
  10 #include "env/mock_env.h"
  11 #include "util/testharness.h"
  12
  13 #include "rocksdb/db.h"
  14 #include "rocksdb/slice.h"
  15 #include "rocksdb/options.h"
  16 #include "util/random.h"
  17 #include <chrono>
  18 #include <ostream>
  19 #include "rocksdb/utilities/transaction_db.h"
  20
  21 class Timer {
  22   typedef std::chrono::high_resolution_clock high_resolution_clock;
  23   typedef std::chrono::milliseconds milliseconds;
  24 public:
  25   explicit Timer(bool run = false)
  26   {
  27     if (run)
  28       Reset();
  29   }
  30   void Reset()
  31   {
  32     _start = high_resolution_clock::now();
  33   }
  34   milliseconds Elapsed() const
  35   {
  36     return std::chrono::duration_cast<milliseconds>(high_resolution_clock::now() - _start);
  37   }
  38   template <typename T, typename Traits>
  39   friend std::basic_ostream<T, Traits>& operator<<(std::basic_ostream<T, Traits>& out, const Timer& timer)
  40   {
  41     return out << timer.Elapsed().count();
  42   }
  43 private:
  44   high_resolution_clock::time_point _start;
  45 };
  46
  47 namespace rocksdb {
  48
  49 class EnvLibradosTest : public testing::Test {
  50 public:
  51   // we will use all of these below
  52   const std::string db_name = "env_librados_test_db";
  53   const std::string db_pool = db_name + "_pool";
  54   const char *keyring = "admin";
  55   const char *config = "../ceph/src/ceph.conf";
  56
  57   EnvLibrados* env_;
  58   const EnvOptions soptions_;
  59
  60   EnvLibradosTest()
  61     : env_(new EnvLibrados(db_name, config, db_pool)) {
  62   }
  63   ~EnvLibradosTest() {
  64     delete env_;
  65     librados::Rados rados;
  66     int ret = 0;
  67     do {
  68       ret = rados.init("admin"); // just use the client.admin keyring
  69       if (ret < 0) { // let's handle any error that might have come back
  70         std::cerr << "couldn't initialize rados! error " << ret << std::endl;
  71         ret = EXIT_FAILURE;
  72         break;
  73       }
  74
  75       ret = rados.conf_read_file(config);
  76       if (ret < 0) {
  77         // This could fail if the config file is malformed, but it'd be hard.
  78         std::cerr << "failed to parse config file " << config
  79                   << "! error" << ret << std::endl;
  80         ret = EXIT_FAILURE;
  81         break;
  82       }
  83
  84       /*
  85        * next, we actually connect to the cluster
  86        */
  87
  88       ret = rados.connect();
  89       if (ret < 0) {
  90         std::cerr << "couldn't connect to cluster! error " << ret << std::endl;
  91         ret = EXIT_FAILURE;
  92         break;
  93       }
  94
  95       /*
  96        * And now we're done, so let's remove our pool and then
  97        * shut down the connection gracefully.
  98        */
  99       int delete_ret = rados.pool_delete(db_pool.c_str());
 100       if (delete_ret < 0) {
 101         // be careful not to
 102         std::cerr << "We failed to delete our test pool!" << db_pool << delete_ret << std::endl;
 103         ret = EXIT_FAILURE;
 104       }
 105     } while (0);
 106   }
 107 };
 108
 109 TEST_F(EnvLibradosTest, Basics) {
 110   uint64_t file_size;
 111   unique_ptr<WritableFile> writable_file;
 112   std::vector<std::string> children;
 113
 114   ASSERT_OK(env_->CreateDir("/dir"));
 115   // Check that the directory is empty.
 116   ASSERT_EQ(Status::NotFound(), env_->FileExists("/dir/non_existent"));
 117   ASSERT_TRUE(!env_->GetFileSize("/dir/non_existent", &file_size).ok());
 118   ASSERT_OK(env_->GetChildren("/dir", &children));
 119   ASSERT_EQ(0U, children.size());
 120
 121   // Create a file.
 122   ASSERT_OK(env_->NewWritableFile("/dir/f", &writable_file, soptions_));
 123   writable_file.reset();
 124
 125   // Check that the file exists.
 126   ASSERT_OK(env_->FileExists("/dir/f"));
 127   ASSERT_OK(env_->GetFileSize("/dir/f", &file_size));
 128   ASSERT_EQ(0U, file_size);
 129   ASSERT_OK(env_->GetChildren("/dir", &children));
 130   ASSERT_EQ(1U, children.size());
 131   ASSERT_EQ("f", children[0]);
 132
 133   // Write to the file.
 134   ASSERT_OK(env_->NewWritableFile("/dir/f", &writable_file, soptions_));
 135   ASSERT_OK(writable_file->Append("abc"));
 136   writable_file.reset();
 137
 138
 139   // Check for expected size.
 140   ASSERT_OK(env_->GetFileSize("/dir/f", &file_size));
 141   ASSERT_EQ(3U, file_size);
 142
 143
 144   // Check that renaming works.
 145   ASSERT_TRUE(!env_->RenameFile("/dir/non_existent", "/dir/g").ok());
 146   ASSERT_OK(env_->RenameFile("/dir/f", "/dir/g"));
 147   ASSERT_EQ(Status::NotFound(), env_->FileExists("/dir/f"));
 148   ASSERT_OK(env_->FileExists("/dir/g"));
 149   ASSERT_OK(env_->GetFileSize("/dir/g", &file_size));
 150   ASSERT_EQ(3U, file_size);
 151
 152   // Check that opening non-existent file fails.
 153   unique_ptr<SequentialFile> seq_file;
 154   unique_ptr<RandomAccessFile> rand_file;
 155   ASSERT_TRUE(
 156     !env_->NewSequentialFile("/dir/non_existent", &seq_file, soptions_).ok());
 157   ASSERT_TRUE(!seq_file);
 158   ASSERT_TRUE(!env_->NewRandomAccessFile("/dir/non_existent", &rand_file,
 159                                          soptions_).ok());
 160   ASSERT_TRUE(!rand_file);
 161
 162   // Check that deleting works.
 163   ASSERT_TRUE(!env_->DeleteFile("/dir/non_existent").ok());
 164   ASSERT_OK(env_->DeleteFile("/dir/g"));
 165   ASSERT_EQ(Status::NotFound(), env_->FileExists("/dir/g"));
 166   ASSERT_OK(env_->GetChildren("/dir", &children));
 167   ASSERT_EQ(0U, children.size());
 168   ASSERT_OK(env_->DeleteDir("/dir"));
 169 }
 170
 171 TEST_F(EnvLibradosTest, ReadWrite) {
 172   unique_ptr<WritableFile> writable_file;
 173   unique_ptr<SequentialFile> seq_file;
 174   unique_ptr<RandomAccessFile> rand_file;
 175   Slice result;
 176   char scratch[100];
 177
 178   ASSERT_OK(env_->CreateDir("/dir"));
 179
 180   ASSERT_OK(env_->NewWritableFile("/dir/f", &writable_file, soptions_));
 181   ASSERT_OK(writable_file->Append("hello "));
 182   ASSERT_OK(writable_file->Append("world"));
 183   writable_file.reset();
 184
 185   // Read sequentially.
 186   ASSERT_OK(env_->NewSequentialFile("/dir/f", &seq_file, soptions_));
 187   ASSERT_OK(seq_file->Read(5, &result, scratch));  // Read "hello".
 188   ASSERT_EQ(0, result.compare("hello"));
 189   ASSERT_OK(seq_file->Skip(1));
 190   ASSERT_OK(seq_file->Read(1000, &result, scratch));  // Read "world".
 191   ASSERT_EQ(0, result.compare("world"));
 192   ASSERT_OK(seq_file->Read(1000, &result, scratch));  // Try reading past EOF.
 193   ASSERT_EQ(0U, result.size());
 194   ASSERT_OK(seq_file->Skip(100));  // Try to skip past end of file.
 195   ASSERT_OK(seq_file->Read(1000, &result, scratch));
 196   ASSERT_EQ(0U, result.size());
 197
 198   // Random reads.
 199   ASSERT_OK(env_->NewRandomAccessFile("/dir/f", &rand_file, soptions_));
 200   ASSERT_OK(rand_file->Read(6, 5, &result, scratch));  // Read "world".
 201   ASSERT_EQ(0, result.compare("world"));
 202   ASSERT_OK(rand_file->Read(0, 5, &result, scratch));  // Read "hello".
 203   ASSERT_EQ(0, result.compare("hello"));
 204   ASSERT_OK(rand_file->Read(10, 100, &result, scratch));  // Read "d".
 205   ASSERT_EQ(0, result.compare("d"));
 206
 207   // Too high offset.
 208   ASSERT_OK(rand_file->Read(1000, 5, &result, scratch));
 209 }
 210
 211 TEST_F(EnvLibradosTest, Locks) {
 212   FileLock* lock = nullptr;
 213   unique_ptr<WritableFile> writable_file;
 214
 215   ASSERT_OK(env_->CreateDir("/dir"));
 216
 217   ASSERT_OK(env_->NewWritableFile("/dir/f", &writable_file, soptions_));
 218
 219   // These are no-ops, but we test they return success.
 220   ASSERT_OK(env_->LockFile("some file", &lock));
 221   ASSERT_OK(env_->UnlockFile(lock));
 222
 223   ASSERT_OK(env_->LockFile("/dir/f", &lock));
 224   ASSERT_OK(env_->UnlockFile(lock));
 225 }
 226
 227 TEST_F(EnvLibradosTest, Misc) {
 228   std::string test_dir;
 229   ASSERT_OK(env_->GetTestDirectory(&test_dir));
 230   ASSERT_TRUE(!test_dir.empty());
 231
 232   unique_ptr<WritableFile> writable_file;
 233   ASSERT_TRUE(!env_->NewWritableFile("/a/b", &writable_file, soptions_).ok());
 234
 235   ASSERT_OK(env_->NewWritableFile("/a", &writable_file, soptions_));
 236   // These are no-ops, but we test they return success.
 237   ASSERT_OK(writable_file->Sync());
 238   ASSERT_OK(writable_file->Flush());
 239   ASSERT_OK(writable_file->Close());
 240   writable_file.reset();
 241 }
 242
 243 TEST_F(EnvLibradosTest, LargeWrite) {
 244   const size_t kWriteSize = 300 * 1024;
 245   char* scratch = new char[kWriteSize * 2];
 246
 247   std::string write_data;
 248   for (size_t i = 0; i < kWriteSize; ++i) {
 249     write_data.append(1, 'h');
 250   }
 251
 252   unique_ptr<WritableFile> writable_file;
 253   ASSERT_OK(env_->CreateDir("/dir"));
 254   ASSERT_OK(env_->NewWritableFile("/dir/g", &writable_file, soptions_));
 255   ASSERT_OK(writable_file->Append("foo"));
 256   ASSERT_OK(writable_file->Append(write_data));
 257   writable_file.reset();
 258
 259   unique_ptr<SequentialFile> seq_file;
 260   Slice result;
 261   ASSERT_OK(env_->NewSequentialFile("/dir/g", &seq_file, soptions_));
 262   ASSERT_OK(seq_file->Read(3, &result, scratch));  // Read "foo".
 263   ASSERT_EQ(0, result.compare("foo"));
 264
 265   size_t read = 0;
 266   std::string read_data;
 267   while (read < kWriteSize) {
 268     ASSERT_OK(seq_file->Read(kWriteSize - read, &result, scratch));
 269     read_data.append(result.data(), result.size());
 270     read += result.size();
 271   }
 272   ASSERT_TRUE(write_data == read_data);
 273   delete[] scratch;
 274 }
 275
 276 TEST_F(EnvLibradosTest, FrequentlySmallWrite) {
 277   const size_t kWriteSize = 1 << 10;
 278   char* scratch = new char[kWriteSize * 2];
 279
 280   std::string write_data;
 281   for (size_t i = 0; i < kWriteSize; ++i) {
 282     write_data.append(1, 'h');
 283   }
 284
 285   unique_ptr<WritableFile> writable_file;
 286   ASSERT_OK(env_->CreateDir("/dir"));
 287   ASSERT_OK(env_->NewWritableFile("/dir/g", &writable_file, soptions_));
 288   ASSERT_OK(writable_file->Append("foo"));
 289
 290   for (size_t i = 0; i < kWriteSize; ++i) {
 291     ASSERT_OK(writable_file->Append("h"));
 292   }
 293   writable_file.reset();
 294
 295   unique_ptr<SequentialFile> seq_file;
 296   Slice result;
 297   ASSERT_OK(env_->NewSequentialFile("/dir/g", &seq_file, soptions_));
 298   ASSERT_OK(seq_file->Read(3, &result, scratch));  // Read "foo".
 299   ASSERT_EQ(0, result.compare("foo"));
 300
 301   size_t read = 0;
 302   std::string read_data;
 303   while (read < kWriteSize) {
 304     ASSERT_OK(seq_file->Read(kWriteSize - read, &result, scratch));
 305     read_data.append(result.data(), result.size());
 306     read += result.size();
 307   }
 308   ASSERT_TRUE(write_data == read_data);
 309   delete[] scratch;
 310 }
 311
 312 TEST_F(EnvLibradosTest, Truncate) {
 313   const size_t kWriteSize = 300 * 1024;
 314   const size_t truncSize = 1024;
 315   std::string write_data;
 316   for (size_t i = 0; i < kWriteSize; ++i) {
 317     write_data.append(1, 'h');
 318   }
 319
 320   unique_ptr<WritableFile> writable_file;
 321   ASSERT_OK(env_->CreateDir("/dir"));
 322   ASSERT_OK(env_->NewWritableFile("/dir/g", &writable_file, soptions_));
 323   ASSERT_OK(writable_file->Append(write_data));
 324   ASSERT_EQ(writable_file->GetFileSize(), kWriteSize);
 325   ASSERT_OK(writable_file->Truncate(truncSize));
 326   ASSERT_EQ(writable_file->GetFileSize(), truncSize);
 327   writable_file.reset();
 328 }
 329
 330 TEST_F(EnvLibradosTest, DBBasics) {
 331   std::string kDBPath = "/tmp/DBBasics";
 332   DB* db;
 333   Options options;
 334   // Optimize RocksDB. This is the easiest way to get RocksDB to perform well
 335   options.IncreaseParallelism();
 336   options.OptimizeLevelStyleCompaction();
 337   // create the DB if it's not already present
 338   options.create_if_missing = true;
 339   options.env = env_;
 340
 341   // open DB
 342   Status s = DB::Open(options, kDBPath, &db);
 343   assert(s.ok());
 344
 345   // Put key-value
 346   s = db->Put(WriteOptions(), "key1", "value");
 347   assert(s.ok());
 348   std::string value;
 349   // get value
 350   s = db->Get(ReadOptions(), "key1", &value);
 351   assert(s.ok());
 352   assert(value == "value");
 353
 354   // atomically apply a set of updates
 355   {
 356     WriteBatch batch;
 357     batch.Delete("key1");
 358     batch.Put("key2", value);
 359     s = db->Write(WriteOptions(), &batch);
 360   }
 361
 362   s = db->Get(ReadOptions(), "key1", &value);
 363   assert(s.IsNotFound());
 364
 365   db->Get(ReadOptions(), "key2", &value);
 366   assert(value == "value");
 367
 368   delete db;
 369 }
 370
 371 TEST_F(EnvLibradosTest, DBLoadKeysInRandomOrder) {
 372   char key[20] = {0}, value[20] = {0};
 373   int max_loop = 1 << 10;
 374   Timer timer(false);
 375   std::cout << "Test size : loop(" << max_loop << ")" << std::endl;
 376   /**********************************
 377             use default env
 378   ***********************************/
 379   std::string kDBPath1 = "/tmp/DBLoadKeysInRandomOrder1";
 380   DB* db1;
 381   Options options1;
 382   // Optimize Rocksdb. This is the easiest way to get RocksDB to perform well
 383   options1.IncreaseParallelism();
 384   options1.OptimizeLevelStyleCompaction();
 385   // create the DB if it's not already present
 386   options1.create_if_missing = true;
 387
 388   // open DB
 389   Status s1 = DB::Open(options1, kDBPath1, &db1);
 390   assert(s1.ok());
 391
 392   rocksdb::Random64 r1(time(nullptr));
 393
 394   timer.Reset();
 395   for (int i = 0; i < max_loop; ++i) {
 396     snprintf(key,
 397              20,
 398              "%16lx",
 399              (unsigned long)r1.Uniform(std::numeric_limits<uint64_t>::max()));
 400     snprintf(value,
 401              20,
 402              "%16lx",
 403              (unsigned long)r1.Uniform(std::numeric_limits<uint64_t>::max()));
 404     // Put key-value
 405     s1 = db1->Put(WriteOptions(), key, value);
 406     assert(s1.ok());
 407   }
 408   std::cout << "Time by default : " << timer << "ms" << std::endl;
 409   delete db1;
 410
 411   /**********************************
 412             use librados env
 413   ***********************************/
 414   std::string kDBPath2 = "/tmp/DBLoadKeysInRandomOrder2";
 415   DB* db2;
 416   Options options2;
 417   // Optimize RocksDB. This is the easiest way to get RocksDB to perform well
 418   options2.IncreaseParallelism();
 419   options2.OptimizeLevelStyleCompaction();
 420   // create the DB if it's not already present
 421   options2.create_if_missing = true;
 422   options2.env = env_;
 423
 424   // open DB
 425   Status s2 = DB::Open(options2, kDBPath2, &db2);
 426   assert(s2.ok());
 427
 428   rocksdb::Random64 r2(time(nullptr));
 429
 430   timer.Reset();
 431   for (int i = 0; i < max_loop; ++i) {
 432     snprintf(key,
 433              20,
 434              "%16lx",
 435              (unsigned long)r2.Uniform(std::numeric_limits<uint64_t>::max()));
 436     snprintf(value,
 437              20,
 438              "%16lx",
 439              (unsigned long)r2.Uniform(std::numeric_limits<uint64_t>::max()));
 440     // Put key-value
 441     s2 = db2->Put(WriteOptions(), key, value);
 442     assert(s2.ok());
 443   }
 444   std::cout << "Time by librados : " << timer << "ms" << std::endl;
 445   delete db2;
 446 }
 447
 448 TEST_F(EnvLibradosTest, DBBulkLoadKeysInRandomOrder) {
 449   char key[20] = {0}, value[20] = {0};
 450   int max_loop = 1 << 6;
 451   int bulk_size = 1 << 15;
 452   Timer timer(false);
 453   std::cout << "Test size : loop(" << max_loop << "); bulk_size(" << bulk_size << ")" << std::endl;
 454   /**********************************
 455             use default env
 456   ***********************************/
 457   std::string kDBPath1 = "/tmp/DBBulkLoadKeysInRandomOrder1";
 458   DB* db1;
 459   Options options1;
 460   // Optimize Rocksdb. This is the easiest way to get RocksDB to perform well
 461   options1.IncreaseParallelism();
 462   options1.OptimizeLevelStyleCompaction();
 463   // create the DB if it's not already present
 464   options1.create_if_missing = true;
 465
 466   // open DB
 467   Status s1 = DB::Open(options1, kDBPath1, &db1);
 468   assert(s1.ok());
 469
 470   rocksdb::Random64 r1(time(nullptr));
 471
 472   timer.Reset();
 473   for (int i = 0; i < max_loop; ++i) {
 474     WriteBatch batch;
 475     for (int j = 0; j < bulk_size; ++j) {
 476       snprintf(key,
 477                20,
 478                "%16lx",
 479                (unsigned long)r1.Uniform(std::numeric_limits<uint64_t>::max()));
 480       snprintf(value,
 481                20,
 482                "%16lx",
 483                (unsigned long)r1.Uniform(std::numeric_limits<uint64_t>::max()));
 484       batch.Put(key, value);
 485     }
 486     s1 = db1->Write(WriteOptions(), &batch);
 487     assert(s1.ok());
 488   }
 489   std::cout << "Time by default : " << timer << "ms" << std::endl;
 490   delete db1;
 491
 492   /**********************************
 493             use librados env
 494   ***********************************/
 495   std::string kDBPath2 = "/tmp/DBBulkLoadKeysInRandomOrder2";
 496   DB* db2;
 497   Options options2;
 498   // Optimize RocksDB. This is the easiest way to get RocksDB to perform well
 499   options2.IncreaseParallelism();
 500   options2.OptimizeLevelStyleCompaction();
 501   // create the DB if it's not already present
 502   options2.create_if_missing = true;
 503   options2.env = env_;
 504
 505   // open DB
 506   Status s2 = DB::Open(options2, kDBPath2, &db2);
 507   assert(s2.ok());
 508
 509   rocksdb::Random64 r2(time(nullptr));
 510
 511   timer.Reset();
 512   for (int i = 0; i < max_loop; ++i) {
 513     WriteBatch batch;
 514     for (int j = 0; j < bulk_size; ++j) {
 515       snprintf(key,
 516                20,
 517                "%16lx",
 518                (unsigned long)r2.Uniform(std::numeric_limits<uint64_t>::max()));
 519       snprintf(value,
 520                20,
 521                "%16lx",
 522                (unsigned long)r2.Uniform(std::numeric_limits<uint64_t>::max()));
 523       batch.Put(key, value);
 524     }
 525     s2 = db2->Write(WriteOptions(), &batch);
 526     assert(s2.ok());
 527   }
 528   std::cout << "Time by librados : " << timer << "ms" << std::endl;
 529   delete db2;
 530 }
 531
 532 TEST_F(EnvLibradosTest, DBBulkLoadKeysInSequentialOrder) {
 533   char key[20] = {0}, value[20] = {0};
 534   int max_loop = 1 << 6;
 535   int bulk_size = 1 << 15;
 536   Timer timer(false);
 537   std::cout << "Test size : loop(" << max_loop << "); bulk_size(" << bulk_size << ")" << std::endl;
 538   /**********************************
 539             use default env
 540   ***********************************/
 541   std::string kDBPath1 = "/tmp/DBBulkLoadKeysInSequentialOrder1";
 542   DB* db1;
 543   Options options1;
 544   // Optimize Rocksdb. This is the easiest way to get RocksDB to perform well
 545   options1.IncreaseParallelism();
 546   options1.OptimizeLevelStyleCompaction();
 547   // create the DB if it's not already present
 548   options1.create_if_missing = true;
 549
 550   // open DB
 551   Status s1 = DB::Open(options1, kDBPath1, &db1);
 552   assert(s1.ok());
 553
 554   rocksdb::Random64 r1(time(nullptr));
 555
 556   timer.Reset();
 557   for (int i = 0; i < max_loop; ++i) {
 558     WriteBatch batch;
 559     for (int j = 0; j < bulk_size; ++j) {
 560       snprintf(key,
 561                20,
 562                "%019lld",
 563                (long long)(i * bulk_size + j));
 564       snprintf(value,
 565                20,
 566                "%16lx",
 567                (unsigned long)r1.Uniform(std::numeric_limits<uint64_t>::max()));
 568       batch.Put(key, value);
 569     }
 570     s1 = db1->Write(WriteOptions(), &batch);
 571     assert(s1.ok());
 572   }
 573   std::cout << "Time by default : " << timer << "ms" << std::endl;
 574   delete db1;
 575
 576   /**********************************
 577             use librados env
 578   ***********************************/
 579   std::string kDBPath2 = "/tmp/DBBulkLoadKeysInSequentialOrder2";
 580   DB* db2;
 581   Options options2;
 582   // Optimize RocksDB. This is the easiest way to get RocksDB to perform well
 583   options2.IncreaseParallelism();
 584   options2.OptimizeLevelStyleCompaction();
 585   // create the DB if it's not already present
 586   options2.create_if_missing = true;
 587   options2.env = env_;
 588
 589   // open DB
 590   Status s2 = DB::Open(options2, kDBPath2, &db2);
 591   assert(s2.ok());
 592
 593   rocksdb::Random64 r2(time(nullptr));
 594
 595   timer.Reset();
 596   for (int i = 0; i < max_loop; ++i) {
 597     WriteBatch batch;
 598     for (int j = 0; j < bulk_size; ++j) {
 599       snprintf(key,
 600                20,
 601                "%16lx",
 602                (unsigned long)r2.Uniform(std::numeric_limits<uint64_t>::max()));
 603       snprintf(value,
 604                20,
 605                "%16lx",
 606                (unsigned long)r2.Uniform(std::numeric_limits<uint64_t>::max()));
 607       batch.Put(key, value);
 608     }
 609     s2 = db2->Write(WriteOptions(), &batch);
 610     assert(s2.ok());
 611   }
 612   std::cout << "Time by librados : " << timer << "ms" << std::endl;
 613   delete db2;
 614 }
 615
 616 TEST_F(EnvLibradosTest, DBRandomRead) {
 617   char key[20] = {0}, value[20] = {0};
 618   int max_loop = 1 << 6;
 619   int bulk_size = 1 << 10;
 620   int read_loop = 1 << 20;
 621   Timer timer(false);
 622   std::cout << "Test size : keys_num(" << max_loop << ", " << bulk_size << "); read_loop(" << read_loop << ")" << std::endl;
 623   /**********************************
 624             use default env
 625   ***********************************/
 626   std::string kDBPath1 = "/tmp/DBRandomRead1";
 627   DB* db1;
 628   Options options1;
 629   // Optimize Rocksdb. This is the easiest way to get RocksDB to perform well
 630   options1.IncreaseParallelism();
 631   options1.OptimizeLevelStyleCompaction();
 632   // create the DB if it's not already present
 633   options1.create_if_missing = true;
 634
 635   // open DB
 636   Status s1 = DB::Open(options1, kDBPath1, &db1);
 637   assert(s1.ok());
 638
 639   rocksdb::Random64 r1(time(nullptr));
 640
 641
 642   for (int i = 0; i < max_loop; ++i) {
 643     WriteBatch batch;
 644     for (int j = 0; j < bulk_size; ++j) {
 645       snprintf(key,
 646                20,
 647                "%019lld",
 648                (long long)(i * bulk_size + j));
 649       snprintf(value,
 650                20,
 651                "%16lx",
 652                (unsigned long)r1.Uniform(std::numeric_limits<uint64_t>::max()));
 653       batch.Put(key, value);
 654     }
 655     s1 = db1->Write(WriteOptions(), &batch);
 656     assert(s1.ok());
 657   }
 658   timer.Reset();
 659   int base1 = 0, offset1 = 0;
 660   for (int i = 0; i < read_loop; ++i) {
 661     base1 = r1.Uniform(max_loop);
 662     offset1 = r1.Uniform(bulk_size);
 663     std::string value1;
 664     snprintf(key,
 665              20,
 666              "%019lld",
 667              (long long)(base1 * bulk_size + offset1));
 668     s1 = db1->Get(ReadOptions(), key, &value1);
 669     assert(s1.ok());
 670   }
 671   std::cout << "Time by default : " << timer << "ms" << std::endl;
 672   delete db1;
 673
 674   /**********************************
 675             use librados env
 676   ***********************************/
 677   std::string kDBPath2 = "/tmp/DBRandomRead2";
 678   DB* db2;
 679   Options options2;
 680   // Optimize RocksDB. This is the easiest way to get RocksDB to perform well
 681   options2.IncreaseParallelism();
 682   options2.OptimizeLevelStyleCompaction();
 683   // create the DB if it's not already present
 684   options2.create_if_missing = true;
 685   options2.env = env_;
 686
 687   // open DB
 688   Status s2 = DB::Open(options2, kDBPath2, &db2);
 689   assert(s2.ok());
 690
 691   rocksdb::Random64 r2(time(nullptr));
 692
 693   for (int i = 0; i < max_loop; ++i) {
 694     WriteBatch batch;
 695     for (int j = 0; j < bulk_size; ++j) {
 696       snprintf(key,
 697                20,
 698                "%019lld",
 699                (long long)(i * bulk_size + j));
 700       snprintf(value,
 701                20,
 702                "%16lx",
 703                (unsigned long)r2.Uniform(std::numeric_limits<uint64_t>::max()));
 704       batch.Put(key, value);
 705     }
 706     s2 = db2->Write(WriteOptions(), &batch);
 707     assert(s2.ok());
 708   }
 709
 710   timer.Reset();
 711   int base2 = 0, offset2 = 0;
 712   for (int i = 0; i < read_loop; ++i) {
 713     base2 = r2.Uniform(max_loop);
 714     offset2 = r2.Uniform(bulk_size);
 715     std::string value2;
 716     snprintf(key,
 717              20,
 718              "%019lld",
 719              (long long)(base2 * bulk_size + offset2));
 720     s2 = db2->Get(ReadOptions(), key, &value2);
 721     if (!s2.ok()) {
 722       std::cout << s2.ToString() << std::endl;
 723     }
 724     assert(s2.ok());
 725   }
 726   std::cout << "Time by librados : " << timer << "ms" << std::endl;
 727   delete db2;
 728 }
 729
 730 class EnvLibradosMutipoolTest : public testing::Test {
 731 public:
 732   // we will use all of these below
 733   const std::string client_name = "client.admin";
 734   const std::string cluster_name = "ceph";
 735   const uint64_t flags = 0;
 736   const std::string db_name = "env_librados_test_db";
 737   const std::string db_pool = db_name + "_pool";
 738   const std::string wal_dir = "/wal";
 739   const std::string wal_pool = db_name + "_wal_pool";
 740   const size_t write_buffer_size = 1 << 20;
 741   const char *keyring = "admin";
 742   const char *config = "../ceph/src/ceph.conf";
 743
 744   EnvLibrados* env_;
 745   const EnvOptions soptions_;
 746
 747   EnvLibradosMutipoolTest() {
 748     env_ = new EnvLibrados(client_name, cluster_name, flags, db_name, config, db_pool, wal_dir, wal_pool, write_buffer_size);
 749   }
 750   ~EnvLibradosMutipoolTest() {
 751     delete env_;
 752     librados::Rados rados;
 753     int ret = 0;
 754     do {
 755       ret = rados.init("admin"); // just use the client.admin keyring
 756       if (ret < 0) { // let's handle any error that might have come back
 757         std::cerr << "couldn't initialize rados! error " << ret << std::endl;
 758         ret = EXIT_FAILURE;
 759         break;
 760       }
 761
 762       ret = rados.conf_read_file(config);
 763       if (ret < 0) {
 764         // This could fail if the config file is malformed, but it'd be hard.
 765         std::cerr << "failed to parse config file " << config
 766                   << "! error" << ret << std::endl;
 767         ret = EXIT_FAILURE;
 768         break;
 769       }
 770
 771       /*
 772        * next, we actually connect to the cluster
 773        */
 774
 775       ret = rados.connect();
 776       if (ret < 0) {
 777         std::cerr << "couldn't connect to cluster! error " << ret << std::endl;
 778         ret = EXIT_FAILURE;
 779         break;
 780       }
 781
 782       /*
 783        * And now we're done, so let's remove our pool and then
 784        * shut down the connection gracefully.
 785        */
 786       int delete_ret = rados.pool_delete(db_pool.c_str());
 787       if (delete_ret < 0) {
 788         // be careful not to
 789         std::cerr << "We failed to delete our test pool!" << db_pool << delete_ret << std::endl;
 790         ret = EXIT_FAILURE;
 791       }
 792       delete_ret = rados.pool_delete(wal_pool.c_str());
 793       if (delete_ret < 0) {
 794         // be careful not to
 795         std::cerr << "We failed to delete our test pool!" << wal_pool << delete_ret << std::endl;
 796         ret = EXIT_FAILURE;
 797       }
 798     } while (0);
 799   }
 800 };
 801
 802 TEST_F(EnvLibradosMutipoolTest, Basics) {
 803   uint64_t file_size;
 804   unique_ptr<WritableFile> writable_file;
 805   std::vector<std::string> children;
 806   std::vector<std::string> v = {"/tmp/dir1", "/tmp/dir2", "/tmp/dir3", "/tmp/dir4", "dir"};
 807
 808   for (size_t i = 0; i < v.size(); ++i) {
 809     std::string dir = v[i];
 810     std::string dir_non_existent = dir + "/non_existent";
 811     std::string dir_f = dir + "/f";
 812     std::string dir_g = dir + "/g";
 813
 814     ASSERT_OK(env_->CreateDir(dir.c_str()));
 815     // Check that the directory is empty.
 816     ASSERT_EQ(Status::NotFound(), env_->FileExists(dir_non_existent.c_str()));
 817     ASSERT_TRUE(!env_->GetFileSize(dir_non_existent.c_str(), &file_size).ok());
 818     ASSERT_OK(env_->GetChildren(dir.c_str(), &children));
 819     ASSERT_EQ(0U, children.size());
 820
 821     // Create a file.
 822     ASSERT_OK(env_->NewWritableFile(dir_f.c_str(), &writable_file, soptions_));
 823     writable_file.reset();
 824
 825     // Check that the file exists.
 826     ASSERT_OK(env_->FileExists(dir_f.c_str()));
 827     ASSERT_OK(env_->GetFileSize(dir_f.c_str(), &file_size));
 828     ASSERT_EQ(0U, file_size);
 829     ASSERT_OK(env_->GetChildren(dir.c_str(), &children));
 830     ASSERT_EQ(1U, children.size());
 831     ASSERT_EQ("f", children[0]);
 832
 833     // Write to the file.
 834     ASSERT_OK(env_->NewWritableFile(dir_f.c_str(), &writable_file, soptions_));
 835     ASSERT_OK(writable_file->Append("abc"));
 836     writable_file.reset();
 837
 838
 839     // Check for expected size.
 840     ASSERT_OK(env_->GetFileSize(dir_f.c_str(), &file_size));
 841     ASSERT_EQ(3U, file_size);
 842
 843
 844     // Check that renaming works.
 845     ASSERT_TRUE(!env_->RenameFile(dir_non_existent.c_str(), dir_g.c_str()).ok());
 846     ASSERT_OK(env_->RenameFile(dir_f.c_str(), dir_g.c_str()));
 847     ASSERT_EQ(Status::NotFound(), env_->FileExists(dir_f.c_str()));
 848     ASSERT_OK(env_->FileExists(dir_g.c_str()));
 849     ASSERT_OK(env_->GetFileSize(dir_g.c_str(), &file_size));
 850     ASSERT_EQ(3U, file_size);
 851
 852     // Check that opening non-existent file fails.
 853     unique_ptr<SequentialFile> seq_file;
 854     unique_ptr<RandomAccessFile> rand_file;
 855     ASSERT_TRUE(
 856       !env_->NewSequentialFile(dir_non_existent.c_str(), &seq_file, soptions_).ok());
 857     ASSERT_TRUE(!seq_file);
 858     ASSERT_TRUE(!env_->NewRandomAccessFile(dir_non_existent.c_str(), &rand_file,
 859                                            soptions_).ok());
 860     ASSERT_TRUE(!rand_file);
 861
 862     // Check that deleting works.
 863     ASSERT_TRUE(!env_->DeleteFile(dir_non_existent.c_str()).ok());
 864     ASSERT_OK(env_->DeleteFile(dir_g.c_str()));
 865     ASSERT_EQ(Status::NotFound(), env_->FileExists(dir_g.c_str()));
 866     ASSERT_OK(env_->GetChildren(dir.c_str(), &children));
 867     ASSERT_EQ(0U, children.size());
 868     ASSERT_OK(env_->DeleteDir(dir.c_str()));
 869   }
 870 }
 871
 872 TEST_F(EnvLibradosMutipoolTest, DBBasics) {
 873   std::string kDBPath = "/tmp/DBBasics";
 874   std::string walPath = "/tmp/wal";
 875   DB* db;
 876   Options options;
 877   // Optimize RocksDB. This is the easiest way to get RocksDB to perform well
 878   options.IncreaseParallelism();
 879   options.OptimizeLevelStyleCompaction();
 880   // create the DB if it's not already present
 881   options.create_if_missing = true;
 882   options.env = env_;
 883   options.wal_dir = walPath;
 884
 885   // open DB
 886   Status s = DB::Open(options, kDBPath, &db);
 887   assert(s.ok());
 888
 889   // Put key-value
 890   s = db->Put(WriteOptions(), "key1", "value");
 891   assert(s.ok());
 892   std::string value;
 893   // get value
 894   s = db->Get(ReadOptions(), "key1", &value);
 895   assert(s.ok());
 896   assert(value == "value");
 897
 898   // atomically apply a set of updates
 899   {
 900     WriteBatch batch;
 901     batch.Delete("key1");
 902     batch.Put("key2", value);
 903     s = db->Write(WriteOptions(), &batch);
 904   }
 905
 906   s = db->Get(ReadOptions(), "key1", &value);
 907   assert(s.IsNotFound());
 908
 909   db->Get(ReadOptions(), "key2", &value);
 910   assert(value == "value");
 911
 912   delete db;
 913 }
 914
 915 TEST_F(EnvLibradosMutipoolTest, DBBulkLoadKeysInRandomOrder) {
 916   char key[20] = {0}, value[20] = {0};
 917   int max_loop = 1 << 6;
 918   int bulk_size = 1 << 15;
 919   Timer timer(false);
 920   std::cout << "Test size : loop(" << max_loop << "); bulk_size(" << bulk_size << ")" << std::endl;
 921   /**********************************
 922             use default env
 923   ***********************************/
 924   std::string kDBPath1 = "/tmp/DBBulkLoadKeysInRandomOrder1";
 925   std::string walPath = "/tmp/wal";
 926   DB* db1;
 927   Options options1;
 928   // Optimize Rocksdb. This is the easiest way to get RocksDB to perform well
 929   options1.IncreaseParallelism();
 930   options1.OptimizeLevelStyleCompaction();
 931   // create the DB if it's not already present
 932   options1.create_if_missing = true;
 933
 934   // open DB
 935   Status s1 = DB::Open(options1, kDBPath1, &db1);
 936   assert(s1.ok());
 937
 938   rocksdb::Random64 r1(time(nullptr));
 939
 940   timer.Reset();
 941   for (int i = 0; i < max_loop; ++i) {
 942     WriteBatch batch;
 943     for (int j = 0; j < bulk_size; ++j) {
 944       snprintf(key,
 945                20,
 946                "%16lx",
 947                (unsigned long)r1.Uniform(std::numeric_limits<uint64_t>::max()));
 948       snprintf(value,
 949                20,
 950                "%16lx",
 951                (unsigned long)r1.Uniform(std::numeric_limits<uint64_t>::max()));
 952       batch.Put(key, value);
 953     }
 954     s1 = db1->Write(WriteOptions(), &batch);
 955     assert(s1.ok());
 956   }
 957   std::cout << "Time by default : " << timer << "ms" << std::endl;
 958   delete db1;
 959
 960   /**********************************
 961             use librados env
 962   ***********************************/
 963   std::string kDBPath2 = "/tmp/DBBulkLoadKeysInRandomOrder2";
 964   DB* db2;
 965   Options options2;
 966   // Optimize RocksDB. This is the easiest way to get RocksDB to perform well
 967   options2.IncreaseParallelism();
 968   options2.OptimizeLevelStyleCompaction();
 969   // create the DB if it's not already present
 970   options2.create_if_missing = true;
 971   options2.env = env_;
 972   options2.wal_dir = walPath;
 973
 974   // open DB
 975   Status s2 = DB::Open(options2, kDBPath2, &db2);
 976   if (!s2.ok()) {
 977     std::cerr << s2.ToString() << std::endl;
 978   }
 979   assert(s2.ok());
 980
 981   rocksdb::Random64 r2(time(nullptr));
 982
 983   timer.Reset();
 984   for (int i = 0; i < max_loop; ++i) {
 985     WriteBatch batch;
 986     for (int j = 0; j < bulk_size; ++j) {
 987       snprintf(key,
 988                20,
 989                "%16lx",
 990                (unsigned long)r2.Uniform(std::numeric_limits<uint64_t>::max()));
 991       snprintf(value,
 992                20,
 993                "%16lx",
 994                (unsigned long)r2.Uniform(std::numeric_limits<uint64_t>::max()));
 995       batch.Put(key, value);
 996     }
 997     s2 = db2->Write(WriteOptions(), &batch);
 998     assert(s2.ok());
 999   }
1000   std::cout << "Time by librados : " << timer << "ms" << std::endl;
1001   delete db2;
1002 }
1003
1004 TEST_F(EnvLibradosMutipoolTest, DBTransactionDB) {
1005   std::string kDBPath = "/tmp/DBTransactionDB";
1006   // open DB
1007   Options options;
1008   TransactionDBOptions txn_db_options;
1009   options.create_if_missing = true;
1010   options.env = env_;
1011   TransactionDB* txn_db;
1012
1013   Status s = TransactionDB::Open(options, txn_db_options, kDBPath, &txn_db);
1014   assert(s.ok());
1015
1016   WriteOptions write_options;
1017   ReadOptions read_options;
1018   TransactionOptions txn_options;
1019   std::string value;
1020
1021   ////////////////////////////////////////////////////////
1022   //
1023   // Simple OptimisticTransaction Example ("Read Committed")
1024   //
1025   ////////////////////////////////////////////////////////
1026
1027   // Start a transaction
1028   Transaction* txn = txn_db->BeginTransaction(write_options);
1029   assert(txn);
1030
1031   // Read a key in this transaction
1032   s = txn->Get(read_options, "abc", &value);
1033   assert(s.IsNotFound());
1034
1035   // Write a key in this transaction
1036   s = txn->Put("abc", "def");
1037   assert(s.ok());
1038
1039   // Read a key OUTSIDE this transaction. Does not affect txn.
1040   s = txn_db->Get(read_options, "abc", &value);
1041
1042   // Write a key OUTSIDE of this transaction.
1043   // Does not affect txn since this is an unrelated key.  If we wrote key 'abc'
1044   // here, the transaction would fail to commit.
1045   s = txn_db->Put(write_options, "xyz", "zzz");
1046
1047   // Commit transaction
1048   s = txn->Commit();
1049   assert(s.ok());
1050   delete txn;
1051
1052   ////////////////////////////////////////////////////////
1053   //
1054   // "Repeatable Read" (Snapshot Isolation) Example
1055   //   -- Using a single Snapshot
1056   //
1057   ////////////////////////////////////////////////////////
1058
1059   // Set a snapshot at start of transaction by setting set_snapshot=true
1060   txn_options.set_snapshot = true;
1061   txn = txn_db->BeginTransaction(write_options, txn_options);
1062
1063   const Snapshot* snapshot = txn->GetSnapshot();
1064
1065   // Write a key OUTSIDE of transaction
1066   s = txn_db->Put(write_options, "abc", "xyz");
1067   assert(s.ok());
1068
1069   // Attempt to read a key using the snapshot.  This will fail since
1070   // the previous write outside this txn conflicts with this read.
1071   read_options.snapshot = snapshot;
1072   s = txn->GetForUpdate(read_options, "abc", &value);
1073   assert(s.IsBusy());
1074
1075   txn->Rollback();
1076
1077   delete txn;
1078   // Clear snapshot from read options since it is no longer valid
1079   read_options.snapshot = nullptr;
1080   snapshot = nullptr;
1081
1082   ////////////////////////////////////////////////////////
1083   //
1084   // "Read Committed" (Monotonic Atomic Views) Example
1085   //   --Using multiple Snapshots
1086   //
1087   ////////////////////////////////////////////////////////
1088
1089   // In this example, we set the snapshot multiple times.  This is probably
1090   // only necessary if you have very strict isolation requirements to
1091   // implement.
1092
1093   // Set a snapshot at start of transaction
1094   txn_options.set_snapshot = true;
1095   txn = txn_db->BeginTransaction(write_options, txn_options);
1096
1097   // Do some reads and writes to key "x"
1098   read_options.snapshot = txn_db->GetSnapshot();
1099   s = txn->Get(read_options, "x", &value);
1100   txn->Put("x", "x");
1101
1102   // Do a write outside of the transaction to key "y"
1103   s = txn_db->Put(write_options, "y", "y");
1104
1105   // Set a new snapshot in the transaction
1106   txn->SetSnapshot();
1107   txn->SetSavePoint();
1108   read_options.snapshot = txn_db->GetSnapshot();
1109
1110   // Do some reads and writes to key "y"
1111   // Since the snapshot was advanced, the write done outside of the
1112   // transaction does not conflict.
1113   s = txn->GetForUpdate(read_options, "y", &value);
1114   txn->Put("y", "y");
1115
1116   // Decide we want to revert the last write from this transaction.
1117   txn->RollbackToSavePoint();
1118
1119   // Commit.
1120   s = txn->Commit();
1121   assert(s.ok());
1122   delete txn;
1123   // Clear snapshot from read options since it is no longer valid
1124   read_options.snapshot = nullptr;
1125
1126   // Cleanup
1127   delete txn_db;
1128   DestroyDB(kDBPath, options);
1129 }
1130
1131 }  // namespace rocksdb
1132
1133 int main(int argc, char** argv) {
1134   ::testing::InitGoogleTest(&argc, argv);
1135   return RUN_ALL_TESTS();
1136 }
1137
1138 #else
1139 #include <stdio.h>
1140
1141 int main(int argc, char** argv) {
1142   fprintf(stderr, "SKIPPED as EnvMirror is not supported in ROCKSDB_LITE\n");
1143   return 0;
1144 }
1145
1146 #endif  // !ROCKSDB_LITE