标签: task-parallel-library

我有一些库(套接字网络)代码,它Task基于提供基于API的待处理请求响应TaskCompletionSource<T>.然而,TPL中的一个烦恼是,似乎不可能阻止同步延续.我会希望能够做的是两种:

• 告诉a TaskCompletionSource<T>不应该允许呼叫者附加TaskContinuationOptions.ExecuteSynchronously,或
• 使用池来设置结果(SetResult/ TrySetResult)以指定TaskContinuationOptions.ExecuteSynchronously应该忽略的方式

具体来说,我遇到的问题是传入的数据正在由专用的阅读器处理,如果调用者可以附加,TaskContinuationOptions.ExecuteSynchronously他们可以阻止阅读器(这不仅影响它们).以前,我通过一些hackery解决了这个问题,它检测是否存在任何延续,如果它们将完成推送到ThreadPool,那么如果调用者已经使工作队列饱和​​,则会产生重大影响,因为完成将不会被处理及时.如果他们使用Task.Wait()(或类似),他们将基本上陷入僵局.同样,这就是读者使用专用线程而不是使用工作者的原因.

所以; 在我尝试唠叨TPL团队之前:我错过了一个选项吗？

关键点:

• 我不希望外部呼叫者能够劫持我的线程
• 我不能使用它ThreadPool作为一个实现,因为它需要在池饱和时工作

以下示例生成输出(排序可能因时间而异):

Continuation on: Main thread
Press [return]
Continuation on: Thread pool

问题在于随机调用者设法在"主线程"上获得延续.在实际代码中,这将打断主要读者; 坏事!

码:

using System;
using System.Threading;
using System.Threading.Tasks;

