Головнев Максим ДЗ2

homework 2/ClusterClient/Clients/ParallelClusterClient.cs

@@ -1,23 +1,44 @@
 using System;
-using System.Collections.Generic;
 using System.Linq;
-using System.Text;
 using System.Threading.Tasks;
 using log4net;
 
 namespace ClusterClient.Clients
 {
-    public class ParallelClusterClient : ClusterClientBase
+    public class ParallelClusterClient(string[] replicaAddresses)
+        : ClusterClientBase(replicaAddresses)
     {
-        public ParallelClusterClient(string[] replicaAddresses) : base(replicaAddresses)
+        public override async Task<string> ProcessRequestAsync(string query, TimeSpan timeout)
         {
-        }
+            var tasks = ReplicaAddresses
+                .Select(address =>
+                {
+                    var request = CreateRequest(address + "?query=" + query);
+                    Log.InfoFormat($"Processing {request.RequestUri}");
+                    return ProcessRequestAsync(request);
+                })
+                .ToList();
 
-        public override Task<string> ProcessRequestAsync(string query, TimeSpan timeout)
-        {
-            throw new NotImplementedException();
+            var timeoutTask = Task.Delay(timeout);
+
+            while (tasks.Count > 0)
+            {
+                var completed = await Task.WhenAny(tasks.Append(timeoutTask));
+
+                if (completed == timeoutTask)
+                    throw new TimeoutException();
+
+                var finishedTask = (Task<string>)completed;
+                tasks.Remove(finishedTask);
+
+                if (finishedTask.Status == TaskStatus.RanToCompletion)
+                    return finishedTask.Result;
+            }
+
+            throw new TimeoutException();
         }
 
-        protected override ILog Log => LogManager.GetLogger(typeof(ParallelClusterClient));
+        protected override ILog Log =>
+            LogManager.GetLogger(typeof(ParallelClusterClient));
     }
-}
+}

homework 2/ClusterClient/Clients/ReplicaStatisticsEntry.cs

@@ -0,0 +1,19 @@
+using System.Collections.Generic;
+using System.Linq;
+
+namespace ClusterClient.Clients;
+
+public class ReplicaStatisticsEntry(string address, int maxHistory = 10)
+{
+    public string Address { get; } = address;
+    private readonly Queue<long> lastTimes = new();
+
+    public void RecordResponseTime(long ms)
+    {
+        lastTimes.Enqueue(ms);
+        if (lastTimes.Count > maxHistory)
+            lastTimes.Dequeue();
+    }
+
+    public double AverageTime => lastTimes.Any() ? lastTimes.Average() : double.MaxValue;
+}

homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs

