openGauss 中的 DCF

• openGauss
• 什么是DCF？
• DCF 搭建
• DCF 说明文档

---

目录

• 1 openGauss
  • 1.1 openGauss 概述
  • 1.2 openGauss 系统架构
  • 1.3 openGauss 代码结构
• 2 什么是DCF？
  • 2.1 DCF 概述
  • 2.2 DCF 架构
• 3 DCF 搭建
• 4 DCF 说明文档

---

1 openGauss

1.1 openGauss 概述

openGauss是关系型数据库，采用客户端/服务器、单进程多线程架构；支持单机和一主多备部署方式，同时支持备机可读、双机高可用等特性。

openGauss有如下基本功能：

• 支持标准SQL
• 支持标准开发接口
• 混合存储引擎支持
• 事务支持
• 软硬件结合
• 智能优化器
• AI支持
• 安全性
• 函数及存储过程支持
• PostgreSQL接口兼容
• 支持SQL hint
• Copy接口支持容错机制

---

1.2 openGauss 系统架构

openGauss主要包含了openGauss服务器、客户端驱动、OM等模块，官方架构如下所示。

openGauss 模块说明:

名称	描述
OM	运维管理模块（Operation Manager）。提供openGauss日常运维、配置管理的管理接口、工具。
客户端驱动	客户端驱动（Client Driver）。负责接收来自应用的访问请求，并向应用返回执行结果；负责与openGauss实例的通信，下发SQL在openGauss实例上执行，并接收命令执行结果。
openGauss主（备）	openGauss主（备）。负责存储业务数据（支持行存、列存、内存表存储）、执行数据查询任务以及向客户端驱动返回执行结果。
Storage	服务器的本地存储资源，持久化存储数据。

---

1.3 openGauss 代码结构

openGauss 代码主要由三部分组成：

• 通信管理
• SQL引擎
• 存储引擎

当服务器的 GaussMaster 线程接收到客户端程序发送过来的服务请求后，会根据收到的信息会立即 fork() 一个子线程，这个子线程对请求进行身份验证成功后成为对应的后端业务处理子线程（gaussdb）。之后该客户端发送的请求将由此业务处理子线程（gaussdb）负责处理。当业务处理子线程（gaussdb）接收到客户端发送过来的查询（SQL）后，会调用openGauss的SQL引擎对SQL语句进行词法解析、语法解析、语义解析、查询重写等处理操作，然后使用查询优化器生成最小代价的查询路径计划。之后，SQL执行器会按照已制定的最优执行计划对SQL语句进行执行，并将执行结果反馈给客户端。

---

2 DCF

2.1 DCF 概述

DCF是openGauss的日志共识框架，DCF全称是Distributed Consensus Framework，即分布式一致性共识框架。解决分布式一致性问题典型算法是Paxos、Raft等，DCF实现了Paxos算法。使用DCF可以提供日志复制、集群高可用等能力。DCF提供了基于Paxos多种角色节点类型，并能进行调整。日志复制支持动态流量调整，支持少数派强起能力，自选主能力。

DCF是一款高性能、高度成熟可靠、易扩展、易使用的独立基础库，其他系统通过接口与DCF简单对接，就能够轻松拥有Paxos算法赋予的强一致、高可用、自动容灾等能力。

---

2.2 DCF 架构

DCF功能架构图如上图所示主要包括：算法模块、存储模块、通信模块、服务层等。

• 接口层：与DB kernel中的日志模块做对接，对外提供读写、查询、注册回调的接口。内核会将DCF看作磁盘进行写入，收到达成共识后返回的成功信息后，继续推进事务。
• 元数据模块：记录与管理集群配置信息，IP端口、角色类型、超时时间等。
• 选举模块：负责选举、心跳维持、状态发布等功能。
• 复制模块：推进日志的复制、分发、达成一致等。
• 存储模块：负责日志数据的持久化。
• 通信模块：负责节点间数据的通信。
• 服务层：提供基础功能。

