BerkeleyDB基本概念

在BerkeleyDB中, 有以下基本概念

1 Environment: 环境, 并不直接作为应用数据的存放容器, 但是为数据访问提供服务. 提供的主要服务有:
  - logging: 日志服务, 记录所有数据修改的操作, 用于事务取消和故障恢复.
  - locking: 锁服务, 为并发访问数据提供控制与支持.
  - memory pool: 缓存服务, 为快速数据访问提供支持.
  - transaction: 事务服务, 使得可以将修改分组为事务, 让修改具有事务特性(原子性, 一致性, 隔离性, 持久性), 同时提供有效的取消操作.

其中logging, locking和memory pool, 是作为单独的子系统存在的, 所以应用程序需要锁,日志或者缓存服务时, 可以直接使用这些功能, 而无需了解庞大的各种数据访问操作.

环境类似于关系数据库系统中的 数据库, 即为快速,可靠的数据访问提供所有的底层支持

在C API中, 环境是用结构DB_ENV来表示的, 其提供的服务和功能是通过该结构的各种函数指针来完成的.

2 Database: 数据库, 用来存放数据的容器, 包含以键/值对形式存在的数据. 在语义上, 相当关系数据库系统中的一个表. 根据数据在磁盘和内存组织方式的不同以及键/值对形式的不同, 数据库提供了5中不同的存取方法:
 
 - DB_BTREE: 数据在内存和磁盘上以变形B+树的形式存放, 键/值都可以是任意的字节串. 特点是, 数据之间表现出空间关联性. 所以用游标顺序遍历时, 数据是按逻辑上的从小到大来返回的.

- DB_HASH: 数据放在不同的hash桶里, hash函数将键值映射为地址. hash函数本身可能不具有逻辑上的含义(比如并不保证会将逻辑上更大的键值映射到地址更高的桶上去). 键/值也都可以是任意的字节串. 但是, 由于数据之间不表现出空间关联性, 用游标遍历时, 是逐个返回各个桶的内容, 而桶内部和桶之间的内容并没有逻辑上大小关系.

- DB_RECNO: 数据存放在Btree中, 键只能为整型的序列号(db_recno_t), 值可以为任意的字节串. 支持定长和不定长的值. 在创建高序列号值时, 低序列对会被隐式创建, 但值实际上为空字节串且不可读取/删除, 除非显式创建.

- DB_HEAP: db-5.2引入, 键为固定的定长结构体(DB_HEAP_RID), 值可以为任意字节串. 与其他不同的在于, 该方式将尽量利用已有空间来存放新数据, 而不像其他方式一样, 会导致数据占用的磁盘空间越来越大, 尽管数据容量在修改过程中无明显变化.

以上4种类型中, 锁的级别都是页级的, 即一旦加锁, 一个数据页上的所有记录都被锁住.

- DB_QUEUE: 类似DB_RECNO, 键只可为整型序列号(db_recno_t). 但不同的是,值为定长字节串. 当高序列号值被创建时, 低序列号值会被自动创建, 而且也是定长字节串, 虽然无法读取与删除. 由于所有数据占用相同长度, 且中间空白自己隐式创建内容, 给出一个序列号, 可以直接算出其值所在的地址, 所以就可以直接将锁加在记录上, 而不是页上. 这样减小了锁的粒度, 提高了并发程度, 当场景适当时, DB_QUEUE可以提供最快的速度和并发能力. 缺点在于自动填充定长记录的方式会导致极大的空间浪费, 如果需要创建的序列号中有很多空隙的话. 所以该方式适用于内容接近定长, 且数据序列号键值有较好连续性的情况.

在C API中, 数据库是用结构DB来表示的, 其提供的服务和功能是通过该结构的各种函数指针来完成的.

3 Cursor: 游标. 就是在数据库上打开的一个记录数据位置且可以用来对数据进行操作的句柄. 由于游标记录数据位置, 所以可以有如下操作:
  获得下一个/上一个, 取最后一个/第一个, 取不同键值的下一个/上一个, 取相同键值的上一个/下一个 等等. 当然, 上一个下一个的含义随数据库本身的数据组织方式不同而不同.

