Merge pull request #35 from zetroot/feature/parallel-observations

+semver: feature

Author: zetroot

src/NTorSpectator.Observer/Program.cs

@@ -50,6 +50,7 @@
     var jobDetail = JobBuilder.Create<SpectatorJob>()
         .WithDescription("Tor spectator job")
         .WithIdentity("tor-spectator")
+        .DisallowConcurrentExecution()
         .Build();
     cfg.AddJob<SpectatorJob>(jobKey: jobDetail.Key, configure: j => {});
     cfg.AddTrigger(t => t.WithCronSchedule("0 0 * * * ?").ForJob(jobDetail));

src/NTorSpectator.Observer/Services/SpectatorJob.cs

@@ -1,3 +1,4 @@
+using System.Collections.Concurrent;
 using System.Diagnostics;
 using NTorSpectator.Observer.TorIntegration;
 using NTorSpectator.Services;
@@ -22,15 +23,13 @@ public class SpectatorJob : IJob
 
     private readonly ILogger<SpectatorJob> _logger;
     private readonly ISitesCatalogue _sitesCatalogue;
-    private readonly TorControlManager _torControl;
-    private readonly ISiteObserver _siteObserver;
-
-    public SpectatorJob(ILogger<SpectatorJob> logger, ISitesCatalogue sitesCatalogue, TorControlManager torControl, ISiteObserver siteObserver)
+    private readonly IServiceProvider _services;
+    
+    public SpectatorJob(ILogger<SpectatorJob> logger, ISitesCatalogue sitesCatalogue, IServiceProvider services)
     {
         _logger = logger;
         _sitesCatalogue = sitesCatalogue;
-        _torControl = torControl;
-        _siteObserver = siteObserver;
+        _services = services;
     }
 
     public async Task Execute(IJobExecutionContext context)
@@ -40,50 +39,74 @@ public async Task Execute(IJobExecutionContext context)
         _logger.LogDebug("Starting sites observations");
         var sites = await _sitesCatalogue.GetAllSites();
         _logger.LogDebug("Got {Count} sites to observe", sites.Count);
-        
-        var siteQueue = new Queue<QueuedSite>(sites.Select(x => new QueuedSite(x, 0)));
-        while(siteQueue.TryDequeue(out var queuedSite))
+        var queuedSites = sites.Select(x => new QueuedSite(x, 0));
+        var siteQueue = new ConcurrentQueue<QueuedSite>(queuedSites);
+        while(true)
         {
             QueueLength.Set(siteQueue.Count);
-            using var _ = _logger.BeginScope(new Dictionary<string, object> { { "HiddenService", queuedSite.Site.SiteUri } });
-            _logger.LogDebug("Starting observations on the next site");
-            try
-            {
-                var observations = await ObserveSite(queuedSite.Site.SiteUri);
-                ObservationsCount.Inc();
-                if (!observations.IsOk)
-                {
-                    _logger.LogDebug("Site observed as not available");
-                    var siteObservationsCount = queuedSite.ObservationsCount;
-                    if (siteObservationsCount < 3)
-                    {
-                        _logger.LogDebug("Site has been observed {Count} times, returning it to queue", siteObservationsCount);
-                        siteQueue.Enqueue(queuedSite with{ObservationsCount = siteObservationsCount + 1});
-                        RetriesCount.Inc();
-                        continue;
-                    }
-                }
-                _logger.LogDebug("Site seems to be up");
-                await _siteObserver.AddNewObservation(queuedSite.Site.SiteUri, observations.IsOk);
-                SiteStatus.WithLabels(queuedSite.Site.SiteUri).Set(observations.IsOk ? 1 : 0);
-                _logger.LogInformation("Site observed");
-            }
-            catch (Exception e)
-            {
-                _logger.LogError(e, "Observation for site failed");
-            }
+            if (!siteQueue.Any())
+                break;
+            var chunk = TryDequeueMax(siteQueue, 4);
+            var tasks = chunk.Select(x => TryCheckSite(x, s => siteQueue.Enqueue(s))).ToArray();
+            await Task.WhenAll(tasks);
         }
         _logger.LogDebug("The queue is finally empty, observations finished");
         sw.Stop();
         TotalSessionDuration.Set(sw.ElapsedMilliseconds);
     }
+
+
+    private QueuedSite[] TryDequeueMax(ConcurrentQueue<QueuedSite> queue, int max)
+    {
+        var result = new List<QueuedSite>(max);
+        for (int i = 0; i < max; ++i)
+        {
+            if (queue.TryDequeue(out var item))
+                result.Add(item);
+        }
+        return result.ToArray();
+    }
 
+    private async Task TryCheckSite(QueuedSite queuedSite, Action<QueuedSite> putBackIntoQueue)
+    {
+        using var _ = _logger.BeginScope(new Dictionary<string, object> { { "HiddenService", queuedSite.Site.SiteUri } });
+        using var scope = _services.CreateScope();
+        var torControl = scope.ServiceProvider.GetRequiredService<TorControlManager>();
+        var siteObserver = scope.ServiceProvider.GetRequiredService<ISiteObserver>();
+        _logger.LogDebug("Starting observations on the next site");
+        try
+        {
+            var observations = await ObserveSite(queuedSite.Site.SiteUri, torControl);
+            ObservationsCount.Inc();
+            if (!observations.IsOk)
+            {
+                _logger.LogDebug("Site observed as not available");
+                var siteObservationsCount = queuedSite.ObservationsCount;
+                if (siteObservationsCount < 3)
+                {
+                    _logger.LogDebug("Site has been observed {Count} times, returning it to queue", siteObservationsCount);
+                    putBackIntoQueue(queuedSite with { ObservationsCount = siteObservationsCount + 1 });
+                    RetriesCount.Inc();
+                    return;
+                }
+            }
+            _logger.LogDebug("Site seems to be up");
+            await siteObserver.AddNewObservation(queuedSite.Site.SiteUri, observations.IsOk);
+            SiteStatus.WithLabels(queuedSite.Site.SiteUri).Set(observations.IsOk ? 1 : 0);
+            _logger.LogInformation("Site observed");
+        }
+        catch (Exception e)
+        {
+            _logger.LogError(e, "Observation for site failed");
+        }
+    }
+
     private record QueuedSite(Site Site, int ObservationsCount);
 
-    private async Task<TorWatchResults> ObserveSite(string site)
+    private async Task<TorWatchResults> ObserveSite(string site, TorControlManager torControl)
     {
         using var _ = RequestDuration.NewTimer();
-        var torReply = await _torControl.HsFetch(site);
+        var torReply = await torControl.HsFetch(site);
         var positive = torReply.Count(x => x.Action == HsDescAction.Received);
         var negative = torReply.Count(x => x.Action == HsDescAction.Failed);
         return new(site, positive, negative);