ceph/src/rocksdb/utilities/env_librados_test.cc

   1 // Copyright (c) Facebook, Inc. and its affiliates. All Rights Reserved.
   2 //  Copyright (c) 2016, Red Hat, Inc.  All rights reserved.
   3 //  This source code is licensed under both the GPLv2 (found in the
   4 //  COPYING file in the root directory) and Apache 2.0 License
   5 //  (found in the LICENSE.Apache file in the root directory).
   6
   7 #ifndef ROCKSDB_LITE
   8
   9 #include "rocksdb/utilities/env_librados.h"
  10 #include <rados/librados.hpp>
  11 #include "env/mock_env.h"
  12 #include "test_util/testharness.h"
  13
  14 #include "rocksdb/db.h"
  15 #include "rocksdb/slice.h"
  16 #include "rocksdb/options.h"
  17 #include "util/random.h"
  18 #include <chrono>
  19 #include <ostream>
  20 #include "rocksdb/utilities/transaction_db.h"
  21
  22 class Timer {
  23   typedef std::chrono::high_resolution_clock high_resolution_clock;
  24   typedef std::chrono::milliseconds milliseconds;
  25 public:
  26   explicit Timer(bool run = false)
  27   {
  28     if (run)
  29       Reset();
  30   }
  31   void Reset()
  32   {
  33     _start = high_resolution_clock::now();
  34   }
  35   milliseconds Elapsed() const
  36   {
  37     return std::chrono::duration_cast<milliseconds>(high_resolution_clock::now() - _start);
  38   }
  39   template <typename T, typename Traits>
  40   friend std::basic_ostream<T, Traits>& operator<<(std::basic_ostream<T, Traits>& out, const Timer& timer)
  41   {
  42     return out << timer.Elapsed().count();
  43   }
  44 private:
  45   high_resolution_clock::time_point _start;
  46 };
  47
  48 namespace ROCKSDB_NAMESPACE {
  49
  50 class EnvLibradosTest : public testing::Test {
  51 public:
  52   // we will use all of these below
  53   const std::string db_name = "env_librados_test_db";
  54   const std::string db_pool = db_name + "_pool";
  55   const char *keyring = "admin";
  56   const char *config = "../ceph/src/ceph.conf";
  57
  58   EnvLibrados* env_;
  59   const EnvOptions soptions_;
  60
  61   EnvLibradosTest()
  62     : env_(new EnvLibrados(db_name, config, db_pool)) {
  63   }
  64   ~EnvLibradosTest() {
  65     delete env_;
  66     librados::Rados rados;
  67     int ret = 0;
  68     do {
  69       ret = rados.init("admin"); // just use the client.admin keyring
  70       if (ret < 0) { // let's handle any error that might have come back
  71         std::cerr << "couldn't initialize rados! error " << ret << std::endl;
  72         ret = EXIT_FAILURE;
  73         break;
  74       }
  75
  76       ret = rados.conf_read_file(config);
  77       if (ret < 0) {
  78         // This could fail if the config file is malformed, but it'd be hard.
  79         std::cerr << "failed to parse config file " << config
  80                   << "! error" << ret << std::endl;
  81         ret = EXIT_FAILURE;
  82         break;
  83       }
  84
  85       /*
  86        * next, we actually connect to the cluster
  87        */
  88
  89       ret = rados.connect();
  90       if (ret < 0) {
  91         std::cerr << "couldn't connect to cluster! error " << ret << std::endl;
  92         ret = EXIT_FAILURE;
  93         break;
  94       }
  95
  96       /*
  97        * And now we're done, so let's remove our pool and then
  98        * shut down the connection gracefully.
  99        */
 100       int delete_ret = rados.pool_delete(db_pool.c_str());
 101       if (delete_ret < 0) {
 102         // be careful not to
 103         std::cerr << "We failed to delete our test pool!" << db_pool << delete_ret << std::endl;
 104         ret = EXIT_FAILURE;
 105       }
 106     } while (0);
 107   }
 108 };
 109
 110 TEST_F(EnvLibradosTest, Basics) {
 111   uint64_t file_size;
 112   std::unique_ptr<WritableFile> writable_file;
 113   std::vector<std::string> children;
 114
 115   ASSERT_OK(env_->CreateDir("/dir"));
 116   // Check that the directory is empty.
 117   ASSERT_EQ(Status::NotFound(), env_->FileExists("/dir/non_existent"));
 118   ASSERT_TRUE(!env_->GetFileSize("/dir/non_existent", &file_size).ok());
 119   ASSERT_OK(env_->GetChildren("/dir", &children));
 120   ASSERT_EQ(0U, children.size());
 121
 122   // Create a file.
 123   ASSERT_OK(env_->NewWritableFile("/dir/f", &writable_file, soptions_));
 124   writable_file.reset();
 125
 126   // Check that the file exists.
 127   ASSERT_OK(env_->FileExists("/dir/f"));
 128   ASSERT_OK(env_->GetFileSize("/dir/f", &file_size));
 129   ASSERT_EQ(0U, file_size);
 130   ASSERT_OK(env_->GetChildren("/dir", &children));
 131   ASSERT_EQ(1U, children.size());
 132   ASSERT_EQ("f", children[0]);
 133
 134   // Write to the file.
 135   ASSERT_OK(env_->NewWritableFile("/dir/f", &writable_file, soptions_));
 136   ASSERT_OK(writable_file->Append("abc"));
 137   writable_file.reset();
 138
 139
 140   // Check for expected size.
 141   ASSERT_OK(env_->GetFileSize("/dir/f", &file_size));
 142   ASSERT_EQ(3U, file_size);
 143
 144
 145   // Check that renaming works.
 146   ASSERT_TRUE(!env_->RenameFile("/dir/non_existent", "/dir/g").ok());
 147   ASSERT_OK(env_->RenameFile("/dir/f", "/dir/g"));
 148   ASSERT_EQ(Status::NotFound(), env_->FileExists("/dir/f"));
 149   ASSERT_OK(env_->FileExists("/dir/g"));
 150   ASSERT_OK(env_->GetFileSize("/dir/g", &file_size));
 151   ASSERT_EQ(3U, file_size);
 152
 153   // Check that opening non-existent file fails.
 154   std::unique_ptr<SequentialFile> seq_file;
 155   std::unique_ptr<RandomAccessFile> rand_file;
 156   ASSERT_TRUE(
 157     !env_->NewSequentialFile("/dir/non_existent", &seq_file, soptions_).ok());
 158   ASSERT_TRUE(!seq_file);
 159   ASSERT_TRUE(!env_->NewRandomAccessFile("/dir/non_existent", &rand_file,
 160                                          soptions_).ok());
 161   ASSERT_TRUE(!rand_file);
 162
 163   // Check that deleting works.
 164   ASSERT_TRUE(!env_->DeleteFile("/dir/non_existent").ok());
 165   ASSERT_OK(env_->DeleteFile("/dir/g"));
 166   ASSERT_EQ(Status::NotFound(), env_->FileExists("/dir/g"));
 167   ASSERT_OK(env_->GetChildren("/dir", &children));
 168   ASSERT_EQ(0U, children.size());
 169   ASSERT_OK(env_->DeleteDir("/dir"));
 170 }
 171
 172 TEST_F(EnvLibradosTest, ReadWrite) {
 173   std::unique_ptr<WritableFile> writable_file;
 174   std::unique_ptr<SequentialFile> seq_file;
 175   std::unique_ptr<RandomAccessFile> rand_file;
 176   Slice result;
 177   char scratch[100];
 178
 179   ASSERT_OK(env_->CreateDir("/dir"));
 180
 181   ASSERT_OK(env_->NewWritableFile("/dir/f", &writable_file, soptions_));
 182   ASSERT_OK(writable_file->Append("hello "));
 183   ASSERT_OK(writable_file->Append("world"));
 184   writable_file.reset();
 185
 186   // Read sequentially.
 187   ASSERT_OK(env_->NewSequentialFile("/dir/f", &seq_file, soptions_));
 188   ASSERT_OK(seq_file->Read(5, &result, scratch));  // Read "hello".
 189   ASSERT_EQ(0, result.compare("hello"));
 190   ASSERT_OK(seq_file->Skip(1));
 191   ASSERT_OK(seq_file->Read(1000, &result, scratch));  // Read "world".
 192   ASSERT_EQ(0, result.compare("world"));
 193   ASSERT_OK(seq_file->Read(1000, &result, scratch));  // Try reading past EOF.
 194   ASSERT_EQ(0U, result.size());
 195   ASSERT_OK(seq_file->Skip(100));  // Try to skip past end of file.
 196   ASSERT_OK(seq_file->Read(1000, &result, scratch));
 197   ASSERT_EQ(0U, result.size());
 198
 199   // Random reads.
 200   ASSERT_OK(env_->NewRandomAccessFile("/dir/f", &rand_file, soptions_));
 201   ASSERT_OK(rand_file->Read(6, 5, &result, scratch));  // Read "world".
 202   ASSERT_EQ(0, result.compare("world"));
 203   ASSERT_OK(rand_file->Read(0, 5, &result, scratch));  // Read "hello".
 204   ASSERT_EQ(0, result.compare("hello"));
 205   ASSERT_OK(rand_file->Read(10, 100, &result, scratch));  // Read "d".
 206   ASSERT_EQ(0, result.compare("d"));
 207
 208   // Too high offset.
 209   ASSERT_OK(rand_file->Read(1000, 5, &result, scratch));
 210 }
 211
 212 TEST_F(EnvLibradosTest, Locks) {
 213   FileLock* lock = nullptr;
 214   std::unique_ptr<WritableFile> writable_file;
 215
 216   ASSERT_OK(env_->CreateDir("/dir"));
 217
 218   ASSERT_OK(env_->NewWritableFile("/dir/f", &writable_file, soptions_));
 219
 220   // These are no-ops, but we test they return success.
 221   ASSERT_OK(env_->LockFile("some file", &lock));
 222   ASSERT_OK(env_->UnlockFile(lock));
 223
 224   ASSERT_OK(env_->LockFile("/dir/f", &lock));
 225   ASSERT_OK(env_->UnlockFile(lock));
 226 }
 227
 228 TEST_F(EnvLibradosTest, Misc) {
 229   std::string test_dir;
 230   ASSERT_OK(env_->GetTestDirectory(&test_dir));
 231   ASSERT_TRUE(!test_dir.empty());
 232
 233   std::unique_ptr<WritableFile> writable_file;
 234   ASSERT_TRUE(!env_->NewWritableFile("/a/b", &writable_file, soptions_).ok());
 235
 236   ASSERT_OK(env_->NewWritableFile("/a", &writable_file, soptions_));
 237   // These are no-ops, but we test they return success.
 238   ASSERT_OK(writable_file->Sync());
 239   ASSERT_OK(writable_file->Flush());
 240   ASSERT_OK(writable_file->Close());
 241   writable_file.reset();
 242 }
 243
 244 TEST_F(EnvLibradosTest, LargeWrite) {
 245   const size_t kWriteSize = 300 * 1024;
 246   char* scratch = new char[kWriteSize * 2];
 247
 248   std::string write_data;
 249   for (size_t i = 0; i < kWriteSize; ++i) {
 250     write_data.append(1, 'h');
 251   }
 252
 253   std::unique_ptr<WritableFile> writable_file;
 254   ASSERT_OK(env_->CreateDir("/dir"));
 255   ASSERT_OK(env_->NewWritableFile("/dir/g", &writable_file, soptions_));
 256   ASSERT_OK(writable_file->Append("foo"));
 257   ASSERT_OK(writable_file->Append(write_data));
 258   writable_file.reset();
 259
 260   std::unique_ptr<SequentialFile> seq_file;
 261   Slice result;
 262   ASSERT_OK(env_->NewSequentialFile("/dir/g", &seq_file, soptions_));
 263   ASSERT_OK(seq_file->Read(3, &result, scratch));  // Read "foo".
 264   ASSERT_EQ(0, result.compare("foo"));
 265
 266   size_t read = 0;
 267   std::string read_data;
 268   while (read < kWriteSize) {
 269     ASSERT_OK(seq_file->Read(kWriteSize - read, &result, scratch));
 270     read_data.append(result.data(), result.size());
 271     read += result.size();
 272   }
 273   ASSERT_TRUE(write_data == read_data);
 274   delete[] scratch;
 275 }
 276
 277 TEST_F(EnvLibradosTest, FrequentlySmallWrite) {
 278   const size_t kWriteSize = 1 << 10;
 279   char* scratch = new char[kWriteSize * 2];
 280
 281   std::string write_data;
 282   for (size_t i = 0; i < kWriteSize; ++i) {
 283     write_data.append(1, 'h');
 284   }
 285
 286   std::unique_ptr<WritableFile> writable_file;
 287   ASSERT_OK(env_->CreateDir("/dir"));
 288   ASSERT_OK(env_->NewWritableFile("/dir/g", &writable_file, soptions_));
 289   ASSERT_OK(writable_file->Append("foo"));
 290
 291   for (size_t i = 0; i < kWriteSize; ++i) {
 292     ASSERT_OK(writable_file->Append("h"));
 293   }
 294   writable_file.reset();
 295
 296   std::unique_ptr<SequentialFile> seq_file;
 297   Slice result;
 298   ASSERT_OK(env_->NewSequentialFile("/dir/g", &seq_file, soptions_));
 299   ASSERT_OK(seq_file->Read(3, &result, scratch));  // Read "foo".
 300   ASSERT_EQ(0, result.compare("foo"));
 301
 302   size_t read = 0;
 303   std::string read_data;
 304   while (read < kWriteSize) {
 305     ASSERT_OK(seq_file->Read(kWriteSize - read, &result, scratch));
 306     read_data.append(result.data(), result.size());
 307     read += result.size();
 308   }
 309   ASSERT_TRUE(write_data == read_data);
 310   delete[] scratch;
 311 }
 312
 313 TEST_F(EnvLibradosTest, Truncate) {
 314   const size_t kWriteSize = 300 * 1024;
 315   const size_t truncSize = 1024;
 316   std::string write_data;
 317   for (size_t i = 0; i < kWriteSize; ++i) {
 318     write_data.append(1, 'h');
 319   }
 320
 321   std::unique_ptr<WritableFile> writable_file;
 322   ASSERT_OK(env_->CreateDir("/dir"));
 323   ASSERT_OK(env_->NewWritableFile("/dir/g", &writable_file, soptions_));
 324   ASSERT_OK(writable_file->Append(write_data));
 325   ASSERT_EQ(writable_file->GetFileSize(), kWriteSize);
 326   ASSERT_OK(writable_file->Truncate(truncSize));
 327   ASSERT_EQ(writable_file->GetFileSize(), truncSize);
 328   writable_file.reset();
 329 }
 330
 331 TEST_F(EnvLibradosTest, DBBasics) {
 332   std::string kDBPath = "/tmp/DBBasics";
 333   DB* db;
 334   Options options;
 335   // Optimize RocksDB. This is the easiest way to get RocksDB to perform well
 336   options.IncreaseParallelism();
 337   options.OptimizeLevelStyleCompaction();
 338   // create the DB if it's not already present
 339   options.create_if_missing = true;
 340   options.env = env_;
 341
 342   // open DB
 343   Status s = DB::Open(options, kDBPath, &db);
 344   assert(s.ok());
 345
 346   // Put key-value
 347   s = db->Put(WriteOptions(), "key1", "value");
 348   assert(s.ok());
 349   std::string value;
 350   // get value
 351   s = db->Get(ReadOptions(), "key1", &value);
 352   assert(s.ok());
 353   assert(value == "value");
 354
 355   // atomically apply a set of updates
 356   {
 357     WriteBatch batch;
 358     batch.Delete("key1");
 359     batch.Put("key2", value);
 360     s = db->Write(WriteOptions(), &batch);
 361   }
 362
 363   s = db->Get(ReadOptions(), "key1", &value);
 364   assert(s.IsNotFound());
 365
 366   db->Get(ReadOptions(), "key2", &value);
 367   assert(value == "value");
 368
 369   delete db;
 370 }
 371
 372 TEST_F(EnvLibradosTest, DBLoadKeysInRandomOrder) {
 373   char key[20] = {0}, value[20] = {0};
 374   int max_loop = 1 << 10;
 375   Timer timer(false);
 376   std::cout << "Test size : loop(" << max_loop << ")" << std::endl;
 377   /**********************************
 378             use default env
 379   ***********************************/
 380   std::string kDBPath1 = "/tmp/DBLoadKeysInRandomOrder1";
 381   DB* db1;
 382   Options options1;
 383   // Optimize Rocksdb. This is the easiest way to get RocksDB to perform well
 384   options1.IncreaseParallelism();
 385   options1.OptimizeLevelStyleCompaction();
 386   // create the DB if it's not already present
 387   options1.create_if_missing = true;
 388
 389   // open DB
 390   Status s1 = DB::Open(options1, kDBPath1, &db1);
 391   assert(s1.ok());
 392
 393   ROCKSDB_NAMESPACE::Random64 r1(time(nullptr));
 394
 395   timer.Reset();
 396   for (int i = 0; i < max_loop; ++i) {
 397     snprintf(key,
 398              20,
 399              "%16lx",
 400              (unsigned long)r1.Uniform(std::numeric_limits<uint64_t>::max()));
 401     snprintf(value,
 402              20,
 403              "%16lx",
 404              (unsigned long)r1.Uniform(std::numeric_limits<uint64_t>::max()));
 405     // Put key-value
 406     s1 = db1->Put(WriteOptions(), key, value);
 407     assert(s1.ok());
 408   }
 409   std::cout << "Time by default : " << timer << "ms" << std::endl;
 410   delete db1;
 411
 412   /**********************************
 413             use librados env
 414   ***********************************/
 415   std::string kDBPath2 = "/tmp/DBLoadKeysInRandomOrder2";
 416   DB* db2;
 417   Options options2;
 418   // Optimize RocksDB. This is the easiest way to get RocksDB to perform well
 419   options2.IncreaseParallelism();
 420   options2.OptimizeLevelStyleCompaction();
 421   // create the DB if it's not already present
 422   options2.create_if_missing = true;
 423   options2.env = env_;
 424
 425   // open DB
 426   Status s2 = DB::Open(options2, kDBPath2, &db2);
 427   assert(s2.ok());
 428
 429   ROCKSDB_NAMESPACE::Random64 r2(time(nullptr));
 430
 431   timer.Reset();
 432   for (int i = 0; i < max_loop; ++i) {
 433     snprintf(key,
 434              20,
 435              "%16lx",
 436              (unsigned long)r2.Uniform(std::numeric_limits<uint64_t>::max()));
 437     snprintf(value,
 438              20,
 439              "%16lx",
 440              (unsigned long)r2.Uniform(std::numeric_limits<uint64_t>::max()));
 441     // Put key-value
 442     s2 = db2->Put(WriteOptions(), key, value);
 443     assert(s2.ok());
 444   }
 445   std::cout << "Time by librados : " << timer << "ms" << std::endl;
 446   delete db2;
 447 }
 448
 449 TEST_F(EnvLibradosTest, DBBulkLoadKeysInRandomOrder) {
 450   char key[20] = {0}, value[20] = {0};
 451   int max_loop = 1 << 6;
 452   int bulk_size = 1 << 15;
 453   Timer timer(false);
 454   std::cout << "Test size : loop(" << max_loop << "); bulk_size(" << bulk_size << ")" << std::endl;
 455   /**********************************
 456             use default env
 457   ***********************************/
 458   std::string kDBPath1 = "/tmp/DBBulkLoadKeysInRandomOrder1";
 459   DB* db1;
 460   Options options1;
 461   // Optimize Rocksdb. This is the easiest way to get RocksDB to perform well
 462   options1.IncreaseParallelism();
 463   options1.OptimizeLevelStyleCompaction();
 464   // create the DB if it's not already present
 465   options1.create_if_missing = true;
 466
 467   // open DB
 468   Status s1 = DB::Open(options1, kDBPath1, &db1);
 469   assert(s1.ok());
 470
 471   ROCKSDB_NAMESPACE::Random64 r1(time(nullptr));
 472
 473   timer.Reset();
 474   for (int i = 0; i < max_loop; ++i) {
 475     WriteBatch batch;
 476     for (int j = 0; j < bulk_size; ++j) {
 477       snprintf(key,
 478                20,
 479                "%16lx",
 480                (unsigned long)r1.Uniform(std::numeric_limits<uint64_t>::max()));
 481       snprintf(value,
 482                20,
 483                "%16lx",
 484                (unsigned long)r1.Uniform(std::numeric_limits<uint64_t>::max()));
 485       batch.Put(key, value);
 486     }
 487     s1 = db1->Write(WriteOptions(), &batch);
 488     assert(s1.ok());
 489   }
 490   std::cout << "Time by default : " << timer << "ms" << std::endl;
 491   delete db1;
 492
 493   /**********************************
 494             use librados env
 495   ***********************************/
 496   std::string kDBPath2 = "/tmp/DBBulkLoadKeysInRandomOrder2";
 497   DB* db2;
 498   Options options2;
 499   // Optimize RocksDB. This is the easiest way to get RocksDB to perform well
 500   options2.IncreaseParallelism();
 501   options2.OptimizeLevelStyleCompaction();
 502   // create the DB if it's not already present
 503   options2.create_if_missing = true;
 504   options2.env = env_;
 505
 506   // open DB
 507   Status s2 = DB::Open(options2, kDBPath2, &db2);
 508   assert(s2.ok());
 509
 510   ROCKSDB_NAMESPACE::Random64 r2(time(nullptr));
 511
 512   timer.Reset();
 513   for (int i = 0; i < max_loop; ++i) {
 514     WriteBatch batch;
 515     for (int j = 0; j < bulk_size; ++j) {
 516       snprintf(key,
 517                20,
 518                "%16lx",
 519                (unsigned long)r2.Uniform(std::numeric_limits<uint64_t>::max()));
 520       snprintf(value,
 521                20,
 522                "%16lx",
 523                (unsigned long)r2.Uniform(std::numeric_limits<uint64_t>::max()));
 524       batch.Put(key, value);
 525     }
 526     s2 = db2->Write(WriteOptions(), &batch);
 527     assert(s2.ok());
 528   }
 529   std::cout << "Time by librados : " << timer << "ms" << std::endl;
 530   delete db2;
 531 }
 532
 533 TEST_F(EnvLibradosTest, DBBulkLoadKeysInSequentialOrder) {
 534   char key[20] = {0}, value[20] = {0};
 535   int max_loop = 1 << 6;
 536   int bulk_size = 1 << 15;
 537   Timer timer(false);
 538   std::cout << "Test size : loop(" << max_loop << "); bulk_size(" << bulk_size << ")" << std::endl;
 539   /**********************************
 540             use default env
 541   ***********************************/
 542   std::string kDBPath1 = "/tmp/DBBulkLoadKeysInSequentialOrder1";
 543   DB* db1;
 544   Options options1;
 545   // Optimize Rocksdb. This is the easiest way to get RocksDB to perform well
 546   options1.IncreaseParallelism();
 547   options1.OptimizeLevelStyleCompaction();
 548   // create the DB if it's not already present
 549   options1.create_if_missing = true;
 550
 551   // open DB
 552   Status s1 = DB::Open(options1, kDBPath1, &db1);
 553   assert(s1.ok());
 554
 555   ROCKSDB_NAMESPACE::Random64 r1(time(nullptr));
 556
 557   timer.Reset();
 558   for (int i = 0; i < max_loop; ++i) {
 559     WriteBatch batch;
 560     for (int j = 0; j < bulk_size; ++j) {
 561       snprintf(key,
 562                20,
 563                "%019lld",
 564                (long long)(i * bulk_size + j));
 565       snprintf(value,
 566                20,
 567                "%16lx",
 568                (unsigned long)r1.Uniform(std::numeric_limits<uint64_t>::max()));
 569       batch.Put(key, value);
 570     }
 571     s1 = db1->Write(WriteOptions(), &batch);
 572     assert(s1.ok());
 573   }
 574   std::cout << "Time by default : " << timer << "ms" << std::endl;
 575   delete db1;
 576
 577   /**********************************
 578             use librados env
 579   ***********************************/
 580   std::string kDBPath2 = "/tmp/DBBulkLoadKeysInSequentialOrder2";
 581   DB* db2;
 582   Options options2;
 583   // Optimize RocksDB. This is the easiest way to get RocksDB to perform well
 584   options2.IncreaseParallelism();
 585   options2.OptimizeLevelStyleCompaction();
 586   // create the DB if it's not already present
 587   options2.create_if_missing = true;
 588   options2.env = env_;
 589
 590   // open DB
 591   Status s2 = DB::Open(options2, kDBPath2, &db2);
 592   assert(s2.ok());
 593
 594   ROCKSDB_NAMESPACE::Random64 r2(time(nullptr));
 595
 596   timer.Reset();
 597   for (int i = 0; i < max_loop; ++i) {
 598     WriteBatch batch;
 599     for (int j = 0; j < bulk_size; ++j) {
 600       snprintf(key,
 601                20,
 602                "%16lx",
 603                (unsigned long)r2.Uniform(std::numeric_limits<uint64_t>::max()));
 604       snprintf(value,
 605                20,
 606                "%16lx",
 607                (unsigned long)r2.Uniform(std::numeric_limits<uint64_t>::max()));
 608       batch.Put(key, value);
 609     }
 610     s2 = db2->Write(WriteOptions(), &batch);
 611     assert(s2.ok());
 612   }
 613   std::cout << "Time by librados : " << timer << "ms" << std::endl;
 614   delete db2;
 615 }
 616
 617 TEST_F(EnvLibradosTest, DBRandomRead) {
 618   char key[20] = {0}, value[20] = {0};
 619   int max_loop = 1 << 6;
 620   int bulk_size = 1 << 10;
 621   int read_loop = 1 << 20;
 622   Timer timer(false);
 623   std::cout << "Test size : keys_num(" << max_loop << ", " << bulk_size << "); read_loop(" << read_loop << ")" << std::endl;
 624   /**********************************
 625             use default env
 626   ***********************************/
 627   std::string kDBPath1 = "/tmp/DBRandomRead1";
 628   DB* db1;
 629   Options options1;
 630   // Optimize Rocksdb. This is the easiest way to get RocksDB to perform well
 631   options1.IncreaseParallelism();
 632   options1.OptimizeLevelStyleCompaction();
 633   // create the DB if it's not already present
 634   options1.create_if_missing = true;
 635
 636   // open DB
 637   Status s1 = DB::Open(options1, kDBPath1, &db1);
 638   assert(s1.ok());
 639
 640   ROCKSDB_NAMESPACE::Random64 r1(time(nullptr));
 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_NAMESPACE::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   std::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     std::unique_ptr<SequentialFile> seq_file;
 854     std::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_NAMESPACE::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_NAMESPACE::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_NAMESPACE
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