更多对于系统的讲解推荐看Gauss松鼠会的视频

---

3 DCF 搭建

3.1 配置 dcf 白名单

切换至omm用户：

su - omm

打开配置文件：

cd /opt/ogdata
vim pg_hba.conf

在做后面添加白名单：

host all all 192.168.1.17/24 trust
host all all 192.168.1.18/24 trust
host all all 192.168.1.19/24 trust

---

3.2 配置dcf参数

将以下信息依次添加到所有主机的/opt/ogdata/postgresql.conf 的最后面

vim /opt/ogdata/postgresql.conf

192.168.1.17 添加下列信息：

port=21000
dcf_node_id = 1
dcf_ssl=off
dcf_data_path = ‘/opt/ogdata/dcf_data’
dcf_log_path= ‘/opt/ogdata/dcf_log’
dcf_config=’[{“stream_id”:1,”node_id”:1,”ip”:”192.168.1.17”,”port”:21000,”role”:”LEADER”},{“stream_id”:1,”node_id”:2,”ip”:”192.168.1.18”,”port”:21000,”role”:”FOLLOWER”},{“stream_id”:1,”node_id”:3,”ip”:”192.168.1.19”,”port”:21000,”role”:”FOLLOWER”}]’
replconninfo1 = ‘localhost=192.168.1.17 localport=21001 localheartbeatport=21005 localservice=21004 remotehost=192.168.1.18 remoteport=21001 remoteheartbeatport=21005 remoteservice=21004’
replconninfo2 = ‘localhost=192.168.1.17 localport=21001 localheartbeatport=21005 localservice=21004 remotehost=192.168.1.19 remoteport=21001 remoteheartbeatport=21005 remoteservice=21004’
enable_dcf = on

192.168.1.18 添加下列信息：

port=21000
dcf_node_id = 2
dcf_ssl=off
dcf_data_path = ‘/opt/ogdata/dcf_data’
dcf_log_path= ‘/opt/ogdata/dcf_log’
dcf_config=’[{“stream_id”:1,”node_id”:1,”ip”:”192.168.1.17”,”port”:21000,”role”:”LEADER”},{“stream_id”:1,”node_id”:2,”ip”:”192.168.1.18”,”port”:21000,”role”:”FOLLOWER”},{“stream_id”:1,”node_id”:3,”ip”:”192.168.1.19”,”port”:21000,”role”:”FOLLOWER”}]’
replconninfo1 = ‘localhost=192.168.1.18 localport=21001 localheartbeatport=21005 localservice=21004 remotehost=192.168.1.17 remoteport=21001 remoteheartbeatport=21005 remoteservice=21004’
replconninfo2 = ‘localhost=192.168.1.18 localport=21001 localheartbeatport=21005 localservice=21004 remotehost=192.168.1.19 remoteport=21001 remoteheartbeatport=21005 remoteservice=21004’
enable_dcf = on

192.168.1.19 添加下列信息：

port=21000
dcf_node_id = 3
dcf_ssl=off
dcf_data_path = ‘/opt/ogdata/dcf_data’
dcf_log_path= ‘/opt/ogdata/dcf_log’
dcf_config=’[{“stream_id”:1,”node_id”:1,”ip”:”192.168.1.17”,”port”:21000,”role”:”LEADER”},{“stream_id”:1,”node_id”:2,”ip”:”192.168.1.18”,”port”:21000,”role”:”FOLLOWER”},{“stream_id”:1,”node_id”:3,”ip”:”192.168.1.19”,”port”:21000,”role”:”FOLLOWER”}]’
replconninfo1 = ‘localhost=192.168.1.19 localport=21001 localheartbeatport=21005 localservice=21004 remotehost=192.168.1.17 remoteport=21001 remoteheartbeatport=21005 remoteservice=21004’
replconninfo2 = ‘localhost=192.168.1.19 localport=21001 localheartbeatport=21005 localservice=21004 remotehost=192.168.1.18 remoteport=21001 remoteheartbeatport=21005 remoteservice=21004’
enable_dcf = on

