标签: task-parallel-library

Stephen Toub在博客中写道

SynchronizationContext和TaskScheduler都是表示"调度程序"的抽象,是您提供一些工作的东西,它决定了运行该工作的时间和地点.有许多不同形式的调度程序.例如,ThreadPool是一个调度程序:您调用ThreadPool.QueueUserWorkItem来提供一个委托来运行,该委托被排队,并且其中一个ThreadPool的线程最终选择并运行该委托.您的用户界面还有一个调度程序:消息泵.

因此System.Reactive.Concurrency.EventLoopScheduler,Reactive Extensions的Dispatcher,ThreadPool,TaskScheduler,SyncrhonizationContext和IScheduler实现在这个意义上都是"调度程序".

他们之间有什么区别？

他们为什么都是必要的？我想我得到EventLoop,Dispatcher,ThreadPool.IScheduler也有很好的解释.
但是TaskScheduler和SyncrhonizationContext仍然不清楚.

Stephen Cleary的优秀文章解释了SyncrhonizationContext,我想我明白了.那么为什么我们需要TaskScheduler,目前尚不清楚.

请解释或指向消息来源.

HttpClient有一个内置超时功能(尽管都是异步的,即超时可以被认为是与http请求功能正交,因此可以由通用异步实用程序处理,但是除此之外)当超时启动时,它会抛出一个TaskCanceledException(包含在内) a AggregateException).

TCE包含CancellationToken等于的CancellationToken.None.

现在,如果我提供HttpClient了CancellationToken我自己的一个并使用它来取消操作完成(或超时),我得到完全相同TaskCanceledException,再次使用CancellationToken.None.

通过仅查看抛出的异常,还有一种方法可以确定超时是否取消了请求,而不必让我自己CancellationToken可以访问检查异常的代码？

PS这可能是一个错误,并CancellationToken以某种方式错误地修复CancellationToken.None？在取消使用自定义CancellationToken的情况下,我希望TaskCanceledException.CancellationToken等于该自定义令牌.

编辑 为了使问题更加清晰,通过访问原始文件CancellationTokenSource,很容易区分超时和用户取消:

origCancellationTokenSource.IsCancellationRequested == true

CancellationToken从异常中获取但是给出了错误的答案:

((TaskCanceledException)e.InnerException).CancellationToken.IsCancellationRequested == false

这是一个最小的例子,由于受欢迎的需求:

