Implementación de multitarea cooperativa en C #

Cuando se trata de multitarea en .Net, la gran mayoría de los casos asume una multitarea preventiva basada en subprocesos del sistema operativo. Pero este artículo se centrará en la implementación de la multitarea cooperativa , con la ayuda de la cual puede crear la apariencia de la operación simultánea de varios métodos utilizando un solo hilo.

Aquí está nuestra sencilla plantilla:

static void Main()
{
    DoWork("A", 4);
    DoWork("B", 3);
    DoWork("C", 2);
    DoWork("D", 1);
}

static void DoWork(string name, int num)
{
    for (int i = 1; i <= num; i++)
    {
        Console.WriteLine($"Work {name}: {i}");
    }
    Console.WriteLine($"Work {name} is completed");
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

4 A, B, C, D . , , , , , .

, , , : “ , - , , , , , ”.

: " C# - ?" ! await! await :

static async ValueTask DoWork(string name, int num)
{
    for (int i = 1; i <= num; i++)
    {
        Console.WriteLine($"Work {name}: {i}");
        await /*Something*/
    }
    Console.WriteLine($"Work {name} is completed");
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

await , , . await-? Task.Delay(), Task.Yield() , . , await, . CooperativeBroker:

private class CooperativeBroker : ICooperativeBroker
{
    private Action? _continuation;

    public void GetResult() 
        => this._continuation = null;

    public bool IsCompleted 
        => false;//Preventing sync completion in async method state machine

    public void OnCompleted(Action continuation)
    {
        this._continuation = continuation;
        this.InvokeContinuation();
    }

    public ICooperativeBroker GetAwaiter() 
        => this;

    public void InvokeContinuation() 
        => this._continuation?.Invoke();
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

C# , await . continuation OnCompleted. , continuation , , continuation , , . . CooperativeContext:

private class CooperativeBroker
{
    private readonly CooperativeContext _cooperativeContext;

    private Action? _continuation;

    public CooperativeBroker(CooperativeContext cooperativeContext)
        => this._cooperativeContext = cooperativeContext;

    ...

    public void OnCompleted(Action continuation)
    {
        this._continuation = continuation;
        this._cooperativeContext.OnCompleted(this);
    }

}

public class CooperativeContext
{
    private readonly List<CooperativeBroker> _brokers = 
        new List<CooperativeBroker>();

    void OnCompleted(CooperativeBroker broker)
    {
        ...
    }
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

OnCompleted :

private void OnCompleted(CooperativeBroker broker)
{
    //       .
    if (this._targetBrokersCount == this._brokers.Count)
    {
        var nextIndex = this._brokers.IndexOf(broker) + 1;
        if (nextIndex == this._brokers.Count)
        {
            nextIndex = 0;
        }

        this._brokers[nextIndex].InvokeContinuation();
    }
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

– , (_targetBrokersCount — ). , , "" .

, :

static void Main()
{
    CooperativeContext.Run(
        b => DoWork(b, "A", 4),
        b => DoWork(b, "B", 3),
        b => DoWork(b, "C", 2),
        b => DoWork(b, "D", 1)
    );
}

static async ValueTask DoWork(CooperativeBroker broker, string name, int num, bool extraWork = false)
{
    for (int i = 1; i <= num; i++)
    {
        Console.WriteLine($"Work {name}: {i}, Thread: {Thread.CurrentThread.ManagedThreadId}");
        await broker;
    }

    Console.WriteLine($"Work {name} is completed, Thread: {Thread.CurrentThread.ManagedThreadId}");
}

public class CooperativeContext
{
    public static void Run(params Func<CooperativeBroker, ValueTask>[] tasks)
    {
        CooperativeContext context = new CooperativeContext(tasks.Length);
        foreach (var task in tasks)
        {
            task(context.CreateBroker());
        }

        ...
    }

    ...

    private int _targetBrokersCount;

    private CooperativeContext(int maxCooperation)
    {
        this._threadId = Thread.CurrentThread.ManagedThreadId;
        this._targetBrokersCount = maxCooperation;
    }

    ...
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, , – , OnCompleted . "", :

public class CooperativeContext
{
    public static void Run(params Func<ICooperativeBroker, ValueTask>[] tasks)
    {
        CooperativeContext context = new CooperativeContext(tasks.Length);
        foreach (var task in tasks)
        {
            task(context.CreateBroker());
        }

        //        ,  
        //   
        while (context._brokers.Count > 0)
        {
            context.ReleaseFirstFinishedBrokerAndInvokeNext();
        }
    }

    ...
    private void ReleaseFirstFinishedBrokerAndInvokeNext()
    {
        // IsNoAction     
        var completedBroker = this._brokers.Find(i => i.IsNoAction)!;

        var index = this._brokers.IndexOf(completedBroker);
        this._brokers.RemoveAt(index);
        this._targetBrokersCount--;

        if (index == this._brokers.Count)
        {
            index = 0;
        }

        if (this._brokers.Count > 0)
        {
            this._brokers[index].InvokeContinuation();
        }
    }    
}

private class CooperativeBroker : ICooperativeBroker
{
    ...
    public bool IsNoAction
        => this._continuation == null;
    ...
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

( ):

static void Main()
{
    CooperativeContext.Run(
        b => DoWork(b, "A", 4),
        b => DoWork(b, "B", 3, extraWork: true),
        b => DoWork(b, "C", 2),
        b => DoWork(b, "D", 1)
    );
}

static async ValueTask DoWork(
    ICooperativeBroker broker, 
    string name, 
    int num, 
    bool extraWork = false)
{
    for (int i = 1; i <= num; i++)
    {
        Console.WriteLine(
               $"Work {name}: {i}, Thread: {Thread.CurrentThread.ManagedThreadId}");
        await broker;
        if (extraWork)
        {
            Console.WriteLine(
                   $"Work {name}: {i} (Extra), Thread: {Thread.CurrentThread.ManagedThreadId}");
            await broker;
        }
    }

    Console.WriteLine(
           $"Work {name} is completed, Thread: {Thread.CurrentThread.ManagedThreadId}");
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

:

Work A: 1, Thread: 1
Work B: 1, Thread: 1
Work C: 1, Thread: 1
Work D: 1, Thread: 1
Work A: 2, Thread: 1
Work B: 1 (Extra), Thread: 1
Work C: 2, Thread: 1
Work D is completed, Thread: 1
Work A: 3, Thread: 1
Work B: 2, Thread: 1
Work C is completed, Thread: 1
Work A: 4, Thread: 1
Work B: 2 (Extra), Thread: 1
Work A is completed, Thread: 1
Work B: 3, Thread: 1
Work B: 3 (Extra), Thread: 1
Work B is completed, Thread: 1
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, , , , .

---

, , , , , C# .

github.

[Update] DistortNeo, ofrece una solución más práctica a este problema.