test/issue377_test.go

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