Implementing Cooperative Multitasking in C #

When it comes to multitasking in .Net, the vast majority of cases assume preemptive multitasking based on operating system threads. But this article will focus on the implementation of cooperative multitasking, with which you can create the appearance of simultaneous operation of several methods using just one single thread.

Here is our simple template:

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, he offers a more practical solution to this problem.