---

3.3 启动opengauss

集群全部节点以standby的模式启动：

gs_ctl start -D /opt/ogdata -M standby

全部节点查看状态：

gs_ctl query -D /opt/ogdata
steam://rungameid/1225570

其中，dcf_replication_info表示当前节点dcf信息。

role：表示当前节点角色，角色一共有如下几种，LEADER、FOLLOWER、LOGGER、PASSIVE、PRE_CANDICATE、CANDIDATE、UNKNOW。从上图可以看出当前节点是LEADER节点。
term：选举任期。
run_mode：DCF运行模式，当前0表示自动选举模式，2表示关闭自动选举模式。
work_mode:DCF工作模式。
hb_interval:DCF节点间心跳间隔时间，单位ms。
elc_timeout:DCF选举超时时间，单位ms。
applied_index:被应用到状态机的日志位置。
commit_index:已被大多数DCF节点保存的日志位置，此commit_index之前日志均已持久化。
first_index:DCF节点保存的首条日志位置，此位置会随着DN调用dcf_truncate而向后推进，之前的日志会被清理。
last_index:DCF节点保存的最后一条日志位置，此日志位置包含DCF节点存储在内存里但是没有持久化的日志，故而last_index >= commit_index。
cluster_min_apply_idx:集群最小已应用的日志位置。
leader_id:leader节点ID。
leader_ip:leader节点IP。
leader_port:leader节点端口，DCF内部使用 。
nodes:集群其他节点信息。

---

DCF 说明文档

一、工程说明

1、编程语言：C

2、编译工程：cmake

3、目录说明：

• DCF：主目录，CMakeLists.txt为主工程入口；
• src:：源代码目录，按子目录划分模块解耦；
• test：测试用例
• build：工程构建脚本

---

二、编译指导

1、概述

编译DCF需要dcf和binarylibs两个组件。

• dcf：dcf的主要代码。可以从开源社区获取。
• binarylibs：依赖的第三方开源软件，你可以直接编译openGauss-third_party代码获取，也可以从开源社区下载已经编译好的并上传的一个副本。

2、操作系统和软件依赖要求

支持以下操作系统：

• CentOS 7.6（x86）
• openEuler-20.03-LTS
  
  适配其他系统，可参照openGauss数据库编译指导
  
  当前DCF依赖第三方软件有securec、lz4、zstd、openssl、cjson;
  编译dcf依赖的第三方软件要求与编译opengauss对依赖的第三方软件要求一致。

3、下载dcf

可以从开源社区下载dcf和openGauss-third_party。
可以通过以下网站获取编译好的binarylibs。
https://opengauss.obs.cn-south-1.myhuaweicloud.com/2.0.0/openGauss-third_party_binarylibs.tar.gz

4、编译第三方软件

在编译dcf之前，需要先编译dcf依赖的开源及第三方软件。这些开源及第三方软件存储在openGauss-third_party代码仓库中，通常只需要构建一次。如果开源软件有更新，需要重新构建软件。

用户也可以直接从binarylibs库中获取开源软件编译和构建的输出文件。

5、代码编译

使用DCF/build/linux/opengauss/build.sh编译代码, 参数说明请见以下表格。

选项	参数	说明
-3rd	[binarylibs path]	指定binarylibs路径。该路径必须是绝对路径。

现在只需使用如下命令即可编译：

[user@linux dcf]$ sh build.sh -3rd [binarylibs path]

完成编译后，动态库生成在DCF/lib目录中

---

三、接口说明

1、API列表

DCF角色定义：
typedef enum en_dcf_role {
DCF_ROLE_UNKNOWN = 0,
DCF_ROLE_LEADER,
DCF_ROLE_FOLLOWER,
DCF_ROLE_LOGGER,
DCF_ROLE_PASSIVE,
DCF_ROLE_PRE_CANDIDATE,
DCF_ROLE_CANDIDATE,
DCF_ROLE_CEIL,
} dcf_role_t;