@@ -1,21 +1,61 @@
 using System;
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.Linq;
 using System.Text;
 using System.Threading.Tasks;
 using log4net;
 
 namespace ClusterClient.Clients
 {
-    public class RoundRobinClusterClient : ClusterClientBase
+    public class RoundRobinClusterClient: ClusterClientBase
     {
+        private readonly Dictionary<string, ReplicaStatisticsEntry> replicaStats;
+
         public RoundRobinClusterClient(string[] replicaAddresses) : base(replicaAddresses)
         {
+            replicaStats = replicaAddresses.ToDictionary(
+                addr => addr,
+                addr => new ReplicaStatisticsEntry(addr)
+            );
         }
-
-        public override Task<string> ProcessRequestAsync(string query, TimeSpan timeout)
+        
+        public override async Task<string> ProcessRequestAsync(string query, TimeSpan timeout)
         {
-            throw new NotImplementedException();
+            var start = Stopwatch.StartNew();
+
+            var orderedReplicas = replicaStats.Values
+                .OrderBy(x => x.AverageTime)
+                .Select(x => x.Address)
+                .ToArray();
+
+            for (var i = 0; i < orderedReplicas.Length; i++)
+            {
+                var remainingTime = timeout - start.Elapsed;
+
+                if (remainingTime <= TimeSpan.Zero)
+                    throw new TimeoutException();
+
+                var remainingReplicas = orderedReplicas.Length - i;
+                var replicaTimeout = TimeSpan.FromMilliseconds(
+                    remainingTime.TotalMilliseconds / remainingReplicas);
+
+                var uri = orderedReplicas[i] + "?query=" + query;
+                var webRequest = CreateRequest(uri);
+
+                Log.InfoFormat($"Processing {webRequest.RequestUri}, timeout {replicaTimeout}");
+
+                var requestTask = ProcessRequestAsync(webRequest);
+                var delayTask = Task.Delay(replicaTimeout);
+
+                var completed = await Task.WhenAny(requestTask, delayTask);
+
+                if (completed != requestTask) continue;
+                if (!requestTask.IsFaulted)
+                    return requestTask.Result;
+            }
+
+            throw new TimeoutException();
         }
 
         protected override ILog Log => LogManager.GetLogger(typeof(RoundRobinClusterClient));

homework 2/ClusterClient/Clients/SmartClusterClient.cs

@@ -1,23 +1,103 @@
 using System;
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.Linq;
-using System.Text;
 using System.Threading.Tasks;
 using log4net;
 
 namespace ClusterClient.Clients
 {
     public class SmartClusterClient : ClusterClientBase
     {
+        private readonly Dictionary<string, ReplicaStatisticsEntry> replicaStats;
+
         public SmartClusterClient(string[] replicaAddresses) : base(replicaAddresses)
         {
+            replicaStats = replicaAddresses.ToDictionary(
+                addr => addr,
+                addr => new ReplicaStatisticsEntry(addr)
+            );
         }
 
-        public override Task<string> ProcessRequestAsync(string query, TimeSpan timeout)
+        protected override ILog Log => LogManager.GetLogger(typeof(SmartClusterClient));
+
+        public override async Task<string> ProcessRequestAsync(string query, TimeSpan timeout)
         {
-            throw new NotImplementedException();
+            var deadline = DateTime.UtcNow + timeout;
+            var runningTasks = new List<(Task<string> task, string address, Stopwatch timer)>();
+
+            var orderedReplicas = replicaStats.Values
+                .OrderBy(x => x.AverageTime)
+                .Select(x => x.Address)
+                .ToArray();
+
+            var addressesLeft = orderedReplicas.Length;
+
+            foreach (var replica in orderedReplicas)
+            {
+                var remaining = deadline - DateTime.UtcNow;
+                if (remaining <= TimeSpan.Zero)
+                    throw new TimeoutException();
+
+                var replicaTimeout = remaining / addressesLeft;
+
+                var request = CreateRequest($"{replica}?query={query}");
+                Log.InfoFormat($"Processing {request.RequestUri}");
+
+                var timer = Stopwatch.StartNew();
+                var task = ProcessRequestAsync(request);
+                runningTasks.Add((task, replica, timer));
+
+                var finished = await WaitOneAsync(runningTasks, replicaTimeout);
+                if (finished != null)
+                {
+                    var (resultTask, resultReplica, resultTimer) = finished.Value;
+                    resultTimer.Stop();
+                    replicaStats[resultReplica].RecordResponseTime(resultTimer.ElapsedMilliseconds);
+                    return await resultTask;
+                }
+
+                addressesLeft--;
+            }
+
+            while (runningTasks.Count > 0)
+            {
+                var remaining = deadline - DateTime.UtcNow;
+                if (remaining <= TimeSpan.Zero)
+                    throw new TimeoutException();
+
+                var finished = await WaitOneAsync(runningTasks, remaining);
+                if (finished == null) continue;
+
+                var (resultTask, resultReplica, resultTimer) = finished.Value;
+                resultTimer.Stop();
+                replicaStats[resultReplica].RecordResponseTime(resultTimer.ElapsedMilliseconds);
+                return await resultTask;
+            }
+
+            throw new TimeoutException();
         }
 
-        protected override ILog Log => LogManager.GetLogger(typeof(SmartClusterClient));
+        private async Task<(Task<string>, string, Stopwatch)?> WaitOneAsync(
+            List<(Task<string> task, string address, Stopwatch timer)> runningTasks,
+            TimeSpan timeout)
+        {
+            if (timeout <= TimeSpan.Zero)
+                return null;
+
+            var delayTask = Task.Delay(timeout);
+            var tasksList = runningTasks.Select(x => x.task).Append(delayTask).ToList();
+
+            var completed = await Task.WhenAny(tasksList);
+
+            if (completed == delayTask)
+                return null;
+
+            var finishedTask = (Task<string>)completed;
+            var result = runningTasks.First(x => x.task == finishedTask);
+            runningTasks.Remove(result);
+
+            return finishedTask.IsCompletedSuccessfully ? (finishedTask, result.address, result.timer) : null;
+        }
     }
-}
+}