在C API中, 游标是用结构DBC来表示的, 其提供的服务和功能是通过该结构的各种函数指针来完成的.

4 Key/Data Item: 数据项, 用来表示BerkeleyDB中的键或者值, 数据项有两个基本的要素, 数据的起始地址和数据的长度. 在存数据时, 用户需要设置此两要素, BerkeleyDB将根据此要素获得数据存入数据库中. 在读数据时, 默认情况下, 用户提供一个初始化(0化)的数据项即可, BerkeleyDB将分配空间存放返回的数据, 并且设置该两要素, 使得应用可以从数据库中获得数据.

在C API中, 数据项是用结构DBT来表示的, 有两个主要的成员data(void *)和size(u_int32_t).

5 Master: 主控站点, 在BerkeleyDB集群中, 处于主控地位的站点. 只有该站点可以进行更新操作, 且其将更新操作发往其他的受控站点.

在BerkeleyDB中, 主控站点实际上就是一个环境(DB_ENV),只是指定其具有了主控属性(DB_REP_MASTER),并且调用了启动集群的操作(DB_ENV->repmgr_start 或 DB_ENV->rep_start).

6 Replica: 受控站点, 在BerkeleyDB集群中, 处于受控地位的站点. 该站点只能接受读数据的操作, 其更新只有通过主控站点分发才能进行. 当主控站点崩溃时, 受控站点将进行选举, 胜利者将成为新的主控站点, 余下的角色不变.

在BerkeleyDB中, 受控站点也是一个环境(DB_ENV),只是指定其具有了受控属性(DB_REP_CLIENT),并且调用了启动集群的操作(DB_ENV->repmgr_start 或 DB_ENV->rep_start).

7 Sequence: 序列. 提供了一个获得严格递增或递减序列的方式. 该序列在多线程环境下和多线程环境下仍然可以保证值的唯一性.  用户可以指定值的范围, 可以指定是递增还是递减, 还可以指定是否回环(到达最大值后是否继续从头开始)

在C API中, 序列是用结构DB_SEQUENCE来表示的, 其提供功能是通过该结构的各种函数指针来完成的.

8 Transaction: 事务. 是一种组织数据操作的方式, 使得数据修改具有ACID特性. 当用一个事务句柄传给若干个操作时, 这些操作就属于同一个事务. 当其中一个出问题, 事务可以回倒保证这些操作就像完全没发生一样, 以此保证原子性. 事务本身就是一个加锁者, 保证属于本事务的所有操作涉及的对象不会被其他的数据修改甚至读取, 以此保证一致性和独立性. 当事务提交时, 操作修改将被flush到磁盘, 以此保证持久性.

在C API中, 事务是用结构DB_TXN来表示的, 其提供功能是通过该结构的各种函数指针来完成的.

下面的示例程序设计到本文中提到的所有概念.

基本逻辑是:

创建3个环境(environment), 形成一个集群(cluster/replication group). 一个作为主控站点(master), 两个作为受控站点(replica).

在 主控站点创建一个序列(sequence), 且创建一个应用数据库(database), 然后以序列为键, 以一个随机长度的字节串作为值, 将此键值对存入应用数据库, 这些存数据的操作将被封装到事务(transaction)中. 完成所有操作后, 等待一段时间, 使得修改得以到达受控站点.

而后分别打开所有站点的应用数据库, 使用游标(cursor)遍历数据库, 并且比较获得的数据.

#include "unistd.h"
#include "fcntl.h"
#include <limits.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <dirent.h>
#include <sys/stat.h>
#include "db.h"

#define CHK_RET(ret, str) do { \
    int _ret = (ret); \
    if (_ret) { \
        fprintf(stderr, "Error(%d, %s): %s\n", \
         _ret, (str), db_strerror(_ret)); \
        exit(1); \
    } \
} while(0)

