This commit is contained in:
benbierens 2024-09-12 14:38:15 +02:00
parent 3c447eb4c5
commit 8e018cbae9
No known key found for this signature in database
GPG Key ID: 877D2C2E09A22F3A
7 changed files with 350 additions and 113 deletions

36
Tools/AutoClient/App.cs Normal file
View File

@ -0,0 +1,36 @@
using Logging;
namespace AutoClient
{
public class App
{
public App(Configuration config)
{
Config = config;
Log = new LogSplitter(
new FileLog(Path.Combine(config.LogPath, "autoclient")),
new ConsoleLog()
);
Generator = CreateGenerator();
CidRepo = new CidRepo(config);
}
public Configuration Config { get; }
public ILog Log { get; }
public IFileGenerator Generator { get; }
public CancellationTokenSource Cts { get; } = new CancellationTokenSource();
public CidRepo CidRepo { get; }
public Performance Performance { get; } = new Performance();
private IFileGenerator CreateGenerator()
{
if (Config.FileSizeMb > 0)
{
return new RandomFileGenerator(Config, Log);
}
return new ImageGenerator(Log);
}
}
}

View File

@ -0,0 +1,85 @@
namespace AutoClient
{
public class CidRepo
{
private readonly Random random = new Random();
private readonly object _lock = new object();
private readonly List<CidEntry> entries = new List<CidEntry>();
private readonly Configuration config;
public CidRepo(Configuration config)
{
this.config = config;
}
public void Add(string nodeId, string cid, long knownSize)
{
lock (_lock)
{
entries.Add(new CidEntry(nodeId, cid, knownSize));
}
}
public void AddEncoded(string originalCid, string encodedCid)
{
lock (_lock)
{
var entry = entries.SingleOrDefault(e => e.Cid == originalCid);
if (entry == null) return;
entry.Encoded = encodedCid;
}
}
public string? GetForeignCid(string myNodeId)
{
lock (_lock)
{
while (true)
{
if (!entries.Any()) return null;
var available = entries.Where(e => e.NodeId != myNodeId).ToArray();
if (!available.Any()) return null;
var i = random.Next(0, available.Length);
var entry = available[i];
if (entry.CreatedUtc < (DateTime.UtcNow + TimeSpan.FromMinutes(config.ContractDurationMinutes)))
{
entries.Remove(entry);
}
else
{
return entry.Cid;
}
}
}
}
public long? GetSizeForCid(string cid)
{
lock (_lock)
{
var entry = entries.SingleOrDefault(e => e.Cid == cid);
if (entry == null) return null;
return entry.KnownSize;
}
}
}
public class CidEntry
{
public CidEntry(string nodeId, string cid, long knownSize)
{
NodeId = nodeId;
Cid = cid;
KnownSize = knownSize;
}
public string NodeId { get; }
public string Cid { get; }
public string Encoded { get; set; } = string.Empty;
public long KnownSize { get; }
public DateTime CreatedUtc { get; } = DateTime.UtcNow;
}
}

View File

