Do t.storage.Flush() on torrent completion, and on storage.Close()

bad_storage.go

@@ -17,6 +17,7 @@ var _ storage.ClientImpl = badStorage{}
 func (bs badStorage) OpenTorrent(*metainfo.Info, metainfo.Hash) (storage.TorrentImpl, error) {
 	return storage.TorrentImpl{
 		Piece: bs.Piece,
+		Flush: func() error { return nil },
 	}, nil
 }
 

mmap_span/mmap_span.go

@@ -19,11 +19,30 @@ func (ms *MMapSpan) Append(mMap mmap.MMap) {
 	ms.mMaps = append(ms.mMaps, mMap)
 }
 
+func (ms *MMapSpan) Flush() (errs []error) {
+	ms.mu.Lock()
+	defer ms.mu.Unlock()
+	for _, mMap := range ms.mMaps {
+		err := mMap.Flush()
+		if err != nil {
+			errs = append(errs, err)
+		}
+	}
+	// This is for issue 211.
+	ms.mMaps = nil
+	ms.InitIndex()
+	return
+}
+
 func (ms *MMapSpan) Close() (errs []error) {
 	ms.mu.Lock()
 	defer ms.mu.Unlock()
 	for _, mMap := range ms.mMaps {
-		err := mMap.Unmap()
+		err := mMap.Flush()
+		if err != nil {
+			errs = append(errs, err)
+		}
+		err = mMap.Unmap()
 		if err != nil {
 			errs = append(errs, err)
 		}

peerconn_test.go

@@ -102,7 +102,7 @@ func BenchmarkConnectionMainReadLoop(b *testing.B) {
 		Length:      1 << 20,
 		PieceLength: 1 << 20,
 	}))
-	t.storage = &storage.Torrent{TorrentImpl: storage.TorrentImpl{Piece: ts.Piece, Close: ts.Close}}
+	t.storage = &storage.Torrent{TorrentImpl: storage.TorrentImpl{Piece: ts.Piece, Close: ts.Close, Flush: func() error { return nil }}}
 	t.onSetInfo()
 	t._pendingPieces.Add(0)
 	r, w := net.Pipe()

storage/bolt.go

@@ -47,6 +47,7 @@ func (me *boltClient) OpenTorrent(_ *metainfo.Info, infoHash metainfo.Hash) (Tor
 	return TorrentImpl{
 		Piece: t.Piece,
 		Close: t.Close,
+		Flush: t.Flush,
 	}, nil
 }
 
@@ -62,3 +63,4 @@ func (me *boltTorrent) Piece(p metainfo.Piece) PieceImpl {
 }
 
 func (boltTorrent) Close() error { return nil }
+func (boltTorrent) Flush() error { return nil }

storage/disabled/disabled.go

@@ -20,6 +20,9 @@ func (c Client) OpenTorrent(info *metainfo.Info, infoHash metainfo.Hash) (storag
 		Close: func() error {
 			return nil
 		},
+		Flush: func() error {
+			return nil
+		},
 		Capacity: &capFunc,
 	}, nil
 }

storage/file.go

@@ -90,6 +90,7 @@ func (fs fileClientImpl) OpenTorrent(info *metainfo.Info, infoHash metainfo.Hash
 	return TorrentImpl{
 		Piece: t.Piece,
 		Close: t.Close,
+		Flush: t.Flush,
 	}, nil
 }
 
@@ -122,6 +123,10 @@ func (fs *fileTorrentImpl) Close() error {
 	return nil
 }
 
+func (fs *fileTorrentImpl) Flush() error {
+	return nil
+}
+
 // A helper to create zero-length files which won't appear for file-orientated storage since no
 // writes will ever occur to them (no torrent data is associated with a zero-length file). The
 // caller should make sure the file name provided is safe/sanitized.

storage/interface.go

