Postgresql中xlog生成和清理逻辑操作

# 在自动的WAL检查点之间的日志文件段的最大数量

checkpoint_segments =

# 在自动WAL检查点之间的最长时间

checkpoint_timeout =

# 缓解io压力

checkpoint_completion_target =

# 日志文件段的保存最小数量，为了备库保留更多段

wal_keep_segments =

# 已完成的WAL段通过archive_command发送到归档存储

archive_mode =

# 强制timeout切换到新的wal段文件

archive_timeout =

max_wal_size =

min_wal_size =

(2 + checkpoint_completion_target) * checkpoint_segments + 1

RemoveOldXlogFiles

> CreateCheckPoint

> CreateRestartPoint

static void

KeepLogSeg(XLogRecPtr recptr, XLogSegNo *logSegNo)

{

  XLogSegNo segno;

  XLogRecPtr keep;

  XLByteToSeg(recptr, segno);

  keep = XLogGetReplicationSlotMinimumLSN();

  /* compute limit for wal_keep_segments first */

  if (wal_keep_segments > 0)

  {

  /* avoid underflow, don 't go below 1 */

  if (segno <= wal_keep_segments)

   segno = 1;

  else

   segno = segno - wal_keep_segments;

  }

  /* then check whether slots limit removal further */

  if (max_replication_slots > 0 && keep != InvalidXLogRecPtr)

  {

  XLogSegNo slotSegNo;

  XLByteToSeg(keep, slotSegNo);

  if (slotSegNo <= 0)

   segno = 1;

  else if (slotSegNo < segno)

   segno = slotSegNo;

  }

  /* don' t delete WAL segments newer than the calculated segment */

  if (segno < *logSegNo)

  *logSegNo = segno;

}

static void

RemoveOldXlogFiles(XLogSegNo segno, XLogRecPtr endptr)

{

   ...

   ...

  while ((xlde = ReadDir(xldir, XLOGDIR)) != NULL )

  {

  /* Ignore files that are not XLOG segments */

  if (strlen(xlde->d_name) != 24 ||

   strspn(xlde->d_name, "0123456789ABCDEF" ) != 24)

   continue ;

  /*

   * We ignore the timeline part of the XLOG segment identifiers in

   * deciding whether a segment is still needed. This ensures that we

   * won't prematurely remove a segment from a parent timeline. We could

   * probably be a little more proactive about removing segments of

   * non-parent timelines, but that would be a whole lot more

   * complicated.

   *

   * We use the alphanumeric sorting property of the filenames to decide

   * which ones are earlier than the lastoff segment.

   */

  if (strcmp(xlde->d_name + 8, lastoff + 8) <= 0)

  {

   if (XLogArchiveCheckDone(xlde->d_name))

         # 归档关闭返回真

         # 存在done文件返回真

         # 存在.ready返回假

         # recheck存在done文件返回真

         # 重建.ready文件返回假

   {

   /* Update the last removed location in shared memory first */

   UpdateLastRemovedPtr(xlde->d_name);

         # 回收 或者 直接删除，清理.done和.ready文件

   RemoveXlogFile(xlde->d_name, endptr);

   }

  }

  }

   ...

   ...

}

static void

pgarch_ArchiverCopyLoop(void)

{

  char xlog [MAX_XFN_CHARS + 1];

   # 拿到最老那个没有被归档的xlog文件名

  while (pgarch_readyXlog(xlog))

  {

  int   failures = 0;

  for (;;)

  {

   /*

   * Do not initiate any more archive commands after receiving

   * SIGTERM, nor after the postmaster has died unexpectedly. The

   * first condition is to try to keep from having init SIGKILL the

   * command, and the second is to avoid conflicts with another

   * archiver spawned by a newer postmaster.

   */

   if (got_SIGTERM || !PostmasterIsAlive())

   return ;

   /*

   * Check for config update . This is so that we'll adopt a new

   * setting for archive_command as soon as possible, even if there

   * is a backlog of files to be archived.

   */

   if (got_SIGHUP)

   {

   got_SIGHUP = false ;

   ProcessConfigFile(PGC_SIGHUP);

   }

   # archive_command没设的话不再执行

       # 我们的command没有设置，走的是这个分支

   if (!XLogArchiveCommandSet())

   {

   /*

    * Change WARNING to DEBUG1, since we will left archive_command empty to

    * let external tools to manage archive

    */

   ereport(DEBUG1,

    (errmsg( "archive_mode enabled, yet archive_command is not set" )));

   return ;

   }

       # 执行归档命令！

   if (pgarch_archiveXlog(xlog))

   {

   # 成功了，把.ready改名为.done

   pgarch_archiveDone(xlog);

   /*

    * Tell the collector about the WAL file that we successfully

    * archived

    */

   pgstat_send_archiver(xlog, false );

   break;  /* out of inner retry loop */

   }

   else

   {

   /*

    * Tell the collector about the WAL file that we failed to

    * archive

    */

   pgstat_send_archiver(xlog, true );

   if (++failures >= NUM_ARCHIVE_RETRIES)

   {

    ereport(WARNING,

     (errmsg( "archiving transaction log file \"%s\" failed too many times, will try again later" ,

      xlog)));

    return ; /* give up archiving for now */

   }

   pg_usleep(1000000L); /* wait a bit before retrying */

   }

  }

  }

}

static void

XLogWrite(XLogwrtRqst WriteRqst, bool flexible)

{

...

       if (finishing_seg)

   {

   issue_xlog_fsync(openLogFile, openLogSegNo);

   /* signal that we need to wakeup walsenders later */

   WalSndWakeupRequest();

   LogwrtResult.Flush = LogwrtResult.Write; /* end of page */

         # 归档打开 && wal_level >= archive

   if (XLogArchivingActive())

           # 生成ready文件

    XLogArchiveNotifySeg(openLogSegNo);

   XLogCtl->lastSegSwitchTime = (pg_time_t) time ( NULL );

...

-- 流复制插槽

-- 如果restart_lsn和当前XLOG相差非常大的字节数, 需要排查slot的订阅者是否能正常接收XLOG,

-- 或者订阅者是否正常. 长时间不将slot的数据取走, pg_xlog目录可能会撑爆

select pg_xlog_location_diff(pg_current_xlog_location(),restart_lsn), *

from pg_replication_slots;

select pg_drop_replication_slot( 'xxx' );

（archive_mode= on archive_command= '' ）