Skip to content

Commit 0ff63c2

Browse files
lilithlilith
committed
Get HybridCache.Benchmark running (still not correctly configured, and should have a direct-to-iblobcache mode).
1 parent 1d82686 commit 0ff63c2

File tree

4 files changed

+338
-197
lines changed

4 files changed

+338
-197
lines changed

‎src/Imazen.Routing/Promises/Pipelines/BlobCachingTestHarness.cs

+120
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,120 @@
1+
using Imazen.Abstractions.BlobCache;
2+
using Imazen.Abstractions.Blobs;
3+
using Imazen.Abstractions.Logging;
4+
using Imazen.Abstractions.Resulting;
5+
using Imazen.Common.Concurrency.BoundedTaskCollection;
6+
using Imazen.Routing.Caching;
7+
using Imazen.Routing.Health;
8+
using Imazen.Routing.Requests;
9+
using Microsoft.Extensions.Hosting;
10+
11+
namespace Imazen.Routing.Promises.Pipelines;
12+
13+
public record BlobCachingTestHarnessOptions(
14+
long? MaxUploadQueueBytes,
15+
MemoryCacheOptions? MemoryCacheOptions,
16+
bool DelayRequestUntilUploadsComplete,
17+
List<List<IBlobCache>> SeriesOfCacheGroups,
18+
List<IBlobCache> SaveToCaches,
19+
Func<IRequestSnapshot, CancellationToken, ValueTask<CodeResult<IBlobWrapper>>> BlobProvider,
20+
LatencyTrackingZone BlobProviderLatencyZone,
21+
IReLogger Logger,
22+
bool LockByUniqueRequest,
23+
bool ShutdownServices)
24+
{
25+
public static BlobCachingTestHarnessOptions TestSingleCacheSync(IBlobCache cache, Func<IRequestSnapshot, CancellationToken, ValueTask<CodeResult<IBlobWrapper>>> blobProvider, IReLogger logger)
26+
{
27+
return new BlobCachingTestHarnessOptions(
28+
null,
29+
null,
30+
true,
31+
new List<List<IBlobCache>> { new List<IBlobCache> { cache } },
32+
new List<IBlobCache> { cache },
33+
blobProvider,
34+
new LatencyTrackingZone("TestBlobProvider", 10000,true),
35+
logger,
36+
false,
37+
true
38+
);
39+
}
40+
}
41+
42+
43+
public class BlobCachingTestHarness: IHostedService
44+
{
45+
BlobCachingTestHarnessOptions options;
46+
BlobPipelineHarness blobPipelineHarness;
47+
BoundedTaskCollection<BlobTaskItem>? uploadQueue;
48+
CacheHealthTracker cacheHealthTracker;
49+
CancellationTokenSource CancellationTokenSource { get; } = new CancellationTokenSource();
50+
public BlobCachingTestHarness(BlobCachingTestHarnessOptions options)
51+
{
52+
this.options = options;
53+
if (options.MaxUploadQueueBytes != null)
54+
{
55+
uploadQueue = new BoundedTaskCollection<BlobTaskItem>(options.MaxUploadQueueBytes.Value, CancellationTokenSource);
56+
// Now ensure caches wait for uploads to write before shutting down.
57+
foreach(var c in options.SaveToCaches)
58+
c.Initialize(new BlobCacheSupportData(() => uploadQueue!.AwaitAllCurrentTasks()));
59+
}
60+
cacheHealthTracker = new CacheHealthTracker(options.Logger);
61+
var cacheEngineOptions = new CacheEngineOptions
62+
{
63+
HealthTracker = cacheHealthTracker,
64+
SeriesOfCacheGroups = options.SeriesOfCacheGroups,
65+
SaveToCaches = options.SaveToCaches,
66+
Logger = options.Logger,
67+
UploadQueue = uploadQueue,
68+
DelayRequestUntilUploadsComplete = options.DelayRequestUntilUploadsComplete,
69+
LockByUniqueRequest = options.LockByUniqueRequest,
70+
BlobFactory = new SimpleReusableBlobFactory()
71+
};
72+
var cacheEngine = new CacheEngine(null, cacheEngineOptions);
73+
blobPipelineHarness = new BlobPipelineHarness(new BlobPipelineHarnessOptions(
74+
cacheEngine,
75+
options.BlobProvider,
76+
options.Logger,
77+
options.BlobProviderLatencyZone));
78+
79+
}
80+
81+
public async ValueTask<CodeResult<IBlobWrapper>> RequestBlobWrapper(string path, string query = "",
82+
CancellationToken cancellationToken = default)
83+
{
84+
return await blobPipelineHarness.RequestBlobWrapper(path, query, cancellationToken);
85+
}
86+
87+
public Task StartAsync(CancellationToken cancellationToken)
88+
{
89+
return Task.CompletedTask;
90+
}
91+
92+
public async Task StopAsync(CancellationToken cancellationToken)
93+
{
94+
if (uploadQueue != null)
95+
{
96+
await uploadQueue.StopAsync(cancellationToken);
97+
}
98+
99+
await cacheHealthTracker.StopAsync(cancellationToken);
100+
if (options.ShutdownServices)
101+
{
102+
var allCaches = options.SeriesOfCacheGroups.SelectMany(x => x).Concat(options.SaveToCaches);
103+
foreach (var cache in allCaches)
104+
{
105+
if (cache is IHostedService service)
106+
{
107+
await service.StopAsync(cancellationToken);
108+
}
109+
}
110+
}
111+
}
112+
113+
public async Task AwaitEnqueuedTasks()
114+
{
115+
if (uploadQueue != null)
116+
{
117+
await uploadQueue.AwaitAllCurrentTasks();
118+
}
119+
}
120+
}

