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         // Don't require fast extension, whether the seeder will provide it or not (so we can test mixed
  55         // cases).
  56         leecherClientConfig.MinPeerExtensions.SetBit(pp.ExtensionBitFast, false)
  57         justOneNetwork(leecherClientConfig)
  58         leecherClient, err := torrent.NewClient(leecherClientConfig)
  59         require.NoError(t, err)
  60         defer leecherClient.Close()
  61         defer testutil.ExportStatusWriter(leecherClient, "l", t)()
  62         info, err := metainfo.UnmarshalInfo()
  63         require.NoError(t, err)
  64         leecherStorage := diskFullStorage{
  65                 pieces: make([]pieceState, info.NumPieces()),
  66                 data:   make([]byte, info.TotalLength()),
  67         }
  68         defer leecherStorage.Close()
  69         leecherTorrent, new, err := leecherClient.AddTorrentSpec(&torrent.TorrentSpec{
  70                 InfoHash: metainfo.HashInfoBytes(),
  71                 Storage:  &leecherStorage,
  72         })
  73         leecherStorage.t = leecherTorrent
  74         require.NoError(t, err)
  75         assert.True(t, new)
  76         seederTorrent, err := seederClient.AddTorrent(metainfo)
  77         require.NoError(t, err)
  78         // Tell the seeder to find the leecher. Is it guaranteed seeders will always try to do this?
  79         seederTorrent.AddClientPeer(leecherClient)
  80         <-leecherTorrent.GotInfo()
  81         r := leecherTorrent.Files()[0].NewReader()
  82         defer r.Close()
  83         // We can't use assertReadAllGreeting here, because the default storage write error handler
  84         // disables data downloads, which now causes Readers to error when they're blocked.
  85         if false {
  86                 assertReadAllGreeting(t, leecherTorrent.NewReader())
  87         } else {
  88                 for func() bool {
  89                         // We don't seem to need to seek, but that's probably just because the storage failure is
  90                         // happening on the first read.
  91                         r.Seek(0, io.SeekStart)
  92                         if err := iotest.TestReader(r, []byte(testutil.GreetingFileContents)); err != nil {
  93                                 t.Logf("got error while reading: %v", err)
  94                                 return true
  95                         }
  96                         return false
  97                 }() {
  98                 }
  99         }
 100         // TODO: Check that PeerConns fastEnabled matches seederFast?
 101         // select {}
 102 }
 103
 104 type pieceState struct {
 105         complete bool
 106 }
 107
 108 type diskFullStorage struct {
 109         pieces                        []pieceState
 110         t                             *torrent.Torrent
 111         defaultHandledWriteChunkError bool
 112         data                          []byte
 113
 114         mu          sync.Mutex
 115         diskNotFull bool
 116 }
 117
 118 func (me *diskFullStorage) Piece(p metainfo.Piece) storage.PieceImpl {
 119         return pieceImpl{
 120                 mip:             p,
 121                 diskFullStorage: me,
 122         }
 123 }
 124
 125 func (me *diskFullStorage) Close() error {
 126         return nil
 127 }
 128
 129 func (d *diskFullStorage) OpenTorrent(info *metainfo.Info, infoHash metainfo.Hash) (storage.TorrentImpl, error) {
 130         return storage.TorrentImpl{Piece: d.Piece, Close: d.Close}, nil
 131 }
 132
 133 type pieceImpl struct {
 134         mip metainfo.Piece
 135         *diskFullStorage
 136 }
 137
 138 func (me pieceImpl) state() *pieceState {
 139         return &me.diskFullStorage.pieces[me.mip.Index()]
 140 }
 141
 142 func (me pieceImpl) ReadAt(p []byte, off int64) (n int, err error) {
 143         off += me.mip.Offset()
 144         return copy(p, me.data[off:]), nil
 145 }
 146
 147 func (me pieceImpl) WriteAt(p []byte, off int64) (int, error) {
 148         off += me.mip.Offset()
 149         if !me.defaultHandledWriteChunkError {
 150                 go func() {
 151                         me.t.SetOnWriteChunkError(func(err error) {
 152                                 log.Printf("got write chunk error to custom handler: %v", err)
 153                                 me.mu.Lock()
 154                                 me.diskNotFull = true
 155                                 me.mu.Unlock()
 156                                 me.t.AllowDataDownload()
 157                         })
 158                         me.t.AllowDataDownload()
 159                 }()
 160                 me.defaultHandledWriteChunkError = true
 161         }
 162         me.mu.Lock()
 163         defer me.mu.Unlock()
 164         if me.diskNotFull {
 165                 return copy(me.data[off:], p), nil
 166         }
 167         return copy(me.data[off:], p[:1]), errors.New("disk full")
 168 }
 169
 170 func (me pieceImpl) MarkComplete() error {
 171         me.state().complete = true
 172         return nil
 173 }
 174
 175 func (me pieceImpl) MarkNotComplete() error {
 176         panic("implement me")
 177 }
 178
 179 func (me pieceImpl) Completion() storage.Completion {
 180         return storage.Completion{
 181                 Complete: me.state().complete,
 182                 Ok:       true,
 183         }
 184 }