Npgsql 在异步环境中无限期锁定线程

Question

我正在努力寻找我在使用 C# 流行的 Postgres 库 Npgsql 时遇到的问题的答案。我不确定这是否完全是 Npgsql 的问题，尽管我怀疑这是因为我的代码非常简单。我看到的问题是：当我有时（并非总是）在 Npgsql 上调用异步方法时，线程会被锁定。据我所知，这个问题是随机发生的。结果是，我在 Microsoft Orleans 环境中运行（这可能与找到解决方案有关），线程无限期锁定，从而使 Orleans 的工作线程之一无法处理工作。当我进行更多的 Npgsql 调用时，这些锁定的线程会堆积起来，最终 Orleans 系统因线程耗尽而阻塞。

所以我真的不知道问题可能是什么，但是因为锁定总是以相同的方法发生，并且因为它似乎发生在 Npgsql 的某些子例程中，所以我认为进一步研究 Npgsql 是公平的。

这是我在奥尔良存储提供程序中使用的代码（处理系统持久层的特殊类。）

var sql = $"SELECT * FROM objects WHERE id = @id";

using (var connection = new NpgsqlConnection(connectionString))
using (var cmd = new NpgsqlCommand(sql, connection))
{
    try
    {
        await connection.OpenAsync();

        cmd.Parameters.AddWithValue("id", id);

        using(var reader = await connection.ExecuteReaderAsync(cmd))
        {
            if (reader.HasRows)
            {
                var objects = await ProtobufSQL.DataReaderToType(modelType, reader);
                var data = objects[0];
                state.Data = data;
            }
    }
    catch (Exception e)
    {
        Log.Error(1, e.Message, e);
    }
}

这是 ProtobufSQL 类的源代码：

public class ProtobufSQL
{
    public static List<Tuple<string, object>> FlattenToSQLColumns(IMessage message, MessageDescriptor descriptor, string prefix = null)
    {
        var fields = descriptor.Fields.InDeclarationOrder();

        var columns = new List<Tuple<string, object>>();

        for (var i = 0; i < fields.Count; i++)
        {
            var field = fields[i];
            var columnName = field.Name.ToLower();

            if (field.Name == "id")
            {
                ByteString bytes = (ByteString)field.Accessor.GetValue(message);
                var uuid = new Guid(bytes.ToByteArray());
                columns.Add(new Tuple<string, object>("id", uuid));
            }
            else if (field.FieldType == FieldType.Message)
            {
                var embeddedDescriptor = field.MessageType;
                var embeddedMessage = field.Accessor.GetValue(message);
                if (field.IsRepeated)
                {
                    throw new Exception("Repeated complex types are not supported, create a foreign key reference in a new object instead.");
                }
                else
                {
                    columns.AddRange(FlattenToSQLColumns((IMessage)embeddedMessage, embeddedDescriptor, $"{columnName}."));
                }
            }
            else if (field.FieldType == FieldType.Group)
            {
                throw new Exception("Groups are not supported by ProtobufSQL.");
            }
            else
            {
                var columnValue = field.Accessor.GetValue(message);
                var key = prefix + columnName;
                if (field.IsRepeated)
                {
                    var enumerableColumnValue = columnValue as IEnumerable;
                    Type listTypeOf = enumerableColumnValue.GetType().GetGenericArguments()[0];
                    Type listType = typeof(List<>).MakeGenericType(listTypeOf);
                    dynamic valueList = Activator.CreateInstance(listType);
                    foreach (var item in enumerableColumnValue)
                    {
                        valueList.Add((dynamic)item);
                    }
                    columns.Add(new Tuple<string, object>(key, valueList.ToArray()));
                }
                else
                {
                    columns.Add(new Tuple<string, object>(key, columnValue));
                }
            }
        }

        return columns;
    }

    public static async Task<IMessage[]> DataReaderToType(Type type, DbDataReader reader)
    {
        var descriptor = (MessageDescriptor)type.GetProperty("Descriptor").GetValue(null);

        IList<IMessage> objects = new List<IMessage>();

        while (await reader.ReadAsync())
        {
            var obj = Activator.CreateInstance(type);
            TraverseDbRow(reader, descriptor, obj);
            objects.Add((IMessage)obj);
        }

        return objects.ToArray();
    }

    private static void TraverseDbRow(DbDataReader reader, MessageDescriptor descriptor, object obj, string prefix = null)
    {
        var fields = descriptor.Fields.InFieldNumberOrder();

        for (var i = 0; i < fields.Count; i++)
        {
            var field = fields[i];
            if (field.FieldType == FieldType.Message)
            {
                if (field.IsRepeated)
                {
                    // Repeated embedded types will be broken out into a separate table,
                    // so there's no need to handle them here.
                }
                else if (field.IsMap)
                {
                    throw new Exception("Maps are not yet supported by ProtobufSQL.");
                }
                else
                {
                    var embeddedDescriptor = field.MessageType;
                    var embeddedObj = Activator.CreateInstance(embeddedDescriptor.ClrType);
                    TraverseDbRow(reader, embeddedDescriptor, embeddedObj, $"{prefix}{field.Name}.");
                }
            }
            else if (field.FieldType == FieldType.Group)
            {
                throw new Exception("Groups are not supported by ProtobufSQL.");
            }
            else
            {
                var columnName = prefix + field.Name;
                try
                {
                    var columnValue = reader[columnName];
                    var propertyInfo = obj.GetType().GetProperty(field.Name, BindingFlags.IgnoreCase | BindingFlags.Public | BindingFlags.Instance);

                    if (field.Name == "id")
                    {
                        var guid = (Guid)columnValue;
                        ByteString bytes = ByteString.CopyFrom(guid.ToByteArray());
                        propertyInfo.SetValue(obj, bytes);
                    }
                    else if (field.IsRepeated)
                    {
                        var repeated = propertyInfo.GetValue(obj);
                        var addRange = repeated.GetType().GetMethod("AddRange");
                        addRange.Invoke(repeated, new object[] { columnValue });
                    }
                    else if (field.IsMap)
                    {
                        throw new Exception("Maps are not yet supported by ProtobufSQL.");
                    }
                    else
                    {
                        propertyInfo.SetValue(obj, Convert.ChangeType(columnValue, propertyInfo.PropertyType));
                    }
                }
                catch (IndexOutOfRangeException e)
                {
                    // columnName was not present in the response
                }
            }
        }
    }
}

我还有线程堆栈被锁定时的屏幕截图：

我真的不知道这一切是怎么回事。希望有人拥有一些知识可以帮助我继续前进！谢谢。

Answer 1

var sql = $"SELECT * FROM objects WHERE id = @id";

using var connection = new NpgsqlConnection(connectionString);
using var cmd = new NpgsqlCommand(sql, connection);
try
{
    await connection.OpenAsync();

    cmd.Parameters.AddWithValue("id", id);

    using var reader = await connection.ExecuteReaderAsync(cmd);
        
    if (reader.HasRows)
        {
            var objects = await ProtobufSQL.DataReaderToType(modelType, reader);
            var data = objects[0];
            state.Data = data;
        }            
    }
    catch (Exception e)
    {
        Log.Error(1, e.Message, e);
    }
    finally
    {
        await connection.CloseAsync();
    }
}

也许你应该关闭连接，这意味着它（就像到数据库的网络连接）没有被关闭，而是返回到池中，以加快连接过程，这在tcp中通常是昂贵的。这样，您将停止溢出数据库的连接池，并且可能会停止尝试新连接来淹没数据库