‎src/Imazen.Routing/Promises/Pipelines/BlobPipelineHarness.cs

+89
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,89 @@
1+
using Imazen.Abstractions.Blobs;
2+
using Imazen.Abstractions.Logging;
3+
using Imazen.Abstractions.Resulting;
4+
using Imazen.Routing.Engine;
5+
using Imazen.Routing.HttpAbstractions;
6+
using Imazen.Routing.Layers;
7+
using Imazen.Routing.Requests;
8+
using Microsoft.Extensions.Logging;
9+
10+
namespace Imazen.Routing.Promises.Pipelines;
11+
12+
public record BlobPipelineHarnessOptions(
13+
IBlobPromisePipeline Pipeline,
14+
Func<IRequestSnapshot, CancellationToken, ValueTask<CodeResult<IBlobWrapper>>> BlobProvider,
15+
IReLogger Logger, LatencyTrackingZone BlobOriginLatencyZone);
16+
17+
public class BlobPipelineHarness
18+
{
19+
readonly RoutingEngine router;
20+
IBlobPromisePipeline pipeline;
21+
IReLogger logger;
22+
23+
public BlobPipelineHarness(RoutingEngine router, IBlobPromisePipeline pipeline, IReLogger logger)
24+
{
25+
this.router = router;
26+
this.pipeline = pipeline;
27+
this.logger = logger;
28+
}
29+
public BlobPipelineHarness(BlobPipelineHarnessOptions options)
30+
{
31+
var routingBuilder = new RoutingBuilder().AddEndpointLayer(
32+
new SimpleLayer("BlobEndpoint", req =>
33+
{
34+
var endpoint =
35+
new PromiseWrappingEndpoint(
36+
new CacheableBlobPromise(req, options.BlobOriginLatencyZone, options.BlobProvider));
37+
return CodeResult<IRoutingEndpoint>.Ok(endpoint);
38+
}, null));
39+
router = routingBuilder.Build(options.Logger);
40+
pipeline = options.Pipeline;
41+
logger = options.Logger;
42+
43+
}
44+
45+
public async ValueTask<CodeResult<IBlobWrapper>> RequestBlobWrapper(string path, string query = "",
46+
CancellationToken cancellationToken = default)
47+
{
48+
var request = new EmptyHttpRequest(path, query);
49+
var mutableRequest = MutableRequest.OriginalRequest(request);
50+
return await Request(mutableRequest, cancellationToken);
51+
}
52+
53+
public async ValueTask<CodeResult<IBlobWrapper>> Request(MutableRequest mutableRequest, CancellationToken cancellationToken = default)
54+
{
55+
var result = await router.RouteToPromiseAsync(mutableRequest, cancellationToken);
56+
if (result == null)
57+
{
58+
return CodeResult<IBlobWrapper>.Err((404, "No route found"));
59+
}
60+
if (result.IsError)
61+
{
62+
return CodeResult<IBlobWrapper>.Err(result.Error);
63+
}
64+
65+
var outerRequest = mutableRequest.OriginatingRequest ?? new EmptyHttpRequest(mutableRequest);
66+
67+
var pipelineResult = await pipeline.GetFinalPromiseAsync(
68+
result.Unwrap(),router, pipeline, outerRequest,cancellationToken);
69+
70+
var finalPromise = pipelineResult.Unwrap();
71+
72+
if (finalPromise.HasDependencies)
73+
{
74+
var dependencyResult = await finalPromise.RouteDependenciesAsync(router, cancellationToken);
75+
if (dependencyResult.IsError)
76+
{
77+
return CodeResult<IBlobWrapper>.Err(dependencyResult.Error);
78+
}
79+
}
80+
var blobResult =
81+
await finalPromise.TryGetBlobAsync(mutableRequest, router, pipeline, cancellationToken);
82+
if (blobResult.IsError)
83+
{
84+
return CodeResult<IBlobWrapper>.Err(blobResult.Error);
85+
}
86+
return CodeResult<IBlobWrapper>.Ok(blobResult.Unwrap());
87+
}
88+
89+
}

‎tests/Imazen.HybridCache.Benchmark/Imazen.HybridCache.Benchmark.csproj

+4
Original file line numberDiff line numberDiff line change
@@ -15,4 +15,8 @@
1515
<PackageReference Include="Microsoft.IO.RecyclableMemoryStream" Version="3.*" />
1616
</ItemGroup>
1717

18+
<ItemGroup>
19+
<Compile Remove="BenchHarness.cs" />
20+
</ItemGroup>
21+
1822
</Project>

0 commit comments

Comments
 (0)