peerconn.go | 7 ++++--- test/issue377_test.go | 101 ++++++++++++++++++++++++++++++++++++++++++----------- torrent.go | 43 ++++++++++++++++++++++++++++++++++++++++++- diff --git a/peerconn.go b/peerconn.go index 9ff1a693c231d4a6ecef0a6f0ebe7889b2dc420a..ee5cdfd986216ee2509098a4d7ea6c496c969cd3 100644 --- a/peerconn.go +++ b/peerconn.go @@ -509,7 +509,7 @@ }) } func (cn *PeerConn) fillWriteBuffer(msg func(pp.Message) bool) { - if !cn.t.networkingEnabled { + if !cn.t.networkingEnabled || cn.t.dataDownloadDisallowed { if !cn.setInterested(false, msg) { return } @@ -1246,9 +1246,10 @@ piece.decrementPendingWrites() if err != nil { - panic(fmt.Sprintf("error writing chunk: %v", err)) + c.logger.Printf("error writing received chunk %v: %v", req, err) t.pendRequest(req) - t.updatePieceCompletion(pieceIndex(msg.Index)) + //t.updatePieceCompletion(pieceIndex(msg.Index)) + t.onWriteChunkErr(err) return nil } diff --git a/test/issue377_test.go b/test/issue377_test.go index 0c2e3837448d33ee24273b6697ec507ae4077318..40ed8916e0ddb72348156c3a94642d5dbd487ec9 100644 --- a/test/issue377_test.go +++ b/test/issue377_test.go @@ -2,7 +2,9 @@ package test import ( "errors" + "log" "os" + "sync" "testing" "github.com/stretchr/testify/assert" @@ -13,12 +15,18 @@ "github.com/anacrolix/torrent/internal/testutil" "github.com/anacrolix/torrent/metainfo" "github.com/anacrolix/torrent/storage" ) + +func justOneNetwork(cc *torrent.ClientConfig) { + cc.DisableTCP = true + cc.DisableIPv4 = true +} func TestReceiveChunkStorageFailure(t *testing.T) { seederDataDir, metainfo := testutil.GreetingTestTorrent() defer os.RemoveAll(seederDataDir) seederClientConfig := torrent.TestingConfig() seederClientConfig.Debug = true + justOneNetwork(seederClientConfig) seederClientStorage := storage.NewMMap(seederDataDir) defer seederClientStorage.Close() seederClientConfig.DefaultStorage = seederClientStorage @@ -29,57 +37,110 @@ require.NoError(t, err) defer testutil.ExportStatusWriter(seederClient, "s")() leecherClientConfig := torrent.TestingConfig() leecherClientConfig.Debug = true + justOneNetwork(leecherClientConfig) leecherClient, err := torrent.NewClient(leecherClientConfig) require.NoError(t, err) defer testutil.ExportStatusWriter(leecherClient, "l")() + info, err := metainfo.UnmarshalInfo() + require.NoError(t, err) + leecherStorage := diskFullStorage{ + pieces: make([]pieceState, info.NumPieces()), + data: make([]byte, info.TotalLength()), + } + defer leecherStorage.Close() leecherTorrent, new, err := leecherClient.AddTorrentSpec(&torrent.TorrentSpec{ InfoHash: metainfo.HashInfoBytes(), - Storage: diskFullStorage{}, + Storage: &leecherStorage, }) + leecherStorage.t = leecherTorrent require.NoError(t, err) assert.True(t, new) seederTorrent, err := seederClient.AddTorrent(metainfo) require.NoError(t, err) // Tell the seeder to find the leecher. Is it guaranteed seeders will always try to do this? seederTorrent.AddClientPeer(leecherClient) - //leecherTorrent.AddClientPeer(seederClient) <-leecherTorrent.GotInfo() assertReadAllGreeting(t, leecherTorrent.NewReader()) } -type diskFullStorage struct{} +type pieceState struct { + complete bool +} + +type diskFullStorage struct { + pieces []pieceState + t *torrent.Torrent + defaultHandledWriteChunkError bool + data []byte + + mu sync.Mutex + diskNotFull bool +} + +func (me *diskFullStorage) Piece(p metainfo.Piece) storage.PieceImpl { + return pieceImpl{ + mip: p, + diskFullStorage: me, + } +} -func (me diskFullStorage) ReadAt(p []byte, off int64) (n int, err error) { - panic("implement me") +func (me diskFullStorage) Close() error { + return nil } -func (me diskFullStorage) WriteAt(p []byte, off int64) (n int, err error) { - return 1, errors.New("disk full") +func (d diskFullStorage) OpenTorrent(info *metainfo.Info, infoHash metainfo.Hash) (storage.TorrentImpl, error) { + return &d, nil +} + +type pieceImpl struct { + mip metainfo.Piece + *diskFullStorage } -func (me diskFullStorage) MarkComplete() error { - panic("implement me") +func (me pieceImpl) state() *pieceState { + return &me.diskFullStorage.pieces[me.mip.Index()] } -func (me diskFullStorage) MarkNotComplete() error { - panic("implement me") +func (me pieceImpl) ReadAt(p []byte, off int64) (n int, err error) { + off += me.mip.Offset() + return copy(p, me.data[off:]), nil } -func (me diskFullStorage) Completion() storage.Completion { - return storage.Completion{ - Complete: false, - Ok: true, +func (me pieceImpl) WriteAt(p []byte, off int64) (int, error) { + off += me.mip.Offset() + if !me.defaultHandledWriteChunkError { + go func() { + me.t.SetOnWriteChunkError(func(err error) { + log.Printf("got write chunk error to custom handler: %v", err) + me.mu.Lock() + me.diskNotFull = true + me.mu.Unlock() + me.t.AllowDataDownload() + }) + me.t.AllowDataDownload() + }() + me.defaultHandledWriteChunkError = true } + me.mu.Lock() + defer me.mu.Unlock() + if me.diskNotFull { + return copy(me.data[off:], p), nil + } + return copy(me.data[off:], p[:1]), errors.New("disk full") } -func (me diskFullStorage) Piece(metainfo.Piece) storage.PieceImpl { - return me +func (me pieceImpl) MarkComplete() error { + me.state().complete = true + return nil } -func (me diskFullStorage) Close() error { +func (me pieceImpl) MarkNotComplete() error { panic("implement me") } -func (d diskFullStorage) OpenTorrent(info *metainfo.Info, infoHash metainfo.Hash) (storage.TorrentImpl, error) { - return d, nil +func (me pieceImpl) Completion() storage.Completion { + return storage.Completion{ + Complete: me.state().complete, + Ok: true, + } } diff --git a/torrent.go b/torrent.go index 74688c0b2614b947206bacc5d309e2eabcf46587..cd136523f0d50c72a6163d2d93626569c0250f66 100644 --- a/torrent.go +++ b/torrent.go @@ -40,7 +40,9 @@ stats ConnStats cl *Client logger log.Logger - networkingEnabled bool + networkingEnabled bool + dataDownloadDisallowed bool + userOnWriteChunkErr func(error) // Determines what chunks to request from peers. requestStrategy requestStrategy @@ -1810,3 +1812,42 @@ func (t *Torrent) requestStrategyCallbacks() requestStrategyCallbacks { return torrentRequestStrategyCallbacks{t} } + +func (t *Torrent) onWriteChunkErr(err error) { + if t.userOnWriteChunkErr != nil { + go t.userOnWriteChunkErr(err) + return + } + t.disallowDataDownloadLocked() +} + +func (t *Torrent) DisallowDataDownload() { + t.cl.lock() + defer t.cl.unlock() + t.disallowDataDownloadLocked() +} + +func (t *Torrent) disallowDataDownloadLocked() { + log.Printf("disallowing data download") + t.dataDownloadDisallowed = true + for c := range t.conns { + c.updateRequests() + } +} + +func (t *Torrent) AllowDataDownload() { + t.cl.lock() + defer t.cl.unlock() + log.Printf("AllowDataDownload") + t.dataDownloadDisallowed = false + for c := range t.conns { + c.updateRequests() + } + +} + +func (t *Torrent) SetOnWriteChunkError(f func(error)) { + t.cl.lock() + defer t.cl.unlock() + t.userOnWriteChunkErr = f +}