From b899e45cbad98b1c7157fdfc4d7cda4e10e11d30 Mon Sep 17 00:00:00 2001 From: maximka200 Date: Tue, 24 Feb 2026 17:28:45 +0500 Subject: [PATCH 1/4] =?UTF-8?q?-=20=D0=BD=D0=B0=D0=BF=D0=B8=D1=81=D0=B0?= =?UTF-8?q?=D0=BB=20=D0=BF=D0=B0=D1=80=D0=B0=D0=BB=D0=BB=D0=B5=D0=BB=D1=8C?= =?UTF-8?q?=D0=BD=D1=8B=D0=B9=20=D0=BA=D0=BB=D0=B8=D0=B5=D0=BD=D1=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Clients/ParallelClusterClient.cs | 41 ++++++++++++++----- 1 file changed, 31 insertions(+), 10 deletions(-) diff --git a/homework 2/ClusterClient/Clients/ParallelClusterClient.cs b/homework 2/ClusterClient/Clients/ParallelClusterClient.cs index 5531800..fcea1de 100644 --- a/homework 2/ClusterClient/Clients/ParallelClusterClient.cs +++ b/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 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 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)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)); } -} +} \ No newline at end of file From 69133599383a7cebb24900d43f61e49e31c01827 Mon Sep 17 00:00:00 2001 From: maximka200 Date: Thu, 26 Feb 2026 14:21:08 +0500 Subject: [PATCH 2/4] =?UTF-8?q?-=20=D1=80=D0=B5=D0=B0=D0=BB=D0=B8=D0=B7?= =?UTF-8?q?=D0=BE=D0=B2=D0=B0=D0=BB=20RoundRobin?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Clients/RoundRobinClusterClient.cs | 37 ++++++++++++++++--- 1 file changed, 31 insertions(+), 6 deletions(-) diff --git a/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs b/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs index 0293628..01b767a 100644 --- a/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs +++ b/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.Diagnostics; using System.Linq; using System.Text; using System.Threading.Tasks; @@ -7,15 +8,39 @@ namespace ClusterClient.Clients { - public class RoundRobinClusterClient : ClusterClientBase + public class RoundRobinClusterClient(string[] replicaAddresses) : ClusterClientBase(replicaAddresses) { - public RoundRobinClusterClient(string[] replicaAddresses) : base(replicaAddresses) + public override async Task ProcessRequestAsync(string query, TimeSpan timeout) { - } + var start = Stopwatch.StartNew(); - public override Task ProcessRequestAsync(string query, TimeSpan timeout) - { - throw new NotImplementedException(); + for (var i = 0; i < ReplicaAddresses.Length; i++) + { + var remainingTime = timeout - start.Elapsed; + + if (remainingTime <= TimeSpan.Zero) + throw new TimeoutException(); + + var remainingReplicas = ReplicaAddresses.Length - i; + var replicaTimeout = TimeSpan.FromMilliseconds( + remainingTime.TotalMilliseconds / remainingReplicas); + + var uri = ReplicaAddresses[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)); From effc2c0242512832e89b286ffd1d33269564adc7 Mon Sep 17 00:00:00 2001 From: maximka200 Date: Thu, 26 Feb 2026 15:12:26 +0500 Subject: [PATCH 3/4] =?UTF-8?q?-=20=D1=80=D0=B5=D0=B0=D0=BB=D0=B8=D0=B7?= =?UTF-8?q?=D0=BE=D0=B2=D0=B0=D0=BB=20SmaerCluster?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Clients/SmartClusterClient.cs | 39 +++++++++++++++---- 1 file changed, 31 insertions(+), 8 deletions(-) diff --git a/homework 2/ClusterClient/Clients/SmartClusterClient.cs b/homework 2/ClusterClient/Clients/SmartClusterClient.cs index eb06d8b..802c669 100644 --- a/homework 2/ClusterClient/Clients/SmartClusterClient.cs +++ b/homework 2/ClusterClient/Clients/SmartClusterClient.cs @@ -1,23 +1,46 @@ using System; using System.Collections.Generic; using System.Linq; -using System.Text; using System.Threading.Tasks; using log4net; namespace ClusterClient.Clients { - public class SmartClusterClient : ClusterClientBase + public class SmartClusterClient(string[] replicaAddresses) : ClusterClientBase(replicaAddresses) { - public SmartClusterClient(string[] replicaAddresses) : base(replicaAddresses) + public override async Task ProcessRequestAsync(string query, TimeSpan timeout) { - } + var tasks = new List>(); + var taskTimeout = Task.Delay(timeout); - public override Task ProcessRequestAsync(string query, TimeSpan timeout) - { - throw new NotImplementedException(); - } + foreach (var address in ReplicaAddresses) + { + var request = CreateRequest($"{address}?query={query}"); + var task = ProcessRequestAsync(request); + tasks.Add(task); + var completed = await Task.WhenAny(task, Task.Delay(timeout / ReplicaAddresses.Length), taskTimeout); + + if (completed == taskTimeout) + throw new TimeoutException(); + if (completed == task && task.IsCompletedSuccessfully) + return task.Result; + } + + while (tasks.Count != 0) + { + var completed = await Task.WhenAny(tasks.Append(taskTimeout)); + if (completed == taskTimeout) + throw new TimeoutException(); + + var finishedTask = (Task)completed; + tasks.Remove(finishedTask); + if (finishedTask.IsCompletedSuccessfully) + return finishedTask.Result; + } + + throw new TimeoutException(); + } protected override ILog Log => LogManager.GetLogger(typeof(SmartClusterClient)); } } From 008e9bee4f5e75b290d0b309ac7b2d11bdef8000 Mon Sep 17 00:00:00 2001 From: maximka200 Date: Thu, 26 Feb 2026 15:45:32 +0500 Subject: [PATCH 4/4] =?UTF-8?q?-=20=D0=BD=D0=B0=D0=BF=D0=B8=D1=81=D0=B0?= =?UTF-8?q?=D0=BB=20=D1=81=D0=BE=D1=80=D1=82=D0=B8=D1=80=D0=BE=D0=B2=D0=BA?= =?UTF-8?q?=D1=83=20=D1=80=D0=B5=D0=BF=D0=BB=D0=B8=D0=BA=20=D0=BF=D0=BE=20?= =?UTF-8?q?=D0=B2=D1=80=D0=B5=D0=BC=D0=B5=D0=BD=D0=B8=20=D0=BE=D1=82=D0=B2?= =?UTF-8?q?=D0=B5=D1=82=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Clients/ReplicaStatisticsEntry.cs | 19 ++++ .../Clients/RoundRobinClusterClient.cs | 25 ++++- .../Clients/SmartClusterClient.cs | 101 ++++++++++++++---- 3 files changed, 118 insertions(+), 27 deletions(-) create mode 100644 homework 2/ClusterClient/Clients/ReplicaStatisticsEntry.cs diff --git a/homework 2/ClusterClient/Clients/ReplicaStatisticsEntry.cs b/homework 2/ClusterClient/Clients/ReplicaStatisticsEntry.cs new file mode 100644 index 0000000..af04a12 --- /dev/null +++ b/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 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; +} \ No newline at end of file diff --git a/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs b/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs index 01b767a..4694833 100644 --- a/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs +++ b/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs @@ -8,24 +8,39 @@ namespace ClusterClient.Clients { - public class RoundRobinClusterClient(string[] replicaAddresses) : ClusterClientBase(replicaAddresses) + public class RoundRobinClusterClient: ClusterClientBase { + private readonly Dictionary replicaStats; + + public RoundRobinClusterClient(string[] replicaAddresses) : base(replicaAddresses) + { + replicaStats = replicaAddresses.ToDictionary( + addr => addr, + addr => new ReplicaStatisticsEntry(addr) + ); + } + public override async Task ProcessRequestAsync(string query, TimeSpan timeout) { var start = Stopwatch.StartNew(); - - for (var i = 0; i < ReplicaAddresses.Length; i++) + + 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 = ReplicaAddresses.Length - i; + var remainingReplicas = orderedReplicas.Length - i; var replicaTimeout = TimeSpan.FromMilliseconds( remainingTime.TotalMilliseconds / remainingReplicas); - var uri = ReplicaAddresses[i] + "?query=" + query; + var uri = orderedReplicas[i] + "?query=" + query; var webRequest = CreateRequest(uri); Log.InfoFormat($"Processing {webRequest.RequestUri}, timeout {replicaTimeout}"); diff --git a/homework 2/ClusterClient/Clients/SmartClusterClient.cs b/homework 2/ClusterClient/Clients/SmartClusterClient.cs index 802c669..f931a38 100644 --- a/homework 2/ClusterClient/Clients/SmartClusterClient.cs +++ b/homework 2/ClusterClient/Clients/SmartClusterClient.cs @@ -1,46 +1,103 @@ using System; using System.Collections.Generic; +using System.Diagnostics; using System.Linq; using System.Threading.Tasks; using log4net; namespace ClusterClient.Clients { - public class SmartClusterClient(string[] replicaAddresses) : ClusterClientBase(replicaAddresses) + public class SmartClusterClient : ClusterClientBase { + private readonly Dictionary replicaStats; + + public SmartClusterClient(string[] replicaAddresses) : base(replicaAddresses) + { + replicaStats = replicaAddresses.ToDictionary( + addr => addr, + addr => new ReplicaStatisticsEntry(addr) + ); + } + + protected override ILog Log => LogManager.GetLogger(typeof(SmartClusterClient)); + public override async Task ProcessRequestAsync(string query, TimeSpan timeout) { - var tasks = new List>(); - var taskTimeout = Task.Delay(timeout); + var deadline = DateTime.UtcNow + timeout; + var runningTasks = new List<(Task task, string address, Stopwatch timer)>(); + + var orderedReplicas = replicaStats.Values + .OrderBy(x => x.AverageTime) + .Select(x => x.Address) + .ToArray(); - foreach (var address in ReplicaAddresses) - { - var request = CreateRequest($"{address}?query={query}"); - var task = ProcessRequestAsync(request); - tasks.Add(task); - var completed = await Task.WhenAny(task, Task.Delay(timeout / ReplicaAddresses.Length), taskTimeout); + var addressesLeft = orderedReplicas.Length; - if (completed == taskTimeout) + foreach (var replica in orderedReplicas) + { + var remaining = deadline - DateTime.UtcNow; + if (remaining <= TimeSpan.Zero) throw new TimeoutException(); - if (completed == task && task.IsCompletedSuccessfully) - return task.Result; - } + 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; + } - while (tasks.Count != 0) + addressesLeft--; + } + + while (runningTasks.Count > 0) { - var completed = await Task.WhenAny(tasks.Append(taskTimeout)); - if (completed == taskTimeout) + var remaining = deadline - DateTime.UtcNow; + if (remaining <= TimeSpan.Zero) throw new TimeoutException(); - var finishedTask = (Task)completed; - tasks.Remove(finishedTask); - if (finishedTask.IsCompletedSuccessfully) - return finishedTask.Result; + 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, Stopwatch)?> WaitOneAsync( + List<(Task 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)completed; + var result = runningTasks.First(x => x.task == finishedTask); + runningTasks.Remove(result); + + return finishedTask.IsCompletedSuccessfully ? (finishedTask, result.address, result.timer) : null; + } } -} +} \ No newline at end of file