Skip to content

Commit 1017c47

Browse files
committed
allow shutdown during network snapshot transfer
1 parent 0f873ea commit 1017c47

1 file changed

Lines changed: 45 additions & 27 deletions

File tree

raft.go

Lines changed: 45 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -1750,37 +1750,55 @@ func (r *Raft) installSnapshot(rpc RPC, req *InstallSnapshotRequest) {
17501750
return
17511751
}
17521752

1753-
// Separately track the progress of streaming a snapshot over the network
1754-
// because this too can take a long time.
1755-
countingRPCReader := newCountingReader(rpc.Reader)
1756-
1757-
// Spill the remote snapshot to disk
1758-
transferMonitor := startSnapshotRestoreMonitor(r.logger, countingRPCReader, req.Size, true)
1759-
n, err := io.Copy(sink, countingRPCReader)
1760-
transferMonitor.StopAndWait()
1761-
if err != nil {
1762-
sink.Cancel()
1763-
r.logger.Error("failed to copy snapshot", "error", err)
1764-
rpcErr = err
1765-
return
1766-
}
1753+
// Spill the remote snapshot to disk.
1754+
// Spawn a goroutine to copy the snapshot, so we can handle a
1755+
// shutdown signal as well.
1756+
diskCopyErrCh := make(chan error, 1)
1757+
go func() {
1758+
// Separately track the progress of streaming a snapshot over the network
1759+
// because this too can take a long time.
1760+
countingRPCReader := newCountingReader(rpc.Reader)
1761+
transferMonitor := startSnapshotRestoreMonitor(r.logger, countingRPCReader, req.Size, true)
1762+
n, err := io.Copy(sink, countingRPCReader)
1763+
transferMonitor.StopAndWait()
1764+
if err != nil {
1765+
sink.Cancel()
1766+
r.logger.Error("failed to copy snapshot", "error", err)
1767+
diskCopyErrCh <- err
1768+
return
1769+
}
17671770

1768-
// Check that we received it all
1769-
if n != req.Size {
1770-
sink.Cancel()
1771-
r.logger.Error("failed to receive whole snapshot",
1772-
"received", hclog.Fmt("%d / %d", n, req.Size))
1773-
rpcErr = fmt.Errorf("short read")
1774-
return
1775-
}
1771+
// Check that we received it all
1772+
if n != req.Size {
1773+
sink.Cancel()
1774+
r.logger.Error("failed to receive whole snapshot",
1775+
"received", hclog.Fmt("%d / %d", n, req.Size))
1776+
diskCopyErrCh <- fmt.Errorf("short read")
1777+
return
1778+
}
17761779

1777-
// Finalize the snapshot
1778-
if err := sink.Close(); err != nil {
1779-
r.logger.Error("failed to finalize snapshot", "error", err)
1780-
rpcErr = err
1780+
// Finalize the snapshot
1781+
if err := sink.Close(); err != nil {
1782+
r.logger.Error("failed to finalize snapshot", "error", err)
1783+
diskCopyErrCh <- err
1784+
return
1785+
}
1786+
r.logger.Info("copied to local snapshot", "bytes", n)
1787+
diskCopyErrCh <- nil
1788+
}()
1789+
1790+
// Wait for snapshot transfer or shutdown
1791+
select {
1792+
case err := <-diskCopyErrCh:
1793+
if err != nil {
1794+
rpcErr = err
1795+
return
1796+
}
1797+
case <-r.shutdownCh:
1798+
sink.Cancel()
1799+
rpcErr = ErrRaftShutdown
17811800
return
17821801
}
1783-
r.logger.Info("copied to local snapshot", "bytes", n)
17841802

17851803
// Restore snapshot
17861804
future := &restoreFuture{ID: sink.ID()}

0 commit comments

Comments
 (0)