test/issue377_test.go

   1 package test
   2
   3 import (
   4         "errors"
   5         "io"
   6         "os"
   7         "sync"
   8         "testing"
   9         "testing/iotest"
  10
  11         "github.com/anacrolix/log"
  12         "github.com/stretchr/testify/assert"
  13         "github.com/stretchr/testify/require"
  14
  15         "github.com/anacrolix/torrent"
  16         "github.com/anacrolix/torrent/internal/testutil"
  17         "github.com/anacrolix/torrent/metainfo"
  18         pp "github.com/anacrolix/torrent/peer_protocol"
  19         "github.com/anacrolix/torrent/storage"
  20 )
  21
  22 func justOneNetwork(cc *torrent.ClientConfig) {
  23         cc.DisableTCP = true
  24         cc.DisableIPv4 = true
  25 }
  26
  27 func TestReceiveChunkStorageFailureSeederFastExtensionDisabled(t *testing.T) {
  28         testReceiveChunkStorageFailure(t, false)
  29 }
  30
  31 func TestReceiveChunkStorageFailure(t *testing.T) {
  32         testReceiveChunkStorageFailure(t, true)
  33 }
  34
  35 func testReceiveChunkStorageFailure(t *testing.T, seederFast bool) {
  36         seederDataDir, metainfo := testutil.GreetingTestTorrent()
  37         defer os.RemoveAll(seederDataDir)
  38         seederClientConfig := torrent.TestingConfig(t)
  39         seederClientConfig.Debug = true
  40         justOneNetwork(seederClientConfig)
  41         seederClientStorage := storage.NewMMap(seederDataDir)
  42         defer seederClientStorage.Close()
  43         seederClientConfig.DefaultStorage = seederClientStorage
  44         seederClientConfig.Seed = true
  45         seederClientConfig.Debug = true
  46         seederClientConfig.Extensions.SetBit(pp.ExtensionBitFast, seederFast)
  47         seederClient, err := torrent.NewClient(seederClientConfig)
  48         require.NoError(t, err)
  49         defer seederClient.Close()
  50         defer testutil.ExportStatusWriter(seederClient, "s", t)()
  51         leecherClientConfig := torrent.TestingConfig(t)
  52         leecherClientConfig.Debug = true
  53         // Don't require fast extension, whether the seeder will provide it or not (so we can test mixed
  54         // cases).
  55         leecherClientConfig.MinPeerExtensions.SetBit(pp.ExtensionBitFast, false)
  56         justOneNetwork(leecherClientConfig)
  57         leecherClient, err := torrent.NewClient(leecherClientConfig)
  58         require.NoError(t, err)
  59         defer leecherClient.Close()
  60         defer testutil.ExportStatusWriter(leecherClient, "l", t)()
  61         info, err := metainfo.UnmarshalInfo()
  62         require.NoError(t, err)
  63         leecherStorage := diskFullStorage{
  64                 pieces: make([]pieceState, info.NumPieces()),
  65                 data:   make([]byte, info.TotalLength()),
  66         }
  67         defer leecherStorage.Close()
  68         leecherTorrent, new, err := leecherClient.AddTorrentSpec(&torrent.TorrentSpec{
  69                 InfoHash: metainfo.HashInfoBytes(),
  70                 Storage:  &leecherStorage,
  71         })
  72         leecherStorage.t = leecherTorrent
  73         require.NoError(t, err)
  74         assert.True(t, new)
  75         seederTorrent, err := seederClient.AddTorrent(metainfo)
  76         require.NoError(t, err)
  77         // Tell the seeder to find the leecher. Is it guaranteed seeders will always try to do this?
  78         seederTorrent.AddClientPeer(leecherClient)
  79         <-leecherTorrent.GotInfo()
  80         r := leecherTorrent.Files()[0].NewReader()
  81         defer r.Close()
  82         // We can't use assertReadAllGreeting here, because the default storage write error handler
  83         // disables data downloads, which now causes Readers to error when they're blocked.
  84         if false {
  85                 assertReadAllGreeting(t, leecherTorrent.NewReader())
  86         } else {
  87                 for func() bool {
  88                         // We don't seem to need to seek, but that's probably just because the storage failure is
  89                         // happening on the first read.
  90                         r.Seek(0, io.SeekStart)
  91                         if err := iotest.TestReader(r, []byte(testutil.GreetingFileContents)); err != nil {
  92                                 t.Logf("got error while reading: %v", err)
  93                                 return true
  94                         }
  95                         return false
  96                 }() {
  97                 }
  98         }
  99         // TODO: Check that PeerConns fastEnabled matches seederFast?
 100         // select {}
 101 }
 102
 103 type pieceState struct {
 104         complete bool
 105 }
 106
 107 type diskFullStorage struct {
 108         pieces                        []pieceState
 109         t                             *torrent.Torrent
 110         defaultHandledWriteChunkError bool
 111         data                          []byte
 112
 113         mu          sync.Mutex
 114         diskNotFull bool
 115 }
 116
 117 func (me *diskFullStorage) Piece(p metainfo.Piece) storage.PieceImpl {
 118         return pieceImpl{
 119                 mip:             p,
 120                 diskFullStorage: me,
 121         }
 122 }
 123
 124 func (me *diskFullStorage) Close() error {
 125         return nil
 126 }
 127
 128 func (d *diskFullStorage) OpenTorrent(info *metainfo.Info, infoHash metainfo.Hash) (storage.TorrentImpl, error) {
 129         return storage.TorrentImpl{Piece: d.Piece, Close: d.Close}, nil
 130 }
 131
 132 type pieceImpl struct {
 133         mip metainfo.Piece
 134         *diskFullStorage
 135 }
 136
 137 func (me pieceImpl) state() *pieceState {
 138         return &me.diskFullStorage.pieces[me.mip.Index()]
 139 }
 140
 141 func (me pieceImpl) ReadAt(p []byte, off int64) (n int, err error) {
 142         off += me.mip.Offset()
 143         return copy(p, me.data[off:]), nil
 144 }
 145
 146 func (me pieceImpl) WriteAt(p []byte, off int64) (int, error) {
 147         off += me.mip.Offset()
 148         if !me.defaultHandledWriteChunkError {
 149                 go func() {
 150                         me.t.SetOnWriteChunkError(func(err error) {
 151                                 log.Printf("got write chunk error to custom handler: %v", err)
 152                                 me.mu.Lock()
 153                                 me.diskNotFull = true
 154                                 me.mu.Unlock()
 155                                 me.t.AllowDataDownload()
 156                         })
 157                         me.t.AllowDataDownload()
 158                 }()
 159                 me.defaultHandledWriteChunkError = true
 160         }
 161         me.mu.Lock()
 162         defer me.mu.Unlock()
 163         if me.diskNotFull {
 164                 return copy(me.data[off:], p), nil
 165         }
 166         return copy(me.data[off:], p[:1]), errors.New("disk full")
 167 }
 168
 169 func (me pieceImpl) MarkComplete() error {
 170         me.state().complete = true
 171         return nil
 172 }
 173
 174 func (me pieceImpl) MarkNotComplete() error {
 175         panic("implement me")
 176 }
 177
 178 func (me pieceImpl) Completion() storage.Completion {
 179         return storage.Completion{
 180                 Complete: me.state().complete,
 181                 Ok:       true,
 182         }
 183 }