access berkeley db parallely

test_bdb.cpp

/*
 * test_bdb.cpp
 *
 * access bdb with DB_THREAD (bad idea, according to
 * https://docs.oracle.com/cd/E17076_04/html/api_reference/CXX/frame_main.html)
 *
 * g++ -o /tmp/test_bdb test_bdb.cpp -ldb_cxx -lboost_thread -lboost_system
 */

#include <db_cxx.h>
#include <cstring>
#include <boost/thread/thread.hpp>

static bool stop = true;

static int opendb(Db* db, std::string& name, u_int32_t flags)
{
  try {
    db->open(NULL, name.c_str(), NULL, DB_BTREE, flags, 0);
  } catch (DbException& e) {
    throw e;
  } catch (std::exception& e) {
    throw e;
  }
}

static int closedb(Db* db)
{
  db->close(0);
}

static void append(Db* db, Dbt& key, Dbt& data)
{
  int ret = db->put(NULL, &key, &data, DB_NOOVERWRITE);
  if (ret == DB_KEYEXIST) {
    db->err(ret, "Put failed because key already exists");
  }
}

static void append_loop(Db* db)
{
  unsigned long index = 0;
  std::string desc("Test data");
  const char* desc_str = desc.c_str();

  Dbt key;
  Dbt data(const_cast<char*>(desc_str), ::strlen(desc_str) + 1);

  do {
    std::cout << "appending..." << index << std::endl;
    Dbt key;
    key.set_data(&index);
    key.set_size(sizeof(unsigned long));
    append(db, key, data);
    index++;
  } while (!stop);
}

static void append_thread_func(Db* db, std::string& name, u_int32_t flags)
{
  append_loop(db);
}

static int query(Db* db, Dbt* key, Dbt* data)
{
  try {
    db->get(NULL, key, data, 0);
  } catch (DbException& e) {
    std::cout << "error" << std::endl;
    return -1;
  }
  return 0;
}

static void query_loop(Db* db)
{
  unsigned long index = 0;

  Dbt key, data;

  do {
    std::cout << "query..." << index << std::endl;

    key.set_data(&index);
    key.set_size(sizeof(unsigned long));

    char desc[1024];
    data.set_data(desc);
    data.set_ulen(sizeof(desc));
    data.set_flags(DB_DBT_USERMEM);

    if(query(db, &key, &data) >= 0) {
      std::cout << desc << std::endl;
      index++;
    }
  } while(!stop);
}

static void query_thread_func(Db* db, std::string& name, u_int32_t flags)
{
  query_loop(db);
}

int main()
{
  u_int32_t env_flags = DB_CREATE | DB_INIT_MPOOL | DB_THREAD;
  u_int32_t db_flags = DB_CREATE | DB_TRUNCATE;
  u_int32_t rd_db_flags = DB_RDONLY;

  std::string dbName("test_bdb.db");
  std::string envHome("/tmp/testEnv");
  DbEnv myEnv(0);
  Db* myDb;

  try {
    myEnv.open(envHome.c_str(), env_flags, 0);
    myDb = new Db(&myEnv, 0);
    stop = false;

    opendb(myDb, dbName, db_flags);
    boost::thread append_thread(append_thread_func, myDb, dbName, db_flags);
    boost::thread query_thread(query_thread_func, myDb, dbName, db_flags);
    ::getchar();
    stop = true;
    append_thread.join();
    query_thread.join();
    append_thread_func(myDb, dbName, db_flags);
    query_thread_func(myDb, dbName, db_flags);
    closedb(myDb);
    delete myDb;
    myEnv.close(0);
  } catch (DbException &e) {
    std::cerr << "Error opening database environment: "
              << envHome << " and database " << dbName << std::endl;
    std::cerr << e.what() << std::endl;
    return -1;
  } catch (std::exception &e) {
    std::cerr << "Error opening database environment: "
              << envHome << " and database " << dbName << std::endl;
    std::cerr << e.what() << std::endl;
    return -1;
  }
}