Berkeley DB示例程序详解(3.1)

/*
* 这个示例程序演示了使用Berkeley DB的replication功能的方法。
*
* Berkeley DB提供了一套基本API和一套replication manager API来使用
* 它的replication功能。
* 前者有更大的灵活性，用户可以选择各种网络协议来实现数据传输，并且使用各种
* 线程库来进行多线程编程。同时由于使用最基本的API， 用户可以定制选举和
* replication系统的内部消息处理方式，非常灵活。所以这也意味着用户需要更多
* 的代码来使用Berkeley DB replication功能，开发难度相对后者较大；
* 后者基于TCP/IP协议和pthread线程库(Windows上面使用win32线程库)，并且按照通常
* 的需求，对选举和内部消息处理进行了通用的处理，并且通过让用户配置
* 策略(policy)和参数的方式提供一定的灵活性和可定制性。
*
* 本程序基于Berkeley DB的replication manager。
*
*
* 名词术语：
*
* site：一个(数据库环境, IP地址, 端口(port))的三元组。所以，多个site可以处于
* 不同的计算机上面，也可以在同一个计算机上面，只要配置了不同的端口并使用不同
* 的数据库环境目录。
*
* replication group：由若干个site组成的集合，其中有且只有一个site 可以写入
* 数据到数据库当中， 称为master；其余的site只可以从中读取数据，称为replica。
*
* master：接受读写请求。如果写入数据，那么这些写入的数据不仅会更新master
* 自己的数据库，还会通过Berkeley DB的
* replication功能，被传播到replication group当中的replica上面。一个
* replication group通过选举或者显式指定的方式，得到master。
*
* replica：接受读请求。从master写入的数据会自动传播到每一个replica上面，从而
* 将replica上面的数据更新。任何一个replica在选举的时候，都有可能成为新的master
*
* 一个replication group作为一个整体来看的话，它比一个单机有
* 更大的数据读写能力，因为所有的site都可以提供数据，自然读数据的吞吐量更大；
* 并且master可以被配置为只写入数据而不必理会读数据请求，由replica提供数据，
* 这样，写入的吞吐量也更大。
*
* replication group比单机的另一个优势，就是高可靠性(high availability)，因为，
* 其中如果任何一个site脱离（断电，网络故障，应用程序错误等），这个组都可以
* 继续服务：如果一个replica脱离，那么
* 还有其他replica可以提供数据，并且这个脱离的replica可以重新加入该组，其数据
* 被自动更新到最新；如果master脱离，那么该组的所有replica举行选举，从中选
* 出新的master。
*/
int
main(argc, argv)
int argc;
char *argv[];
{
DB_ENV *dbenv;
SETUP_DATA setup_info;
repsite_t *site_list;
APP_DATA my_app_data;
thread_t ckp_thr, lga_thr;
supthr_args sup_args;
u_int32_t start_policy;
int i, ret, t_ret;
memset(&setup_info, 0, sizeof(SETUP_DATA));
setup_info.progname = progname;
memset(&my_app_data, 0, sizeof(APP_DATA));
dbenv = NULL;
ret = 0;
/*
* 默认使用选举的方法选出master。替代方案就是指定某一个site是master，
* 其他site是replica。
*/
start_policy = DB_REP_ELECTION;
/* 创建一个拥有replication功能的数据库环境，但还不打开。 */
if ((ret = create_env(progname, &dbenv)) != 0)
goto err;
dbenv->app_private = &my_app_data;
/*
* 配置replication消息处理回调函数。
* 有任何replication消息，Berkeley DB都会调用该函数来处理。
*/
(void)dbenv->set_event_notify(dbenv, event_callback);
/* Parse command line and perform common replication setup. */
if ((ret = common_rep_setup(dbenv, argc, argv, &setup_info)) != 0)
goto err;
/*
* Perform repmgr-specific setup based on command line options.
* 如果显式指定了master，那么就使用DB_REP_MASTER把本site设置为master，
* 并用DB_REP_CLIENT把其他site设置为replica。
*/
if (setup_info.role == MASTER)
start_policy = DB_REP_MASTER;
else if (setup_info.role == CLIENT)
start_policy = DB_REP_CLIENT;
/*
* 设置本地site信息，也就是IP地址和端口信息。因为site是一个
* （数据库环境，IP地址，port）的三元组，而数据库环境我们已经创建好
* 了。
*/
if ((ret = dbenv->repmgr_set_local_site(dbenv, setup_info.self.host,
setup_info.self.port, 0)) != 0) {
fprintf(stderr, "Could not set listen address (%d)./n", ret);
goto err;
}
site_list = setup_info.site_list;
/*
* 添加其他site的信息到本site。如果显式指定master，那么一个replica
* 一定要知道master的信息；如果选举，那么每一个site要知道其他所有
* site的信息。
*
* 两个site可以成为peer关系，通过在下面的函数中设置
* DB_REPMGR_PEER来指定。一个replica可以被它的peer更新到最新，这样
* 就为master去掉了更新过时的replica的负担。
*/
for (i = 0; i < setup_info.remotesites; i++) {
if ((ret = dbenv->repmgr_add_remote_site(dbenv,
site_list[i].host, site_list[i].port, NULL,
site_list[i].peer ? DB_REPMGR_PEER : 0)) != 0) {
dbenv->err(dbenv, ret,
"Could not add site %s:%d", site_list[i].host,
(int)site_list[i].port);
goto err;
}
}
/*
* Configure heartbeat timeouts so that repmgr monitors the
* health of the TCP connection. Master sites broadcast a heartbeat
* at the frequency specified by the DB_REP_HEARTBEAT_SEND timeout.
* Client sites wait for message activity the length of the
* DB_REP_HEARTBEAT_MONITOR timeout before concluding that the
* connection to the master is lost. The DB_REP_HEARTBEAT_MONITOR
* timeout should be longer than the DB_REP_HEARTBEAT_SEND timeout.
*/
if ((ret = dbenv->rep_set_timeout(dbenv, DB_REP_HEARTBEAT_SEND,
5000000)) != 0)
dbenv->err(dbenv, ret,
"Could not set heartbeat send timeout./n");
if ((ret = dbenv->rep_set_timeout(dbenv, DB_REP_HEARTBEAT_MONITOR,
10000000)) != 0)
dbenv->err(dbenv, ret,
"Could not set heartbeat monitor timeout./n");
/*
* The following repmgr features may also be useful to your
* application. See Berkeley DB documentation for more details.
* - Two-site strict majority rule - In a two-site replication
* group, require both sites to be available to elect a new
* master.
* - Timeouts - Customize the amount of time repmgr waits
* for such things as waiting for acknowledgements or attempting
* to reconnect to other sites.
* - Site list - return a list of sites currently known to repmgr.
*/
/* 打开数据库环境。*/
if ((ret = env_init(dbenv, setup_info.home)) != 0)
goto err;
/* Start checkpoint and log archive threads. */
sup_args.dbenv = dbenv;
sup_args.shared = &my_app_data.shared_data;
if ((ret = start_support_threads(dbenv, &sup_args, &ckp_thr,
&lga_thr)) != 0)
goto err;
if ((ret = dbenv->repmgr_start(dbenv, 3, start_policy)) != 0)
goto err;
if ((ret = doloop(dbenv, &my_app_data.shared_data)) != 0) {
dbenv->err(dbenv, ret, "Client failed");
goto err;
}
/* Finish checkpoint and log archive threads. */
if ((ret = finish_support_threads(&ckp_thr, &lga_thr)) != 0)
goto err;
/*
* We have used the DB_TXN_NOSYNC environment flag for improved
* performance without the usual sacrifice of transactional durability,
* as discussed in the "Transactional guarantees" page of the Reference
* Guide: if one replication site crashes, we can expect the data to
* exist at another site. However, in case we shut down all sites
* gracefully, we push out the end of the log here so that the most
* recent transactions don't mysteriously disappear.
*/
if ((ret = dbenv->log_flush(dbenv, NULL)) != 0) {
dbenv->err(dbenv, ret, "log_flush");
goto err;
}
err:
if (dbenv != NULL &&
(t_ret = dbenv->close(dbenv, 0)) != 0) {
fprintf(stderr, "failure closing env: %s (%d)/n",
db_strerror(t_ret), t_ret);
if (ret == 0)
ret = t_ret;
}
return (ret);
}
/*
* 事件通知函数。当使用replication manager的时候，该事件已经被
* replication manager 适当地处理好了，所以在这个回调函数当中，
* 我们只需要简单地更新状态或者打印消息即可。
*/
static void
event_callback(dbenv, which, info)
DB_ENV *dbenv;
u_int32_t which;
void *info;
{
APP_DATA *app = dbenv->app_private;
SHARED_DATA *shared = &app->shared_data;
info = NULL; /* Currently unused. */
switch (which) {
case DB_EVENT_REP_CLIENT:
shared->is_master = 0;
shared->in_client_sync = 1;
break;
case DB_EVENT_REP_MASTER:
shared->is_master = 1;
shared->in_client_sync = 0;
break;
case DB_EVENT_REP_NEWMASTER:
shared->in_client_sync = 1;
break;
case DB_EVENT_REP_PERM_FAILED:
/*
* Did not get enough acks to guarantee transaction
* durability based on the configured ack policy. This
* transaction will be flushed to the master site's
* local disk storage for durability.
*/
printf(
"Insufficient acknowledgements to guarantee transaction durability./n");
break;
case DB_EVENT_REP_STARTUPDONE:
shared->in_client_sync = 0;
break;
default:
dbenv->errx(dbenv, "ignoring event %d", which);
}
}

未完，下文见 Berkeley DB示例程序详解(3.2)

本文出自 “赵伟 计算机软件技术博客” 博客，请务必保留此出处http://davidzhao.blog.51cto.com/4548102/1225772