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

Помогите, пожалуйста, с пониманием асинхронной реализации Tcp клиент-сервера. Читал статьи на MDSN, гуглил. Но эта куча BeginWrite, BeginRead, коллбэков просто выносит мозг. Правда ли то, что несмотря на задающийся размер буфера в BeginRead, могут прийти меньше или больше данных? И как тогда с этим справляться? Как вообще эффективно реализовать простейший клиент-сервер, если первыми двумя байтами идет "код действия", а затем произвольное кол-во байт, в которых содержатся строки, int, ushort и т.д., если часть пакета может где-то застрять или прийти излишек? С синхронной реализацией все ок, но ест много ЦП т.к. цикл while.

Ответ

Сейчас принято делать всю асинхронность через 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(); } } }