public void foo()
{
    makeRequest().ContinueWith(task =>
    {
        try
        {
            var result = task.Result;
            // do something with the result;
        }
        catch (Exception e)
        {
            TaskCanceledException innerException = e.InnerException as TaskCanceledException;
            bool …

下面是我遇到问题的代码的简化版本.当我在控制台应用程序中运行它时,它按预期工作.所有查询都是并行运行的,并Task.WaitAll()在完成后返回.

但是,当此代码在Web应用程序中运行时,请求才会挂起.当我附加一个调试器并中断所有时,它表明执行正在等待Task.WaitAll().第一项任务已经完成,但其他任务从未完成.

我无法弄清楚它在ASP.NET中运行时挂起的原因,但在控制台应用程序中运行良好.

public Foo[] DoWork(int[] values)
{
    int count = values.Length;
    Task[] tasks = new Task[count];

    for (int i = 0; i < count; i++)
    {
        tasks[i] = GetFooAsync(values[i]);
    }

    try
    {
        Task.WaitAll(tasks);
    }
    catch (AggregateException)
    {
        // Handle exceptions
    }

    return ...
}

public async Task<Foo> GetFooAsync(int value)
{
    Foo foo = null;

    Func<Foo, Task> executeCommand = async (command) =>
    {
        foo = new Foo();

        using (SqlDataReader reader = await command.ExecuteReaderAsync())
        {
            ReadFoo(reader, foo); …

假设我有以下类和它依赖的接口:

public class MyController
{
    private IRepository _repository;
    public MyController(IRepository repository)
    {
        _repository = repository;
    }

    public async Task MethodUnderTest(int someId)
    {
        var o = await _repository.FindById(someId);
        // update o
        await _repository.Commit();
    }
}

public interface IRepository
{
    Task Commit();
}

当我对这个方法进行单元测试时,我可以执行以下操作(使用xUnit和Rhino Mocks):

[Fact]
public async Task MyTest()
{
    IRepository repositoryStub = MockRepository.GenerateStub<IRepository>();

    MyController controller = new MyController(repositoryStub);

    await controller.MethodUnderTest(1);
}

这会因System.NullReferenceException而失败:对象引用未设置为对象的实例.

使用以下StackTrace:

UnitTest.MyController.<MethodUnderTest>d__0.MoveNext() in 
\UnitTest\Class1.cs:line 35
--- End of stack trace from previous location where …

我发现TaskCompletionSource.SetResult();在返回之前调用等待任务的代码.在我的情况下,导致死锁.

这是一个在普通版本中启动的简化版本 Thread

void ReceiverRun()
    while (true)
    {
        var msg = ReadNextMessage();
        TaskCompletionSource<Response> task = requests[msg.RequestID];

        if(msg.Error == null)
            task.SetResult(msg);
        else
            task.SetException(new Exception(msg.Error));
    }
}

代码的"异步"部分看起来像这样.

await SendAwaitResponse("first message");
SendAwaitResponse("second message").Wait();

Wait实际上嵌套在非异步调用中.

SendAwaitResponse(简化)

public static Task<Response> SendAwaitResponse(string msg)
{
    var t = new TaskCompletionSource<Response>();
    requests.Add(GetID(msg), t);
    stream.Write(msg);
    return t.Task;
}

我的假设是第二个SendAwaitResponse将在ThreadPool线程中执行,但它会在为ReceiverRun创建的线程中继续.

无论如何设置任务的结果而不继续等待代码？

应用程序是一个控制台应用程序.

我正在观看异步的禅:最佳性能的最佳实践和Stephen Toub开始谈论任务缓存,而不是缓存任务作业的结果,你自己缓存任务.据我所知,为每项工作开始一项新任务是昂贵的,应该尽量减少.在28:00左右,他展示了这种方法:

private static ConcurrentDictionary<string, string> s_urlToContents;

public static async Task<string> GetContentsAsync(string url)
{
    string contents;
    if(!s_urlToContents.TryGetValue(url, out contents))
    {
        var response = new HttpClient().GetAsync(url);
        contents = response.EnsureSuccessStatusCode().Content.ReadAsString();
        s_urlToContents.TryAdd(url, contents);
    }
    return contents;
}

首先看起来像是一个很好的思考方法,你缓存结果,我没有考虑缓存获取内容的工作.

而且他展示了这种方法:

private static ConcurrentDictionary<string, Task<string>> s_urlToContents;

public static Task<string> GetContentsAsync(string url)
{
    Task<string> contents;
    if(!s_urlToContents.TryGetValue(url, out contents))
    {
        contents = GetContentsAsync(url);
        contents.ContinueWith(t => s_urlToContents.TryAdd(url, t); },
        TaskContinuationOptions.OnlyOnRanToCompletion |
        TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
    }
    return contents;
}

private static async Task<string> GetContentsAsync(string url)
{ …

我遇到了为我的应用程序进行单元测试而创建的ac#class的问题,特别是围绕System.Threading.Tasks.Task对象的问题.

我有一个这样的对象列表,我想要同步执行每个对象.

我打电话给以下人:

myTask.RunSynchronously();

当我这样做时,我总是得到以下错误,我不知道为什么我可以解决它.

System.InvalidOperationException:可能不会在已启动的任务上调用RunSynchronously.

有人有任何想法吗？

(这可能与ASP.NET MVC4异步控制器问题重复- 为什么要使用？,但关于webapi,我不同意那里的答案)

假设我有一个长时间运行的SQL请求.它的数据应序列化为JSON并发送到浏览器(作为xhr请求的响应).示例代码:

public class DataController : ApiController
{
    public Task<Data> Get()
    {
        return LoadDataAsync(); // Load data asynchronously?
    }
}

当我执行$ .getJson('api/data',...)时会发生什么(请参阅此海报http://www.asp.net/posters/web-api/ASP.NET-Web-API-Poster. pdf):

1. [IIS] IIS接受请求.
2. [IIS] IIS等待来自托管池的一个线程[THREAD](http://msdn.microsoft.com/en-us/library/0ka9477y ( v=vs.110 ) .aspx)并开始工作.
3. [THREAD] Webapi在该线程和其他类中创建新的DataController对象.
4. [THREAD]使用任务并行lib在[THREAD2]中启动sql-query
5. [THREAD]返回托管池,准备进行其他处理
6. [THREAD2]与sql驱动程序一起工作,准备好读取数据并调用[THREAD3]来回复xhr请求
7. [THREAD3]发送响应.

如果出现问题,请随时纠正我.

在上面的问题中,他们说,点和利润是,[THREAD2]不是来自托管池,但MSDN文章(上面的链接)说

默认情况下,并行库类型类似Task并Task<TResult>使用线程池线程来运行任务.

所以我得出结论,所有三个线程都来自托管池.

此外,如果我使用同步方法,我仍然只使用一个线程(来自宝贵的线程池)保持我的服务器响应.

那么,从1个线程交换到3个线程的实际意义是什么？为什么不在线程池中最大化线程？

有没有明显有用的方法来使用异步控制器？

我正在处理一个托管对象在async方法中间过早完成的情况.

这是一个业余爱好家庭自动化项目(Windows 8.1,.NET 4.5.1),我向非托管第三方DLL提供C#回调.在某个传感器事件时调用回调.

为了处理这个事件,我使用async/await了一个简单的自定义awaiter(而不是TaskCompletionSource).我这样做的部分原因是为了减少不必要的分配数量,但主要是出于好奇心作为学习练习.

下面是我所拥有的非常剥离的版本,使用Win32计时器队列计时器来模拟非托管事件源.让我们从输出开始:

Press Enter to exit...
Awaiter()
tick: 0
tick: 1
~Awaiter()
tick: 2
tick: 3
tick: 4

请注意我的等待者在第二次打勾后如何最终确定.这是出乎意料的.

代码(控制台应用程序):

using System;
using System.Runtime.InteropServices;
using System.Threading;
using System.Threading.Tasks;

namespace ConsoleApplication
{
    class Program
    {
        static async Task TestAsync()
        {
            var awaiter = new Awaiter();
            //var hold = GCHandle.Alloc(awaiter);

            WaitOrTimerCallbackProc callback = (a, b) =>
                awaiter.Continue();

            IntPtr timerHandle;
            if (!CreateTimerQueueTimer(out timerHandle, 
                    IntPtr.Zero, 
                    callback, 
                    IntPtr.Zero, 500, 500, 0))
                throw new System.ComponentModel.Win32Exception( …

在使用时,Parallel.ForEach我们可以选择定义Parallel选项并设置Max Degree of Parallelism,如:

Parallel.ForEach(values, new ParallelOptions {MaxDegreeOfParallelism = number}, value = > {
    // Do Work
})

但在做PLINQ时:

Tabel.AsEnumberable()
     .AsParallel()
     .Where(//Logic)

我无法找到一种方法来设置MaxDegreeOfParallelism.我也在网上抬头,但没有找到任何东西.有没有人找到解决方法呢？任何帮助表示赞赏.