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