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:
Timer(bool run=false)26   explicit Timer(bool run = false)
27   {
28     if (run)
29       Reset();
30   }
Reset()31   void Reset()
32   {
33     _start = high_resolution_clock::now();
34   }
Elapsed() const35   milliseconds Elapsed() const
36   {
37     return std::chrono::duration_cast<milliseconds>(high_resolution_clock::now() - _start);
38   }
39   template <typename T, typename Traits>
operator <<(std::basic_ostream<T,Traits> & out,const Timer & timer)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 
EnvLibradosTest()61   EnvLibradosTest()
62     : env_(new EnvLibrados(db_name, config, db_pool)) {
63   }
~EnvLibradosTest()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 
TEST_F(EnvLibradosTest,Basics)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 
TEST_F(EnvLibradosTest,ReadWrite)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 
TEST_F(EnvLibradosTest,Locks)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 
TEST_F(EnvLibradosTest,Misc)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 
TEST_F(EnvLibradosTest,LargeWrite)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 
TEST_F(EnvLibradosTest,FrequentlySmallWrite)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 
TEST_F(EnvLibradosTest,Truncate)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 
TEST_F(EnvLibradosTest,DBBasics)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 
TEST_F(EnvLibradosTest,DBLoadKeysInRandomOrder)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 
TEST_F(EnvLibradosTest,DBBulkLoadKeysInRandomOrder)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 
TEST_F(EnvLibradosTest,DBBulkLoadKeysInSequentialOrder)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 
TEST_F(EnvLibradosTest,DBRandomRead)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 
EnvLibradosMutipoolTest()747   EnvLibradosMutipoolTest() {
748     env_ = new EnvLibrados(client_name, cluster_name, flags, db_name, config, db_pool, wal_dir, wal_pool, write_buffer_size);
749   }
~EnvLibradosMutipoolTest()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 
TEST_F(EnvLibradosMutipoolTest,Basics)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 
TEST_F(EnvLibradosMutipoolTest,DBBasics)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 
TEST_F(EnvLibradosMutipoolTest,DBBulkLoadKeysInRandomOrder)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 
TEST_F(EnvLibradosMutipoolTest,DBTransactionDB)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 
main(int argc,char ** argv)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 
main(int argc,char ** argv)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
1147