• int dcf_set_param(const char *param_name, const char *param_value);

功能说明：设置DCF配置参数
参数说明：param_name是需要设置的参数名称，param_value是需要设置的参数值。
参数名称有以下类型：
“ELECTION_TIMEOUT” –选举超时时间，单位ms
“HEARTBEAT_INTERVAL” –心跳间隔，单位ms
“RUN_MODE” –运行模式，ELECTION_AUTO或ELECTION_MANUAL
“INSTANCE_NAME” –实例名称
“DATA_PATH” –数据文件路径
“LOG_PATH” –日志文件路径
“LOG_LEVEL” –最大日志级别”RUN_ERR|RUN_WAR|RUN_INF|DEBUG_ERR|DEBUG_WAR|DEBUG_INF|MEC|OPER|TRACE|PROFILE”,
需要开启自定义级别,从上述字符串中选取并使用|分割;
默认级别”RUN_ERR|RUN_WAR|DEBUG_ERR|OPER”
若需要关闭日志打印，配置”NONE”
“LOG_BACKUP_FILE_COUNT” –日志备份文件数
“MAX_LOG_FILE_SIZE” –日志文件最大size，单位MB
“LOG_FILE_PERMISSION” –日志文件权限,权限不高于700
“LOG_PATH_PERMISSION” –日志路径权限,权限不高于700
“MEC_AGENT_THREAD_NUM” –通信agent线程数量
“MEC_REACTOR_THREAD_NUM” –通信reactor线程数量
“MEC_CHANNEL_NUM” –通信通道数量
“MEM_POOL_INIT_SIZE” –共用buddy pool的初始size
“MEM_POOL_MAX_SIZE” –共用buddy pool的最大size
“COMPRESS_ALGORITHM” –通信压缩算法, 0:COMPRESS_NONE, 1:COMPRESS_ZSTD, 2:COMPRESS_LZ4
“COMPRESS_LEVEL” –压缩级别
“SOCKET_TIMEOUT” –socket收发报文超时时间，单位ms
“CONNECT_TIMEOUT” –连接超时时间，单位ms
“REP_APPEND_THREAD_NUM” –leader节点发送日志的线程数
“MEC_FRAGMENT_SIZE” –通信消息buffer size
“STG_POOL_INIT_SIZE” –存储pool初始size
“STG_POOL_MAX_SIZE” –存储pool最大size，存储有读写两个pool，这里是单个pool的size
“MEC_POOL_MAX_SIZE” –通信pool最大size，通信有收发两个pool，这里是单个pool的size
“FLOW_CONTROL_CPU_THRESHOLD” – CPU使用率超过此值时会对passive节点的日志同步进行流控，单位%
“FLOW_CONTROL_NET_QUEUE_MESSAGE_NUM_THRESHOLD” – DCF发送日志队列中消息超过此值时会对passive节点的日志同步进行流控
“FLOW_CONTROL_DISK_RAWAIT_THRESHOLD” – 磁盘读延时超过此值时会对passive节点的日志同步进行流控，单位us
“DN_FLOW_CONTROL_RTO” – DN流控参数，结合接口dcf_pause_rep使用
“DN_FLOW_CONTROL_RPO” – DN流控参数，结合接口dcf_pause_rep使用

• int dcf_get_param(const char *param_name, const char *param_value, unsigned int size);

功能说明：设置DCF配置参数
参数说明：param_name是需要设置的参数名称，参数名称如dcf_set_param中参数param_name一致
param_value是获取的参数值，需提前分配内存
size是param_value的大小

• int dcf_register_after_writer(usr_cb_after_writer_t cb_func);

