1

Столкнулся с магией, пока разбирал вот этот вопрос, кстати, проблему, о которой пишет автор, я не смог воспроизвести.

Взял код своего TCP сервера. Привожу его тут полностью. Пытался создать минимальный воспроизводимый пример, но не смог воспроизвести проблему на нем.

class Program
{
    public static async Task Main(string[] args)
    {
        int port = 85;
        Console.WriteLine("Запуск сервера....");
        using (TcpServer server = new TcpServer(port))
        {
            Task servertask = server.ListenAsync();
            Console.WriteLine($"Слушаем клиента на порту: {port}");
            while (true)
            {
                string input = Console.ReadLine();
                if (input == "stop")
                {
                    Console.WriteLine("Остановка сервера...");
                    server.Stop();
                    break;
                }
            }
            await servertask;
        }
        Console.WriteLine("Нажмите любую клавишу для выхода...");
        Console.ReadKey(true);

    }
}

class TcpServer : IDisposable
{
    private readonly TcpListener _listener;
    private readonly List<Connection> _clients;
    private readonly object _lock = new object();
    bool disposed;

    public TcpServer(int port)
    {
        _listener = new TcpListener(IPAddress.Any, port);
        _clients = new List<Connection>();
    }

    public async Task ListenAsync()
    {
        try
        {
            _listener.Start();
            Console.WriteLine("Сервер стартовал на " + _listener.LocalEndpoint);
            while (true)
            {
                TcpClient client = await _listener.AcceptTcpClientAsync();
                Console.WriteLine("Подключение: " + client.Client.RemoteEndPoint + " > " + client.Client.LocalEndPoint);
                lock (_lock)
                {
                    _clients.Add(new Connection(client, RemoveClient));
                }
            }
        }
        catch (SocketException)
        {
            Console.WriteLine("Сервер остановлен.");
        }
    }

    private void RemoveClient(Connection client)
    {
        lock (_lock)
        {
            _clients.Remove(client);
            client.Dispose();
            Console.WriteLine(_clients.Count);
        }
    }

    public void Stop()
    {
        _listener.Stop();
    }

    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this);
    }

    private void Dispose(bool disposing)
    {
        if (disposed)
            throw new ObjectDisposedException(typeof(TcpServer).FullName);
        disposed = true;
        _listener.Stop();
        if (disposing)
        {
            lock (_lock)
            {
                Console.WriteLine(_clients.Count);
                if (_clients.Count > 0)
                {
                    Console.WriteLine("Отключаю клиентов...");
                    foreach (Connection client in _clients)
                    {
                        client.Dispose();
                    }
                    Console.WriteLine("Клиенты отключены.");
                }
            }
        }
    }

    ~TcpServer() => Dispose(false);
}

class Connection : IDisposable
{
    private readonly TcpClient _client;
    private readonly NetworkStream _stream;
    private readonly EndPoint _remoteEndPoint;
    private readonly Task _readingTask;
    private readonly Task _writingTask;
    private readonly Action<Connection> _disposeCallback;
    private readonly Channel<string> _channel;

    public Connection(TcpClient client, Action<Connection> disposeCallback)
    {
        _client = client;
        _stream = client.GetStream();
        _remoteEndPoint = client.Client.RemoteEndPoint;
        _disposeCallback = disposeCallback;
        _channel = Channel.CreateUnbounded<string>();
        _readingTask = RunReadingLoop();
        _writingTask = RunWritingLoop();
    }

    private async Task RunReadingLoop()
    {
        try
        {
            byte[] headerBuffer = new byte[4];
            while (true)
            {
                int bytesReceived = await _stream.ReadAsync(headerBuffer, 0, 4);
                if (bytesReceived != 4)
                    break;
                int length = BinaryPrimitives.ReadInt32LittleEndian(headerBuffer);
                byte[] buffer = new byte[length];
                int count = 0;
                while (count < length)
                {
                    bytesReceived = await _stream.ReadAsync(buffer, count, buffer.Length - count);
                    count += bytesReceived;
                }
                string message = Encoding.UTF8.GetString(buffer);
                Console.WriteLine($"<< {_remoteEndPoint}: {message}");
                await SendMessageAsync($"Echo: {message}");
            }
            Console.WriteLine($"Клиент {_remoteEndPoint} отключился.");
            _stream.Close();
        }
        catch (IOException ex)
        {
            Console.WriteLine($"Подключение к {_remoteEndPoint} закрыто сервером: {ex.Message}");
        }
        catch (Exception ex)
        {
            Console.WriteLine(ex.GetType().Name + ": " + ex.Message);
        }
        _disposeCallback(this);
    }

