parallel - task run async c#

Thread Join() hace que Task.RunSynchronously no termine

Question

Llamar a _thread.Join() hace que el bucle GetConsumingEnumerable quede bloqueado en el último elemento. ¿Por qué ocurre este comportamiento?

  public abstract class ActorBase : IDisposable
  {
    private readonly BlockingCollection<Task> _queue = new BlockingCollection<Task>(new ConcurrentQueue<Task>());
    private readonly Thread _thread;
    private bool _isDisposed;

    protected ActorBase()
    {
      _thread = new Thread(ProcessMessages);
      _thread.Start();
    }

    protected void QueueTask(Task task)
    {
      if (_isDisposed)
      {
        throw new Exception("Actor was disposed, cannot queue task.");
      }
      _queue.Add(task);
    }

    private void ProcessMessages()
    {
      foreach (var task in _queue.GetConsumingEnumerable())
      {
        task.RunSynchronously();
      }
    }

    public void Dispose()
    {
      _isDisposed = true;
      _queue.CompleteAdding();
      _thread.Join();
    }
  }

  public class SampleActor : ActorBase
  {
    private string GetThreadStatus()
    {
      Thread.Sleep(500);
      return string.Format("Running on thread {0}", Thread.CurrentThread.ManagedThreadId);
    }

    public async Task<string> GetThreadStatusAsync()
    {
      var task = new Task<string>(GetThreadStatus);
      QueueTask(task);
      return await task;
    }
  }

  class Program
  {
    public static async Task Run()
    {
      using (var sa = new SampleActor())
      {
        for (int i = 0; i < 3; i++)
        {
          Console.WriteLine(await sa.GetThreadStatusAsync());
        }
      }
    }

    public static void Main(string[] args)
    {
      Console.WriteLine("Main thread id {0}", Thread.CurrentThread.ManagedThreadId);
      var task = Task.Run(async ()=> { await Run(); });
      task.Wait();
    }
  }

El contexto para este enfoque es que necesito asegurarme de que todas las operaciones se ejecuten en una secuencia del sistema operativo, lo que permitiría a una parte de la aplicación usar credenciales diferentes a la del hilo principal.

Answer 1

Podría ser tentador hacer una simple comprobación para ver si el hilo que intentas Join es Thread.CurrentThread , pero sería incorrecto.

Además, creo que todo el enfoque ( programar y ejecutar objetos de Task fríos con un programador personalizado que no cumpla con TPL) es incorrecto . Debería utilizar un planificador de tareas amigable para TPL, similar al StaTaskScheduler Stephen Toub. O ejecute un SynchronizationContext personalizado para su hilo que sirva como actor (como AsyncPump de Toub) y use TaskScheduler.FromCurrentSynchronizationContext y Task.Factory.StartNew para programar tareas con su planificador personalizado (o use Task.Start(TaskScheduler) si tiene que lidiar con el frío Tareas).

De esta forma, tendrá control total sobre dónde se ejecutan las tareas y sus continuaciones, así como sobre la alineación de tareas .

Answer 2

async-await funciona con continuaciones. Para ser eficiente y reducir la programación de estas continuaciones, generalmente se ejecuta en el mismo subproceso que completó la tarea anterior.

Eso significa que, en su caso, su hilo especial no solo está ejecutando las tareas, sino que también ejecuta todas las continuación después de estas tareas (el propio bucle for ). Puede ver eso al imprimir el id. De subproceso:

using (var sa = new SampleActor())
{
    for (int i = 0; i < 3; i++)
    {
        Console.WriteLine(await sa.GetThreadStatusAsync());
        Console.WriteLine("Continue on thread :" + Thread.CurrentThread.ManagedThreadId);
    }
}

Cuando se completa el ciclo for y se SampleActor el SampleActor se llama a Thread.Join desde el mismo hilo que intentas unir, por lo que se llega a un punto muerto. Tu situación se reduce a esto:

public static void Main()
{
    Thread thread = null;
    thread = new Thread(() =>
    {
        Thread.Sleep(100);
        thread.Join();
        Console.WriteLine("joined");
    });
    thread.Start();
}

En .Net 4.6 puede resolver esto con TaskCreationOptions.RunContinuationsAsynchronously pero en la versión actual puede especificar el TaskScheduler predeterminado:

public Task<string> GetThreadStatusAsync()
{
    var task = new Task<string>(GetThreadStatus);
    QueueTask(task);
    return task.ContinueWith(task1 => task1.GetAwaiter().GetResult(), TaskScheduler.Default);
}