Skip to content

Commit 9e01890

Browse files
committed
allow shutdown during network snapshot transfer
1 parent 0f873ea commit 9e01890

1 file changed

Lines changed: 44 additions & 27 deletions

File tree

raft.go

Lines changed: 44 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -1750,37 +1750,54 @@ 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 Spawn a goroutine to copy the
1754+
// snapshot to disk, so we can respond to the shutdown as well.
1755+
diskCopyErrCh := make(chan error, 1)
1756+
go func() {
1757+
// Separately track the progress of streaming a snapshot over the network
1758+
// because this too can take a long time.
1759+
countingRPCReader := newCountingReader(rpc.Reader)
1760+
transferMonitor := startSnapshotRestoreMonitor(r.logger, countingRPCReader, req.Size, true)
1761+
n, err := io.Copy(sink, countingRPCReader)
1762+
transferMonitor.StopAndWait()
1763+
if err != nil {
1764+
sink.Cancel()
1765+
r.logger.Error("failed to copy snapshot", "error", err)
1766+
diskCopyErrCh <- err
1767+
return
1768+
}
17671769

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

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

17851802
// Restore snapshot
17861803
future := &restoreFuture{ID: sink.ID()}

0 commit comments

Comments
 (0)