@ -1,45 +1,57 @@
using CodexOpenApi; using CodexOpenApi;
using Logging; using Logging;
using static Org.BouncyCastle.Math.EC.ECCurve;
using Utils; using Utils;
namespace AutoClient namespace AutoClient
{ {
public class CodexUser public class CodexUser
{ {
private readonly ILog log; private readonly App app;
private readonly CodexApi codex; private readonly CodexApi codex;
private readonly HttpClient client; private readonly HttpClient client;
private readonly Address address; private readonly Address address;
private readonly IFileGenerator generator; private readonly List<Purchaser> purchasers = new List<Purchaser>();
private readonly Configuration config; private Task starterTask = Task.CompletedTask;
private readonly CancellationToken cancellationToken; private readonly string nodeId = Guid.NewGuid().ToString();
public CodexUser(ILog log, CodexApi codex, HttpClient client, Address address, IFileGenerator generator, Configuration config, CancellationToken cancellationToken) public CodexUser(App app, CodexApi codex, HttpClient client, Address address)
{ {
this.log = log; this.app = app;
this.codex = codex; this.codex = codex;
this.client = client; this.client = client;
this.address = address; this.address = address;
this.generator = generator;
this.config = config;
this.cancellationToken = cancellationToken;
} }
public async Task Run() public void Start(int index)
{ {
var purchasers = new List<Purchaser>(); for (var i = 0; i < app.Config.NumConcurrentPurchases; i++)
for (var i = 0; i < config.NumConcurrentPurchases; i++)
{ {
purchasers.Add(new Purchaser(new LogPrefixer(log, $"({i}) "), client, address, codex, config, generator, cancellationToken)); purchasers.Add(new Purchaser(app, nodeId, new LogPrefixer(app.Log, $"({i}) "), client, address, codex));
} }
var delayPerPurchaser = TimeSpan.FromMinutes(config.ContractDurationMinutes) / config.NumConcurrentPurchases; var delayPerPurchaser =
TimeSpan.FromSeconds(10 * index) +
TimeSpan.FromMinutes(app.Config.ContractDurationMinutes) / app.Config.NumConcurrentPurchases;
starterTask = Task.Run(() => StartPurchasers(delayPerPurchaser));
}
private async Task StartPurchasers(TimeSpan delayPerPurchaser)
{
foreach (var purchaser in purchasers) foreach (var purchaser in purchasers)
{ {
purchaser.Start(); purchaser.Start();
await Task.Delay(delayPerPurchaser); await Task.Delay(delayPerPurchaser);
} }
} }
public void Stop()
{
starterTask.Wait();
foreach (var purchaser in purchasers)
{
purchaser.Stop();
}
}
} }
} }

View File