#define TEST_ASSERT(expr) do { \
    if(!(expr)) { \
        printf("Assert Failure(%s:%d): "#expr, __FILE__, __LINE__); \
        exit(1); \
    } \
}while(0)

int test1(int test_cnt);

int main(int argc, char *argv[]) {
    int ret = 0, cnt;

    cnt = 1000;
    if (argc >= 2) {
        cnt = atol(argv[1]);
    }

    ret = test1(cnt);
}

static int open_env(const char *envdir, DB_ENV **envp);
static int setup_site(DB_ENV *dbenv, u_int lport, u_int rport1,
    u_int rport2, int is_master);
static int open_db(DB_ENV *dbenv, const char *fname, const char *dname,
    u_int32_t oflag, DB **dbp);
static int open_sequence(DB *db, DBT *keyp, DB_SEQUENCE **seqp);
static int verify_datadb(DB_ENV *menv, DB_ENV *cenv1, DB_ENV *cenv2,
    const char *fname, const char *dname);

int test1(int test_cnt) {
    DB_ENV *menv, *cenv1, *cenv2, *dbenv;
    DB *seqdb, *datadb;
    const char *mdir, *cdir1, *cdir2;
    const char *seqdbname, *datadbname, *seqkey;
    DB_SITE *site;
    DB_SEQUENCE *seq;
    db_seq_t seqv;
    char buf[100];
    u_int32_t envo_flag, dbo_flag;
    DBT seqdbt, kdbt, ddbt;
    DB_TXN *txn;


    int i, ret;


    mdir = "TESTDIR/masterdir";
    cdir1 = "TESTDIR/clientdir1";
    cdir2 = "TESTDIR/clientdir2";
    seqdbname = "seq.db";
    datadbname = "data.db";
    seqkey = "sequence1";

    for (i = 0; i < sizeof(buf); i++) {
        buf[i] = (char)(i+1);
    }
    srand(time(NULL));


    dbo_flag = DB_THREAD | DB_AUTO_COMMIT;

    /* Open the Environment(s) */
    CHK_RET((ret = open_env(mdir, &menv)), "open_env(mdir)");
    CHK_RET((ret = open_env(cdir1, &cenv1)), "open_env(cdir1)");
    CHK_RET((ret = open_env(cdir2, &cenv2)), "open_env(cdir2)");

    /*
     * Set up the cluster
     * menv is the master
     * cenv1 and cenv2 are replicas
     */
    CHK_RET((ret = setup_site(menv, 30101, 30102, 30103, 1)), "setup_site(menv)");
    CHK_RET((ret = setup_site(cenv1, 30102, 30101, 30103, 0)), "setup_site(cenv1)");
    CHK_RET((ret = setup_site(cenv2, 30103, 30101, 30102, 0)), "setup_site(cenv2)");

    dbenv = menv;

    /* Sequence needs a underlying database, so create and open it */
    CHK_RET((ret = open_db(dbenv, seqdbname, NULL, DB_CREATE, &seqdb)), "open_db(seqdb)");

    /* Create the application database */
    CHK_RET((ret = open_db(dbenv, datadbname, NULL, DB_CREATE, &datadb)), "open_db(datadb)");

    /* Create and opent he sequence */
    memset(&seqdbt, 0, sizeof(DBT));
    seqdbt.data = seqkey;
    seqdbt.size = strlen(seqkey);
    CHK_RET((ret = open_sequence(seqdb, &seqdbt, &seq)), "open_sequence(seqdb)");

    memset(&kdbt, 0, sizeof(DBT));
    kdbt.data = &seqv;
    kdbt.size = sizeof(db_seq_t);
    memset(&ddbt, 0, sizeof(DBT));
    ddbt.data = buf;

    /*
     * Put the data into the application database,
     * The actions are protected by transactions
     */
    txn = NULL;
    CHK_RET((ret = dbenv->txn_begin(dbenv, NULL, &txn, 0)), "dbenv->txn_begin");
    for (i = 0; i < test_cnt; i++) {
        CHK_RET((ret = seq->get(seq, NULL, rand()%5+1, &seqv, 0)), "seq->get");
        if ((i + 1) % 37 == 0) {
            CHK_RET((txn->commit(txn, 0)), "txn->commit");
            CHK_RET((ret = dbenv->txn_begin(dbenv, NULL, &txn, 0)),
             "dbenv->txn_begin");
        }
        ddbt.size = rand() % 99 + 1;
        ret = datadb->put(datadb, txn, &kdbt, &ddbt, 0);
        if (ret != 0) {
            CHK_RET((ret = txn->abort(txn)), "txn->abort");
            CHK_RET((ret = dbenv->txn_begin(dbenv, NULL, &txn, 0)),
             "dbenv->txn_begin");
        }
    }
    CHK_RET((txn->commit(txn, 0)), "txn->commit");

    sleep(test_cnt/50);

    /*
     * Verify the content in the cluster
     * We make sure the items are identical in master and replicas
     */
    CHK_RET((ret = verify_datadb(menv, cenv1, cenv2, datadbname, NULL)),
     "verify_datadb");

    CHK_RET((ret = datadb->close(datadb, 0)), "datadb->close");
    CHK_RET((ret = seq->close(seq, 0)), "seq->close");
    CHK_RET((ret = seqdb->close(seqdb, 0)), "seqdb->close");
    CHK_RET((ret = cenv2->close(cenv2, 0)), "cenv2->close");
    CHK_RET((ret = cenv1->close(cenv1, 0)), "cenv1->close");
    CHK_RET((ret = menv->close(menv, 0)), "menv->close");
    return ret;
}

