运行自定义线程池上的 ValueTasks

Question

我希望在自定义线程池上执行一堆 ValueTask 返回函数 - 即在一堆我自己产生和处理的线程上，而不是默认的 ThreadPool .

意思是，这些函数的所有同步位，包括任何潜在的任务延续，都应该在我的自定义线程池上执行。

从概念上讲，类似于：

class Example
{
    async ValueTask DoStuff(int something)
    {
        // .. do some stuff in here, might complete synchronously or not, who knows ..
    }

    private void Test()
    {
        for (int i = 0; i < 1_000; i++)
        {
            Func<ValueTask> method = () => DoStuff(1);
            MyThreadPool.Queue(method);
        }
    }
}

最好的方法是什么？

我目前的做法是这样的：

class Example
{
    async ValueTask DoStuff(int something)
    {
        // .. do some stuff in here, might complete synchronously or not, who knows ..
    }

    private void Test()
    {
        SynchronizationContext myContext = new MyCustomThreadPoolSynchronisationContext();
        TaskScheduler myScheduler;
        
        var prevCtx = SynchronizationContext.Current;
        try
        {
            SynchronizationContext.SetSynchronizationContext(myContext);
            myScheduler = TaskScheduler.FromCurrentSynchronizationContext();
        }
        finally
        {
            SynchronizationContext.SetSynchronizationContext(prevCtx);
        }

        var myTaskFactory = new TaskFactory(myScheduler);
        
        
        for (int i = 0; i < 1_000; i++)
        {
            myTaskFactory.StartNew(() => DoStuff(i).AsTask());
        }
    }
}

这似乎可行，但必须将 ValueTask 转换为 Task 并将其存放到 TaskFactory 感觉异常笨拙。并且必须安装我的同步上下文，只是为了能够推迟适当的 TaskScheduler（然后立即回到旧的同步上下文）感觉也很糟糕。

我目前的方法是否存在概念上的缺陷？

更好的是，有没有更好、更简单的方法来做到这一点？

Answer 1

我读到的关于创建自定义线程池的所有内容都说不要。

另一种方法是在共享线程池上使用自定义 TaskScheduler。

您可以这样使用 TaskScheduler class：

static async Task Main(string[] args)
{
    // Create a scheduler that uses four threads.
    LimitedConcurrencyLevelTaskScheduler lcts = new LimitedConcurrencyLevelTaskScheduler(4);
    List<Task> tasks = new List<Task>();

    TaskFactory factory = new TaskFactory(lcts);
    CancellationTokenSource cts = new CancellationTokenSource(10000);

    // Start 20 tasks that will run 4 threads at 100% CPU
    for (var i = 0; i < 20; i++)
        tasks.Add(factory.StartNew(() => {
            while (true)
                if (cts.Token.IsCancellationRequested)
                    break;
        },cts.Token));

    await Task.WhenAll(tasks);
    cts.Dispose();
}

运行自定义线程池上的 ValueTasks

Run ValueTasks on a custom thread pool

c#

asynchronous

async-await

taskscheduler

valuetask

运行 自定义线程池上的 ValueTasks

Run ValueTasks on a custom thread pool

c#

asynchronous

async-await

taskscheduler

valuetask

运行自定义线程池上的 ValueTasks