G>>

_> /*skipped*/
_>

var len1 = network.Read(buf1,0, buf1.Length);
var len2 = file.Read(buf2,0, buf2.Length);
someF(buf1,len1, buf2, len2)

var result = await Task.WhenAll(network.ReadAsync(buf1,0, buf1.Length),file.ReadAsync(buf2,0, buf2.Length));
someF(buf1,result[0], buf2, result[1])

def download_and_update_statistics(url):
    x = statistics
    y = download_and_calculate_statistics(url)
    statistics =  x + y

EP>

EP>// must happen atomically
EP>{
EP>    half_modify(structure);
EP>    stream >> x;
EP>    finish_modification(structure);
EP>}
EP>

В этом случае если внутри "stream >> x" произойдёт yield, то инварианты могут быть поломаны. Но, если есть такой код — то ты его и на обычных потоках не сможешь запустить, корутины тут не причём

EP>>>std::getline тут стандартная, не модифицированная. Точно также можно протащить yield и через любую другую функцию

    Task task;
    while(!finished)
    {
        main_tasks.pop(task);
        task();
    }

EP>>>>>При этом код использующий этот поток не поменяется, а в случае с C# await — нужно будет патчить весь call-stack

EP>>>С такими же рассуждениями можно прийти к мысли что и yield, и await/async не нужны

EP>Ну ты говоришь что преобразование синхронного кода в асинхронный это механическая операция. Так и преобразование кода с await/yield — точно такая же механическая операция

EP>Слушай, ну если ты не нацелен на конструктив, давай прекратим, ок? По второй ссылке Boost.Asio — самое что ни на есть асинхронное I/O, с тем самым IOCP под капотом

EP>>>Ты мне предлагаешь реализовать свои обвёртки поверх голого IOCP? Зачем?

EP>Подожди, то есть ты думаешь что для C++ нет библиотек асинхронным API

EP>Детский сад — для того чтобы пользоваться библиотекой знать её код не обязательно. В нормальных библиотеках есть чётко определённый интерфес

EP>Это привязка не к устройству, а к интерфейсу

G>var len1 = network.Read(buf1,0, buf1.Length);
G>var len2 = file.Read(buf2,0, buf2.Length);
G>someF(buf1,len1, buf2, len2)
G>

G>var result = await Task.WhenAll(network.ReadAsync(buf1,0, buf1.Length),file.ReadAsync(buf2,0, buf2.Length));
G>someF(buf1,result[0], buf2, result[1])
G>

vector<char> buf1, buf2;
network.Read(buf1);
file.Read(buf2);
F(buf1, buf2);

vector<char> buf1, buf2;
int c=2;
network.StartAsyncRead(buf1, [&]{if(--c==0) CallFromUI(__coro);});
file.StartAsyncRead(buf2, [&]{if(--c==0) CallFromUI(__coro);});
__yield();
F(buf1, buf2);

// Вот код, с которым все шоколадно в синхронном однопоточном коде
value = read()
write(next(value))

// а вот однопоточныый вариант, но асинхронный
нить 1 value = read()
нить 2 value = read()
нить 2 write(next(value)) 
нить 1 write(next(value)) // нарушение инварианта

// и вот
нить 1 value = read()
нить 2 value = read()
нить 1 write(next(value))
нить 2 write(next(value)) // нарушение инварианта

static void ComplexFunc(Stream src, Stream dst)
{
    var buf = new byte[256];
    while(src.Position < src.Length)
    {
        var len = src.Read(buf,0,buf.Length);
        dst.Write(buf, 0, len);
    }
}

private void button1_Click(object sender, EventArgs e)
{
    for (int i = 0; i < 100; i++)
    {
        using (var src = File.OpenRead("TextFile1.txt"))
        using (var dst = File.Create("TextFile2.txt"))
            ComplexFunc(src, dst);
        Console.WriteLine("Iteration " + i);
    }
}

private void button1_Click(object sender, EventArgs e)
{
    scheduler.CreateFiber((s, yield) => { 
        for (int i = 0; i < 100; i++)
        {
            using (var src = new AsyncStream(File.OpenRead("TextFile1.txt"), yield))
            using (var dst = new AsyncStream(File.Create("TextFile2.txt"), yield))
                ComplexFunc(src, dst);
            Console.WriteLine("Iteration " + i);
        }
    });
}

