当前位置：首页 > news >正文

GPDB - 高可用 - 流复制状态

news 来源：原创 2024/5/14 13:46:32

GPDB - 高可用 - 流复制状态

GPDB的高可用基于流复制，通过FTS进行自动故障切换。自动故障切换需要根据primary-mirror流复制的各种状态进行判断。本节就聊聊primary-mirror流复制的各种状态。同样适用于PgSQL

1、WalSndState

typedef enum WalSndState
{WALSNDSTATE_STARTUP = 0,WALSNDSTATE_BACKUP,WALSNDSTATE_CATCHUP,WALSNDSTATE_STREAMING,WALSNDSTATE_STOPPING
} WalSndState;

WalSndState保存的是wal sender进程的状态信息，变量值如上代码。

WALSNDSTATE_STARTUP表示启动状态；

WALSNDSTATE_BACKUP表示备份状态

WALSNDSTATE_CATCHUP表示追赶状态

WALSNDSTATE_STREAMING表示流复制状态

WALSNDSTATE_STOPPING表示wal sender即将退出

2、什么时候切换到WALSNDSTATE_STOPPING

1）集群shutdown有三种方式：smart、fast、immediate

三种标记值分别为：

#define      SmartShutdown  1
#define      FastShutdown  2
#define      ImmediateShutdown  3

Smart shutdown：不允许有新连接，待已有连接全部结束后关闭数据库；

Fast shutdown：不允许新连接，向所有活跃的服务进程发送SIGTERM信号，让他们立即退出，之后等待所有子进程退出并关闭数据库

Immediate shutdown：不允许新连接，主进程postgres向所有子进程发送SIGQUIT信号并立即退出，所有子进程也会立即退出。下次启动会回放WAL日志进行恢复。

2）如果shutdown模式不为immediate，则集群shutdown的时候，postgres主进程会向checkpoint进程发送SIGUSR2信号：

3）checkpoint进程的SIGUSR2信号处理函数为ReqShutdownHandler，从上图的代码逻辑可见，ReqShutdownHandler会将shutdown_requested置为true，并唤醒MyLatch。

4）checkpoint进程接着调用ShutdownXLog，然后proc_exit(0)退出checkpoint进程。

5）ShutdownXLog函数调用WalSndInitStopping向所有sender进程发送SIGUSR1信号；然后调用WalSndWaitStopping等待所有sender进程退出，每个10ms判断一次。

6）sender进程SIGUSR1信号处理函数procsignal_sigusr1_handler检查信号来自PROCSIG_WALSND_INIT_STOPPING，然后将got_STOPPING置为true

7）流复制的sender处理完SIGUSR1信号后，继续返回信号前处理流程。Sender的发送日志函数为XLogSendPhysical，此时got_STOOPING已为true，所以调用WalSndSetState将walsnd->state切换到WALSNDSTATE_STOPPING状态，然后调用FTSReplicationStatusUpdateForWalState更新WAL复制状态

8）另外当sender进程从WalSndLoop退出后（replication_active置为false），这个时候，Wal sender进程才接收到信号，HandleWalSndInitStopping中也可以看到，会向自己发送SIGTERM信号，信号处理函数die，即退出进程（因为流复制终止了，不必管它了）。

9）若，sender进程还没从WalSndLoop退出（replication_active置为true），这个时候，Wal sender进程接收到信号，HandleWalSndInitStopping中也可以看到，他会设置got_STOPPING为true，让WAL sender进程发送完WAL后退出WalSndLoop循环后调用proc_exit自行退出。

2、sender进程什么时候退出？

书接上文，产生个问题：WalSndLoop何时退出？若没有shutdown，何时再发起流复制？

Wal sender进程接收到mirror发来的start replication命令后，进入StartReplication开始流复制。

1）WalSndLoop循环中，通过XLogSendPhysical函数不断发送WAL

2）XLogSendPhysical函数发送WAL达到一个时间线的末尾节点位置时，向mirror的receiver进程发送CopyDone消息，即开头为‘c’的消息，并将streamingDoneSending变量改为true

3）receiver进程的入口函数WalReceiverMain，通过walrcv_receive::libpqrcv_receive不断接收WAL日志和消息。当接收到发来的CopyDone消息后返回-1

4）接着，返回到WalReceiverMain函数中，当walrcv_receive返回-1后，一路下来会退出接收消息和日志的循环，并进入walrcv_endstreaming再向primary发送个CopyDone消息

