WAL parallel replay

Terms

• data block: A unit of data storage.

• WAL: An abbreviation for write-ahead logging.

• task node: A node in the parallel execution framework that receives and executes a single subtask.

• task tag: An identifier that categorizes subtasks. Subtasks with the same tag have execution order dependencies.

• hold list: A linked list that each child process in the parallel execution framework uses to schedule and execute replay subtasks.

How it works

• Overview

  A single WAL record can modify multiple data blocks.

  1. Assume that the i-th WAL log has an LSN of LSNi​ and modifies m data blocks. Then, the list of data blocks modified by the i-th WAL log is defined as Blocki​=[Blocki,0​,Blocki,1​,...,Blocki,m​].

  2. The minimum replay subtask is defined as Taski,j​=LSNi​−>Blocki,j​, which represents the replay of the i-th WAL log on the data block Blocki,j​.

  3. Therefore, a WAL log that modifies m blocks can be represented as a set of m replay sub-tasks: TASKi,∗​=[Taski,0​,Taski,1​,...,Taski,m​].

  4. Therefore, multiple WAL logs can be represented as a series of replay subtasks: TASK∗,∗​=[Task0,∗​,Task1,∗​,...,TaskN,∗​].

  In the log replay subtask set Task∗,∗​, the execution of each subtask is not always dependent on the execution result of a preceding subtask.

  Assume the set of replay subtasks is as follows: TASK∗,∗​=[Task0,∗​,Task1,∗​,Task2,∗​], where:

  • Task0,∗​=[Task0,0​,Task0,1​,Task0,2​]

  • Task1,∗​=[Task1,0​,Task1,1​]

  • Task2,∗​=[Task2,0​]

  And, Block_0,0=Block_1,0, Block_0,1=Block_1,1, and Block_0,2=Block_2,0.

  In this case, three sets of subtasks can be replayed in parallel: [Task_0,0,Task_1,0], [Task_0,1,Task_1,1], and [Task_0,2,Task_2,0].

  In summary, many subtask sequences within the complete set of WAL replay tasks can run in parallel without affecting the consistency of the final result. Based on this principle, PolarDB introduces a parallel task execution framework and applies it to the WAL replay process.

• Parallel task execution framework

  The framework divides a segment of shared memory equally among concurrent processes. Each resulting segment, allocated to one process, acts as a circular queue. You can configure the depth of each circular queue with parameters.

  • Dispatcher process

    • Controls concurrent scheduling by dispatching tasks to specific processes.

    • Removes completed tasks from the queue.

  • process pool

    Each process in the pool retrieves tasks from its corresponding circular queue and decides whether to execute them based on their state.

  • Task

    The circular queue contains task nodes. Each task node can be in one of five states: Idle, Running, Hold, Finished, or Removed.

    • Idle: The task node has no assigned task.

    • Running: The task node is assigned a task that is waiting for execution or is currently being executed.

    • Hold: The task depends on a preceding task and must wait for that task to finish.

    • Finished: The task has been completed by a process in the process pool.

    • Removed: When the Dispatcher process finds a task in the Finished state, it implies all its prerequisite tasks are also finished. The Dispatcher process then changes the state to Removed, indicating it has cleared the task and its prerequisites from the management structure. This process ensures that the Dispatcher handles dependent tasks in the correct order.

    The Dispatcher process performs transitions marked by black lines, while the parallel replay process pool performs those marked by orange lines.

  • Dispatcher process

    The Dispatcher process uses three key data structures: a Task HashMap, a Task Running Queue, and Task Idle Nodes.

    • The Task HashMap stores the hash mappings between task tags and their corresponding task execution lists.

      • Each task is assigned a specific task tag. If two tasks have a dependency, they share the same task tag.

      • When dispatching tasks, if a task node has a prerequisite, the Dispatcher process marks its state as Hold, causing it to wait until the prerequisite is executed.

    • The Task Running Queue tracks all tasks that are currently being executed.

    • Task Idle Nodes records the task nodes from different processes in the process pool that are currently in the Idle state.

    The Dispatcher process uses the following scheduling strategy:

    • If a task to be executed has the same task tag as another task already in execution, the new task is preferentially assigned to the process running the last task in that task tag's linked list. This strategy aims to keep dependent tasks on the same process to reduce inter-process synchronization overhead.

    • If the preferred process queue is full or no task with the same Task Tag is running, a process is selected sequentially from the process pool, and a task node in the Idle state is retrieved from that process to run the task. The purpose is to distribute tasks as evenly as possible among different processes.

    

  • process pool

    This parallel execution targets tasks of the same type that have the same task node data structure. When the process pool is initialized, configure SchedContext to specify the function pointer that is responsible for executing specific tasks:

    • TaskStartup: The initialization action required before a process can execute a task.

    • TaskHandler: The function responsible for executing a specific task based on the provided task node.

    • TaskCleanup: The cleanup action required before an execution process exits.

    

    A process in the process pool retrieves a task node from a circular queue. If the current status of the task node is Hold, the process inserts the task node at the tail of the hold list. If the status of the task node is Running, the process calls the TaskHandler to execute it. If the TaskHandler fails to execute, the process sets a default retry count of 3 for the task node and inserts it at the head of the hold list.

    The process first searches the hold list from the head to retrieve an executable Task. If the Task's status is Running and its waiting call count is 0, the process executes the Task. If the Task's status is Running but its waiting call count is greater than 0, the process decrements the waiting call count by 1.

• WAL parallel replay

  LogIndex data records the mapping between WAL records and the data blocks they modify, and it supports LSN-based lookups. During continuous WAL replay on a standby node, PolarDB introduces the parallel task execution framework and uses LogIndex data to parallelize WAL replay tasks, which speeds up data synchronization on the standby node.

  Workflow

  • The Startup process parses WAL records and builds LogIndex data but does not replay the WAL records themselves.

  • The LogIndex BGW background replay process acts as the Dispatcher process for the parallel task execution framework. It uses LSNs to retrieve LogIndex data, creates replay subtasks, and assigns them to the parallel replay process pool.

  • Processes within the parallel replay process pool execute replay subtasks, applying a single log record to a data block.

  • When a backend process actively reads a data block, it uses the page tag to look up the LogIndex data. It retrieves the linked list of LSNs that modified the data block and replays the full chain of log records on that block.

  • The Dispatcher process uses the LSN to retrieve LogIndex data, enumerates the PageTag and the corresponding LSN in the LogIndex insertion order, and constructs a {LSN -> PageTag} mapping to form the corresponding task node.

  • The page tag is used as the task tag for the task node.

  • The enumerated task nodes are then dispatched to the child processes in the parallel execution framework's process pool for replay.

Enable WAL parallel replay

polar_enable_parallel_replay_standby_mode = ON