功能说明：注册leader节点写入数据成功的回调函数
参数说明：回调函数形式如下，其中stream_id是分组编号，相同编号的组成一个一致性group;
index是落盘日志的index; buf是落盘的日志buf; size是落盘的日志size;
key是落盘日志的key，可以唯一标识一条日志; error_no是错误码
typedef int (*usr_cb_after_writer_t)(unsigned int stream_id, unsigned long long index,
const char *buf, unsigned int size, unsigned long long key, int error_no);

• int dcf_register_consensus_notify(usr_cb_consensus_notify_t cb_func);

功能说明：注册follower节点写入数据成功的回调函数
参数说明：回调函数形式如下，参数解释同上
typedef int (*usr_cb_consensus_notify_t)(unsigned int stream_id, unsigned long long index,
const char *buf, unsigned int size, unsigned long long key);

• int dcf_register_status_notify(usr_cb_status_notify_t cb_func);

功能说明：注册节点角色变化的回调函数
参数说明：回调函数形式如下，new_role是节点新角色
typedef int (*usr_cb_status_notify_t)(unsigned int stream_id, dcf_role_t new_role);

• int dcf_register_log_output(usr_cb_log_output_t cb_func);

功能说明：注册日志输出的回调函数
参数说明：回调函数形式如下，log_type是日志类型，LOG_RUN、LOG_DEBUG等; log_level是日志级别，LEVEL_ERROR、LEVEL_WARN等;
code_file_name是代码文件名，如__FILE__; code_line_num是代码行号，如__LINE__;
module_name是模块名，如”DCF”; format, …是格式化字符串
typedef void (*usr_cb_log_output_t)(int log_type, int log_level, const char *code_file_name,
unsigned int code_line_num, const char *module_name, const char *format, …);

• int dcf_register_exception_report(usr_cb_exception_notify_t cb_func);

功能说明：注册异常处理函数
参数说明：回调函数形式如下，dcf_exception_t异常类型，见dcf_interface.h中定义
typedef int(*usr_cb_exception_notify_t)(unsigned int stream_id, dcf_exception_t exception);

• int dcf_register_election_notify(usr_cb_election_notify_t cb_func);

功能说明：注册选举leader变化的回调函数
参数说明：回调函数形式如下，new_leader 是新主的nodeid
typedef int (*usr_cb_election_notify_t)(unsigned int stream_id, unsigned int new_leader);

• int dcf_register_msg_proc(usr_cb_msg_proc_t cb_func);

功能说明：注册选举leader变化的回调，follower调用函数
参数说明：回调函数形式如下，
typedef int (usr_cb_msg_proc_t)(unsigned int stream_id, unsigned int src_node_id, const char msg,
unsigned int msg_size);

• int dcf_start(unsigned int node_id, const char *cfg_str);

功能说明：启动工作线程
参数说明：node_id是节点id; cfg_str是集群节点列表，按照json字符串的格式进行配置，每个json item的配置信息包括stream_id/node_id/ip/port/role;
例如三个节点”[{
“stream_id”:1,
“node_id”:1,
“ip”:”127.0.0.1”,
“port”:1711,
“role”:”LEADER”
},{
“stream_id”:1,
“node_id”:2,
“ip”:”127.0.0.1”,
“port”:1712,
“role”:”FOLLOWER”
},{
“stream_id”:1,
“node_id”:3,
“ip”:”127.0.0.1”,
“port”:1713,
“role”:”FOLLOWER”
}]”

• int dcf_write(unsigned int stream_id, const char* buffer, unsigned int length, unsigned long long key, unsigned long long *index);

功能说明：写入数据，仅leader节点调用
参数说明：buffer是待写入数据的buffer; length是待写入数据的size; key是待写入数据的key，可以唯一标识一条日志; index是leader分配的日志index

• int dcf_read(unsigned int stream_id, unsigned long long index, char *buffer, unsigned int length);

功能说明：查询已写入的数据，成功返回实际读到的字节数，失败返回ERROR(-1)
参数说明：参考前述

• int dcf_stop();

功能说明：停止工作线程
参数说明：

• int dcf_truncate(unsigned int stream_id, unsigned long long first_index_kept);

