2020年新年大赛！

Question

aepot

Asked:2022-08-20 21:13:28 +0000 UTC2022-08-20 21:13:28 +0000 UTC 2022-08-20 21:13:28 +0000 UTC

lock 不为 List<T> 提供线程安全？

772

我在整理这个问题时遇到了魔法，顺便说一下，我无法重现作者所写的问题。

我拿了我的 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，但我有一个列表，它在一个流中只是空的，而在另一个流中则没有。

在调试和发布版本中可重现，无论有无调试器。

1 个回答

Voted

aepot · Answer 1 · 2022-08-20T22:43:23Z

我想通了，问题突然出现在这一行

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();

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

也就是说，锁定、波动性和列表结果与它无关。问题出在 dotnet 本身的网络部分的细微差别上。ReadAsync如果它在客户端关闭连接并进入真正的等待之前启动，则不会出现错误。也就是说，Console.ReadKey()在客户端应用程序的末尾添加一些可以解决问题。哦，那些插座...

换句话说，不需要从构造函数中运行任何可怕的代码，即使它是异步的。我告诉大家这个，但我自己绊倒了。

lock 不为 List<T> 提供线程安全？

我看不懂措辞

请求的模块“del”不提供名为“default”的导出

"!+tab" 在 HTML 的 vs 代码中不起作用

我正在尝试解决“猜词”的问题。Python

可以使用哪些命令将当前指针移动到指定的提交而不更改工作目录中的文件？

Python解析野莓

问题：“警告：检查最新版本的 pip 时出错。”

帮助编写一个用值填充变量的循环。解决这个问题

尽管依赖数组为空，但在渲染上调用了 2 次 useEffect

数据不通过 Telegram.WebApp.sendData 发送

lock 不为​​ List<T> 提供线程安全？

1 个回答

相关问题

lock 不为 List<T> 提供线程安全？