static int open_env(const char *envdir, DB_ENV **envp) {
    DB_ENV *dbenv = NULL;
    int ret = 0;
    mode_t dirmode = S_IRWXU | S_IRWXG;
    u_int32_t envo_flag;

    envo_flag = DB_INIT_MPOOL | DB_INIT_TXN | DB_INIT_LOG |
     DB_INIT_LOCK | DB_INIT_REP | DB_THREAD | DB_CREATE;

    mkdir(envdir, dirmode);
    CHK_RET((ret = db_env_create(&dbenv, 0)), "db_env_create");
    CHK_RET((ret = dbenv->set_cachesize(dbenv, 0, 16*1048576, 1)),
     "dbenv->set_cachesize");
    CHK_RET((ret = dbenv->open(dbenv, envdir, envo_flag, 0644)), "dbenv->open");
    *envp = dbenv;
    return ret;
}

static int setup_site(DB_ENV *dbenv, u_int lport, u_int rport1,
    u_int rport2, int is_master) {
    u_int ports[3];
    int i, ret;
    DB_SITE *site = NULL;
    u_int32_t role;

    ports[0] = lport;
    ports[1] = rport1;
    ports[2] = rport2;

    /* Set up the information for other sites in the cluster */
    for (i = 0; i < 3; i++) {
        CHK_RET((ret = dbenv->repmgr_site(dbenv, "127.0.0.1",
         ports[i], &site, 0)), "dbenv->repmgr_site");
        if (i == 0) {
            CHK_RET((ret = site->set_config(site, DB_LOCAL_SITE, 1)),
             "site->set_config(DB_LOCAL_SITE)");
            if (is_master) {
                CHK_RET((ret = site->set_config(site, DB_GROUP_CREATOR, 1)),
                 "site->set_config(DB_GROUP_CREATOR");
            }
        } else {
            CHK_RET((ret = site->set_config(site, DB_BOOTSTRAP_HELPER, 1)),
             "site->set_config(DB_BOOTSTRAP_HELPER)");
        }
        CHK_RET((ret = site->close(site)), "site->close");
        site = NULL;
    }

    role = DB_REP_CLIENT;
    if (is_master)
        role = DB_REP_MASTER;

    /* Start the cluster, retry some times if not up in time */
    for (i = 0; i < 10; i++) {
        ret = dbenv->repmgr_start(dbenv, 3, role);
        if (ret != DB_REP_UNAVAIL)
            break;
    }
    CHK_RET(ret, "dbenv->repmgr_start");

    return ret;
}