功能说明：丢弃索引first_index_kept之前的日志
参数说明：first_index_kept是保留的第一个日志index

• int dcf_set_applied_index(unsigned int stream_id, unsigned long long index);

功能说明：设置applied index,在函数dcf_start调用前调用
参数说明：index是日志index

• int dcf_get_cluster_min_applied_idx(unsigned int stream_id, unsigned long long* index);

功能说明：获取集群所有节点最小的applied index
参数说明：*index是获取到的最小applied index

• int dcf_get_leader_last_index(unsigned int stream_id, unsigned long long* index);

功能说明：查询leader节点的last index
参数说明：返回值index为last index

• int dcf_get_last_index(unsigned int stream_id, unsigned long long* index);

功能说明：查询当前节点的last index
参数说明：返回值index为last index

• int dcf_get_node_last_disk_index(unsigned int stream_id, unsigned int node_id, unsigned long long* index);

功能说明：获取node_id节点的last disk index，只可在leader调用。成功返回SUCCESS，失败返回ERROR
参数说明：*index为获取到的last disk index。

• int dcf_query_cluster_info(char* buffer, unsigned int length);

功能说明：查询集群信息，streamlist、node等
参数说明：buffer是查询信息输出空间; length是最大输出长度; 函数返回值是实际输出长度
例如三个节点的cluster查询信息：
{
“local_node_id”:1,
“stream_list”:[{“stream_id”:1,”local_node_id”:1,”role”:”FOLLOWER”,”term”:3,”work_mode”:0,
“applied_index”:0,”commit_index”:0,”first_index”:1,”last_index”:5733936,
“leader_id”:3,”leader_ip”:”127.0.0.1”,”leader_port”:1713,
“nodes”:[{“node_id”:1,”ip”:”127.0.0.1”,”port”:1711,”role”:”FOLLOWER”},
{“node_id”:2,”ip”:”127.0.0.1”,”port”:1712,”role”:”FOLLOWER”},
{“node_id”:3,”ip”:”127.0.0.1”,”port”:1713,”role”:”LEADER”}]
}]
}

• int dcf_query_stream_info(unsigned int stream_id, char *buffer, unsigned int length);

功能说明：查询stream信息
参数说明：stream_id是待查询stream的id; buffer是查询信息输出空间; length是最大输出长度; 函数返回值是实际输出长度
例如三个节点的stream查询信息：
{
“stream_id”:1,”local_node_id”:3,”role”:”FOLLOWER”,”term”:2,”work_mode”:0,
“applied_index”:0,”commit_index”:0,”first_index”:1,”last_index”:0,
“leader_id”:2,”leader_ip”:”127.0.0.1”,”leader_port”:1712,
“nodes”:[{“node_id”:1,”ip”:”127.0.0.1”,”port”:1711,”role”:”FOLLOWER”},
{“node_id”:2,”ip”:”127.0.0.1”,”port”:1712,”role”:”LEADER”},
{“node_id”:3,”ip”:”127.0.0.1”,”port”:1713,”role”:”FOLLOWER”}]
}

• int dcf_query_leader_info(unsigned int stream_id, char *ip, unsigned int ip_len, unsigned int *port, unsigned int *node_id);

功能说明：查询leader信息
参数说明：ip是输出leader ip的buffer; ip_len是ip buffer长度; port输出leader的port; node_id输出leader的node_id

• int dcf_get_errorno();

功能说明：获取错误码
参数说明：

• const char* dcf_get_error(int code);

功能说明：获取错误信息
参数说明：code错误码

• const char *dcf_get_version();

功能说明：获取版本信息
参数说明：

• int dcf_add_member(unsigned int stream_id, unsigned int node_id, const char *ip, unsigned int port, dcf_role_t role, unsigned int wait_timeout_ms);