class AsyncStream:Stream
{
    Stream baseStream;
    Action<bool> yield;

    public AsyncStream(Stream baseStream, Action<bool> yield)
    {
        this.baseStream = baseStream;
        this.yield = yield;
    }

    public override int Read(byte[] buffer, int offset, int count)
    {
        var task = baseStream.ReadAsync(buffer, offset, count);

        task.GetAwaiter().OnCompleted(() => yield(true));
        yield(false);
        return task.Result;            
    }

    public override void Write(byte[] buffer, int offset, int count)
    {
        var task = baseStream.WriteAsync(buffer, offset, count);
        task.GetAwaiter().OnCompleted(() => yield(true));
        yield(false);
        return;
    }
}

public class FiberScheduler
{
    private ConcurrentQueue<Fiber> queue = new ConcurrentQueue<Fiber>();

    public uint PrimaryId { get; private set; }


    public FiberScheduler()
    {
        this.PrimaryId = UnmanagedFiberAPI.ConvertThreadToFiber(0);
        if (this.PrimaryId == 0)
        {
            throw new Win32Exception();
        }
    }

    public Fiber CreateFiber(Action<FiberScheduler, Action<bool>> action)
    {
        var fiber = new Fiber(action, this);
        queue.Enqueue(fiber);
        return fiber;
    }

    public void Delete(Fiber fiber)
    {
        fiber.Delete();
        UnmanagedFiberAPI.DeleteFiber(fiber.Id);
            
    }

    internal void Switch(Fiber fib, bool enqueue)
    {
        if (enqueue)
        {
            queue.Enqueue(fib);
        }
        UnmanagedFiberAPI.SwitchToFiber(this.PrimaryId);
    }

    public void Next()
    {
        Fiber fib;
        if (queue.TryDequeue(out fib))
        {
            switch (fib.Status)
            {
                case FiberStatus.Error:
                case FiberStatus.Started:
                case FiberStatus.Done:
                    UnmanagedFiberAPI.SwitchToFiber(fib.Id);
                    break;
                default:
                    break;
            }
        }
    }
}

public enum FiberStatus
{
    Error,
    Started,
    Done,
    Deleted
}
public class Fiber
{
    private Action<FiberScheduler, Action<bool>> action;

    public uint Id { get; private set; }

    public FiberScheduler Scheduler { get; private set; }

    public FiberStatus Status { get; internal set; }


    internal Fiber(Action<FiberScheduler, Action<bool>> action, FiberScheduler scheduler)
    {
        this.action = action;

        Scheduler = scheduler;
        Status = FiberStatus.Started;

        UnmanagedFiberAPI.LPFIBER_START_ROUTINE lpFiber = new UnmanagedFiberAPI.LPFIBER_START_ROUTINE(FiberRunnerProc);
        Id = UnmanagedFiberAPI.CreateFiber(0, lpFiber, 0);

        if(Id == 0)
        {
            throw new Win32Exception();
        }
    }

    private uint FiberRunnerProc(uint lpParam)
    {
        uint status = 0;
        try
        {
            action(this.Scheduler, Yield);
            Status = FiberStatus.Done;
        }

        catch (Exception)
        {
            status = 1;
            Status = FiberStatus.Error; 
            throw;
        }

        finally
        {
            Yield(false);
            UnmanagedFiberAPI.DeleteFiber((uint)Id);
            Status = FiberStatus.Deleted;
        }

        return status;
    }

    private void Yield(bool enqueue)
    {
        this.Scheduler.Switch(this, enqueue);
            
    }

    internal void Delete()
    {
        this.Status = FiberStatus.Deleted;            
    }
}

internal static class UnmanagedFiberAPI
{
    public delegate uint LPFIBER_START_ROUTINE(uint param);

    [DllImport("Kernel32.dll")]
    public static extern uint ConvertThreadToFiber(uint lpParameter);

    [DllImport("Kernel32.dll")]
    public static extern void SwitchToFiber(uint lpFiber);

    [DllImport("Kernel32.dll")]
    public static extern void DeleteFiber(uint lpFiber);

    [DllImport("Kernel32.dll")]
    public static extern uint CreateFiber(uint dwStackSize, LPFIBER_START_ROUTINE lpStartAddress, uint lpParameter);
}

