11 "github.com/anacrolix/missinggo/v2/bitmap"
12 "github.com/frankban/quicktest"
13 "github.com/stretchr/testify/assert"
14 "github.com/stretchr/testify/require"
15 "golang.org/x/time/rate"
17 "github.com/anacrolix/torrent"
18 "github.com/anacrolix/torrent/internal/testutil"
19 "github.com/anacrolix/torrent/storage"
22 type LeecherStorageTestCase struct {
24 Factory StorageFactory
28 type StorageFactory func(string) storage.ClientImplCloser
30 func TestLeecherStorage(t *testing.T, ls LeecherStorageTestCase) {
32 for _, ss := range []struct {
36 {"File", storage.NewFile},
37 {"Mmap", storage.NewMMap},
39 t.Run(fmt.Sprintf("%sSeederStorage", ss.name), func(t *testing.T) {
40 for _, responsive := range []bool{false, true} {
41 t.Run(fmt.Sprintf("Responsive=%v", responsive), func(t *testing.T) {
42 t.Run("NoReadahead", func(t *testing.T) {
43 testClientTransfer(t, testClientTransferParams{
44 Responsive: responsive,
46 LeecherStorage: ls.Factory,
47 GOMAXPROCS: ls.GoMaxProcs,
50 for _, readahead := range []int64{-1, 0, 1, 2, 9, 20} {
51 t.Run(fmt.Sprintf("readahead=%v", readahead), func(t *testing.T) {
52 testClientTransfer(t, testClientTransferParams{
54 Responsive: responsive,
57 LeecherStorage: ls.Factory,
58 GOMAXPROCS: ls.GoMaxProcs,
68 type ConfigureClient struct {
69 Config func(cfg *torrent.ClientConfig)
70 Client func(cl *torrent.Client)
73 type testClientTransferParams struct {
77 LeecherStorage func(string) storage.ClientImplCloser
78 // TODO: Use a generic option type. This is the capacity of the leecher storage for determining
79 // whether it's possible for the leecher to be Complete. 0 currently means no limit.
80 LeecherStorageCapacity int64
81 SeederStorage func(string) storage.ClientImplCloser
82 SeederUploadRateLimiter *rate.Limiter
83 LeecherDownloadRateLimiter *rate.Limiter
84 ConfigureSeeder ConfigureClient
85 ConfigureLeecher ConfigureClient
88 LeecherStartsWithoutMetadata bool
91 // Creates a seeder and a leecher, and ensures the data transfers when a read
92 // is attempted on the leecher.
93 func testClientTransfer(t *testing.T, ps testClientTransferParams) {
94 prevGOMAXPROCS := runtime.GOMAXPROCS(ps.GOMAXPROCS)
95 newGOMAXPROCS := prevGOMAXPROCS
96 if ps.GOMAXPROCS > 0 {
97 newGOMAXPROCS = ps.GOMAXPROCS
100 quicktest.Check(t, runtime.GOMAXPROCS(prevGOMAXPROCS), quicktest.ContentEquals, newGOMAXPROCS)
103 greetingTempDir, mi := testutil.GreetingTestTorrent()
104 defer os.RemoveAll(greetingTempDir)
105 // Create seeder and a Torrent.
106 cfg := torrent.TestingConfig(t)
109 // Less than a piece, more than a single request.
110 cfg.MaxAllocPeerRequestDataPerConn = 4
111 // Some test instances don't like this being on, even when there's no cache involved.
112 cfg.DropMutuallyCompletePeers = false
113 if ps.SeederUploadRateLimiter != nil {
114 cfg.UploadRateLimiter = ps.SeederUploadRateLimiter
116 // cfg.ListenAddr = "localhost:4000"
117 if ps.SeederStorage != nil {
118 storage := ps.SeederStorage(greetingTempDir)
119 defer storage.Close()
120 cfg.DefaultStorage = storage
122 cfg.DataDir = greetingTempDir
124 if ps.ConfigureSeeder.Config != nil {
125 ps.ConfigureSeeder.Config(cfg)
127 seeder, err := torrent.NewClient(cfg)
128 require.NoError(t, err)
129 if ps.ConfigureSeeder.Client != nil {
130 ps.ConfigureSeeder.Client(seeder)
132 defer testutil.ExportStatusWriter(seeder, "s", t)()
133 seederTorrent, _, _ := seeder.AddTorrentSpec(torrent.TorrentSpecFromMetaInfo(mi))
134 // Run a Stats right after Closing the Client. This will trigger the Stats
135 // panic in #214 caused by RemoteAddr on Closed uTP sockets.
136 defer seederTorrent.Stats()
138 // Adding a torrent and setting the info should trigger piece checks for everything
139 // automatically. Wait until the seed Torrent agrees that everything is available.
140 <-seederTorrent.Complete.On()
141 // Create leecher and a Torrent.
142 leecherDataDir := t.TempDir()
143 cfg = torrent.TestingConfig(t)
144 // See the seeder client config comment.
145 cfg.DropMutuallyCompletePeers = false
146 if ps.LeecherStorage == nil {
147 cfg.DataDir = leecherDataDir
149 storage := ps.LeecherStorage(leecherDataDir)
150 defer storage.Close()
151 cfg.DefaultStorage = storage
153 if ps.LeecherDownloadRateLimiter != nil {
154 cfg.DownloadRateLimiter = ps.LeecherDownloadRateLimiter
158 if ps.ConfigureLeecher.Config != nil {
159 ps.ConfigureLeecher.Config(cfg)
161 leecher, err := torrent.NewClient(cfg)
162 require.NoError(t, err)
163 defer leecher.Close()
164 if ps.ConfigureLeecher.Client != nil {
165 ps.ConfigureLeecher.Client(leecher)
167 defer testutil.ExportStatusWriter(leecher, "l", t)()
168 leecherTorrent, new, err := leecher.AddTorrentSpec(func() (ret *torrent.TorrentSpec) {
169 ret = torrent.TorrentSpecFromMetaInfo(mi)
171 if ps.LeecherStartsWithoutMetadata {
176 require.NoError(t, err)
177 assert.False(t, leecherTorrent.Complete.Bool())
180 //// This was used when observing coalescing of piece state changes.
181 //logPieceStateChanges(leecherTorrent)
183 // Now do some things with leecher and seeder.
184 added := leecherTorrent.AddClientPeer(seeder)
185 assert.False(t, leecherTorrent.Seeding())
186 // The leecher will use peers immediately if it doesn't have the metadata. Otherwise, they
187 // should be sitting idle until we demand data.
188 if !ps.LeecherStartsWithoutMetadata {
189 assert.EqualValues(t, added, leecherTorrent.Stats().PendingPeers)
191 if ps.LeecherStartsWithoutMetadata {
192 <-leecherTorrent.GotInfo()
194 r := leecherTorrent.NewReader()
196 go leecherTorrent.SetInfoBytes(mi.InfoBytes)
201 r.SetReadahead(ps.Readahead)
203 assertReadAllGreeting(t, r)
204 info, err := mi.UnmarshalInfo()
205 require.NoError(t, err)
206 canComplete := ps.LeecherStorageCapacity == 0 || ps.LeecherStorageCapacity >= info.TotalLength()
208 // Reading from a cache doesn't refresh older pieces until we fail to read those, so we need
209 // to force a refresh since we just read the contents from start to finish.
210 go leecherTorrent.VerifyData()
213 <-leecherTorrent.Complete.On()
215 <-leecherTorrent.Complete.Off()
217 assert.NotEmpty(t, seederTorrent.PeerConns())
218 leecherPeerConns := leecherTorrent.PeerConns()
219 if cfg.DropMutuallyCompletePeers {
220 // I don't think we can assume it will be empty already, due to timing.
221 // assert.Empty(t, leecherPeerConns)
223 assert.NotEmpty(t, leecherPeerConns)
226 for _, pc := range leecherPeerConns {
227 completed := pc.PeerPieces().GetCardinality()
228 t.Logf("peer conn %v has %v completed pieces", pc, completed)
229 if completed == bitmap.BitRange(leecherTorrent.Info().NumPieces()) {
234 t.Errorf("didn't find seeder amongst leecher peer conns")
237 seederStats := seederTorrent.Stats()
238 assert.True(t, 13 <= seederStats.BytesWrittenData.Int64())
239 assert.True(t, 8 <= seederStats.ChunksWritten.Int64())
241 leecherStats := leecherTorrent.Stats()
242 assert.True(t, 13 <= leecherStats.BytesReadData.Int64())
243 assert.True(t, 8 <= leecherStats.ChunksRead.Int64())
245 // Try reading through again for the cases where the torrent data size
246 // exceeds the size of the cache.
247 assertReadAllGreeting(t, r)
250 func assertReadAllGreeting(t *testing.T, r io.ReadSeeker) {
251 pos, err := r.Seek(0, io.SeekStart)
252 assert.NoError(t, err)
253 assert.EqualValues(t, 0, pos)
254 quicktest.Check(t, iotest.TestReader(r, []byte(testutil.GreetingFileContents)), quicktest.IsNil)