Как исправить ошибку End of Stream encountered before parsing was completed?

Question

[NameOf Var]

Имеется два .NET приложения — клиент и сервер. Клиент по TCP раз в пять секунд отправляет хеш настроек на сервер (используется TcpClient). Сервер в свою очередь ожидает получения TCP запроса от клиента (TcpListener) и при получении сверяет хеш настроек клиента с хешем текущих настроек. Если хеши различаются, то сервер формирует ответ клиенту и по TCP отправляет агенту новые настройки в бинарном формате (через BinaryFormatter). Клиент в свою очередь ожидает в отдельном потоке новые настройки и как только настройки прибывают, он десериализует их BinaryFormatter'ом и применяет их. Если настроек на сервере мало, то всё отрабатывает удачно. Но как только увеличивается количество настроек, то клиент падает при попытке десериализовать объект (Ошибка: Конец потока обнаружен до завершения разбора или End of Stream encountered before parsing was completed).
Итак, краткая схема работы.
1. При запуске клиента открывается один TCPListener на порту 8765 и в бесконечном цикле while(true) проверяется условие: если (listener.Pending()), то читаем ответ от сервера, десериализуем его и применяем полученные настройки от сервера.
2. Во втором потоке клиента каждые пять секунд шлется TCP запрос на сервер, в этом запросе хранится хеш текущих настроек клиента.
3. При запуске сервера открывается один TcpListener на порту 8766 и и в бесконечном цикле while(true) проверяется условие: если (listener.Pending()), то сравниваем хеш настроек клиента с текущем хешем настроек на сервере. Если хеши различаются, то на порт 8765 отправляем новые настройки на клиент.

== Клиент ==

Код создания TcpListener'a

try
            {
                listener = TcpListener.Create(ListenPortNumber);
                listener.Start();
            }
            catch (Exception ex)
            {
                _listenToServerThrLive = false;
            }

            while (_listenToServerThrLive)
            {
                try
                {
                    if (listener.Pending())
                    {
                        ThreadPool.QueueUserWorkItem(HandleServerRespose, listener);
                    }
                }
                catch (Exception ex)
                {
                    Thread.Sleep(100);
                    continue; // --> while(_listenToServerThrLive)
                }

                Thread.Sleep(100);
            }

HandleServerRespose

private void HandleServerRespose(object objListener)
        {
            TcpListener listener = objListener as TcpListener;

            try
            {
                ResponseData hbRespData = null;
            byte[] data = new byte[256];

            using (MemoryStream memStream = new MemoryStream())
            {
                using (TcpClient client = listener.AcceptTcpClient())
                {
                    using (NetworkStream stream = client.GetStream())
                    {
                        do
                        {
                            int bytes = stream.Read(data, 0, data.Length);
                            memStream.Write(data, 0, bytes);
                        } while (stream.DataAvailable);
                    }
                }

                var frmt = new BinaryFormatter();
                memStream.Flush();
                memStream.Position = 0;
                hbRespData = (ResponseData)frmt.Deserialize(memStream); // здесь вылетает ошибка "End of Stream encountered before parsing was completed"
            }

                if (resp.NeedSettingsUpdate)
                {
                    _settingsProvider.UpdateSettings();
                }
            }
            catch (Exception exc)
            {

            }
        }

== Сервер ==

Создание TcpListener'a

listener = TcpListener.Create(_settings.AgentHeartbeatListenPort);
 listener.Start();
 while (_heartbeatsListenerThrLive)
 {
try
                {
                    if (listener.Pending())
                    {
                        TcpClient client = listener.AcceptTcpClient();
                        ThreadPool.QueueUserWorkItem(RequestsHandler, client);
                    }
                }
                catch (Exception ex)
                {
                    continue; // --> while (_heartbeatsListenerThrLive)
                }

                Thread.Sleep(100);
            }
}

Обработка запроса от клиента

private void RequestsHandler(object objClient)
        {
            IPAddress clientIpAddress = IPAddress.None;
            HeartbeatData hbdata = null;

            using (TcpClient client = objClient as TcpClient)
            {
                clientIpAddress = ((IPEndPoint)client.Client.RemoteEndPoint).Address.MapToIPv4();

                try
                {
                    hbdata = ReadHeartbeatData(client);
                }
                catch (Exception exc)
                {
                    _log.Error("Unable to read client heartbeat data.", exc);
                }
                if (!hbdata.SettingsHash.Equals(commonHash, StringComparison.InvariantCultureIgnoreCase))
                {
                     try
                    {
                        SendHeartbeatResponse(clientIpAddress, ClientHeartbeatResponsePort, response);
                    }
                    catch (Exception exc) { }

SendHeartbeatResponse

private void SendHeartbeatResponse(IPAddress ipAddress, int portNumber, ResponseData response)
        {
            lock (lockMarker)
            {
                using (TcpClient tcpClient = new TcpClient())
                {
                    tcpClient.Connect(ipAddress, portNumber);
                    using (NetworkStream stream = tcpClient.GetStream())
                    {
                        using (var binStream = new MemoryStream())
                        {
                            var frmt = new BinaryFormatter();
                            frmt.Serialize(binStream, response);
                            binStream.Flush();
                            binStream.Position = 0;
                            binStream.CopyTo(stream);
                        }

                        tcpClient.Close();
                    }
                }
            }
        }

Comments

[freeExec]

Очень похоже, что вам не приходят все данные. Поэкспериментируйте на тексте, например перегоните всё в json.

[NameOf Var]

freeExec, в чем может быть причина того, что не приходят все данные? Есть ли у TCP ограничение на количество передаваемых байт? Либо может быть файрвол как-то ограничивает?

Answer 1

[Александр Юдаков]

При передаче данных по сети иногда бывают паузы.

Ваш код ждет первой такой паузы:

while (stream.DataAvailable)

...и, не дожидаясь остальных данных, завершает чтение.

Читать надо не пока DataAvailable, а пока не придет нужное количество байт. Для этого длину сообщения нужно сообщить клиенту перед отправкой самого сообщения.

Comments

[NameOf Var]

а каким образом нужно это сообщить?

[NameOf Var]

Скорее всего причина ошибок из-за большого объема передаваемых данных, т.к. этот код работает при отрпавке малого количества данных

[Александр Юдаков]

NameOf Var, тем же образом, которым вы отправляете тело сообщения.

Т.е сначала сервер отправляет 32-битное целое (4 байта), где указана длина тела сообщения, а потом уже само тело.

Клиент сначала читает 4 байта - длину тела сообщения, а потом - читает указанное число байт - получает тело.

[Александр Юдаков]

NameOf Var, на маленьких объемах данных код работает не потому, что он правильно написан, а потому что вам просто везет и пауз при передаче данных не возникает. На каком-нибудь мобильном канале связи паузы будут и на маленьких объемах.