test/issue377_test.go

   1 package test
   2
   3 import (
   4         "errors"
   5         "io"
   6         "io/ioutil"
   7         "os"
   8         "sync"
   9         "testing"
  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()
  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()
  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                         output, err := ioutil.ReadAll(r)
  90                         if err != nil {
  91                                 t.Logf("got error while reading: %v", err)
  92                                 return true
  93                         }
  94                         assert.EqualValues(t, testutil.GreetingFileContents, output)
  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 &d, 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 }