15 "github.com/anacrolix/missinggo/v2/bitmap"
16 "github.com/anacrolix/missinggo/v2/filecache"
17 "github.com/anacrolix/torrent"
18 "github.com/anacrolix/torrent/internal/testutil"
19 "github.com/anacrolix/torrent/storage"
20 sqliteStorage "github.com/anacrolix/torrent/storage/sqlite"
21 "github.com/frankban/quicktest"
22 "golang.org/x/time/rate"
24 "github.com/stretchr/testify/assert"
25 "github.com/stretchr/testify/require"
28 type testClientTransferParams struct {
32 LeecherStorage func(string) storage.ClientImplCloser
33 // TODO: Use a generic option type. This is the capacity of the leecher storage for determining
34 // whether it's possible for the leecher to be Complete. 0 currently means no limit.
35 LeecherStorageCapacity int64
36 SeederStorage func(string) storage.ClientImplCloser
37 SeederUploadRateLimiter *rate.Limiter
38 LeecherDownloadRateLimiter *rate.Limiter
39 ConfigureSeeder ConfigureClient
40 ConfigureLeecher ConfigureClient
43 LeecherStartsWithoutMetadata bool
46 func assertReadAllGreeting(t *testing.T, r io.ReadSeeker) {
47 pos, err := r.Seek(0, io.SeekStart)
48 assert.NoError(t, err)
49 assert.EqualValues(t, 0, pos)
50 quicktest.Check(t, iotest.TestReader(r, []byte(testutil.GreetingFileContents)), quicktest.IsNil)
53 // Creates a seeder and a leecher, and ensures the data transfers when a read
54 // is attempted on the leecher.
55 func testClientTransfer(t *testing.T, ps testClientTransferParams) {
58 prevGOMAXPROCS := runtime.GOMAXPROCS(ps.GOMAXPROCS)
59 newGOMAXPROCS := prevGOMAXPROCS
60 if ps.GOMAXPROCS > 0 {
61 newGOMAXPROCS = ps.GOMAXPROCS
64 quicktest.Check(t, runtime.GOMAXPROCS(prevGOMAXPROCS), quicktest.ContentEquals, newGOMAXPROCS)
67 greetingTempDir, mi := testutil.GreetingTestTorrent()
68 defer os.RemoveAll(greetingTempDir)
69 // Create seeder and a Torrent.
70 cfg := torrent.TestingConfig(t)
73 // Some test instances don't like this being on, even when there's no cache involved.
74 cfg.DropMutuallyCompletePeers = false
75 if ps.SeederUploadRateLimiter != nil {
76 cfg.UploadRateLimiter = ps.SeederUploadRateLimiter
78 // cfg.ListenAddr = "localhost:4000"
79 if ps.SeederStorage != nil {
80 storage := ps.SeederStorage(greetingTempDir)
82 cfg.DefaultStorage = storage
84 cfg.DataDir = greetingTempDir
86 if ps.ConfigureSeeder.Config != nil {
87 ps.ConfigureSeeder.Config(cfg)
89 seeder, err := torrent.NewClient(cfg)
90 require.NoError(t, err)
91 if ps.ConfigureSeeder.Client != nil {
92 ps.ConfigureSeeder.Client(seeder)
94 defer testutil.ExportStatusWriter(seeder, "s", t)()
95 seederTorrent, _, _ := seeder.AddTorrentSpec(torrent.TorrentSpecFromMetaInfo(mi))
96 // Run a Stats right after Closing the Client. This will trigger the Stats
97 // panic in #214 caused by RemoteAddr on Closed uTP sockets.
98 defer seederTorrent.Stats()
100 // Adding a torrent and setting the info should trigger piece checks for everything
101 // automatically. Wait until the seed Torrent agrees that everything is available.
102 <-seederTorrent.Complete.On()
103 // Create leecher and a Torrent.
104 leecherDataDir := t.TempDir()
105 cfg = torrent.TestingConfig(t)
106 // See the seeder client config comment.
107 cfg.DropMutuallyCompletePeers = false
108 if ps.LeecherStorage == nil {
109 cfg.DataDir = leecherDataDir
111 storage := ps.LeecherStorage(leecherDataDir)
112 defer storage.Close()
113 cfg.DefaultStorage = storage
115 if ps.LeecherDownloadRateLimiter != nil {
116 cfg.DownloadRateLimiter = ps.LeecherDownloadRateLimiter
120 if ps.ConfigureLeecher.Config != nil {
121 ps.ConfigureLeecher.Config(cfg)
123 leecher, err := torrent.NewClient(cfg)
124 require.NoError(t, err)
125 defer leecher.Close()
126 if ps.ConfigureLeecher.Client != nil {
127 ps.ConfigureLeecher.Client(leecher)
129 defer testutil.ExportStatusWriter(leecher, "l", t)()
130 leecherTorrent, new, err := leecher.AddTorrentSpec(func() (ret *torrent.TorrentSpec) {
131 ret = torrent.TorrentSpecFromMetaInfo(mi)
133 if ps.LeecherStartsWithoutMetadata {
138 require.NoError(t, err)
139 assert.False(t, leecherTorrent.Complete.Bool())
142 //// This was used when observing coalescing of piece state changes.
143 //logPieceStateChanges(leecherTorrent)
145 // Now do some things with leecher and seeder.
146 added := leecherTorrent.AddClientPeer(seeder)
147 assert.False(t, leecherTorrent.Seeding())
148 // The leecher will use peers immediately if it doesn't have the metadata. Otherwise, they
149 // should be sitting idle until we demand data.
150 if !ps.LeecherStartsWithoutMetadata {
151 assert.EqualValues(t, added, leecherTorrent.Stats().PendingPeers)
153 if ps.LeecherStartsWithoutMetadata {
154 <-leecherTorrent.GotInfo()
156 r := leecherTorrent.NewReader()
158 go leecherTorrent.SetInfoBytes(mi.InfoBytes)
163 r.SetReadahead(ps.Readahead)
165 assertReadAllGreeting(t, r)
166 info, err := mi.UnmarshalInfo()
167 require.NoError(t, err)
168 canComplete := ps.LeecherStorageCapacity == 0 || ps.LeecherStorageCapacity >= info.TotalLength()
170 // Reading from a cache doesn't refresh older pieces until we fail to read those, so we need
171 // to force a refresh since we just read the contents from start to finish.
172 go leecherTorrent.VerifyData()
175 <-leecherTorrent.Complete.On()
177 <-leecherTorrent.Complete.Off()
179 assert.NotEmpty(t, seederTorrent.PeerConns())
180 leecherPeerConns := leecherTorrent.PeerConns()
181 if cfg.DropMutuallyCompletePeers {
182 // I don't think we can assume it will be empty already, due to timing.
183 // assert.Empty(t, leecherPeerConns)
185 assert.NotEmpty(t, leecherPeerConns)
188 for _, pc := range leecherPeerConns {
189 completed := pc.PeerPieces().GetCardinality()
190 t.Logf("peer conn %v has %v completed pieces", pc, completed)
191 if completed == bitmap.BitRange(leecherTorrent.Info().NumPieces()) {
196 t.Errorf("didn't find seeder amongst leecher peer conns")
199 seederStats := seederTorrent.Stats()
200 assert.True(t, 13 <= seederStats.BytesWrittenData.Int64())
201 assert.True(t, 8 <= seederStats.ChunksWritten.Int64())
203 leecherStats := leecherTorrent.Stats()
204 assert.True(t, 13 <= leecherStats.BytesReadData.Int64())
205 assert.True(t, 8 <= leecherStats.ChunksRead.Int64())
207 // Try reading through again for the cases where the torrent data size
208 // exceeds the size of the cache.
209 assertReadAllGreeting(t, r)
212 type fileCacheClientStorageFactoryParams struct {
217 func newFileCacheClientStorageFactory(ps fileCacheClientStorageFactoryParams) storageFactory {
218 return func(dataDir string) storage.ClientImplCloser {
219 fc, err := filecache.NewCache(dataDir)
223 var sharedCapacity *int64
225 sharedCapacity = &ps.Capacity
226 fc.SetCapacity(ps.Capacity)
232 storage.NewResourcePiecesOpts(
233 fc.AsResourceProvider(),
234 storage.ResourcePiecesOpts{
235 Capacity: sharedCapacity,
237 ioutil.NopCloser(nil),
242 type storageFactory func(string) storage.ClientImplCloser
244 func TestClientTransferDefault(t *testing.T) {
245 testClientTransfer(t, testClientTransferParams{
246 LeecherStorage: newFileCacheClientStorageFactory(fileCacheClientStorageFactoryParams{}),
250 func TestClientTransferDefaultNoMetadata(t *testing.T) {
251 testClientTransfer(t, testClientTransferParams{
252 LeecherStorage: newFileCacheClientStorageFactory(fileCacheClientStorageFactoryParams{}),
253 LeecherStartsWithoutMetadata: true,
257 func TestClientTransferRateLimitedUpload(t *testing.T) {
258 started := time.Now()
259 testClientTransfer(t, testClientTransferParams{
260 // We are uploading 13 bytes (the length of the greeting torrent). The
261 // chunks are 2 bytes in length. Then the smallest burst we can run
262 // with is 2. Time taken is (13-burst)/rate.
263 SeederUploadRateLimiter: rate.NewLimiter(11, 2),
265 require.True(t, time.Since(started) > time.Second)
268 func TestClientTransferRateLimitedDownload(t *testing.T) {
269 testClientTransfer(t, testClientTransferParams{
270 LeecherDownloadRateLimiter: rate.NewLimiter(512, 512),
271 ConfigureSeeder: ConfigureClient{
272 Config: func(cfg *torrent.ClientConfig) {
273 // If we send too many keep alives, we consume all the leechers available download
274 // rate. The default isn't exposed, but a minute is pretty reasonable.
275 cfg.KeepAliveTimeout = time.Minute
281 func testClientTransferSmallCache(t *testing.T, setReadahead bool, readahead int64) {
282 testClientTransfer(t, testClientTransferParams{
283 LeecherStorage: newFileCacheClientStorageFactory(fileCacheClientStorageFactoryParams{
285 // Going below the piece length means it can't complete a piece so
286 // that it can be hashed.
289 LeecherStorageCapacity: 5,
290 SetReadahead: setReadahead,
291 // Can't readahead too far or the cache will thrash and drop data we
293 Readahead: readahead,
295 // These tests don't work well with more than 1 connection to the seeder.
296 ConfigureLeecher: ConfigureClient{
297 Config: func(cfg *torrent.ClientConfig) {
298 cfg.DropDuplicatePeerIds = true
299 // cfg.DisableIPv6 = true
300 // cfg.DisableUTP = true
306 func TestClientTransferSmallCachePieceSizedReadahead(t *testing.T) {
307 testClientTransferSmallCache(t, true, 5)
310 func TestClientTransferSmallCacheLargeReadahead(t *testing.T) {
311 testClientTransferSmallCache(t, true, 15)
314 func TestClientTransferSmallCacheDefaultReadahead(t *testing.T) {
315 testClientTransferSmallCache(t, false, -1)
318 type leecherStorageTestCase struct {
324 func TestClientTransferVarious(t *testing.T) {
326 for _, ls := range []leecherStorageTestCase{
327 {"Filecache", newFileCacheClientStorageFactory(fileCacheClientStorageFactoryParams{}), 0},
328 {"Boltdb", storage.NewBoltDB, 0},
329 {"SqliteDirect", func(s string) storage.ClientImplCloser {
330 path := filepath.Join(s, "sqlite3.db")
331 var opts sqliteStorage.NewDirectStorageOpts
333 cl, err := sqliteStorage.NewDirectStorage(opts)
340 t.Run(fmt.Sprintf("LeecherStorage=%s", ls.name), func(t *testing.T) {
342 for _, ss := range []struct {
346 {"File", storage.NewFile},
347 {"Mmap", storage.NewMMap},
349 t.Run(fmt.Sprintf("%sSeederStorage", ss.name), func(t *testing.T) {
350 for _, responsive := range []bool{false, true} {
351 t.Run(fmt.Sprintf("Responsive=%v", responsive), func(t *testing.T) {
352 t.Run("NoReadahead", func(t *testing.T) {
353 testClientTransfer(t, testClientTransferParams{
354 Responsive: responsive,
356 LeecherStorage: ls.f,
357 GOMAXPROCS: ls.gomaxprocs,
360 for _, readahead := range []int64{-1, 0, 1, 2, 9, 20} {
361 t.Run(fmt.Sprintf("readahead=%v", readahead), func(t *testing.T) {
362 testClientTransfer(t, testClientTransferParams{
364 Responsive: responsive,
366 Readahead: readahead,
367 LeecherStorage: ls.f,
368 GOMAXPROCS: ls.gomaxprocs,
380 // Check that after completing leeching, a leecher transitions to a seeding
381 // correctly. Connected in a chain like so: Seeder <-> Leecher <-> LeecherLeecher.
382 func TestSeedAfterDownloading(t *testing.T) {
383 greetingTempDir, mi := testutil.GreetingTestTorrent()
384 defer os.RemoveAll(greetingTempDir)
386 cfg := torrent.TestingConfig(t)
388 cfg.DataDir = greetingTempDir
389 seeder, err := torrent.NewClient(cfg)
390 require.NoError(t, err)
392 defer testutil.ExportStatusWriter(seeder, "s", t)()
393 seederTorrent, ok, err := seeder.AddTorrentSpec(torrent.TorrentSpecFromMetaInfo(mi))
394 require.NoError(t, err)
396 seederTorrent.VerifyData()
398 cfg = torrent.TestingConfig(t)
400 cfg.DataDir = t.TempDir()
401 leecher, err := torrent.NewClient(cfg)
402 require.NoError(t, err)
403 defer leecher.Close()
404 defer testutil.ExportStatusWriter(leecher, "l", t)()
406 cfg = torrent.TestingConfig(t)
408 cfg.DataDir = t.TempDir()
409 leecherLeecher, _ := torrent.NewClient(cfg)
410 require.NoError(t, err)
411 defer leecherLeecher.Close()
412 defer testutil.ExportStatusWriter(leecherLeecher, "ll", t)()
413 leecherGreeting, ok, err := leecher.AddTorrentSpec(func() (ret *torrent.TorrentSpec) {
414 ret = torrent.TorrentSpecFromMetaInfo(mi)
418 require.NoError(t, err)
420 llg, ok, err := leecherLeecher.AddTorrentSpec(func() (ret *torrent.TorrentSpec) {
421 ret = torrent.TorrentSpecFromMetaInfo(mi)
425 require.NoError(t, err)
427 // Simultaneously DownloadAll in Leecher, and read the contents
428 // consecutively in LeecherLeecher. This non-deterministically triggered a
429 // case where the leecher wouldn't unchoke the LeecherLeecher.
430 var wg sync.WaitGroup
436 quicktest.Check(t, iotest.TestReader(r, []byte(testutil.GreetingFileContents)), quicktest.IsNil)
438 done := make(chan struct{})
440 go leecherGreeting.AddClientPeer(seeder)
441 go leecherGreeting.AddClientPeer(leecherLeecher)
445 leecherGreeting.DownloadAll()
451 type ConfigureClient struct {
452 Config func(cfg *torrent.ClientConfig)
453 Client func(cl *torrent.Client)