Orleans.SyncWork 8.1.12

此包的目的是公开一个抽象基类，以便 Orleans 能够在没有过载的情况下工作与长时间运行、CPU 密集型、同步工作。

使用开源 <a href="https://jb.gg/OpenSourceSupport"><img src="docs/images/Rider_icon.svg" width=25 height=25></a> 许可证构建，感谢 JetBrains！

构建

该项目主要是以 .net3 为目标进行构建的，虽然有多个主要版本发布支持 .net6、.net7 和 .net8；根据包版本（应与 .net 版本一致）。

需求

• .NET 6.0 SDK
• .NET 7.0 SDK
• .NET 8.0 SDK
• dotnet-format

项目概述

此仓库中有几个项目，所有项目都旨在证明或测试 NuGet 包 https://nuget.net.cn/packages/Orleans.SyncWork/ 是否如其所称那样工作。

请注意，此项目的重大修订版与 Orleans 的主要版本保持一致，因此项目并不一定遵循 SemVer，但我们尽量这样做。如果引入了破坏性变更，释放说明中应提供破坏性变更的描述以及如何针对它进行实施的说明。

此仓库中的项目包括

• Orleans.SyncWork
• Orleans.SyncWork.Tests
• Orleans.SyncWork.Demo.Api
• Orleans_sync_work.Demo.Api.Benchmark
• Orleans.SyncWork.Demo.Services

Orleans.SyncWork

项目核心。本项目以抽象基类SyncWorker的形式，实现了“长时间运行、CPU密集型、同步工作”的抽象，该类实现了ISyncWorker接口。

当你识别到长时间运行的工作时，可以扩展基类SyncWorker，提供独特的针对长时间运行工作的TRequest和TResponse。这允许你创建尽可能多的ISyncWork<TRequest, TResponse>实现，以满足你的所有长时间运行CPU密集型需求！（至少这是我们的愿望。）

SyncWork的“基本流程”

• 开始
• 轮询GetStatus，直到收到Completed或Faulted状态
• 根据GetStatus的返回，调用GetResult或GetException

此包对Orleans提出了一些“要求”

• 为了避免过载Orleans，我们引入了LimitedConcurrencyLevelTaskScheduler。此任务调度器会注册（可以是手动或通过提供的扩展方法），为正在设置的silo设置最大并发级别。这个最大并发级别必须允许空闲线程，否则Orleans服务器将被过载。在测试中，一般规则是Environment.ProcessorCount - 2最大并发。重要的是CPU不应该完全“满载”，否则由于阻塞同步工作，正常的Orleans异步消息可能无法通过 - 这将导致超时。

• 阻塞grain是状态的，当前以Guid作为键。如果在需要多个长时间运行工作的场景下，每个grain都应该使用其唯一的身份初始化。

• 阻塞grain可能不能进一步调度其他阻塞grain。这尚未在存储库中进行测试，但可以推断，在有限并发调度器的情况下，以下场景可能导致死锁

  • Grain A是长时间运行
  • Grain B是长时间运行
  • Grain A初始化并触发Grain B
  • Grain A无法完成工作直到收到Grain B的结果

  在上面的场景中，如果“Grain A”正在“积极工作”并触发“Grain B”，但“Grain A”无法完成工作直到“Grain B”完成，而由于有限的并发，“Grain B”无法开始其工作，您就遇到了有限并发任务调度器永远无法完成“Grain A”的工作的情况。

  这句话有些长，但希望这个观点可以以一种合理的方式传达。可能有一种避免上述场景的方法，但我还没有深入研究。

使用方法

创建一个实现ISyncWorker<TRequest, TResult>和一个IGrainWith...Key接口的grain接口。然后创建一个新的类，该类扩展SyncWorker<TRequest, TResult>抽象类，并实现新引入的接口。

public interface IPasswordVerifierGrain
    : ISyncWorker<PasswordVerifierRequest, PasswordVerifierResult>, IGrainWithGuidKey;

public class PasswordVerifierGrain : SyncWorker<PasswordVerifierRequest, PasswordVerifierResult>, IPasswordVerifierGrain
{
    private readonly IPasswordVerifier _passwordVerifier;

    public PasswordVerifier(
        ILogger<PasswordVerifier> logger,
        LimitedConcurrencyLevelTaskScheduler limitedConcurrencyLevelTaskScheduler,
        IPasswordVerifier passwordVerifier) : base(logger, limitedConcurrencyLevelTaskScheduler)
    {
        _passwordVerifier = passwordVerifier;
    }

    protected override async Task<PasswordVerifierResult> PerformWork(
        PasswordVerifierRequest request, GrainCancellationToken grainCancellationToken)
    {
        var verifyResult = await _passwordVerifier.VerifyPassword(request.PasswordHash, request.Password);

        return new PasswordVerifierResult()
        {
            IsValid = verifyResult
        };
    }
}