static int open_db(DB_ENV *dbenv, const char *fname, const char *dname, u_int32_t oflag, DB **dbp) {
    DB *db;
    u_int32_t dbo_flag;
    int ret;

    db = NULL;
    dbo_flag = DB_AUTO_COMMIT | DB_THREAD | oflag;

    CHK_RET((ret = db_create(&db, dbenv, 0)), "db_create(dbenv)");
    CHK_RET((ret = db->open(db, NULL, fname, dname, DB_BTREE, dbo_flag, 0644)),
     "db->open(DB_BTREE)");
    *dbp = db;
    return ret;
}

static int open_sequence(DB *db, DBT *keyp, DB_SEQUENCE **seqp) {
    DB_SEQUENCE *seq;
    int ret;


    CHK_RET((ret = db_sequence_create(&seq, db, 0)), "db_sequence_create");
    /* Increment and allow wrap */
    CHK_RET((ret = seq->set_flags(seq, DB_SEQ_INC | DB_SEQ_WRAP)),
     "seq->set_flags(DB_SEQ_INC|DB_SEQ_WRAP)");
    CHK_RET((ret = seq->set_range(seq, 1, UINT_MAX)), "seq->set_range(1, UINT_MAX)");
    CHK_RET((ret = seq->initial_value(seq, 1)), "seq->initial_value(1)");
    CHK_RET((ret = seq->open(seq, NULL, keyp, DB_CREATE|DB_THREAD)), "seq->open");
    *seqp = seq;
    return ret;
}

static int verify_datadb(DB_ENV *menv, DB_ENV *cenv1, DB_ENV *cenv2,
    const char *fname, const char *dname) {
    DB_ENV *dbenvs[3];
    DB *dbs[3];
    DBC *dbcs[3];
    int ret, i, rets[3];
    DBT kdbts[3], ddbts[3];

    dbenvs[0] = menv;
    dbenvs[1] = cenv1;
    dbenvs[2] = cenv2;
    memset(kdbts, 0, sizeof(kdbts));
    memset(ddbts, 0, sizeof(ddbts));

    for (i = 0; i < 3; i++) {
        CHK_RET((ret = open_db(dbenvs[i], fname, dname, DB_RDONLY, &dbs[i])),
         "open_db(no_create)");
        CHK_RET((ret = dbs[i]->cursor(dbs[i], NULL, &dbcs[i], 0)), "DB->cursor");
    }

    /*
     * Open a cursor per database, and get the data using the cursor, then compare
     * the items. The items should be identical
     */
    while (1) {
        for (i = 0; i < 3; i++) {
            rets[i] = dbcs[i]->get(dbcs[i], &kdbts[i], &ddbts[i], DB_NEXT);
            TEST_ASSERT((rets[i] == 0 || rets[i] == DB_NOTFOUND));
        }
        for (i = 1; i < 3; i++) {
            TEST_ASSERT((rets[i] == rets[0]));
            if (rets[i] == DB_NOTFOUND)
                continue;
            TEST_ASSERT((kdbts[i].size = kdbts[0].size));
            TEST_ASSERT((memcmp(kdbts[i].data, kdbts[0].data, kdbts[0].size) == 0));
            TEST_ASSERT((ddbts[i].size = ddbts[0].size));
            TEST_ASSERT((memcmp(ddbts[i].data, ddbts[0].data, ddbts[0].size) == 0));
        }
        if (rets[0] == DB_NOTFOUND)
            break;
    }

    for (i = 0; i < 3; i++) {
        CHK_RET((ret = dbcs[i]->close(dbcs[i])), "DBC->close");
        CHK_RET((ret = dbs[i]->close(dbs[i], 0)), "DB->close");
    }

    return ret;
}