From ed7ea3e29bef6cbd18ed6d2f5b9eb919e540c34b Mon Sep 17 00:00:00 2001 From: shept Date: Wed, 4 Mar 2026 03:46:24 +0500 Subject: [PATCH 1/4] implement ParallelClusterClient.cs --- .../Clients/ParallelClusterClient.cs | 26 ++++++++++++++++--- 1 file changed, 23 insertions(+), 3 deletions(-) diff --git a/homework 2/ClusterClient/Clients/ParallelClusterClient.cs b/homework 2/ClusterClient/Clients/ParallelClusterClient.cs index 5531800..ec8080f 100644 --- a/homework 2/ClusterClient/Clients/ParallelClusterClient.cs +++ b/homework 2/ClusterClient/Clients/ParallelClusterClient.cs @@ -13,11 +13,31 @@ public ParallelClusterClient(string[] replicaAddresses) : base(replicaAddresses) { } - public override Task ProcessRequestAsync(string query, TimeSpan timeout) + public override async Task ProcessRequestAsync(string query, TimeSpan timeout) { - throw new NotImplementedException(); + var requests = ReplicaAddresses + .Select(uri => CreateRequest(uri + "?query=" + query)) + .Select(ProcessRequestAsync) + .ToList(); + + var delay = Task.Delay(timeout); + + while (requests.Count > 0) + { + var completedTask = await Task.WhenAny(requests.Concat([delay])); + + if (completedTask == delay) + throw new TimeoutException(); + + requests.Remove((Task)completedTask); + + if (completedTask.IsCompletedSuccessfully) + return await (Task)completedTask; + } + + return null; } protected override ILog Log => LogManager.GetLogger(typeof(ParallelClusterClient)); } -} +} \ No newline at end of file From 0bd3dd6cc426eb1c08516415e1d58ece5099718e Mon Sep 17 00:00:00 2001 From: shept Date: Wed, 4 Mar 2026 14:54:11 +0500 Subject: [PATCH 2/4] implement RoundRobinClusterClient.cs --- .../Clients/RoundRobinClusterClient.cs | 33 +++++++++++++++---- 1 file changed, 27 insertions(+), 6 deletions(-) diff --git a/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs b/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs index 0293628..f1fffd1 100644 --- a/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs +++ b/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs @@ -1,7 +1,5 @@ using System; -using System.Collections.Generic; -using System.Linq; -using System.Text; +using System.Diagnostics; using System.Threading.Tasks; using log4net; @@ -13,11 +11,34 @@ public RoundRobinClusterClient(string[] replicaAddresses) : base(replicaAddresse { } - public override Task ProcessRequestAsync(string query, TimeSpan timeout) + public override async Task ProcessRequestAsync(string query, TimeSpan timeout) { - throw new NotImplementedException(); + var stopwatch = Stopwatch.StartNew(); + var remainingReplicas = ReplicaAddresses.Length; + + foreach (var replicaAddress in ReplicaAddresses) + { + var timeoutPerReplica = (timeout - stopwatch.Elapsed) / remainingReplicas--; + + var request = CreateRequest(replicaAddress + "?query=" + query); + var requestTask = ProcessRequestAsync(request); + var delay = Task.Delay(timeoutPerReplica); + + var completedTask = await Task.WhenAny(requestTask, delay); + + if (completedTask == delay) + continue; + + if (!completedTask.IsCompletedSuccessfully) + continue; + + return await requestTask; + } + + throw new TimeoutException(); } protected override ILog Log => LogManager.GetLogger(typeof(RoundRobinClusterClient)); } -} + +} \ No newline at end of file From e780764a9079f351cc1731cb70ffa06f51657a62 Mon Sep 17 00:00:00 2001 From: shept Date: Wed, 4 Mar 2026 15:30:14 +0500 Subject: [PATCH 3/4] implement SmartClusterClient.cs --- .../Clients/SmartClusterClient.cs | 44 ++++++++++++++++++- 1 file changed, 42 insertions(+), 2 deletions(-) diff --git a/homework 2/ClusterClient/Clients/SmartClusterClient.cs b/homework 2/ClusterClient/Clients/SmartClusterClient.cs index eb06d8b..9e63fc9 100644 --- a/homework 2/ClusterClient/Clients/SmartClusterClient.cs +++ b/homework 2/ClusterClient/Clients/SmartClusterClient.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.Diagnostics; using System.Linq; using System.Text; using System.Threading.Tasks; @@ -13,9 +14,48 @@ public SmartClusterClient(string[] replicaAddresses) : base(replicaAddresses) { } - public override Task ProcessRequestAsync(string query, TimeSpan timeout) + public override async Task ProcessRequestAsync(string query, TimeSpan timeout) { - throw new NotImplementedException(); + var stopwatch = Stopwatch.StartNew(); + var remainingReplicas = ReplicaAddresses.Length; + var pendingRequests = new List>(); + + foreach (var replicaAddress in ReplicaAddresses) + { + var timeoutPerReplica = (timeout - stopwatch.Elapsed) / remainingReplicas--; + + pendingRequests.Add(ProcessRequestAsync(CreateRequest(replicaAddress + "?query=" + query))); + + var delay = Task.Delay(timeoutPerReplica); + var completedTask = await Task.WhenAny(pendingRequests.Append(delay)) as Task; + + if (completedTask == null) + continue; + + pendingRequests.Remove(completedTask); + + if (completedTask.IsCompletedSuccessfully) + return await completedTask; + } + + while (pendingRequests.Count > 0) + { + var remainingTime = timeout - stopwatch.Elapsed; + if (remainingTime <= TimeSpan.Zero) + break; + + var completedTask = await Task.WhenAny(pendingRequests.Append(Task.Delay(remainingTime))) as Task; + + if (completedTask == null) + break; + + pendingRequests.Remove(completedTask); + + if (completedTask.IsCompletedSuccessfully) + return await completedTask; + } + + throw new TimeoutException(); } protected override ILog Log => LogManager.GetLogger(typeof(SmartClusterClient)); From 91d194b313b1cf5fab8e9f5f69fef894fd0cfc76 Mon Sep 17 00:00:00 2001 From: shept Date: Wed, 4 Mar 2026 17:06:35 +0500 Subject: [PATCH 4/4] add replicas statistics --- .../Clients/RoundRobinClusterClient.cs | 18 ++++++++++++++---- .../Clients/SmartClusterClient.cs | 18 ++++++++++++++---- 2 files changed, 28 insertions(+), 8 deletions(-) diff --git a/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs b/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs index f1fffd1..3e077fd 100644 --- a/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs +++ b/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs @@ -1,5 +1,8 @@ using System; +using System.Collections.Concurrent; +using System.Collections.Generic; using System.Diagnostics; +using System.Linq; using System.Threading.Tasks; using log4net; @@ -7,16 +10,23 @@ namespace ClusterClient.Clients { public class RoundRobinClusterClient : ClusterClientBase { + private readonly ConcurrentDictionary _replicaTimings = new(); + public RoundRobinClusterClient(string[] replicaAddresses) : base(replicaAddresses) { + foreach (var replicaAddress in replicaAddresses) + { + _replicaTimings.TryAdd(replicaAddress, 0); + } } public override async Task ProcessRequestAsync(string query, TimeSpan timeout) { var stopwatch = Stopwatch.StartNew(); + var sortedReplicas = ReplicaAddresses.OrderBy(a => _replicaTimings.GetValueOrDefault(a, long.MaxValue)); var remainingReplicas = ReplicaAddresses.Length; - foreach (var replicaAddress in ReplicaAddresses) + foreach (var replicaAddress in sortedReplicas) { var timeoutPerReplica = (timeout - stopwatch.Elapsed) / remainingReplicas--; @@ -25,13 +35,14 @@ public override async Task ProcessRequestAsync(string query, TimeSpan ti var delay = Task.Delay(timeoutPerReplica); var completedTask = await Task.WhenAny(requestTask, delay); + _replicaTimings[replicaAddress] = stopwatch.ElapsedMilliseconds; if (completedTask == delay) continue; - + if (!completedTask.IsCompletedSuccessfully) continue; - + return await requestTask; } @@ -40,5 +51,4 @@ public override async Task ProcessRequestAsync(string query, TimeSpan ti protected override ILog Log => LogManager.GetLogger(typeof(RoundRobinClusterClient)); } - } \ No newline at end of file diff --git a/homework 2/ClusterClient/Clients/SmartClusterClient.cs b/homework 2/ClusterClient/Clients/SmartClusterClient.cs index 9e63fc9..0b2a22e 100644 --- a/homework 2/ClusterClient/Clients/SmartClusterClient.cs +++ b/homework 2/ClusterClient/Clients/SmartClusterClient.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Concurrent; using System.Collections.Generic; using System.Diagnostics; using System.Linq; @@ -10,24 +11,32 @@ namespace ClusterClient.Clients { public class SmartClusterClient : ClusterClientBase { + private readonly ConcurrentDictionary _replicaTimings = new(); + public SmartClusterClient(string[] replicaAddresses) : base(replicaAddresses) { + foreach (var replicaAddress in replicaAddresses) + { + _replicaTimings.TryAdd(replicaAddress, 0); + } } public override async Task ProcessRequestAsync(string query, TimeSpan timeout) { var stopwatch = Stopwatch.StartNew(); + var sortedReplicas = ReplicaAddresses.OrderBy(a => _replicaTimings.GetValueOrDefault(a, long.MaxValue)); var remainingReplicas = ReplicaAddresses.Length; var pendingRequests = new List>(); - foreach (var replicaAddress in ReplicaAddresses) + foreach (var replicaAddress in sortedReplicas) { var timeoutPerReplica = (timeout - stopwatch.Elapsed) / remainingReplicas--; - + pendingRequests.Add(ProcessRequestAsync(CreateRequest(replicaAddress + "?query=" + query))); var delay = Task.Delay(timeoutPerReplica); var completedTask = await Task.WhenAny(pendingRequests.Append(delay)) as Task; + _replicaTimings[replicaAddress] = stopwatch.ElapsedMilliseconds; if (completedTask == null) continue; @@ -44,7 +53,8 @@ public override async Task ProcessRequestAsync(string query, TimeSpan ti if (remainingTime <= TimeSpan.Zero) break; - var completedTask = await Task.WhenAny(pendingRequests.Append(Task.Delay(remainingTime))) as Task; + var completedTask = + await Task.WhenAny(pendingRequests.Append(Task.Delay(remainingTime))) as Task; if (completedTask == null) break; @@ -60,4 +70,4 @@ public override async Task ProcessRequestAsync(string query, TimeSpan ti protected override ILog Log => LogManager.GetLogger(typeof(SmartClusterClient)); } -} +} \ No newline at end of file