    public async Task SendMessageAsync(string message)
    {
        Console.WriteLine($">> {_remoteEndPoint}: {message}");
        await _channel.Writer.WriteAsync(message);
    }

    private async Task RunWritingLoop()
    {
        await foreach (string message in _channel.Reader.ReadAllAsync())
        {
            byte[] buffer = Encoding.UTF8.GetBytes(message);
            byte[] header = new byte[4];
            BinaryPrimitives.WriteInt32LittleEndian(header, buffer.Length);
            await _stream.WriteAsync(header, 0, header.Length);
            await _stream.WriteAsync(buffer, 0, buffer.Length);
        }
    }

    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this);
    }

    bool disposed;
    protected virtual void Dispose(bool disposing)
    {
        if (disposed)
            throw new ObjectDisposedException(typeof(Connection).FullName);
        disposed = true;
        if (_client.Connected)
        {
            _channel.Writer.Complete();
            _stream.Close();
            Task.WaitAll(_readingTask, _writingTask);
        }
        if (disposing)
        {
            _client.Dispose();
        }
    }

    ~Connection() => Dispose(false);
}

Запускаю сервер, все ок.

Затем беру нерабочий клиент, по факту он для этого сервера отправляет 0 как длину пакета и затем закрывает соединение.

class Program
{
    static void Main(string[] args)
    {
        var tcpPeer = new TcpPeer("127.0.0.1", 85);
        try
        {
            tcpPeer.Write(Encoding.Unicode.GetBytes("Hello world"));
        }
        catch (Exception e)
        {
            Console.WriteLine(e.ToString());
                
        }

    }
}

class TcpPeer : IPeer
{
    private TcpClient _client;
    private NetworkStream _stream;
    public TcpPeer(string ip, int port)
    {
        _client = new TcpClient(ip, port);
        _stream = _client.GetStream();
    }
    public byte[] Read()
    {
        byte[] data = new byte[8192];

        int len = _stream.Read(data, 0, data.Length);

        Array.Resize(ref data, len);
        return data;
    }
    public void Write(byte[] msg)
    {
        //и тут
        _stream.Write(msg, 0, msg.Length);
    }
}

interface IPeer
{
    byte[] Read();
    void Write(byte[] data);
}

Сервер корректно реагирует на закрытие соединения, затем я даю серверу команду stop и схватываю ObjectDisposedException, а всё потому что уже закрытое подключение пытаюсь закрыть еще раз. В коде сервера обратите внимание на 2 места, где Console.WriteLine(_clients.Count);.

Здесь все проблемы в классе TcpServer. Магия заключается в том, что в метод RemoveClient отрабатывает раньше, и _clients.Count содержит там 0. А вот метод Dispose, выполняемый значительно позднее, пусть это будет 10 секунд или больше, выдает для _clients.Count в консоль 1.

Вывод в консоль. До строчики "Сервер остановлен." вывод нормальный, а вот дальше - магия!

Запуск сервера....
Сервер стартовал на 0.0.0.0:85
Слушаем клиента на порту: 85
Подключение: 127.0.0.1:61468 > 127.0.0.1:85
Подключение к 127.0.0.1:61468 закрыто сервером: Unable to read data from the transport connection: An existing connection was forcibly closed by the remote host..
0
stop
Остановка сервера...
Сервер остановлен.
1
Отключаю клиентов...
Unhandled exception. System.ObjectDisposedException: Cannot access a disposed object.
Object name: 'ConsoleTcpServer.Connection'.
   at ConsoleTcpServer.Connection.Dispose(Boolean disposing) in C:\Source\ConsoleTcpServer\ConsoleTcpServer\Program.cs:line 206
   at ConsoleTcpServer.Connection.Dispose() in C:\Source\ConsoleTcpServer\ConsoleTcpServer\Program.cs:line 198
   at ConsoleTcpServer.TcpServer.Dispose(Boolean disposing) in C:\Source\ConsoleTcpServer\ConsoleTcpServer\Program.cs:line 110
   at ConsoleTcpServer.TcpServer.Dispose() in C:\Source\ConsoleTcpServer\ConsoleTcpServer\Program.cs:line 90
   at ConsoleTcpServer.Program.Main(String[] args) in C:\Source\ConsoleTcpServer\ConsoleTcpServer\Program.cs:line 31
   at ConsoleTcpServer.Program.<Main>(String[] args)

С чем это связано? Почему список в одном потоке поменялся, а в другом нет? Волатильность? Но как это объяснить? Складывается иллюзия, что поле ведет себя как [ThreadStatic], но в этом случае был бы null, а у меня список есть, просто в одном потоке он пустой, а в другом - нет.

