ceph/src/test/ObjectMap/test_keyvaluedb_atomicity.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 #include <pthread.h>
   3 #include "include/buffer.h"
   4 #include "kv/KeyValueDB.h"
   5 #include <sys/types.h>
   6 #include <dirent.h>
   7 #include <string>
   8 #include <vector>
   9 #include "include/memory.h"
  10 #include <boost/scoped_ptr.hpp>
  11 #include <iostream>
  12 #include <sstream>
  13 #include "stdlib.h"
  14 #include "global/global_context.h"
  15
  16 using namespace std;
  17
  18 const string CONTROL_PREFIX = "CONTROL";
  19 const string PRIMARY_PREFIX = "PREFIX";
  20 const int NUM_COPIES = 100;
  21 const int NUM_THREADS = 30;
  22
  23 string prefix_gen(int i) {
  24   stringstream ss;
  25   ss << PRIMARY_PREFIX << "_" << i << std::endl;
  26   return ss.str();
  27 }
  28
  29 int verify(KeyValueDB *db) {
  30   // Verify
  31   {
  32     map<int, KeyValueDB::Iterator> iterators;
  33     for (int i = 0; i < NUM_COPIES; ++i) {
  34       iterators[i] = db->get_iterator(prefix_gen(i));
  35       iterators[i]->seek_to_first();
  36     }
  37     while (iterators.rbegin()->second->valid()) {
  38       for (map<int, KeyValueDB::Iterator>::iterator i = iterators.begin();
  39            i != iterators.end();
  40            ++i) {
  41         assert(i->second->valid());
  42         assert(i->second->key() == iterators.rbegin()->second->key());
  43         bufferlist r = i->second->value();
  44         bufferlist l = iterators.rbegin()->second->value();
  45         i->second->next();
  46       }
  47     }
  48     for (map<int, KeyValueDB::Iterator>::iterator i = iterators.begin();
  49          i != iterators.end();
  50          ++i) {
  51       assert(!i->second->valid());
  52     }
  53   }
  54   return 0;
  55 }
  56
  57 void *write(void *_db) {
  58   KeyValueDB *db = static_cast<KeyValueDB*>(_db);
  59   std::cout << "Writing..." << std::endl;
  60   for (int i = 0; i < 12000; ++i) {
  61     if (!(i % 10)) {
  62       std::cout << "Iteration: " << i << std::endl;
  63     }
  64     int key_num = rand();
  65     stringstream key;
  66     key << key_num << std::endl;
  67     map<string, bufferlist> to_set;
  68     stringstream val;
  69     val << i << std::endl;
  70     bufferptr bp(val.str().c_str(), val.str().size() + 1);
  71     to_set[key.str()].push_back(bp);
  72
  73     KeyValueDB::Transaction t = db->get_transaction();
  74     for (int j = 0; j < NUM_COPIES; ++j) {
  75       t->set(prefix_gen(j), to_set);
  76     }
  77     assert(!db->submit_transaction(t));
  78   }
  79   return 0;
  80 }
  81
  82 int main() {
  83   char *path = getenv("OBJECT_MAP_PATH");
  84   boost::scoped_ptr< KeyValueDB > db;
  85   if (!path) {
  86     std::cerr << "No path found, OBJECT_MAP_PATH undefined" << std::endl;
  87     return 0;
  88   }
  89   string strpath(path);
  90   std::cerr << "Using path: " << strpath << std::endl;
  91   KeyValueDB *store = KeyValueDB::create(g_ceph_context, "leveldb", strpath);
  92   assert(!store->create_and_open(std::cerr));
  93   db.reset(store);
  94
  95   verify(db.get());
  96
  97   vector<pthread_t> threads(NUM_THREADS);
  98   for (vector<pthread_t>::iterator i = threads.begin();
  99        i != threads.end();
 100        ++i) {
 101     pthread_create(&*i, 0, &write, static_cast<void *>(db.get()));
 102   }
 103   for (vector<pthread_t>::iterator i = threads.begin();
 104        i != threads.end();
 105        ++i) {
 106     void *tmp;
 107     pthread_join(*i, &tmp);
 108   }
 109   verify(db.get());
 110 }