Азанов Андрей

homework 2/ClusterClient/Clients/ClusterClientBase.cs

@@ -1,6 +1,8 @@
 using System;
+using System.Collections.Concurrent;
 using System.Diagnostics;
 using System.IO;
+using System.Linq;
 using System.Net;
 using System.Text;
 using System.Threading.Tasks;
@@ -10,6 +12,8 @@ namespace ClusterClient.Clients
 {
     public abstract class ClusterClientBase
     {
+        protected static ConcurrentDictionary<string, ResponseStatistics> ResponseStatisticsHistory { get; } = [];
+
         protected string[] ReplicaAddresses { get; set; }
 
         protected ClusterClientBase(string[] replicaAddresses)
@@ -40,5 +44,27 @@ protected async Task<string> ProcessRequestAsync(WebRequest request)
                 return result;
             }
         }
+
+        protected Task<string> CreateAndRunRequestTask(string replicaAddress, string query)
+        {
+            var uri = new UriBuilder(replicaAddress)
+            {
+                Query = $"query={query}"
+            };
+            var request = CreateRequest(uri.ToString());
+            Log.InfoFormat($"Processing {request.RequestUri}");
+            return ProcessRequestAsync(request);
+        }
+
+        protected string[] GetFilteredAndSortedAddresses()
+        {
+            return ReplicaAddresses
+                .Where(t => !ResponseStatisticsHistory.TryGetValue(t, out var statistics)
+                            || statistics.IsSuccessful)
+                .OrderBy(t => ResponseStatisticsHistory.TryGetValue(t, out var statistics)
+                    ? statistics.ResponseTime
+                    : TimeSpan.Zero)
+                .ToArray();
+        }
     }
 }

homework 2/ClusterClient/Clients/ParallelClusterClient.cs

@@ -1,23 +1,38 @@
 using System;
-using System.Collections.Generic;
 using System.Linq;
-using System.Text;
 using System.Threading.Tasks;
 using log4net;
 
-namespace ClusterClient.Clients
+namespace ClusterClient.Clients;
+
+public class ParallelClusterClient : ClusterClientBase
 {
-    public class ParallelClusterClient : ClusterClientBase
+    public ParallelClusterClient(string[] replicaAddresses) : base(replicaAddresses)
     {
-        public ParallelClusterClient(string[] replicaAddresses) : base(replicaAddresses)
-        {
-        }
+    }
 
-        public override Task<string> ProcessRequestAsync(string query, TimeSpan timeout)
+    public override async Task<string> ProcessRequestAsync(string query, TimeSpan timeout)
+    {
+        var requestTasks = ReplicaAddresses
+            .Select(t => CreateAndRunRequestTask(t, query))
+            .ToHashSet();
+        var timeoutTask = Task.Delay(timeout);
+
+        while (requestTasks.Count > 0 && !timeoutTask.IsCompleted)
         {
-            throw new NotImplementedException();
+            var completedTask = await Task.WhenAny(requestTasks.Append(timeoutTask));
+
+            if (completedTask == timeoutTask)
+                throw new TimeoutException();
+
+            if (completedTask.IsCompletedSuccessfully)
+                return await (Task<string>)completedTask;
+
+            requestTasks.Remove((Task<string>)completedTask);
         }
 
-        protected override ILog Log => LogManager.GetLogger(typeof(ParallelClusterClient));
+        throw new TimeoutException();
     }
-}
+
+    protected override ILog Log => LogManager.GetLogger(typeof(ParallelClusterClient));
+}

homework 2/ClusterClient/Clients/ResponseStatistics.cs

@@ -0,0 +1,5 @@
+using System;
+
+namespace ClusterClient.Clients;
+
+public record ResponseStatistics(bool IsSuccessful, TimeSpan ResponseTime);

homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs

@@ -1,23 +1,45 @@
 using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
+using System.Diagnostics;
 using System.Threading.Tasks;
 using log4net;
 