5）primary的ProcessRepliesIfAny处理mirror发来的消息，当接收到CopyDone消息后，将streamingDoneReceiving改为true

6）返回WalSndLoop循环，当streamingDoneSending和streamingDoneReceiving都为true时退出循环

总结一句话：primary发完一个时间线内的WAL，切换下一个时间线时，会退出发送WAL日志的循环stop streaming；当然mirror的receiver进程发起下一个时间线的日志拉取，即再次调用libpqrcv_startstreaming函数向primary发送START_REPLICATION命令后，primary仍旧会再次进入WalSndLoop循环发送WAL日志。

3、什么时候进入WALSNDSTATE_BACKUP？

exec_replication_command：进行基础备份的时候

exec_replication_command：进行基础备份的时候switch (cmd_node->type){case T_BaseBackupCmd:PreventInTransactionBlock(true, "BASE_BACKUP");SendBaseBackup((BaseBackupCmd *) cmd_node);|  parse_basebackup_options(cmd->options, &opt);|  WalSndSetState(WALSNDSTATE_BACKUP);|  perform_base_backup(&opt);break;...}

进行基础备份，也就是构建mirror的时候进入该状态。

4、什么时候进入WALSNDSTATE_STARTUP？

1）sender进程刚fork出来，InitWalSenderSlot初始化的时候

2）WalSndLoop进程退出后又进入startup状态，因为下个时间线的复制即将开始

3）sender进程遇到ERROR故障，跳回到PostgresMain回退操作处，回退事务后，进入WalSndErrorCleanup，若没有stop则重新设置为startup状态，等待接收start replication命令重新开始复制。

PostgresMainif (am_walsender)InitWalSender();//sender进程的初始化|--  InitWalSenderSlot|--  for (i = 0; i < max_wal_senders; i++){|    WalSnd     *walsnd = &WalSndCtl->walsnds[i];|    SpinLockAcquire(&walsnd->mutex);|    if (walsnd->pid != 0){|      //找一个空闲的slot|      SpinLockRelease(&walsnd->mutex);|      continue;|    }else{|      walsnd->pid = MyProcPid;|      walsnd->state = WALSNDSTATE_STARTUP;|      ...|      break;|    }|  }|--  on_shmem_exit(WalSndKill, 0);

StartReplication:sender的WalSndLoop退出后又进入startup状态WalSndLoop(XLogSendLogical);...if (got_STOPPING)proc_exit(0);WalSndSetState(WALSNDSTATE_STARTUP);EndCommand("COPY 0", DestRemote);

PostgresMain//sender进程遇到ERROR报错,sender进程需要再次start replication才能进入传输walif (sigsetjmp(local_sigjmp_buf, 1) != 0){AbortCurrentTransaction();if (am_walsender)WalSndErrorCleanup();|--  if (got_STOPPING || got_SIGUSR2)|    proc_exit(0);|--  WalSndSetState(WALSNDSTATE_STARTUP);...for (;;){firstchar = ReadCommand(&input_message);switch (firstchar){case 'Q':{if (am_walsender){if (!exec_replication_command(query_string))exec_simple_query(query_string);}else if (am_ftshandler)HandleFtsMessage(query_string);else if (am_faulthandler)HandleFaultMessage(query_string);elseexec_simple_query(query_string);send_ready_for_query = true;break;}case 'M': ...}}

5、什么时候进入WALSNDSTATE_CATCHUP？

开始流复制前，设置成catchup状态。

StartReplication:开始流复制前WalSndSetState(WALSNDSTATE_CATCHUP);/* Send a CopyBothResponse message, and start streaming */pq_beginmessage(&buf, 'W');pq_sendbyte(&buf, 0);pq_sendint16(&buf, 0);pq_endmessage(&buf);pq_flush();WalSndLoop(XLogSendLogical);...

6、什么时候进入WALSNDSTATE_STREAMING？

当前时间线内没有要发送的日志了，并且没有下一个时间线需要切换发送日志，则将其改为streaming状态。

WalSndLoopfor (;;){if (!pq_is_send_pending())send_data();elseWalSndCaughtUp = false;...//现在没有要发送的了if (WalSndCaughtUp && !pq_is_send_pending()){if (MyWalSnd->state == WALSNDSTATE_CATCHUP)WalSndSetState(WALSNDSTATE_STREAMING);}...}