From 46edc8f3a5cfa9d8ac1c5e74b7f62d2e8eb14922 Mon Sep 17 00:00:00 2001 From: Foma Medvedev Date: Mon, 23 Feb 2026 15:10:51 +0500 Subject: [PATCH 1/5] =?UTF-8?q?=D1=80=D0=B5=D0=B0=D0=BB=D0=B8=D0=B7=D0=BE?= =?UTF-8?q?=D0=B2=D0=B0=D0=BB=20=D0=BF=D0=B0=D1=80=D0=B0=D0=BB=D0=BB=D0=B5?= =?UTF-8?q?=D0=BB=D1=8C=D0=BD=D0=BE=D0=B3=D0=BE=20=D0=BA=D0=BB=D0=B8=D0=B5?= =?UTF-8?q?=D0=BD=D1=82=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Clients/ParallelClusterClient.cs | 29 +++++++++++++++++-- 1 file changed, 27 insertions(+), 2 deletions(-) diff --git a/homework 2/ClusterClient/Clients/ParallelClusterClient.cs b/homework 2/ClusterClient/Clients/ParallelClusterClient.cs index 5531800..8e55141 100644 --- a/homework 2/ClusterClient/Clients/ParallelClusterClient.cs +++ b/homework 2/ClusterClient/Clients/ParallelClusterClient.cs @@ -1,7 +1,9 @@ using System; using System.Collections.Generic; using System.Linq; +using System.Net; using System.Text; +using System.Threading; using System.Threading.Tasks; using log4net; @@ -13,9 +15,32 @@ 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 tasks = new List>(); + for (var i = 0; i < ReplicaAddresses.Length; ++i) + { + var request = CreateRequest(ReplicaAddresses[i] + "?query=" + query); + tasks.Add(ProcessRequestAsync(request)); + } + + var timeoutTask = Task.Delay(timeout).ContinueWith(_ => throw new TimeoutException()); + + while (tasks.Count > 0) + { + var completedTask = await Task.WhenAny(tasks.Cast().Concat(new[] {timeoutTask})); + if (completedTask == timeoutTask) + throw new TimeoutException(); + try + { + return await (Task)completedTask; + } + catch + { + tasks.Remove((Task)completedTask); + } + } + throw new TimeoutException(); } protected override ILog Log => LogManager.GetLogger(typeof(ParallelClusterClient)); From f0f6c746a545b86b49491651b8f5e3fe03db0517 Mon Sep 17 00:00:00 2001 From: Foma Medvedev Date: Mon, 23 Feb 2026 15:34:41 +0500 Subject: [PATCH 2/5] =?UTF-8?q?=D1=81=D0=B4=D0=B5=D0=BB=D0=B0=D0=BB=20Roun?= =?UTF-8?q?dRobin-=D0=BA=D0=BB=D0=B8=D0=B5=D0=BD=D1=82=D0=B0=20(=D0=BF?= =?UTF-8?q?=D0=BE=D0=BA=D0=B0=20=D0=B1=D0=B5=D0=B7=20=D0=B8=D1=81=D1=82?= =?UTF-8?q?=D0=BE=D1=80=D0=B8=D0=B8)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Clients/RoundRobinClusterClient.cs | 17 +++++++++++++++-- 1 file changed, 15 insertions(+), 2 deletions(-) diff --git a/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs b/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs index 0293628..a50761c 100644 --- a/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs +++ b/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs @@ -13,9 +13,22 @@ 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 timeoutForReplica = timeout / ReplicaAddresses.Length; + foreach (var replicaAddress in ReplicaAddresses) + { + var timeoutTask = Task.Delay(timeoutForReplica); + var request = CreateRequest(replicaAddress + "?query=" + query); + var requestResult = await Task.WhenAny(ProcessRequestAsync(request), timeoutTask); + if (requestResult != timeoutTask) + { + var result = (Task)requestResult; + if (result.Status == TaskStatus.RanToCompletion) + return result.Result; + } + } + throw new TimeoutException(); } protected override ILog Log => LogManager.GetLogger(typeof(RoundRobinClusterClient)); From 0c1ec8fd353578ec69673d56df627070157323fa Mon Sep 17 00:00:00 2001 From: Foma Medvedev Date: Mon, 23 Feb 2026 16:15:58 +0500 Subject: [PATCH 3/5] =?UTF-8?q?=D1=81=D0=B4=D0=B5=D0=BB=D0=B0=D0=BB=20Smar?= =?UTF-8?q?t-=D0=BA=D0=BB=D0=B8=D0=B5=D0=BD=D1=82=D0=B0=20(=D0=BF=D0=BE?= =?UTF-8?q?=D0=BA=D0=B0=20=D0=B1=D0=B5=D0=B7=20=D0=B8=D1=81=D1=82=D0=BE?= =?UTF-8?q?=D1=80=D0=B8=D0=B8)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Clients/SmartClusterClient.cs | 20 +++++++++++++++++-- 1 file changed, 18 insertions(+), 2 deletions(-) diff --git a/homework 2/ClusterClient/Clients/SmartClusterClient.cs b/homework 2/ClusterClient/Clients/SmartClusterClient.cs index eb06d8b..3be4bf6 100644 --- a/homework 2/ClusterClient/Clients/SmartClusterClient.cs +++ b/homework 2/ClusterClient/Clients/SmartClusterClient.cs @@ -13,9 +13,25 @@ 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 timeoutForReplica = timeout / ReplicaAddresses.Length; + var requests = new List>(); + foreach (var replicaAddress in ReplicaAddresses) + { + var timeoutTask = Task.Delay(timeoutForReplica); + var request = CreateRequest(replicaAddress + "?query=" + query); + requests.Add(ProcessRequestAsync(request)); + var requestResult = await Task.WhenAny(requests.Cast().Concat(new[] {timeoutTask})); + if (requestResult != timeoutTask) + { + var result = (Task)requestResult; + if (result.Status == TaskStatus.RanToCompletion) + return result.Result; + requests.Remove(result); + } + } + throw new TimeoutException(); } protected override ILog Log => LogManager.GetLogger(typeof(SmartClusterClient)); From 3d0eb6110debab3d4e0ced5dc10065ab310ca351 Mon Sep 17 00:00:00 2001 From: Foma Medvedev Date: Mon, 23 Feb 2026 18:17:17 +0500 Subject: [PATCH 4/5] =?UTF-8?q?=D0=B4=D0=BE=D0=B1=D0=B0=D0=B2=D0=B8=D0=BB?= =?UTF-8?q?=20=D0=B8=D1=81=D1=82=D0=BE=D1=80=D0=B8=D1=8E=20=D0=B4=D0=BB?= =?UTF-8?q?=D1=8F=20Smart=20=D0=B8=20RoundRobin?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Clients/RoundRobinClusterClient.cs | 27 +++++++++++++--- .../Clients/SmartClusterClient.cs | 31 ++++++++++++++----- .../ClusterClient/ReplicasStatistics.cs | 28 +++++++++++++++++ 3 files changed, 75 insertions(+), 11 deletions(-) create mode 100644 homework 2/ClusterClient/ReplicasStatistics.cs diff --git a/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs b/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs index a50761c..6d2ee65 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; @@ -9,23 +10,41 @@ namespace ClusterClient.Clients { public class RoundRobinClusterClient : ClusterClientBase { + private readonly ReplicasStatistics _replicasStatistics; public RoundRobinClusterClient(string[] replicaAddresses) : base(replicaAddresses) { + _replicasStatistics = new ReplicasStatistics(replicaAddresses); } public override async Task ProcessRequestAsync(string query, TimeSpan timeout) { - var timeoutForReplica = timeout / ReplicaAddresses.Length; - foreach (var replicaAddress in ReplicaAddresses) + var timer = Stopwatch.StartNew(); + var sortedReplicas = ReplicaAddresses + .OrderBy(x => _replicasStatistics.GetStats(x)) + .ToArray(); + var sentTasks = new Dictionary, (string, DateTime)>(); + for (int i = 0; i < sortedReplicas.Length; i++) { + var remainingTime = timeout - timer.Elapsed; + var timeoutForReplica = TimeSpan.FromTicks(remainingTime.Ticks / (sortedReplicas.Length - i)); var timeoutTask = Task.Delay(timeoutForReplica); - var request = CreateRequest(replicaAddress + "?query=" + query); - var requestResult = await Task.WhenAny(ProcessRequestAsync(request), timeoutTask); + var request = CreateRequest(sortedReplicas[i] + "?query=" + query); + var task = ProcessRequestAsync(request); + sentTasks.Add(task, (address: sortedReplicas[i], startTime: DateTime.Now)); + var requestResult = await Task.WhenAny(task, timeoutTask); if (requestResult != timeoutTask) { + var replicaData = sentTasks[task]; + var time = DateTime.Now - replicaData.Item2; var result = (Task)requestResult; if (result.Status == TaskStatus.RanToCompletion) + { + _replicasStatistics.UpdateStats(replicaData.Item1, time.TotalMilliseconds); + sentTasks.Remove(task); return result.Result; + } + _replicasStatistics.UpdateStats(replicaData.Item1, timeout.TotalMilliseconds); + sentTasks.Remove(task); } } throw new TimeoutException(); diff --git a/homework 2/ClusterClient/Clients/SmartClusterClient.cs b/homework 2/ClusterClient/Clients/SmartClusterClient.cs index 3be4bf6..d19c2ad 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; @@ -9,31 +10,47 @@ namespace ClusterClient.Clients { public class SmartClusterClient : ClusterClientBase { + private readonly ReplicasStatistics _replicasStatistics; + public SmartClusterClient(string[] replicaAddresses) : base(replicaAddresses) { + _replicasStatistics = new ReplicasStatistics(replicaAddresses); } public override async Task ProcessRequestAsync(string query, TimeSpan timeout) { - var timeoutForReplica = timeout / ReplicaAddresses.Length; - var requests = new List>(); - foreach (var replicaAddress in ReplicaAddresses) + var timer = Stopwatch.StartNew(); + var sortedReplicas = ReplicaAddresses + .OrderBy(x => _replicasStatistics.GetStats(x)) + .ToArray(); + var requests = new Dictionary, (string, DateTime)>(); + for (int i = 0; i < sortedReplicas.Length; i++) { + var remainingTime = timeout - timer.Elapsed; + var timeoutForReplica = TimeSpan.FromTicks(remainingTime.Ticks / (sortedReplicas.Length - i)); var timeoutTask = Task.Delay(timeoutForReplica); - var request = CreateRequest(replicaAddress + "?query=" + query); - requests.Add(ProcessRequestAsync(request)); - var requestResult = await Task.WhenAny(requests.Cast().Concat(new[] {timeoutTask})); + var request = CreateRequest(sortedReplicas[i] + "?query=" + query); + requests.Add(ProcessRequestAsync(request), (sortedReplicas[i], DateTime.Now)); + var requestResult = await Task.WhenAny(requests.Keys.Cast().Concat(new[] { timeoutTask })); if (requestResult != timeoutTask) { var result = (Task)requestResult; + var replicaData = requests[result]; + var time = DateTime.Now - replicaData.Item2; if (result.Status == TaskStatus.RanToCompletion) + { + _replicasStatistics.UpdateStats(replicaData.Item1, time.TotalMilliseconds); + requests.Remove(result); return result.Result; + } + _replicasStatistics.UpdateStats(replicaData.Item1, time.TotalMilliseconds); requests.Remove(result); } } + throw new TimeoutException(); } protected override ILog Log => LogManager.GetLogger(typeof(SmartClusterClient)); } -} +} \ No newline at end of file diff --git a/homework 2/ClusterClient/ReplicasStatistics.cs b/homework 2/ClusterClient/ReplicasStatistics.cs new file mode 100644 index 0000000..7903d91 --- /dev/null +++ b/homework 2/ClusterClient/ReplicasStatistics.cs @@ -0,0 +1,28 @@ +using System.Collections.Concurrent; +using System.Linq; + +namespace ClusterClient; + +public class ReplicasStatistics +{ + private const double ConfidenceFactor = 0.8; + private readonly ConcurrentDictionary _replicaStatistics; + + public ReplicasStatistics(string[] replicaAddresses) + { + _replicaStatistics = new ConcurrentDictionary(); + foreach (var replicaAddress in replicaAddresses) + _replicaStatistics.TryAdd(replicaAddress, 0); + } + + public void UpdateStats(string replicaAddress, double workTime) + { + _replicaStatistics[replicaAddress] = + _replicaStatistics[replicaAddress] * ConfidenceFactor + workTime * (1 - ConfidenceFactor); + } + + public double GetStats(string replicaAddress) + { + return _replicaStatistics.TryGetValue(replicaAddress, out double value) ? value : 0; + } +} \ No newline at end of file From 5daa9fc2cbf054fbb99e13de1cdd3ae81eb6663a Mon Sep 17 00:00:00 2001 From: Foma Medvedev Date: Mon, 2 Mar 2026 13:14:00 +0500 Subject: [PATCH 5/5] =?UTF-8?q?=D0=B8=D1=81=D0=BF=D1=80=D0=B0=D0=B2=D0=B8?= =?UTF-8?q?=D0=BB=20=D0=B7=D0=B0=D0=BC=D0=B5=D1=87=D0=B0=D0=BD=D0=B8=D1=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../ClusterClient/Clients/ParallelClusterClient.cs | 6 +++--- .../ClusterClient/Clients/RoundRobinClusterClient.cs | 6 ++++++ .../ClusterClient/Clients/SmartClusterClient.cs | 11 ++++++++++- homework 2/ClusterClient/ReplicasStatistics.cs | 7 +++++-- 4 files changed, 24 insertions(+), 6 deletions(-) diff --git a/homework 2/ClusterClient/Clients/ParallelClusterClient.cs b/homework 2/ClusterClient/Clients/ParallelClusterClient.cs index 8e55141..3dd32bc 100644 --- a/homework 2/ClusterClient/Clients/ParallelClusterClient.cs +++ b/homework 2/ClusterClient/Clients/ParallelClusterClient.cs @@ -18,17 +18,17 @@ public ParallelClusterClient(string[] replicaAddresses) : base(replicaAddresses) public override async Task ProcessRequestAsync(string query, TimeSpan timeout) { var tasks = new List>(); + var timeoutTask = Task.Delay(timeout); + for (var i = 0; i < ReplicaAddresses.Length; ++i) { var request = CreateRequest(ReplicaAddresses[i] + "?query=" + query); tasks.Add(ProcessRequestAsync(request)); } - var timeoutTask = Task.Delay(timeout).ContinueWith(_ => throw new TimeoutException()); - while (tasks.Count > 0) { - var completedTask = await Task.WhenAny(tasks.Cast().Concat(new[] {timeoutTask})); + var completedTask = await Task.WhenAny(tasks.Concat(new[] {timeoutTask})); if (completedTask == timeoutTask) throw new TimeoutException(); try diff --git a/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs b/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs index 6d2ee65..ddc93b6 100644 --- a/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs +++ b/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs @@ -26,7 +26,11 @@ public override async Task ProcessRequestAsync(string query, TimeSpan ti for (int i = 0; i < sortedReplicas.Length; i++) { var remainingTime = timeout - timer.Elapsed; + if (remainingTime <= TimeSpan.Zero) + throw new TimeoutException(); var timeoutForReplica = TimeSpan.FromTicks(remainingTime.Ticks / (sortedReplicas.Length - i)); + if (timeoutForReplica <= TimeSpan.Zero) + throw new TimeoutException(); var timeoutTask = Task.Delay(timeoutForReplica); var request = CreateRequest(sortedReplicas[i] + "?query=" + query); var task = ProcessRequestAsync(request); @@ -46,6 +50,8 @@ public override async Task ProcessRequestAsync(string query, TimeSpan ti _replicasStatistics.UpdateStats(replicaData.Item1, timeout.TotalMilliseconds); sentTasks.Remove(task); } + else + _replicasStatistics.UpdateStats(sortedReplicas[i], timeoutForReplica.TotalMilliseconds); } throw new TimeoutException(); } diff --git a/homework 2/ClusterClient/Clients/SmartClusterClient.cs b/homework 2/ClusterClient/Clients/SmartClusterClient.cs index d19c2ad..4c2cfb0 100644 --- a/homework 2/ClusterClient/Clients/SmartClusterClient.cs +++ b/homework 2/ClusterClient/Clients/SmartClusterClient.cs @@ -27,11 +27,15 @@ public override async Task ProcessRequestAsync(string query, TimeSpan ti for (int i = 0; i < sortedReplicas.Length; i++) { var remainingTime = timeout - timer.Elapsed; + if (remainingTime <= TimeSpan.Zero) + throw new TimeoutException(); var timeoutForReplica = TimeSpan.FromTicks(remainingTime.Ticks / (sortedReplicas.Length - i)); + if (timeoutForReplica <= TimeSpan.Zero) + throw new TimeoutException(); var timeoutTask = Task.Delay(timeoutForReplica); var request = CreateRequest(sortedReplicas[i] + "?query=" + query); requests.Add(ProcessRequestAsync(request), (sortedReplicas[i], DateTime.Now)); - var requestResult = await Task.WhenAny(requests.Keys.Cast().Concat(new[] { timeoutTask })); + var requestResult = await Task.WhenAny(requests.Keys.Concat(new[] { timeoutTask })); if (requestResult != timeoutTask) { var result = (Task)requestResult; @@ -41,11 +45,16 @@ public override async Task ProcessRequestAsync(string query, TimeSpan ti { _replicasStatistics.UpdateStats(replicaData.Item1, time.TotalMilliseconds); requests.Remove(result); + foreach (var badRequest in requests) + _replicasStatistics.UpdateStats(badRequest.Value.Item1, timeoutForReplica.TotalMilliseconds); + requests.Clear(); return result.Result; } _replicasStatistics.UpdateStats(replicaData.Item1, time.TotalMilliseconds); requests.Remove(result); } + else + _replicasStatistics.UpdateStats(sortedReplicas[i], timeoutForReplica.TotalMilliseconds); } throw new TimeoutException(); diff --git a/homework 2/ClusterClient/ReplicasStatistics.cs b/homework 2/ClusterClient/ReplicasStatistics.cs index 7903d91..b14b73a 100644 --- a/homework 2/ClusterClient/ReplicasStatistics.cs +++ b/homework 2/ClusterClient/ReplicasStatistics.cs @@ -17,8 +17,11 @@ public ReplicasStatistics(string[] replicaAddresses) public void UpdateStats(string replicaAddress, double workTime) { - _replicaStatistics[replicaAddress] = - _replicaStatistics[replicaAddress] * ConfidenceFactor + workTime * (1 - ConfidenceFactor); + _replicaStatistics.AddOrUpdate( + replicaAddress, + workTime, + (_, previousValue) => previousValue * ConfidenceFactor + workTime * (1 - ConfidenceFactor) + ); } public double GetStats(string replicaAddress)