Merge pull request #572 from alanmcgovern/fix-tracker-announce-concurrency

[core] Refactor how announce requests are handled.

Author: alanmcgovern

src/MonoTorrent.Client/MonoTorrent.Client/Managers/TrackerManager.cs

@@ -44,10 +44,12 @@ namespace MonoTorrent.Trackers
     /// </summary>
     class TrackerManager : ITrackerManager
     {
+        static readonly int MaxConcurrentAnnounces = 15;
+
         public event EventHandler<AnnounceResponseEventArgs>? AnnounceComplete;
         public event EventHandler<ScrapeResponseEventArgs>? ScrapeComplete;
 
-        public SemaphoreSlim AnnounceLimiter { get; }
+        ReusableExclusiveSemaphore AnnounceLimiter { get; }
 
         Factories Factories { get; }
 
@@ -77,7 +79,7 @@ class TrackerManager : ITrackerManager
         /// <param name="isPrivate">True if adding/removing tracker should be disallowed.</param>
         internal TrackerManager (Factories factories, ITrackerRequestFactory requestFactory, IEnumerable<IEnumerable<string>> announces, bool isPrivate)
         {
-            AnnounceLimiter = new SemaphoreSlim (10);
+            AnnounceLimiter = new ReusableExclusiveSemaphore ();
             Factories = factories;
             RequestFactory = requestFactory;
             Private = isPrivate;
@@ -154,49 +156,64 @@ public async ReusableTask<bool> RemoveTrackerAsync (ITracker tracker)
         public ReusableTask AnnounceAsync (CancellationToken token)
             => AnnounceAsync (TorrentEvent.None, token);
 
-        public async ReusableTask AnnounceAsync (TorrentEvent clientEvent, CancellationToken token)
-        {
-            // If the user initiates an Announce we need to go to the correct thread to process it.
-            await ClientEngine.MainLoop;
+        public ReusableTask AnnounceAsync (TorrentEvent clientEvent, CancellationToken token)
+            => AnnounceAsync (clientEvent, null, token);
 
-            var args = RequestFactory.CreateAnnounce (clientEvent);
-            var announces = new List<Task> ();
-            for (int i = 0; i < Tiers.Count; i++) {
-                var task = AnnounceTierAsync (Tiers[i], args, token);
-                if (task.IsCompleted)
-                    await task;
-                else
-                    announces.Add (task.AsTask ());
-            }
+        public ReusableTask AnnounceAsync (ITracker tracker, CancellationToken token)
+            => AnnounceAsync (TorrentEvent.None, tracker ?? throw new ArgumentNullException (nameof (tracker)), token);
 
-            if (announces.Count > 0)
-                await Task.WhenAll (announces);
-        }
-        async ReusableTask AnnounceTierAsync (TrackerTier tier, AnnounceRequest args, CancellationToken token)
+        async ReusableTask AnnounceAsync (TorrentEvent clientEvent, ITracker? tracker, CancellationToken token)
         {
-            using (await AnnounceLimiter.EnterAsync ())
-                await tier.AnnounceAsync (args, token);
-        }
-
-        public async ReusableTask AnnounceAsync (ITracker tracker, CancellationToken token)
-        {
-            Check.Tracker (tracker);
-
             // If the user initiates an Announce we need to go to the correct thread to process it.
             await ClientEngine.MainLoop;
 
-            try {
+            // Check if there are any in-progress announce requests being handled.
+            if (!AnnounceLimiter.TryEnter (out ReusableExclusiveSemaphore.Releaser releaser)) {
+                // There is an in-progress announce, and this is a regular recurring announce attempt, so just bail out.
+                if (tracker is null && clientEvent == TorrentEvent.None)
+                    return;
+
+                // If we get here it means there's an in-progress announce *and* this is a special event. Either the user
+                // has announced to a specific tracker, or we have a special 'TorrentEvent' to announce to the tracker.
+                // Wait for the in-progress announce to complete, then run this one!
+                releaser = await AnnounceLimiter.EnterAsync ();
+            }
+
+            using var autorelease = releaser;
+            token.ThrowIfCancellationRequested ();
+
+            // This is either a regularly scheduled announce, or a specific TorrentEvent is being sent to the tracker.
+            if (tracker is null) {
+                // Create and re-use the announce args for all trackers and tiers
+                var args = RequestFactory.CreateAnnounce (clientEvent);
+
+                // Capture a list of Tiers in case the user adds/removes any mid-announce.
+                var pendingTiers = new Queue<TrackerTier> (Tiers);
+                var activeAnnounces = new List<Task> ();
+
+                while (pendingTiers.Count > 0) {
+                    if (activeAnnounces.Count == MaxConcurrentAnnounces) {
+                        var completed = await Task.WhenAny (activeAnnounces);
+                        activeAnnounces.Remove (completed);
+                        await completed;
+                    }
+
+                    // The announce *might* fast-path and exit immediately if that
+                    // tracker has been announced to recently.
+                    var task = pendingTiers.Dequeue ().AnnounceAsync (args, token);
+                    if (task.IsCompleted)
+                        await task;
+                    else
+                        activeAnnounces.Add (task.AsTask ());
+                }
+                await Task.WhenAll (activeAnnounces);
+            } else {
+                // This only occurs when the user has manually announced to a specific tracker.
                 var trackerTier = Tiers.First (t => t.Trackers.Contains (tracker));
                 AnnounceRequest args = RequestFactory.CreateAnnounce (TorrentEvent.None);
-                await AnnounceTrackerAsync (trackerTier, args, tracker, token);
-            } catch {
+                await trackerTier.AnnounceAsync (args, tracker, token);
             }
         }
-        async ReusableTask AnnounceTrackerAsync (TrackerTier tier, AnnounceRequest args, ITracker tracker, CancellationToken token)
-        {
-            using (await AnnounceLimiter.EnterAsync ())
-                await tier.AnnounceAsync (args, tracker, token);
-        }
 
         public async ReusableTask ScrapeAsync (CancellationToken token)
         {

src/Tests/Tests.MonoTorrent.Client/MonoTorrent.Client/TestRig.cs

@@ -121,7 +121,7 @@ public ReusableTask SetMaximumOpenFilesAsync (int maximumOpenFiles)
         }
     }
 
-    class CustomTracker : ITrackerConnection
+    class CustomTrackerConnection : ITrackerConnection
     {
         public List<DateTime> AnnouncedAt = new List<DateTime> ();
         public List<AnnounceRequest> AnnounceParameters = new List<AnnounceRequest> ();
@@ -135,7 +135,7 @@ class CustomTracker : ITrackerConnection
 
         readonly List<Peer> peers = new List<Peer> ();
 
-        public CustomTracker (Uri uri)
+        public CustomTrackerConnection (Uri uri)
         {
             Uri = uri;
         }
@@ -419,7 +419,7 @@ public async Task RecreateManager ()
                 .WithDhtListenerCreator (port => new NullDhtListener ())
                 .WithLocalPeerDiscoveryCreator (() => new ManualLocalPeerListener ())
                 .WithPeerConnectionListenerCreator (endpoint => new CustomListener ())
-                .WithTrackerCreator ("custom", uri => new Tracker (new CustomTracker (uri)))
+                .WithTrackerCreator ("custom", uri => new Tracker (new CustomTrackerConnection (uri)))
                 ;
 
             Engine = new ClientEngine (EngineSettingsBuilder.CreateForTests (

src/Tests/Tests.MonoTorrent.Client/MonoTorrent.Client/TrackerManagerTests.cs

@@ -43,35 +43,54 @@
 
 namespace MonoTorrent.Client
 {
-    class TrackerWithConnection : Tracker
+    class CustomTracker : Tracker
     {
-        public CustomTracker Connection { get; }
+        public CustomTrackerConnection Connection { get; }
 
-        public TrackerWithConnection (ITrackerConnection connection)
+        public CustomTracker (CustomTrackerConnection connection)
             : base (connection)
         {
-            Connection = (CustomTracker) connection;
+            Connection = connection;
         }
     }
-    class DefaultTracker : ITrackerConnection
+
+    class RateLimitingTracker : Tracker
+    {
+        public RateLimitingTrackerConnection Connection { get; }
+
+        public RateLimitingTracker (RateLimitingTrackerConnection connection)
+            : base (connection)
+        {
+            Connection = (RateLimitingTrackerConnection) connection;
+        }
+    }
+
+    class RateLimitingTrackerConnection : ITrackerConnection
     {
         public bool CanScrape => true;
 
         public Uri Uri => new Uri ("http://tracker:5353/announce");
 
-        public DefaultTracker ()
+        public List<TaskCompletionSource<TrackerState>> PendingAnnounces { get; } = new List<TaskCompletionSource<TrackerState>> ();
+        public List<TaskCompletionSource<TrackerState>> PendingScrapes { get; } = new List<TaskCompletionSource<TrackerState>> ();
+
+        public RateLimitingTrackerConnection ()
         {
 
         }
 
-        public ReusableTask<AnnounceResponse> AnnounceAsync (AnnounceRequest parameters, CancellationToken token)
+        public async ReusableTask<AnnounceResponse> AnnounceAsync (AnnounceRequest parameters, CancellationToken token)
         {
-            return ReusableTask.FromResult (new AnnounceResponse (TrackerState.Ok));
+            var tcs = new TaskCompletionSource<TrackerState> ();
+            PendingAnnounces.Add (tcs);
+            return new AnnounceResponse (await tcs.Task);
         }
 
-        public ReusableTask<ScrapeResponse> ScrapeAsync (ScrapeRequest parameters, CancellationToken token)
+        public async ReusableTask<ScrapeResponse> ScrapeAsync (ScrapeRequest parameters, CancellationToken token)
         {
-            return ReusableTask.FromResult (new ScrapeResponse (TrackerState.Ok));
+            var tcs = new TaskCompletionSource<TrackerState> ();
+            PendingScrapes.Add (tcs);
+            return new ScrapeResponse (await tcs.Task);
         }
     }
 
@@ -110,17 +129,17 @@ public ScrapeRequest CreateScrape ()
         };
 
         TrackerManager trackerManager;
-        IList<List<TrackerWithConnection>> trackers;
+        IList<List<CustomTracker>> trackers;
 
         Factories Factories { get; set; }
 
         [SetUp]
         public void Setup ()
         {
             Factories = Factories.Default
-                .WithTrackerCreator ("custom", uri => new TrackerWithConnection (new CustomTracker (uri)));
+                .WithTrackerCreator ("custom", uri => new CustomTracker (new CustomTrackerConnection (uri)));
             trackerManager = new TrackerManager (Factories, new RequestFactory (), trackerUrls, true);
-            trackers = trackerManager.Tiers.Select (t => t.Trackers.Cast<TrackerWithConnection> ().ToList ()).ToList ();
+            trackers = trackerManager.Tiers.Select (t => t.Trackers.Cast<CustomTracker> ().ToList ()).ToList ();
         }
 
         [Test]
@@ -231,6 +250,47 @@ public async Task AnnounceFailed ()
             Assert.AreEqual (0, trackers[1][1].Connection.AnnouncedAt.Count, "#8");
         }
 
+        [Test]
+        public async Task Announce_RateLimitedAnnounceAttempts ()
+        {
+            var factories = Factories.Default
+                .WithTrackerCreator ("custom", uri => new RateLimitingTracker (new RateLimitingTrackerConnection ()));
+
+            var tier = new[] { new[] { $"custom://tracker/announce" } };
+            var trackerManager = new TrackerManager (factories, new RequestFactory (), tier, true);
+            var trackers = trackerManager.Tiers.Select (t => t.Trackers.Cast<RateLimitingTracker> ().ToList ()).ToList ();
+
+            // only 1 concurrent regular announce can run at a time. 
+            var announce = trackerManager.AnnounceAsync (CancellationToken.None);
+
+            // These should all early-exit
+            for (int i = 0; i < 3; i++)
+                await trackerManager.AnnounceAsync (CancellationToken.None).WithTimeout (TimeSpan.FromSeconds (10));
+            for (int i = 0; i < 3; i++)
+                await trackerManager.AnnounceAsync (TorrentEvent.None, CancellationToken.None).WithTimeout (TimeSpan.FromSeconds (10));
+
+            Assert.IsFalse (announce.IsCompleted);
+        }
+
+        [Test]
+        public void Announce_RateLimitedTierAnnounces ()
+        {
+            var factories = Factories.Default
+                .WithTrackerCreator ("custom", uri => new RateLimitingTracker (new RateLimitingTrackerConnection ()));
+
+            // Create 100 tracker tiers.
+            var urls = Enumerable.Range (0, 100).Select (t => new[] { $"custom://tracker{t}/announce" }).ToArray ();
+            var trackerManager = new TrackerManager (factories, new RequestFactory (), urls, true);
+            var trackers = trackerManager.Tiers.Select (t => t.Trackers.Cast<RateLimitingTracker> ().ToList ()).ToList ();
+
+            var cts = new CancellationTokenSource (TimeSpan.FromSeconds (10));
+            var announce = trackerManager.AnnounceAsync (CancellationToken.None);
+            while (trackers.SelectMany (t => t).Where (t => t.Connection.PendingAnnounces.Count == 1).Count () != 15) {
+                Thread.Sleep (1);
+                cts.Token.ThrowIfCancellationRequested ();
+            }
+        }
+
         [Test]
         public async Task CurrentTracker ()
         {
@@ -273,7 +333,7 @@ public async Task AnnounceTwice_SendStartedOnce ()
         [Test]
         public void Defaults ()
         {
-            var tracker = new Tracker (new DefaultTracker ());
+            var tracker = new CustomTracker (new CustomTrackerConnection (new Uri("http://tester/announce")));
             Assert.AreEqual (TimeSpan.FromMinutes (3), tracker.MinUpdateInterval, "#1");
             Assert.AreEqual (TimeSpan.FromMinutes (30), tracker.UpdateInterval, "#2");
             Assert.IsNotNull (tracker.WarningMessage, "#3");