static class Program
{
    static void Identify()
    {
        var thread = Thread.CurrentThread;
        string name = thread.IsThreadPoolThread
            ? "Thread pool" : thread.Name;
        if (string.IsNullOrEmpty(name))
            name = …

在我的一个有点聚合的项目中,我从网上解析了feed,podcast等.

如果我使用顺序方法,考虑到大量资源,处理所有资源需要相当长的时间(因为网络问题和类似的东西);

foreach(feed in feeds)
{
   read_from_web(feed)
   parse(feed)
}

所以我想实现并发,并且无法决定我是否应该使用ThreadPools来处理工作线程,或者只是依靠TPL来对它进行排序.

ThreadPools肯定会用工作线程处理我的工作,我会得到我所期望的(在多核CPU环境中,其他核心也将被利用).

但我仍然想考虑TPL,因为它推荐的方法,但我有点担心它.首先,我知道TPL使用ThreadPools,但增加了额外的决策层.我主要关心的是单核环境存在的情况.如果我没错,TPL从一开始就用一个数量的工作线程开始,这些工作线程数等于可用的CPU核心数.我担心TPL会产生与我的IO绑定案例的顺序方法类似的结果.

因此,对于IO绑定操作(在我的情况下从Web读取资源),最好是使用ThreadPools来控制事物,还是更好地依赖于TPL？TPL也可以用于IO绑定场景吗？

更新:我主要担心的是 - 在单核CPU环境中,TPL只是表现得像顺序方式,还是会提供并发性？我已经阅读了使用Microsoft .NET的并行编程,所以这本书却找不到确切的答案.

注意:这是我之前的问题的重新措辞[ 是否可以将线程并发和并行一起使用？这是错误的措辞.

我有一些异步代码,我想添加一个CancellationToken.但是,有许多实现不需要这样做,所以我想有一个默认参数 - 也许是CancellationToken.None.然而,

Task<x> DoStuff(...., CancellationToken ct = null)

产量

类型''的值不能用作默认参数,因为没有标准转换来键入'System.Threading.CancellationToken'

和

Task<x> DoStuff(...., CancellationToken ct = CancellationToken.None)

'ct'的默认参数值必须是编译时常量

有没有办法让CancellationToken有一个默认值？

我从一个函数开始一个新的任务,但我不希望它在同一个线程上运行.我不关心它运行在哪个线程上,只要它是一个不同的线程(因此这个问题中给出的信息没有帮助).

我保证TestLock在允许Task t再次输入之前,以下代码将始终退出吗？如果没有,建议的设计模式是什么来防止再次出现？

object TestLock = new object();

public void Test(bool stop = false) {
    Task t;
    lock (this.TestLock) {
        if (stop) return;
        t = Task.Factory.StartNew(() => { this.Test(stop: true); });
    }
    t.Wait();
}

编辑:基于Jon Skeet和Stephen Toub的以下答案,确定性地防止重入的一种简单方法是传递CancellationToken,如此扩展方法所示:

public static Task StartNewOnDifferentThread(this TaskFactory taskFactory, Action action) 
 {
    return taskFactory.StartNew(action: action, cancellationToken: new CancellationToken());
}

我得到了这个程序,它给我语法错误"System.Threading.Tasks.task不包含Run的定义."

我正在使用VB 2010 .NET 4.0任何想法？在.net 4.0中运行的任何替换？

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace ChatApp
{
class ChatProg
{
    static void Main(string[] args)
    {
        Task<int> wakeUp = DoWorkAsync(2000,"Waking up");
        Task.WaitAll(wakeUp);
    }

    static Task<int> DoWorkAsync(int milliseconds, string name)
    {

        //error appears below on word Run
        return Task.Run(() =>
            {
                Console.WriteLine("* starting {0} work", name);
                Thread.Sleep(milliseconds);
                Console.WriteLine("* {0} work one", name);
                return 1;
            });
    }
}
}

使用Thread非常简单

 Thread thread = new Thread(MethodWhichRequiresSTA);
 thread.SetApartmentState(ApartmentState.STA);  

如何使用WPF应用程序中的任务完成相同的操作？这是一些代码:

Task.Factory.StartNew
  (
    () => 
    {return "some Text";}
  )
   .ContinueWith(r => AddControlsToGrid(r.Result));  

我收到了一个InvalidOperationException

调用线程必须是STA,因为许多UI组件都需要这个.

我想await在BlockingCollection<T>.Take()异步的结果,所以我不阻止线程.寻找这样的事情:

var item = await blockingCollection.TakeAsync();

我知道我可以这样做:

var item = await Task.Run(() => blockingCollection.Take());

但是有点杀死了整个想法,因为另一个线程ThreadPool被阻止了.

还有其他选择吗？

我正在开展一个多任务网络项目,我是新手Threading.Tasks.我实现了一个简单的Task.Factory.StartNew(),我想知道我该怎么做Task.Run()？

这是基本代码:

Task.Factory.StartNew(new Action<object>(
(x) =>
{
    // Do something with 'x'
}), rawData);

我System.Threading.Tasks.Task在对象浏览器中查看,我找不到Action<T>类似的参数.只有Action这需要void参数,没有类型.

只有两件事情similiar:static Task Run(Action action)和static Task Run(Func<Task> function),但不能发布与两个参数(一个或多个).

是的,我知道我可以,但创建一个简单的扩展方法,我的主要问题是,我们可以把它写在一行用Task.Run()？

我正在尝试反序列化http://api.usa.gov/jobs/search.json?query=nursing+jobs使用.NET 4.0任务模式返回的JSON .它返回此JSON('加载JSON数据'@ http://jsonviewer.stack.hu/).

[
  {
    "id": "usajobs:353400300",
    "position_title": "Nurse",
    "organization_name": "Indian Health Service",
    "rate_interval_code": "PA",
    "minimum": 42492,
    "maximum": 61171,
    "start_date": "2013-10-01",
    "end_date": "2014-09-30",
    "locations": [
      "Gallup, NM"
    ],
    "url": "https://www.usajobs.gov/GetJob/ViewDetails/353400300"
  },
  {
    "id": "usajobs:359509200",
    "position_title": "Nurse",
    "organization_name": "Indian Health Service",
    "rate_interval_code": "PA",
    "minimum": 42913,
    "maximum": 61775,
    "start_date": "2014-01-16",
    "end_date": "2014-12-31",
    "locations": [
      "Gallup, NM"
    ],
    "url": "https://www.usajobs.gov/GetJob/ViewDetails/359509200"
  },
  ...
]

指数行动:

  public class HomeController : Controller
  {
    public ActionResult Index()
    {
      Jobs model = null;
      var …

有了System.Threading.Tasks.Task<TResult>,我必须管理可能抛出的异常.我正在寻找最好的方法.到目前为止,我已经创建了一个基类来管理调用中的所有未捕获的异常.ContinueWith(...)

我想知道是否有更好的方法可以做到这一点.或者即使这是一个很好的方法.

public class BaseClass
{
    protected void ExecuteIfTaskIsNotFaulted<T>(Task<T> e, Action action)
    {
        if (!e.IsFaulted) { action(); }
        else
        {
            Dispatcher.CurrentDispatcher.BeginInvoke(new Action(() =>
            {
                /* I display a window explaining the error in the GUI 
                 * and I log the error.
                 */
                this.Handle.Error(e.Exception);
            }));            
        }
    }
}   

public class ChildClass : BaseClass
{
    public void DoItInAThread()
    {
        var context = TaskScheduler.FromCurrentSynchronizationContext();
        Task.Factory.StartNew<StateObject>(() => this.Action())
                    .ContinueWith(e => this.ContinuedAction(e), context);
    }

    private void ContinuedAction(Task<StateObject> e)
    {
        this.ExecuteIfTaskIsNotFaulted(e, () => …