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