_>vector<char> buf1, buf2;
_>network.Read(buf1);
_>file.Read(buf2);
_>F(buf1, buf2);
_>

_>vector<char> buf1, buf2;
_>int c=2;
_>network.StartAsyncRead(buf1, [&]{if(--c==0) CallFromUI(__coro);});
_>file.StartAsyncRead(buf2, [&]{if(--c==0) CallFromUI(__coro);});
_>__yield();
_>F(buf1, buf2);
_>

A>def download_and_update_statistics(url):
A>    x = statistics
A>    y = download_and_calculate_statistics(url)
A>    statistics =  x + y
A>

EP>>

EP>>// must happen atomically
EP>>{
EP>>    half_modify(structure);
EP>>    stream >> x;
EP>>    finish_modification(structure);
EP>>}
EP>>

В этом случае если внутри "stream >> x" произойдёт yield, то инварианты могут быть поломаны. Но, если есть такой код — то ты его и на обычных потоках не сможешь запустить, корутины тут не причём

Придумай хотя бы пару примеров поломки инвариантов в функции которая принимает пользовательский stream, и гарантированно работает в многопоточной среде.

EP>>>>std::getline тут стандартная, не модифицированная. Точно также можно протащить yield и через любую другую функцию

for(auto &&f : fs)
    cout << await f << ":\tafter end" << endl;
// or
await when_all(fs);

G>    Task task;
G>    while(!finished)
G>    {
G>        main_tasks.pop(task);
G>        task();
G>    }
G>

    // Custom scheduling is not required - can be integrated
    // to other systems transparently
    main_tasks.push([]
    {
        asynchronous([]
        {
            return async_user_handler(),
                   finished = true;
        });
    });

    Task task;
    while(!finished)
    {
        main_tasks.pop(task);
        task();
    }

EP>>Ну ты говоришь что преобразование синхронного кода в асинхронный это механическая операция. Так и преобразование кода с await/yield — точно такая же механическая операция

// было:
void task()
{
    MessageBox(0, TEXT("on max"), TEXT(""), MB_OK);
    async_something([]
    {
        MessageBox(0, TEXT("on min"), TEXT(""), MB_OK);
    });
}
// стало:
void task()
{
    // without syntax sugar
    typedef boost::coroutines::coroutine<void()> Coro;
    auto &&red_portal = make_shared<Coro>();
    *red_portal = Coro([=](Coro::caller_type &blue_portal)
    {
        MessageBox(0, TEXT("on max"), TEXT(""), MB_OK);
        async_something( [=]{ (*red_portal)(); } );
        blue_portal();
        MessageBox(0, TEXT("on min"), TEXT(""), MB_OK);
    });
}

G>В .NET очень легко — берешь dll и достаешь исходники.

EP>>Слушай, ну если ты не нацелен на конструктив, давай прекратим, ок? По второй ссылке Boost.Asio — самое что ни на есть асинхронное I/O, с тем самым IOCP под капотом

I>Ну значит UI Loop из моего примера вызовется чудом, например сканированием адресного пространства на предмет наличия ::GetMessage и ::TranslateMessage или, как вариант, короутина догадается WinMain вызвать.

EP>>Ну ты говоришь что преобразование синхронного кода в асинхронный это механическая операция. Так и преобразование кода с await/yield — точно такая же механическая операция
G>Во-первых это не так из-за message pump.

EP>>Вот ты тоже не знал, не верил, отрицал. Зато теперь "И что с того ?" — большой прогресс

// было:
void task()
{
    MessageBox(0, TEXT("on max"), TEXT(""), MB_OK);
    async_something([]
    {
        MessageBox(0, TEXT("on min"), TEXT(""), MB_OK);
    });
}
// стало:
void task()
{
    // without syntax sugar
    typedef boost::coroutines::coroutine<void()> Coro;
    auto &&red_portal = make_shared<Coro>();
    *red_portal = Coro([=](Coro::caller_type &blue_portal)
    {
        MessageBox(0, TEXT("on max"), TEXT(""), MB_OK);
        async_something( [=]{ (*red_portal)(); } );
        blue_portal();
        MessageBox(0, TEXT("on min"), TEXT(""), MB_OK);
    });
}