Воспроизводится в Debug и Release сборке, с отладчиком и без.

5
  • Может это? 3 абзац. Или вот этот ответ
    – Aarnihauta
    20 июн 2022 в 13:24
  • Мне вот не нравится, что тут async поверх этого всего местами. Нутром чую, как-то тут слишком всего понамешано. Но что именно не так не могу понять.
    – CrazyElf
    20 июн 2022 в 13:30
  • @Aarnihauta меня интересует не столько то, как именно это полечить, способов много, включая так же те самые потокобезопасные коллекции, а то, в чем же именно причина?
    – aepot
    20 июн 2022 в 13:35
  • Ещё мне не нравится что _disposeCallback(this); лежит не в finally, например. Но может это и ничего.
    – CrazyElf
    20 июн 2022 в 13:35
  • @CrazyElf оно отрабатывает как надо, нет смысла пихать вызов сallback в finally, так как все исключения будут перехвачены и выведены в консоль до того как дело дойдет до выполнения этой строчки. В контексте данного вопроса к вызову колбэка вопросов нет, он вызывается корректно. Да, код сервера может кривоват, но вопрос сейчас не в этом, а в List<T> и почему он так себя ведет в многопоточке.
    – aepot
    20 июн 2022 в 13:39

1 ответ 1

4

Разобрался, проблема была внезапно вот в этой строчке

int bytesReceived = await _stream.ReadAsync(headerBuffer, 0, 4);

Дело в том что этот await не сработал, так как клиент закрыл соединение раньше, чем вернулся Task в вызывающий метод, то есть метод ReadAsync отработал синхронно. А исключение, то самое ObjectDisposedException возникло еще ДО того как строчка _clients.Add(...) отработала. Вследствие чего я пытался удалить клиента из коллекции еще до того как он был туда добавлен (_clients.Remove() конечно там возвращал false, но я это не проверял).

Решение проблемы - добавить await Task.Yield() до начала ожидания данных, то есть принудить машину состояний вернуть Task до того как метод начнет выполнение.

private async Task RunReadingLoop()
{
    await Task.Yield();

    // ... остальной код ...
}

То есть локи, волатильность и список, оказались не при чем. Проблема оказалась в нюансах работы сетевой части самого дотнета. Ошибки бы не было, если бы ReadAsync стартовал до того как клиент закрыл бы соединение и вошел в реальное ожидание. То есть добавление какого-нибудь Console.ReadKey() в конец клиентского приложения решило бы проблему. Ох уж эти сокеты...

Другими словами, нечего всякий страшный код запускать из конструктора, даже если он асинхронный. Всем это говорю, а сам споткнулся.

10
  • 1
    Последняя фраза больше похожа на крик души :)
    – Frehzy
    20 июн 2022 в 14:52
  • 1
    @Frehzy здесь скорее крик в том, что NetworkStream.ReadAsync какого-то черта оказался синхронный. Минимальный напишу и зарепорчу баг в Microsoft, а пока что-то там похожих багов найти не могу.
    – aepot
    20 июн 2022 в 14:57
  • 3
    "Дело в том что этот await не работает, то есть он синхронно вешает поток" - i.stack.imgur.com/C6gQK.jpg Почему-то имеет место тенденция воспринимать асинхронный код, как неблокирующий, хотя это принципиально неверно. А еще в вопрос все-таки стоило бы добавить минимальный воспроизводимый пример 20 июн 2022 в 15:40
  • 1
    @user7860670 чтобы его добавить, стоило сначала понимать, в чем проблема. хотя это принципиально неверно - сами Microsoft это заявляли и даже 2 года фиксили эти же проблемы в System.IO, и ведь пофиксили в 6-м дотнете. То есть они как бы к этому и идут. И даже сокет, который по своей природе синхронный, можно было бы обвешать нужным количеством костылей, чтобы публичная API работала как заявлено. А пока выглядит как тяп-ляп-в-продакшн. Впрочем, ничего нового.
    – aepot
    20 июн 2022 в 15:45
  • 1
    @user7860670 зря я наклеветал на .NET. Проблема была в том что соединение закрывалось до того как стартовал ReadAsync, поэтому он отрабатывал синхронно. Обновил ответ, вот теперь я точно разобрался. принципиально неверно - если вы про то что любой асинхронный метод может завершиться синхронно согласно своей внутренней какой-то логике, то да, это нормально, я просто сначала не так понял.
    – aepot
    20 июн 2022 в 17:10

Ваш ответ

Нажимая на кнопку «Отправить ответ», вы соглашаетесь с нашими условиями использования и подтверждаете, что прочитали и поняли наши политику конфиденциальности и нормы поведения.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.