test/issue377_test.go

   1 package test
   2
   3 import (
   4         "errors"
   5         "os"
   6         "sync"
   7         "testing"
   8
   9         "github.com/anacrolix/log"
  10         pp "github.com/anacrolix/torrent/peer_protocol"
  11
  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         "github.com/anacrolix/torrent/storage"
  19 )
  20
  21 func justOneNetwork(cc *torrent.ClientConfig) {
  22         cc.DisableTCP = true
  23         cc.DisableIPv4 = true
  24 }
  25
  26 func TestReceiveChunkStorageFailureSeederFastExtensionDisabled(t *testing.T) {
  27         testReceiveChunkStorageFailure(t, false)
  28 }
  29
  30 func TestReceiveChunkStorageFailure(t *testing.T) {
  31         testReceiveChunkStorageFailure(t, true)
  32 }
  33
  34 func testReceiveChunkStorageFailure(t *testing.T, seederFast bool) {
  35         seederDataDir, metainfo := testutil.GreetingTestTorrent()
  36         defer os.RemoveAll(seederDataDir)
  37         seederClientConfig := torrent.TestingConfig()
  38         seederClientConfig.Debug = true
  39         justOneNetwork(seederClientConfig)
  40         seederClientStorage := storage.NewMMap(seederDataDir)
  41         defer seederClientStorage.Close()
  42         seederClientConfig.DefaultStorage = seederClientStorage
  43         seederClientConfig.Seed = true
  44         seederClientConfig.Debug = true
  45         seederClientConfig.Extensions.SetBit(pp.ExtensionBitFast, seederFast)
  46         seederClient, err := torrent.NewClient(seederClientConfig)
  47         require.NoError(t, err)
  48         defer seederClient.Close()
  49         defer testutil.ExportStatusWriter(seederClient, "s", t)()
  50         leecherClientConfig := torrent.TestingConfig()
  51         leecherClientConfig.Debug = true
  52         justOneNetwork(leecherClientConfig)
  53         leecherClient, err := torrent.NewClient(leecherClientConfig)
  54         require.NoError(t, err)
  55         defer leecherClient.Close()
  56         defer testutil.ExportStatusWriter(leecherClient, "l", t)()
  57         info, err := metainfo.UnmarshalInfo()
  58         require.NoError(t, err)
  59         leecherStorage := diskFullStorage{
  60                 pieces: make([]pieceState, info.NumPieces()),
  61                 data:   make([]byte, info.TotalLength()),
  62         }
  63         defer leecherStorage.Close()
  64         leecherTorrent, new, err := leecherClient.AddTorrentSpec(&torrent.TorrentSpec{
  65                 InfoHash: metainfo.HashInfoBytes(),
  66                 Storage:  &leecherStorage,
  67         })
  68         leecherStorage.t = leecherTorrent
  69         require.NoError(t, err)
  70         assert.True(t, new)
  71         seederTorrent, err := seederClient.AddTorrent(metainfo)
  72         require.NoError(t, err)
  73         // Tell the seeder to find the leecher. Is it guaranteed seeders will always try to do this?
  74         seederTorrent.AddClientPeer(leecherClient)
  75         <-leecherTorrent.GotInfo()
  76         assertReadAllGreeting(t, leecherTorrent.NewReader())
  77         // TODO: Check that PeerConns fastEnabled matches seederFast?
  78         //select {}
  79 }
  80
  81 type pieceState struct {
  82         complete bool
  83 }
  84
  85 type diskFullStorage struct {
  86         pieces                        []pieceState
  87         t                             *torrent.Torrent
  88         defaultHandledWriteChunkError bool
  89         data                          []byte
  90
  91         mu          sync.Mutex
  92         diskNotFull bool
  93 }
  94
  95 func (me *diskFullStorage) Piece(p metainfo.Piece) storage.PieceImpl {
  96         return pieceImpl{
  97                 mip:             p,
  98                 diskFullStorage: me,
  99         }
 100 }
 101
 102 func (me diskFullStorage) Close() error {
 103         return nil
 104 }
 105
 106 func (d diskFullStorage) OpenTorrent(info *metainfo.Info, infoHash metainfo.Hash) (storage.TorrentImpl, error) {
 107         return &d, nil
 108 }
 109
 110 type pieceImpl struct {
 111         mip metainfo.Piece
 112         *diskFullStorage
 113 }
 114
 115 func (me pieceImpl) state() *pieceState {
 116         return &me.diskFullStorage.pieces[me.mip.Index()]
 117 }
 118
 119 func (me pieceImpl) ReadAt(p []byte, off int64) (n int, err error) {
 120         off += me.mip.Offset()
 121         return copy(p, me.data[off:]), nil
 122 }
 123
 124 func (me pieceImpl) WriteAt(p []byte, off int64) (int, error) {
 125         off += me.mip.Offset()
 126         if !me.defaultHandledWriteChunkError {
 127                 go func() {
 128                         me.t.SetOnWriteChunkError(func(err error) {
 129                                 log.Printf("got write chunk error to custom handler: %v", err)
 130                                 me.mu.Lock()
 131                                 me.diskNotFull = true
 132                                 me.mu.Unlock()
 133                                 me.t.AllowDataDownload()
 134                         })
 135                         me.t.AllowDataDownload()
 136                 }()
 137                 me.defaultHandledWriteChunkError = true
 138         }
 139         me.mu.Lock()
 140         defer me.mu.Unlock()
 141         if me.diskNotFull {
 142                 return copy(me.data[off:], p), nil
 143         }
 144         return copy(me.data[off:], p[:1]), errors.New("disk full")
 145 }
 146
 147 func (me pieceImpl) MarkComplete() error {
 148         me.state().complete = true
 149         return nil
 150 }
 151
 152 func (me pieceImpl) MarkNotComplete() error {
 153         panic("implement me")
 154 }
 155
 156 func (me pieceImpl) Completion() storage.Completion {
 157         return storage.Completion{
 158                 Complete: me.state().complete,
 159                 Ok:       true,
 160         }
 161 }