Асинхронный Tcp клиент-сервер

#c_sharp #асинхронность #tcp

Помогите, пожалуйста, с пониманием асинхронной реализации Tcp клиент-сервера. Читал
статьи на MDSN, гуглил. Но эта куча BeginWrite, BeginRead, коллбэков просто выносит мозг.


Правда ли то, что несмотря на задающийся размер буфера в BeginRead, могут прийти
меньше или больше данных? И как тогда с этим справляться?


Как вообще эффективно реализовать простейший клиент-сервер, если первыми двумя байтами
идет "код действия", а затем произвольное кол-во байт, в которых содержатся строки,
int, ushort и т.д., если часть пакета может где-то застрять или прийти излишек?


С синхронной реализацией все ок, но ест много ЦП т.к. цикл while.    

Ответы

Ответ 1

Сейчас принято делать всю асинхронность через async/await. Вот несколько примеров:
[1], [2], я надёргаю кусков из них.

Сервер слушает входящие сообщения, и при приходе запускает на обработку. Обработка
бежит параллельно, а сервер продолжает слушать дальше в цикле (AcceptTcpClientAsync()
и дальше). Получится что-то такое:

Весь сервер:

void RunServer()
{
    var tcpListener = TcpListener.Create(<порт>);
    tcpListener.Start();
    while (true) // тут какое-то разумное условие выхода
    {
        var tcpClient = await tcpListener.AcceptTcpClientAsync();
        processClientTearOff(tcpClient); // await не нужен
    }
}

async Task processClientTearOff(TcpClient c)
{
    using (var client = new Client(c))
        await client.ProcessAsync();
}


Обработчик одного клиентского запроса:

class Client : IDisposable
{
    NetworkStream s;

    public Client(TcpClient c)
    {
        s = client.GetStream();
    }

    public void Dispose()
    {
        s.Dispose();
    }

    async Task ReadFromStreamAsync(int nbytes)
    {
        var buf = new byte[nbytes];
        var readpos = 0;
        while (readpos < nbytes)
            readpos += await s.ReadAsync(buf, readpos, nbytes - readpos);
        return buf;
    }

    public async Task ProcessAsync()
    {
        var actionBuffer = await ReadFromStreamAsync(2);
        var action = (ActionEnum)BitConverter.ToInt16(actionBuffer, 0);
        switch (action)
        {
            // логика в зависимости от кода команды
        }
    }
}


Если сервер надо останавливать, вам придётся дождаться окончания работы запущенных
Task'ов:

async void RunServer()
{
    var tcpListener = TcpListener.Create(<порт>);
    tcpListener.Start();
    while (можно продолжать)
    {
        var tcpClient = await tcpListener.AcceptTcpClientAsync();
        processClient(tcpClient); // await не нужен
    }
    await Task.WaitAll(activeClientTasks.ToList()); // нужна копия
}

HashSet activeClientTasks = new HashSet();
async Task processClient(TcpClient c)
{
    using (var client = new Client(c))
    {
        Task task = null;
        try
        {
            task = client.ProcessAsync();
            activeClientTasks.Add(task);
            await task;
        }
        finally
        {
            if (task != null)
                activeClientTasks.Remove(task);
        }
    }
}




Вот реализация таймаута (набросал, код не запускал, возможны ошибки):

async Task ReadFromStreamAsync(int nbytes, CancellationToken ct)
{
    var buf = new byte[nbytes];
    var readpos = 0;
    while (readpos < nbytes)
        readpos += await s.ReadAsync(buf, readpos, nbytes - readpos, ct);
    return buf;
}

async Task ReadWithTimeout(int n)
{
    using (var cts = new CancellationTokenSource())
    {
        var readTask = ReadFromStreamAsync(n, cts.Token);
        var timeoutTask = Task.Delay(1000);
        await Task.WhenAny(readTask, timeoutTask);
        if (!readTask.IsCompleted)
        {
            cts.Cancel(); // cancel read task
            return null;
        }
        else
        {
            var bytes = readTask.Result;
            return Decode(bytes);
        }
    }
}




Обновление:

Следуя совету @Pavel Mayorov, последний метод можно переписать проще и изящнее:

async Task ReadWithTimeout(int n) // (*)
{
    using (var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(1000)))
    {
        try
        {
            return Decode(await ReadFromStreamAsync(n, cts.Token));
        }
        catch (OperationCanceledException)
        {
            return null;
        }
    }
}




Обновление:

К сожалению, несмотря свою изящность, метод с обрывом чтения (я имею в виду вариант,
обозначенный (*)) не работает из-за бага в BCL. Я попробую уточнить код. Кажется, хорошей
идеей является закрыть клиент полностью, в соответствии со старой семантикой WinAPI.
Вот сообщение на Microsoft Connect, проголосуйте за исправление!



Обновление: Поправил код, руководствуясь предыдущим обновлением. Теперь при обрыве
по таймауту возвращается не null, а бросается TimeoutException (после которого клиент
нужно закрывать).

async Task ReadWithTimeout(int n)
{
    using (var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(1000)))
    {
        var token = cts.Token;
        try
        {
            // при обрыве ожидания по токену закрываем клиент
            // с выбросом ObjectDisposedException
            using (token.Register(client.Close))
                return Decode(await ReadFromStreamAsync(n, cts.Token));
        }
        catch (ObjectDisposedException) when (token.IsCancellationRequested)
        {
            throw new TimeoutException();
        }
    }
}