public class PasswordVerifierRequest
{
    public string Password { get; set; }
    public string PasswordHash { get; set; }
}

public class PasswordVerifierResult
{
    public bool IsValid { get; set; }
}

运行grain

var request = new PasswordVerifierRequest()
{
    Password = "my super neat password that's totally secure because it's super long",
    PasswordHash = "$2a$11$vBzJ4Ewx28C127AG5x3kT.QCCS8ai0l4JLX3VOX3MzHRkF4/A5twy"
}
var passwordVerifyGrain = grainFactory.GetGrain<IPasswordVerifierGrain>(Guid.NewGuid());
var result = await passwordVerifyGrain.StartWorkAndPollUntilResult(request);

上述StartWorkAndPollUntilResult是包中定义的扩展方法（在SyncWorkerExtensions中定义），会启动、轮询，并在工作完成后最终获取结果或异常。似乎这里还有改进的空间，特别是关于测试意外场景、基于配置的轮询等。

Orleans.SyncWork.Tests

Orleans.SyncWork的工作的单元测试项目。这些测试提供了一个用于测试整个测试周期的“TestCluster”，用于测试与grains的交互。

特别是其中一个测试会将10k个grain一次性投放到集群中，所有这些grain在我的机器上都是长时间运行（每个大约200ms） - 如果有限的并发任务调度器没有正确地与SyncWork基类一起工作，这足以使集群过载。

待办事项：这里还可以增加一些单元测试来记录行为。

Orleans.SyncWork.Demo.Api

这是ISyncWork<TRequest, TResult>的演示。此项目同时用作Orleans Silo和客户端。通常，你会在与集群不同的地方建立节点，用于向客户端提供服务。由于我们只有一个节点用于测试，因此此项目同时充当Silo主机和客户端。

使用API还将启动< apoyo="https://github.com/OrleansContrib/OrleansDashboard" rel="noopener noreferrer nofollow">OrleansDashboard。你可以在其中看到端点被调用的示例，这里收到了10k个密码验证请求

Swagger UI也提供用于测试端点的API。

Orleans_sync_work.Demo.Api.Benchmark

利用Benchmark DotNet，创建了一个基准测试类来测试集群是否崩溃，并查看我们正在处理的计时情况。

以下是在撰写本文时使用的基准测试

public class Benchy
{
    const int TotalNumberPerBenchmark = 100;
    private readonly IPasswordVerifier _passwordVerifier = new Services.PasswordVerifier();
    private readonly PasswordVerifierRequest _request = new PasswordVerifierRequest()
    {
        Password = PasswordConstants.Password,
        PasswordHash = PasswordConstants.PasswordHash
    };

    [Benchmark]
    public void Serial()
    {
        for (var i = 0; i < TotalNumberPerBenchmark; i++)
        {
            _passwordVerifier.VerifyPassword(PasswordConstants.PasswordHash, PasswordConstants.Password);
        }
    }

    [Benchmark]
    public async Task MultipleTasks()
    {
        var tasks = new List<Task>();
        for (var i = 0; i < TotalNumberPerBenchmark; i++)
        {
            tasks.Add(_passwordVerifier.VerifyPassword(PasswordConstants.PasswordHash, PasswordConstants.Password));
        }

        await Task.WhenAll(tasks);
    }

    [Benchmark]
    public async Task MultipleParallelTasks()
    {
        var tasks = new List<Task>();

        Parallel.For(0, TotalNumberPerBenchmark, i =>
        {
            tasks.Add(_passwordVerifier.VerifyPassword(PasswordConstants.PasswordHash, PasswordConstants.Password));
        });

        await Task.WhenAll(tasks);
    }

    [Benchmark]
    public async Task OrleansTasks()
    {
        var siloHost = await BenchmarkingSIloHost.GetSiloHost();
        var grainFactory = siloHost.Services.GetRequiredService<IGrainFactory>();
        var tasks = new List<Task>();
        for (var i = 0; i < TotalNumberPerBenchmark; i++)
        {
            var grain = grainFactory.GetGrain<IPasswordVerifierGrain>(Guid.NewGuid());
            tasks.Add(grain.StartWorkAndPollUntilResult(_request));
        }

        await Task.WhenAll(tasks);
    }
}

以下是结果

当然，请注意，在上面的Orleans任务中是在本地集群内有限使用的。在更真实的情况下，当你有一个有多个节点的集群时，你可能会获得更好的计时，尽管你可能需要更多处理网络延迟。

Orleans.SyncWork.Demo.Services

此项目定义了几个grain来演示通过Web API、基准测试和测试的Orleans.SyncWork包的工作原理。