Recover pipeline from brutal shutdown (checkpointing)

[chubei]

Precondition

DAG is partially consistent, which means

• For each node in the DAG, we're able to read its checkpoint.
• For each node in the DAG, its checkpoint is less than or equal to all its parents.
• We can find a source node whose checkpoint is greater than or euqal to all other source nodes.

Checkpoint

A checkpoint is a HashMap from source node handles to (txn id, seq number) pairs. We define checkpoint's compare function as follows

• If the keys are not exactly the same, ordering is undefined.
• For each key value pair, compare the values.
  • If all values compare Equal, return Equal.
  • If some or all values compare Less and other values compare Equal, return Less.
  • If some or all values compare Greater and other values compare Equal, return Greater.
  • Otherwise ordering is undefined.

Postcondition

DAG is fully consistent, which means

• All nodes' checkpoints are equal to that of the maximum node.

Algorithm

• Find the maximum source node, call its checkpoint target.
• Traverse all sources.
  • Check the (txn id, seq_number) pair of this source in its checkpoint, if it's less than that in target, run Source Replay subroutine.
  • Now this source's checkpoint is guaranteed to be target.
• Topologically sort the DAG, traverse all the nodes.
  • Now all the node's parents' checkpoints are guaranteed to be equal to target.
  • If the node's checkpoint is equal to target, skip.
  • Run Receiver Replay subroutine.
  • Now this node's checkpoint is guaranteed to be target.

Transaction Log

Source Replay and Receiver Replay subroutines read from and write to transaction logs.

Multiple transaction log entries are maintained for each output port. A transaction log entry looks like following

struct TransactionLogEntry {
  /// Exclusive.
  from: Checkpoint,
  /// Inclusive.
  to: Checkpoint,
  /// Data.
  ops: Vec<Operation>,
}

We guarantee entry.from < entry.to.

If we sort all the transaction log entries by from, we guarantee entries[i + 1].from == entries[i].to.

A node's checkpoint is entries.last().unwrap().to.

Source Replay

Input

• current checkpoint
• target checkpoint
• A dyn Source.

Precondition

• current < target.
• current contains an entry for this node. Call its value (current txn, current_seq).
• target contains an entry for this node. Call its values (target_txn, target_seq).

Postcondition

• Transaction log entry from current to target for all output ports are added to the database.

Algorithm

Start the source from (current txn, current_seq), when it reaches (target_txn, target_seq), commit the transaction log entry and stop the source.

Receiver Replay

Input

• A dyn Processor or dyn Sink.
• current checkpoint.
• target checkpoint.
• Transaction logs of input ports.

Precondition

• For each input port, there exists transaction log entries such that entries.first().unwrap().from == current, entries.last().unwrap().to == targetandentries[i + 1].from == entries[i].to. Call such log entries "replay entries from currenttotarget`".

Postcondition

• Transaction log entry from current to target for all output ports are added to the database.

Algorithm

• For each input port, find replay entries from current to target.
• Process ops in all the entries and commit the transaction log entry.

@snork-alt Check this?

[chubei]

Maybe this is unnecessarily complicated. With the EpochManager in place, nodes can differ by at most one commit.

[snork-alt]

@chubei Why not maintain a log of all put() and del() directly from the SharedTransaction? This way we do not need to serialize data twice since we already get the raw bytes written and deleted as arguments.

[chubei]

I think put and del of SharedTransaction are not only used for record writing? Meta data writes also go through SharedTransaction, and any node specific data. I was going to put the log implementation in RecordWriter. Can also avoid double serialization.

[snork-alt]

put() and del() are the only way for each processor to update their internal state. That's why I was thinking to put it there. Metadata of each node needs to be checkpointed as well. In any case, I think both approaches will work. My idea was to create a tx log of all put() and del() operations that are applied on the state, your idea is more about capturing the operations flowing into each port, that will eventually be passed to the processor and trigger the state updates (put() and del()). The end result is teh same

[chubei]

I don't understand how the put and del logs work. Say we have two processors p1 and p2, p1 is the sole parent of p2. Now we're in the situation where p1's checkpoint is src: 100, p2's checkpoint is src: 50. I believed that we need the output of p1 from src: 50 to src: 100 and let p2 process them. But you said we can make p2 catch up to src: 100 by only using the put and del logs of p1?