Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

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

Merged
merged 22 commits into from
Jul 7, 2022
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions mmap_span/mmap_span.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,18 @@ func (ms *MMapSpan) Append(mMap mmap.MMap) {
ms.mMaps = append(ms.mMaps, mMap)
}

func (ms *MMapSpan) Flush() (errs []error) {
ms.mu.RLock()
defer ms.mu.RUnlock()
for _, mMap := range ms.mMaps {
err := mMap.Flush()
if err != nil {
errs = append(errs, err)
}
}
return
}

func (ms *MMapSpan) Close() (errs []error) {
ms.mu.Lock()
defer ms.mu.Unlock()
Expand Down Expand Up @@ -69,6 +81,7 @@ func (ms *MMapSpan) locateCopy(copyArgs func(remainingArgument, mmapped []byte)
_n := copyBytes(copyArgs(p, mMapBytes))
p = p[_n:]
n += _n

if segments.Int(_n) != e.Length {
panic(fmt.Sprintf("did %d bytes, expected to do %d", _n, e.Length))
}
Expand Down
6 changes: 6 additions & 0 deletions piece.go
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,12 @@ func (p *Piece) Storage() storage.Piece {
return p.t.storage.Piece(p.Info())
}

func (p *Piece) Flush() {
if p.t.storage.Flush != nil {
_ = p.t.storage.Flush()
}
}

func (p *Piece) pendingChunkIndex(chunkIndex chunkIndexType) bool {
return !p.chunkIndexDirty(chunkIndex)
}
Expand Down
1 change: 1 addition & 0 deletions storage/interface.go
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
9 changes: 8 additions & 1 deletion storage/mmap.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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
Expand Down
4 changes: 4 additions & 0 deletions torrent.go
Original file line number Diff line number Diff line change
Expand Up @@ -2007,7 +2007,11 @@ func (t *Torrent) pieceHashed(piece pieceIndex, passed bool, hashIoErr error) {
c._stats.incrementPiecesDirtiedGood()
}
t.clearPieceTouchers(piece)
hasDirty := p.hasDirtyChunks()
t.cl.unlock()
if hasDirty {
p.Flush() // You can be synchronous here!
}
err := p.Storage().MarkComplete()
if err != nil {
t.logger.Printf("%T: error marking piece complete %d: %s", t.storage, piece, err)
Expand Down