@ -11,16 +11,16 @@ namespace AutoClient
public class ImageGenerator : IFileGenerator public class ImageGenerator : IFileGenerator
{ {
private LogSplitter log; private readonly ILog log;
public ImageGenerator(LogSplitter log) public ImageGenerator(ILog log)
{ {
this.log = log; this.log = log;
} }
public async Task<string> Generate() public async Task<string> Generate()
{ {
log.Log("Fetching random image from picsum.photos..."); log.Debug("Fetching random image from picsum.photos...");
var httpClient = new HttpClient(); var httpClient = new HttpClient();
var thing = await httpClient.GetStreamAsync("https://picsum.photos/3840/2160"); var thing = await httpClient.GetStreamAsync("https://picsum.photos/3840/2160");

View File

@ -0,0 +1,45 @@
namespace AutoClient
{
public class Performance
{
internal void DownloadFailed(Exception ex)
{
throw new NotImplementedException();
}
internal void DownloadSuccessful(long? size, TimeSpan time)
{
throw new NotImplementedException();
}
internal void StorageContractCancelled()
{
throw new NotImplementedException();
}
internal void StorageContractErrored(string error)
{
throw new NotImplementedException();
}
internal void StorageContractFinished()
{
throw new NotImplementedException();
}
internal void StorageContractStarted()
{
throw new NotImplementedException();
}
internal void UploadFailed(Exception exc)
{
throw new NotImplementedException();
}
internal void UploadSuccessful(long length, TimeSpan time)
{
throw new NotImplementedException();
}
}
}

View File

@ -1,24 +1,15 @@
using ArgsUniform; using ArgsUniform;
using AutoClient; using AutoClient;
using CodexOpenApi; using CodexOpenApi;
using Core;
using Logging;
using Nethereum.Model;
using Utils; using Utils;
public class Program public class Program
{ {
private readonly CancellationTokenSource cts; private readonly App app;
private readonly Configuration config;
private readonly LogSplitter log;
private readonly IFileGenerator generator;
public Program(CancellationTokenSource cts, Configuration config, LogSplitter log, IFileGenerator generator) public Program(Configuration config)
{ {
this.cts = cts; app = new App(config);
this.config = config;
this.log = log;
this.generator = generator;
} }
public static async Task Main(string[] args) public static async Task Main(string[] args)
@ -34,30 +25,31 @@ public class Program
throw new Exception("Number of concurrent purchases must be > 0"); throw new Exception("Number of concurrent purchases must be > 0");
} }
var log = new LogSplitter( var p = new Program(config);
new FileLog(Path.Combine(config.LogPath, "autoclient")), await p.Run();
new ConsoleLog()
);
var generator = CreateGenerator(config, log);
var p = new Program(cts, config, log, generator);
await p.Run(args);
cts.Token.WaitHandle.WaitOne();
log.Log("Done.");
} }
public async Task Run(string[] args) public async Task Run()
{ {
var codexUsers = CreateUsers(); var codexUsers = await CreateUsers();
var i = 0;
foreach (var user in codexUsers)
{
user.Start(i);
i++;
}
app.Cts.Token.WaitHandle.WaitOne();
foreach (var user in codexUsers) user.Stop();
app.Log.Log("Done");
} }
private async Task<CodexUser[]> CreateUsers() private async Task<CodexUser[]> CreateUsers()
{ {
var endpointStrs = config.CodexEndpoints.Split(";", StringSplitOptions.RemoveEmptyEntries); var endpointStrs = app.Config.CodexEndpoints.Split(";", StringSplitOptions.RemoveEmptyEntries);
var result = new List<CodexUser>(); var result = new List<CodexUser>();
foreach (var e in endpointStrs) foreach (var e in endpointStrs)
@ -79,21 +71,24 @@ public class Program
port: port port: port
); );
log.Log($"Start. Address: {address}");
var client = new HttpClient(); var client = new HttpClient();
var codex = new CodexApi(client); var codex = new CodexApi(client);
codex.BaseUrl = $"{address.Host}:{address.Port}/api/codex/v1"; codex.BaseUrl = $"{address.Host}:{address.Port}/api/codex/v1";
app.Log.Log($"Checking Codex at {address}...");
await CheckCodex(codex); await CheckCodex(codex);
app.Log.Log("OK");
return new CodexUser(); return new CodexUser(
app,
codex,
client,
address
);
} }
private async Task CheckCodex(CodexApi codex) private async Task CheckCodex(CodexApi codex)
{ {
log.Log("Checking Codex...");
try try
{ {
var info = await codex.GetDebugInfoAsync(); var info = await codex.GetDebugInfoAsync();
@ -101,20 +96,11 @@ public class Program
} }
catch (Exception ex) catch (Exception ex)
{ {
log.Log($"Codex not OK: {ex}"); app.Log.Error($"Codex not OK: {ex}");
throw; throw;
} }
} }
private static IFileGenerator CreateGenerator(Configuration config, LogSplitter log)
{
if (config.FileSizeMb > 0)
{
return new RandomFileGenerator(config, log);
}
return new ImageGenerator(log);
}
private static void PrintHelp() private static void PrintHelp()
{ {
Console.WriteLine("Generates fake data and creates Codex storage contracts for it."); Console.WriteLine("Generates fake data and creates Codex storage contracts for it.");

View File

@ -8,36 +8,76 @@ namespace AutoClient
{ {
public class Purchaser public class Purchaser
{ {
private readonly App app;
private readonly string nodeId;
private readonly ILog log; private readonly ILog log;
private readonly HttpClient client; private readonly HttpClient client;
private readonly Address address; private readonly Address address;
private readonly CodexApi codex; private readonly CodexApi codex;
private readonly Configuration config; private Task workerTask = Task.CompletedTask;
private readonly IFileGenerator generator;
private readonly CancellationToken ct;
public Purchaser(ILog log, HttpClient client, Address address, CodexApi codex, Configuration config, IFileGenerator generator, CancellationToken ct) public Purchaser(App app, string nodeId, ILog log, HttpClient client, Address address, CodexApi codex)
{ {
this.app = app;
this.nodeId = nodeId;
this.log = log; this.log = log;
this.client = client; this.client = client;
this.address = address; this.address = address;
this.codex = codex; this.codex = codex;
this.config = config;
this.generator = generator;
this.ct = ct;
} }
public void Start() public void Start()
{ {
Task.Run(Worker); workerTask = Task.Run(Worker);
}
public void Stop()
{
workerTask.Wait();
} }
private async Task Worker() private async Task Worker()
{ {
while (!ct.IsCancellationRequested) log.Log("Worker started.");
while (!app.Cts.Token.IsCancellationRequested)
{
try
{ {
var pid = await StartNewPurchase(); var pid = await StartNewPurchase();
await WaitTillFinished(pid); await WaitTillFinished(pid);
await DownloadForeignCid();
}
catch (Exception ex)
{
log.Error("Worker failed with: " + ex);
await Task.Delay(TimeSpan.FromHours(6));
}
}
}
private async Task DownloadForeignCid()
{
var cid = app.CidRepo.GetForeignCid(nodeId);
if (cid == null) return;
var size = app.CidRepo.GetSizeForCid(cid);
if (cid == null) return;
try
{
var sw = System.Diagnostics.Stopwatch.StartNew();
var filename = Guid.NewGuid().ToString().ToLowerInvariant();
{
using var fileStream = File.OpenWrite(filename);
var fileResponse = await codex.DownloadNetworkAsync(cid);
fileResponse.Stream.CopyTo(fileStream);
}
var time = sw.Elapsed;
File.Delete(filename);
app.Performance.DownloadSuccessful(size, time);
}
catch (Exception ex)
{
app.Performance.DownloadFailed(ex);
} }
} }
@ -50,72 +90,96 @@ namespace AutoClient
private async Task<string> CreateFile() private async Task<string> CreateFile()
{ {
return await generator.Generate(); return await app.Generator.Generate();
} }
private async Task<ContentId> UploadFile(string filename) private async Task<ContentId> UploadFile(string filename)
{ {
// Copied from CodexNode :/
using var fileStream = File.OpenRead(filename); using var fileStream = File.OpenRead(filename);
try
{
var info = new FileInfo(filename);
var sw = System.Diagnostics.Stopwatch.StartNew();
var cid = await UploadStream(fileStream);
var time = sw.Elapsed;
app.Performance.UploadSuccessful(info.Length, time);
app.CidRepo.Add(nodeId, cid.Id, info.Length);
return cid;
}
catch (Exception exc)
{
app.Performance.UploadFailed(exc);
throw;
}
}
log.Log($"Uploading file {filename}..."); private async Task<ContentId> UploadStream(FileStream fileStream)
var response = await codex.UploadAsync(fileStream, ct); {
log.Debug($"Uploading file...");
var response = await codex.UploadAsync(fileStream, app.Cts.Token);
if (string.IsNullOrEmpty(response)) FrameworkAssert.Fail("Received empty response."); if (string.IsNullOrEmpty(response)) FrameworkAssert.Fail("Received empty response.");
if (response.StartsWith("Unable to store block")) FrameworkAssert.Fail("Node failed to store block."); if (response.StartsWith("Unable to store block")) FrameworkAssert.Fail("Node failed to store block.");
log.Log($"Uploaded file. Received contentId: '{response}'."); log.Debug($"Uploaded file. Received contentId: '{response}'.");
return new ContentId(response); return new ContentId(response);
} }
private async Task<string> RequestStorage(ContentId cid) private async Task<string> RequestStorage(ContentId cid)
{ {
log.Log("Requesting storage for " + cid.Id); log.Debug("Requesting storage for " + cid.Id);
var result = await codex.CreateStorageRequestAsync(cid.Id, new StorageRequestCreation() var result = await codex.CreateStorageRequestAsync(cid.Id, new StorageRequestCreation()
{ {
Collateral = config.RequiredCollateral.ToString(), Collateral = app.Config.RequiredCollateral.ToString(),
Duration = (config.ContractDurationMinutes * 60).ToString(), Duration = (app.Config.ContractDurationMinutes * 60).ToString(),
Expiry = (config.ContractExpiryMinutes * 60).ToString(), Expiry = (app.Config.ContractExpiryMinutes * 60).ToString(),
Nodes = config.NumHosts, Nodes = app.Config.NumHosts,
Reward = config.Price.ToString(), Reward = app.Config.Price.ToString(),
ProofProbability = "15", ProofProbability = "15",
Tolerance = config.HostTolerance Tolerance = app.Config.HostTolerance
}, ct); }, app.Cts.Token);
log.Log("Purchase ID: " + result); log.Debug("Purchase ID: " + result);
var encoded = await GetEncodedCid(result);
app.CidRepo.AddEncoded(cid.Id, encoded);
return result; return result;
} }
private async Task<string?> GetPurchaseState(string pid) private async Task<string> GetEncodedCid(string pid)
{ {
try try
{
var sp = await GetStoragePurchase(pid)!;
return sp.Request.Content.Cid;
}
catch (Exception ex)
{
log.Error(ex.ToString());
throw;
}
}
private async Task<StoragePurchase?> GetStoragePurchase(string pid)
{ {
// openapi still don't match code. // openapi still don't match code.
var str = await client.GetStringAsync($"{address.Host}:{address.Port}/api/codex/v1/storage/purchases/{pid}"); var str = await client.GetStringAsync($"{address.Host}:{address.Port}/api/codex/v1/storage/purchases/{pid}");
if (string.IsNullOrEmpty(str)) return null; if (string.IsNullOrEmpty(str)) return null;
var sp = JsonConvert.DeserializeObject<StoragePurchase>(str)!; return JsonConvert.DeserializeObject<StoragePurchase>(str);
log.Log($"Purchase {pid} is {sp.State}");
if (!string.IsNullOrEmpty(sp.Error)) log.Log($"Purchase {pid} error is {sp.Error}");
return sp.State;
}
catch
{
return null;
}
} }
private async Task WaitTillFinished(string pid) private async Task WaitTillFinished(string pid)
{ {
log.Log("Waiting...");
try try
{ {
var emptyResponseTolerance = 10; var emptyResponseTolerance = 10;
while (true) while (!app.Cts.Token.IsCancellationRequested)
{ {
var status = (await GetPurchaseState(pid))?.ToLowerInvariant(); var purchase = await GetStoragePurchase(pid);
if (string.IsNullOrEmpty(status)) if (purchase == null)
{ {
await FixedShortDelay();
emptyResponseTolerance--; emptyResponseTolerance--;
if (emptyResponseTolerance == 0) if (emptyResponseTolerance == 0)
{ {
@ -123,20 +187,29 @@ namespace AutoClient
await ExpiryTimeDelay(); await ExpiryTimeDelay();
return; return;
} }
continue;
} }
else var status = purchase.State.ToLowerInvariant();
if (status.Contains("cancel"))
{ {
if (status.Contains("cancel") || app.Performance.StorageContractCancelled();
status.Contains("error") || return;
status.Contains("finished")) }
if (status.Contains("error"))
{ {
app.Performance.StorageContractErrored(purchase.Error);
return;
}
if (status.Contains("finished"))
{
app.Performance.StorageContractFinished();
return; return;
} }
if (status.Contains("started")) if (status.Contains("started"))
{ {
app.Performance.StorageContractStarted();
await FixedDurationDelay(); await FixedDurationDelay();
} }
}
await FixedShortDelay(); await FixedShortDelay();
} }
@ -150,17 +223,17 @@ namespace AutoClient
private async Task FixedDurationDelay() private async Task FixedDurationDelay()
{ {
await Task.Delay(config.ContractDurationMinutes * 60 * 1000, ct); await Task.Delay(app.Config.ContractDurationMinutes * 60 * 1000, app.Cts.Token);
} }
private async Task ExpiryTimeDelay() private async Task ExpiryTimeDelay()
{ {
await Task.Delay(config.ContractExpiryMinutes * 60 * 1000, ct); await Task.Delay(app.Config.ContractExpiryMinutes * 60 * 1000, app.Cts.Token);
} }
private async Task FixedShortDelay() private async Task FixedShortDelay()
{ {
await Task.Delay(15 * 1000, ct); await Task.Delay(15 * 1000, app.Cts.Token);
} }
} }
} }