-namespace ClusterClient.Clients
+namespace ClusterClient.Clients;
+
+public class RoundRobinClusterClient : ClusterClientBase
 {
-    public class RoundRobinClusterClient : ClusterClientBase
+    public RoundRobinClusterClient(string[] replicaAddresses) : base(replicaAddresses)
     {
-        public RoundRobinClusterClient(string[] replicaAddresses) : base(replicaAddresses)
-        {
-        }
+    }
 
-        public override Task<string> ProcessRequestAsync(string query, TimeSpan timeout)
+    public override async Task<string> ProcessRequestAsync(string query, TimeSpan timeout)
+    {
+        var preparedAddresses = GetFilteredAndSortedAddresses();
+        var goodAddressesCount = preparedAddresses.Length;
+        var sw = new Stopwatch();
+        foreach (var replicaAddress in preparedAddresses)
         {
-            throw new NotImplementedException();
+            if (ResponseStatisticsHistory.TryGetValue(replicaAddress, out var statistics) && !statistics.IsSuccessful)
+            {
+                goodAddressesCount--;
+                continue;
+            }
+
+            sw.Restart();
+            var requestTask = CreateAndRunRequestTask(replicaAddress, query);
+            var completedTask = await Task.WhenAny(requestTask, Task.Delay(timeout / goodAddressesCount));
+
+            if (!completedTask.IsCompletedSuccessfully)
+                goodAddressesCount--;
+
+            ResponseStatisticsHistory[replicaAddress] =
+                new ResponseStatistics(completedTask.IsCompletedSuccessfully, sw.Elapsed);
+
+            if (requestTask.IsCompletedSuccessfully)
+                return requestTask.Result;
         }
 
-        protected override ILog Log => LogManager.GetLogger(typeof(RoundRobinClusterClient));
+        throw new TimeoutException();
     }
-}
+
+    protected override ILog Log => LogManager.GetLogger(typeof(RoundRobinClusterClient));
+}

homework 2/ClusterClient/Clients/SmartClusterClient.cs

@@ -1,23 +1,54 @@
 using System;
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.Linq;
-using System.Text;
 using System.Threading.Tasks;
 using log4net;
 
-namespace ClusterClient.Clients
+namespace ClusterClient.Clients;
+
+public class SmartClusterClient : ClusterClientBase
 {
-    public class SmartClusterClient : ClusterClientBase
+    public SmartClusterClient(string[] replicaAddresses) : base(replicaAddresses)
     {
-        public SmartClusterClient(string[] replicaAddresses) : base(replicaAddresses)
-        {
-        }
+    }
 
-        public override Task<string> ProcessRequestAsync(string query, TimeSpan timeout)
+    public override async Task<string> ProcessRequestAsync(string query, TimeSpan timeout)
+    {
+        var preparedAddresses = GetFilteredAndSortedAddresses();
+        var goodAddressesCount = preparedAddresses.Length;
+        var requestTasks = new HashSet<Task<string>>();
+        foreach (var replicaAddress in preparedAddresses)
         {
-            throw new NotImplementedException();
+            if (ResponseStatisticsHistory.TryGetValue(replicaAddress, out var statistics) && !statistics.IsSuccessful)
+            {
+                goodAddressesCount--;
+                continue;
+            }
+
+            var sw = Stopwatch.StartNew();
+            var requestTask = CreateAndRunRequestTask(replicaAddress, query);
+            _ = requestTask.ContinueWith(t => ResponseStatisticsHistory[replicaAddress]
+                = new ResponseStatistics(t.IsCompletedSuccessfully, sw.Elapsed));
+            requestTasks.Add(requestTask);
+            var timeoutTask = Task.Delay(timeout / goodAddressesCount);
+            var completedTask = await Task.WhenAny(requestTasks.Append(timeoutTask));
+
+            if (completedTask == timeoutTask)
+            {
+                ResponseStatisticsHistory[replicaAddress] = new ResponseStatistics(true, sw.Elapsed);
+                continue;
+            }
+
+            if (completedTask.IsCompletedSuccessfully)
+                return await (Task<string>)completedTask;
+
+            goodAddressesCount--;
+            requestTasks.Remove((Task<string>)completedTask);
         }
 
-        protected override ILog Log => LogManager.GetLogger(typeof(SmartClusterClient));
+        throw new TimeoutException();
     }
-}
+
+    protected override ILog Log => LogManager.GetLogger(typeof(SmartClusterClient));
+}