功能说明：添加节点，只可在leader调用。成功返回SUCCESS(0)，失败返回ERROR(-1)，超时返回TIMEOUT(1),超时最终也可能成功，可以重试。
参数说明：node_id是待添加节点id; ip是待添加节点ip; port是待添加节点port，调用者需保证port可用; role是待添加节点角色; wait_timeout_ms是超时时间，单位ms

• int dcf_remove_member(unsigned int stream_id, unsigned int node_id, unsigned int wait_timeout_ms);

功能说明：删除节点，只可在leader调用。成功返回SUCCESS(0)，失败返回ERROR(-1)，超时返回TIMEOUT(1),超时最终也可能成功，可以重试。
参数说明：node_id是待删除节点id; wait_timeout_ms是超时时间，单位ms

• int dcf_change_member_role(unsigned int stream_id, unsigned int node_id, dcf_role_t new_role, unsigned int wait_timeout_ms);

功能说明：改变节点角色，在leader调用可改变其他节点角色，在follower节点调用只能改变自己角色。成功返回SUCCESS(0)，失败返回ERROR(-1)，超时返回TIMEOUT(1),超时最终也可能成功，可以重试。
参数说明：node_id为被修改角色节点id; new_role是节点新角色

• int dcf_promote_leader(unsigned int stream_id, unsigned int node_id, unsigned int wait_timeout_ms);

功能说明：推选指定节点为leader。在leader调用可推选其他节点，在follower节点调用只能推选自己。
参数说明：node_id为被推选节点id; wait_timeout_ms是超时时间，单位ms，为0表示不阻塞leader直接发起推选。

• int dcf_timeout_notify(unsigned int stream_id, unsigned int node_id);

功能说明：外部触发超时
参数说明：stream_id≠0表示触发指定stream_id超时，stream_id=0表示触发所有stream_id超时

int int dcf_set_work_mode(unsigned int stream_id, dcf_work_mode_t work_mode, unsigned int vote_num)；

功能说明：设置运行模式（正常、少数派）
参数说明：work_mode为正常或少数派，如果是少数派模式，需指定票数。

• int dcf_query_statistics_info(char *buffer, unsigned int length);

功能说明：获取统计信息，需要日志级别开启PROFILE。
参数说明：buffer是查询信息输出空间; length是最大输出长度

• int dcf_check_if_all_logs_applied(unsigned int stream_id, unsigned int *all_applied);

功能说明：一般在节点升主时使用，检查当前欲升主节点的dcf日志是否都完成apply。调用成功返回SUCCESS，失败返回ERROR，调用成功后可从all_applied获取结果。
参数说明：all_applied为获取到的结果，0表示日志没有都完成apply，非0表示日志都完成apply。

• int dcf_send_msg(unsigned int stream_id, unsigned int dest_node_id, const char* msg, unsigned int msg_size);

功能说明：用于节点间对指定节点发送消息。调用成功返回SUCCESS，失败返回ERROR。
参数说明：dest_node_id为指定节点，msg表示待发送的消息，msg_size表示消息大小。

• int dcf_broadcast_msg(unsigned int stream_id, const char* msg, unsigned int msg_size);

功能说明：用于对除当前节点外所有节点广播发送消息。调用成功返回SUCCESS，失败返回ERROR。
参数说明：msg表示待发送的消息，msg_size表示消息大小。

• int dcf_pause_rep(unsigned int stream_id, unsigned int node_id, unsigned int time_us);

功能说明：对指定节点暂停日志复制。调用成功返回SUCCESS，失败返回ERROR。
参数说明：node_id指定暂停的节点; time_us是暂停时间(不超过1s)，单位us。

2、DEMO示例

参见：DCF/test/test_main目录

---

四、测试工程

1、编译

2、执行测试用例

待续…

---

五、应用实例

1、GaussDB(for openGauss)使能paxos特性实践

具体可参考：https://gitee.com/opengauss/blog/blob/master/content/zh/post/yanghaiyan/openGauss%E4%BD%BF%E8%83%BDpaxos%E7%89%B9%E6%80%A7%E5%AE%9E%E8%B7%B5.md