diff --git a/homework 2/ClusterClient/Clients/ClusterClientBase.cs b/homework 2/ClusterClient/Clients/ClusterClientBase.cs index 23a2ffd..24fcee6 100644 --- a/homework 2/ClusterClient/Clients/ClusterClientBase.cs +++ b/homework 2/ClusterClient/Clients/ClusterClientBase.cs @@ -6,39 +6,41 @@ using System.Threading.Tasks; using log4net; -namespace ClusterClient.Clients +namespace ClusterClient.Clients; + +public abstract class ClusterClientBase { - public abstract class ClusterClientBase + protected readonly ReplicaStats ReplicaStats; + + protected ClusterClientBase(string[] replicaAddresses) { - protected string[] ReplicaAddresses { get; set; } + ReplicaAddresses = replicaAddresses; + ReplicaStats = new ReplicaStats(replicaAddresses); + } - protected ClusterClientBase(string[] replicaAddresses) - { - ReplicaAddresses = replicaAddresses; - } + protected string[] ReplicaAddresses { get; set; } + protected abstract ILog Log { get; } - public abstract Task ProcessRequestAsync(string query, TimeSpan timeout); - protected abstract ILog Log { get; } + public abstract Task ProcessRequestAsync(string query, TimeSpan timeout); - protected static HttpWebRequest CreateRequest(string uriStr) - { - var request = WebRequest.CreateHttp(Uri.EscapeUriString(uriStr)); - request.Proxy = null; - request.KeepAlive = true; - request.ServicePoint.UseNagleAlgorithm = false; - request.ServicePoint.ConnectionLimit = 100500; - return request; - } + protected static HttpWebRequest CreateRequest(string uriStr) + { + var request = WebRequest.CreateHttp(Uri.EscapeUriString(uriStr)); + request.Proxy = null; + request.KeepAlive = true; + request.ServicePoint.UseNagleAlgorithm = false; + request.ServicePoint.ConnectionLimit = 100500; + return request; + } - protected async Task ProcessRequestAsync(WebRequest request) + protected async Task ProcessRequestAsync(WebRequest request) + { + var timer = Stopwatch.StartNew(); + using (var response = await request.GetResponseAsync()) { - var timer = Stopwatch.StartNew(); - using (var response = await request.GetResponseAsync()) - { - var result = await new StreamReader(response.GetResponseStream(), Encoding.UTF8).ReadToEndAsync(); - Log.InfoFormat("Response from {0} received in {1} ms", request.RequestUri, timer.ElapsedMilliseconds); - return result; - } + var result = await new StreamReader(response.GetResponseStream(), Encoding.UTF8).ReadToEndAsync(); + Log.InfoFormat("Response from {0} received in {1} ms", request.RequestUri, timer.ElapsedMilliseconds); + return result; } } } \ No newline at end of file diff --git a/homework 2/ClusterClient/Clients/ParallelClusterClient.cs b/homework 2/ClusterClient/Clients/ParallelClusterClient.cs index 5531800..fe2cc71 100644 --- a/homework 2/ClusterClient/Clients/ParallelClusterClient.cs +++ b/homework 2/ClusterClient/Clients/ParallelClusterClient.cs @@ -1,23 +1,48 @@ using System; -using System.Collections.Generic; using System.Linq; -using System.Text; +using System.Threading; using System.Threading.Tasks; +using Fclp.Internals.Extensions; 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) - { - } + } + + protected override ILog Log => LogManager.GetLogger(typeof(ParallelClusterClient)); + + public override async Task ProcessRequestAsync(string query, TimeSpan timeout) + { + var webRequests = ReplicaAddresses + .Select(replicaAddress => CreateRequest(replicaAddress + "?query=" + query)) + .ToList(); + + webRequests.ForEach(r => Log.InfoFormat($"Processing {r.RequestUri}")); + + var timeoutTask = Task.Delay(timeout); + + var tasks = webRequests + .Select(ProcessRequestAsync) + .Append(timeoutTask) + .ToList(); + + while (tasks.Count > 1) + { + var completedTask = await Task.WhenAny(tasks); + + if (completedTask == timeoutTask) + throw new TimeoutException(); + + tasks.Remove(completedTask); - public override Task ProcessRequestAsync(string query, TimeSpan timeout) - { - throw new NotImplementedException(); + if (completedTask.Status == TaskStatus.RanToCompletion) + return await (Task)completedTask; } - protected override ILog Log => LogManager.GetLogger(typeof(ParallelClusterClient)); + throw new TimeoutException(); } -} +} \ No newline at end of file diff --git a/homework 2/ClusterClient/Clients/RandomClusterClient.cs b/homework 2/ClusterClient/Clients/RandomClusterClient.cs index 4de13f3..6a79aa7 100644 --- a/homework 2/ClusterClient/Clients/RandomClusterClient.cs +++ b/homework 2/ClusterClient/Clients/RandomClusterClient.cs @@ -2,34 +2,33 @@ using System.Threading.Tasks; using log4net; -namespace ClusterClient.Clients +namespace ClusterClient.Clients; + +public class RandomClusterClient : ClusterClientBase { - public class RandomClusterClient : ClusterClientBase + private readonly Random random = new(); + + public RandomClusterClient(string[] replicaAddresses) + : base(replicaAddresses) { - private readonly Random random = new Random(); + } - public RandomClusterClient(string[] replicaAddresses) - : base(replicaAddresses) - { - } + protected override ILog Log => LogManager.GetLogger(typeof(RandomClusterClient)); - public override async Task ProcessRequestAsync(string query, TimeSpan timeout) - { - var uri = ReplicaAddresses[random.Next(ReplicaAddresses.Length)]; + public override async Task ProcessRequestAsync(string query, TimeSpan timeout) + { + var uri = ReplicaAddresses[random.Next(ReplicaAddresses.Length)]; - var webRequest = CreateRequest(uri + "?query=" + query); - - Log.InfoFormat($"Processing {webRequest.RequestUri}"); + var webRequest = CreateRequest(uri + "?query=" + query); - var resultTask = ProcessRequestAsync(webRequest); + Log.InfoFormat($"Processing {webRequest.RequestUri}"); - await Task.WhenAny(resultTask, Task.Delay(timeout)); - if (!resultTask.IsCompleted) - throw new TimeoutException(); + var resultTask = ProcessRequestAsync(webRequest); - return resultTask.Result; - } + await Task.WhenAny(resultTask, Task.Delay(timeout)); + if (!resultTask.IsCompleted) + throw new TimeoutException(); - protected override ILog Log => LogManager.GetLogger(typeof(RandomClusterClient)); + return resultTask.Result; } } \ No newline at end of file diff --git a/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs b/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs index 0293628..e30b4b8 100644 --- a/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs +++ b/homework 2/ClusterClient/Clients/RoundRobinClusterClient.cs @@ -1,23 +1,52 @@ 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 RoundRobinClusterClient : ClusterClientBase { - public class RoundRobinClusterClient : ClusterClientBase + public RoundRobinClusterClient(string[] replicaAddresses) : base(replicaAddresses) { - public RoundRobinClusterClient(string[] replicaAddresses) : base(replicaAddresses) - { - } + } + + protected override ILog Log => LogManager.GetLogger(typeof(RoundRobinClusterClient)); - public override Task ProcessRequestAsync(string query, TimeSpan timeout) + public override async Task ProcessRequestAsync(string query, TimeSpan timeout) + { + var badRequestCounter = 0; + var sw = new Stopwatch(); + var orderedReplicaAddresses = ReplicaAddresses + .OrderBy(r => ReplicaStats.Stats[r]) + .ToList(); + + foreach (var address in orderedReplicaAddresses) { - throw new NotImplementedException(); + var webRequest = CreateRequest(address + "?query=" + query); + + Log.InfoFormat($"Processing {webRequest.RequestUri}"); + + var task = ProcessRequestAsync(webRequest); + var requestTimeout = timeout / (ReplicaAddresses.Length - badRequestCounter); + var timeoutTask = Task.Delay(requestTimeout); + + sw.Restart(); + var completedTask = await Task.WhenAny(task, timeoutTask); + sw.Stop(); + + timeout -= sw.Elapsed; + + if (completedTask != timeoutTask && completedTask.Status == TaskStatus.RanToCompletion) + { + ReplicaStats.Update(address, sw.Elapsed.Milliseconds); + return await (Task)completedTask; + } + + badRequestCounter++; } - protected override ILog Log => LogManager.GetLogger(typeof(RoundRobinClusterClient)); + throw new TimeoutException(); } -} +} \ No newline at end of file diff --git a/homework 2/ClusterClient/Clients/SmartClusterClient.cs b/homework 2/ClusterClient/Clients/SmartClusterClient.cs index eb06d8b..fbd34da 100644 --- a/homework 2/ClusterClient/Clients/SmartClusterClient.cs +++ b/homework 2/ClusterClient/Clients/SmartClusterClient.cs @@ -1,23 +1,63 @@ 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) - { - } + } + + protected override ILog Log => LogManager.GetLogger(typeof(SmartClusterClient)); - public override Task ProcessRequestAsync(string query, TimeSpan timeout) + public override async Task ProcessRequestAsync(string query, TimeSpan timeout) + { + var badRequestCounter = 0; + var sw = new Stopwatch(); + var tasks = new List>(); + var orderedReplicaAddresses = ReplicaAddresses + .OrderBy(r => ReplicaStats.Stats[r]) + .ToList(); + + foreach (var address in orderedReplicaAddresses) { - throw new NotImplementedException(); + var webRequest = CreateRequest(address + "?query=" + query); + + Log.InfoFormat($"Processing {webRequest.RequestUri}"); + + var task = ProcessRequestAsync(webRequest); + var requestTimeout = timeout / (ReplicaAddresses.Length - badRequestCounter); + var timeoutTask = Task.Delay(requestTimeout); + tasks.Add(task); + + sw.Restart(); + var anyTask = Task.WhenAny(tasks); + var completedTask = await Task.WhenAny(anyTask, timeoutTask); + sw.Stop(); + + timeout -= sw.Elapsed; + + if (completedTask != timeoutTask) + { + var result = await anyTask; + if (result.IsFaulted) + tasks.Remove(result); + + if (result.Status == TaskStatus.RanToCompletion) + { + ReplicaStats.Update(address, sw.Elapsed.Milliseconds); + return await result; + } + } + + badRequestCounter++; } - protected override ILog Log => LogManager.GetLogger(typeof(SmartClusterClient)); + throw new TimeoutException(); } -} +} \ No newline at end of file diff --git a/homework 2/ClusterClient/ClusterClient.csproj b/homework 2/ClusterClient/ClusterClient.csproj index b99eac3..f2255fe 100644 --- a/homework 2/ClusterClient/ClusterClient.csproj +++ b/homework 2/ClusterClient/ClusterClient.csproj @@ -1,29 +1,29 @@  - - net8.0 - Exe - ..\ - true - false - - - - PreserveNewest - - - - - 2.0.8 - - - 1.0.25 - - - - - - - PreserveNewest - - + + net8.0 + Exe + ..\ + true + false + + + + PreserveNewest + + + + + 2.0.8 + + + 1.0.25 + + + + + + + PreserveNewest + + \ No newline at end of file diff --git a/homework 2/ClusterClient/Program.cs b/homework 2/ClusterClient/Program.cs index 6397123..fb1babe 100644 --- a/homework 2/ClusterClient/Program.cs +++ b/homework 2/ClusterClient/Program.cs @@ -1,95 +1,94 @@ using System; -using System.Text; using System.Diagnostics; using System.IO; using System.Linq; using System.Reflection; +using System.Text; using System.Threading.Tasks; using ClusterClient.Clients; using Fclp; using log4net; using log4net.Config; -namespace ClusterClient +namespace ClusterClient; + +internal class Program { - class Program + private static readonly ILog Log = LogManager.GetLogger(typeof(Program)); + + private static void Main(string[] args) { - static void Main(string[] args) - { - Encoding.RegisterProvider(System.Text.CodePagesEncodingProvider.Instance); - XmlConfigurator.Configure(LogManager.GetRepository(Assembly.GetCallingAssembly()), new FileInfo("log4net.config")); + Encoding.RegisterProvider(CodePagesEncodingProvider.Instance); + XmlConfigurator.Configure(LogManager.GetRepository(Assembly.GetCallingAssembly()), + new FileInfo("log4net.config")); - if (!TryGetReplicaAddresses(args, out var replicaAddresses)) - return; + if (!TryGetReplicaAddresses(args, out var replicaAddresses)) + return; - try + try + { + var clients = new ClusterClientBase[] { - var clients = new ClusterClientBase[] - { - new RandomClusterClient(replicaAddresses), - }; + new RandomClusterClient(replicaAddresses) + }; - var queries = new[] - { - "lorem", "ipsum", "dolor", "sit", "amet", "consectetuer", - "adipiscing", "elit", "sed", "diam", "nonummy", "nibh", "euismod", - "tincidunt", "ut", "laoreet", "dolore", "magna", "aliquam", "erat" - }; + var queries = new[] + { + "lorem", "ipsum", "dolor", "sit", "amet", "consectetuer", + "adipiscing", "elit", "sed", "diam", "nonummy", "nibh", "euismod", + "tincidunt", "ut", "laoreet", "dolore", "magna", "aliquam", "erat" + }; - foreach (var client in clients) + foreach (var client in clients) + { + Console.WriteLine("Testing {0} started", client.GetType()); + Task.WaitAll(queries.Select(async query => { - Console.WriteLine("Testing {0} started", client.GetType()); - Task.WaitAll(queries.Select( - async query => - { - var timer = Stopwatch.StartNew(); - try - { - await client.ProcessRequestAsync(query, TimeSpan.FromSeconds(6)); + var timer = Stopwatch.StartNew(); + try + { + await client.ProcessRequestAsync(query, TimeSpan.FromSeconds(6)); - Console.WriteLine("Processed query \"{0}\" in {1} ms", query, timer.ElapsedMilliseconds); - } - catch (TimeoutException) - { - Console.WriteLine("Query \"{0}\" timeout ({1} ms)", query, timer.ElapsedMilliseconds); - } - }).ToArray()); - Console.WriteLine("Testing {0} finished", client.GetType()); - } - } - catch (Exception e) - { - Log.Fatal(e); - Console.WriteLine($"E :{e}"); + Console.WriteLine("Processed query \"{0}\" in {1} ms", query, timer.ElapsedMilliseconds); + } + catch (TimeoutException) + { + Console.WriteLine("Query \"{0}\" timeout ({1} ms)", query, timer.ElapsedMilliseconds); + } + }).ToArray()); + Console.WriteLine("Testing {0} finished", client.GetType()); } } - - private static bool TryGetReplicaAddresses(string[] args, out string[] replicaAddresses) + catch (Exception e) { - var argumentsParser = new FluentCommandLineParser(); - string[] result = {}; + Log.Fatal(e); + Console.WriteLine($"E :{e}"); + } + } - argumentsParser.Setup(CaseType.CaseInsensitive, "f", "file") - .WithDescription("Path to the file with replica addresses") - .Callback(fileName => result = File.ReadAllLines(fileName)) - .Required(); + private static bool TryGetReplicaAddresses(string[] args, out string[] replicaAddresses) + { + var argumentsParser = new FluentCommandLineParser(); + string[] result = { }; - argumentsParser.SetupHelp("?", "h", "help") - .Callback(text => Console.WriteLine(text)); + argumentsParser.Setup(CaseType.CaseInsensitive, "f", "file") + .WithDescription("Path to the file with replica addresses") + .Callback(fileName => result = File.ReadAllLines(fileName)) + .Required(); - var parsingResult = argumentsParser.Parse(args); + argumentsParser.SetupHelp("?", "h", "help") + .Callback(text => Console.WriteLine(text)); - if (parsingResult.HasErrors) - { - argumentsParser.HelpOption.ShowHelp(argumentsParser.Options); - replicaAddresses = null; - return false; - } + var parsingResult = argumentsParser.Parse(args); - replicaAddresses = result; - return !parsingResult.HasErrors; + if (parsingResult.HasErrors) + { + argumentsParser.HelpOption.ShowHelp(argumentsParser.Options); + replicaAddresses = null; + return false; } - private static readonly ILog Log = LogManager.GetLogger(typeof(Program)); + replicaAddresses = result; + return !parsingResult.HasErrors; } -} +} \ No newline at end of file diff --git a/homework 2/ClusterClient/ReplicaStats.cs b/homework 2/ClusterClient/ReplicaStats.cs new file mode 100644 index 0000000..c93b2f5 --- /dev/null +++ b/homework 2/ClusterClient/ReplicaStats.cs @@ -0,0 +1,22 @@ +using System.Collections.Concurrent; +using Fclp.Internals.Extensions; + +namespace ClusterClient; + +public class ReplicaStats +{ + private const float Coefficient = 0.2f; + public readonly ConcurrentDictionary Stats = new(); + + public ReplicaStats(string[] replicas) + { + replicas.ForEach(r => Stats.TryAdd(r, 0)); + } + + public void Update(string replica, int latency) + { + var oldValue = Stats[replica]; + var ema = Stats[replica] * (1 - Coefficient) + latency * Coefficient; + Stats.TryUpdate(replica, ema, oldValue); + } +} \ No newline at end of file diff --git a/homework 2/ClusterClient/log4net.config b/homework 2/ClusterClient/log4net.config index 67c9cd7..982323d 100644 --- a/homework 2/ClusterClient/log4net.config +++ b/homework 2/ClusterClient/log4net.config @@ -1,19 +1,19 @@  - - - - - - - - - - -
-