@@ -22,6 +22,7 @@ type TorrentCapacity *func() (cap int64, capped bool)
 type TorrentImpl struct {
 	Piece func(p metainfo.Piece) PieceImpl
 	Close func() error
+	Flush func() error
 	// Storages that share the same space, will provide equal pointers. The function is called once
 	// to determine the storage for torrents sharing the same function pointer, and mutated in
 	// place.

storage/mmap.go

@@ -41,7 +41,7 @@ func (s *mmapClientImpl) OpenTorrent(info *metainfo.Info, infoHash metainfo.Hash
 		span:     span,
 		pc:       s.pc,
 	}
-	return TorrentImpl{Piece: t.Piece, Close: t.Close}, err
+	return TorrentImpl{Piece: t.Piece, Close: t.Close, Flush: t.Flush}, err
 }
 
 func (s *mmapClientImpl) Close() error {
@@ -71,6 +71,13 @@ func (ts *mmapTorrentStorage) Close() error {
 	}
 	return nil
 }
+func (ts *mmapTorrentStorage) Flush() error {
+	errs := ts.span.Flush()
+	if len(errs) > 0 {
+		return errs[0]
+	}
+	return nil
+}
 
 type mmapStoragePiece struct {
 	pc PieceCompletionGetSetter

storage/piece-resource.go

@@ -49,12 +49,16 @@ func (piecePerResourceTorrentImpl) Close() error {
 	return nil
 }
 
+func (piecePerResourceTorrentImpl) Flush() error {
+	return nil
+}
+
 func (s piecePerResource) OpenTorrent(info *metainfo.Info, infoHash metainfo.Hash) (TorrentImpl, error) {
 	t := piecePerResourceTorrentImpl{
 		s,
 		make([]sync.RWMutex, info.NumPieces()),
 	}
-	return TorrentImpl{Piece: t.Piece, Close: t.Close}, nil
+	return TorrentImpl{Piece: t.Piece, Close: t.Close, Flush: t.Flush}, nil
 }
 
 func (s piecePerResourceTorrentImpl) Piece(p metainfo.Piece) PieceImpl {

storage/sqlite/direct.go

@@ -40,7 +40,7 @@ type client struct {
 
 func (c *client) OpenTorrent(*metainfo.Info, metainfo.Hash) (storage.TorrentImpl, error) {
 	t := torrent{c.Cache}
-	return storage.TorrentImpl{Piece: t.Piece, Close: t.Close, Capacity: &c.capacity}, nil
+	return storage.TorrentImpl{Piece: t.Piece, Close: t.Close, Capacity: &c.capacity, Flush: t.Flush}, nil
 }
 
 type torrent struct {
@@ -59,6 +59,9 @@ func (t torrent) Piece(p metainfo.Piece) storage.PieceImpl {
 func (t torrent) Close() error {
 	return nil
 }
+func (t torrent) Flush() error {
+	return nil
+}
 
 type piece struct {
 	sb squirrel.Blob

test/issue377_test.go

@@ -126,8 +126,12 @@ func (me *diskFullStorage) Close() error {
 	return nil
 }
 
+func (me *diskFullStorage) Flush() error {
+	return nil
+}
+
 func (d *diskFullStorage) OpenTorrent(info *metainfo.Info, infoHash metainfo.Hash) (storage.TorrentImpl, error) {
-	return storage.TorrentImpl{Piece: d.Piece, Close: d.Close}, nil
+	return storage.TorrentImpl{Piece: d.Piece, Close: d.Close, Flush: d.Flush}, nil
 }
 
 type pieceImpl struct {

torrent.go

@@ -2461,7 +2461,11 @@ func (t *Torrent) pieceRequestIndexOffset(piece pieceIndex) RequestIndex {
 }
 
 func (t *Torrent) updateComplete() {
+	before := t.Complete.Bool()
 	t.Complete.SetBool(t.haveAllPieces())
+	if !before && t.Complete.Bool() {
+		_ = t.storage.Flush()
+	}
 }
 
 func (t *Torrent) cancelRequest(r RequestIndex) *Peer {