Fatal error. System.AccessViolationException !

[weiwxg]

在使用C#驱动TDengine.Connector进行频繁读数据，经常会触发如下的异常，并导致容器异常重启

connect failed, reason: Unable to establish connection

Fatal error. System.AccessViolationException: Attempted to read or write protected memory. This is often an indication that other memory is corrupt.
Repeat 2 times:
--------------------------------
   at TDengineDriver.TDengine.Connect(System.String, System.String, System.String, System.String, Int16)
--------------------------------
   at TdEngine.Infrastructure.TDEngineHelper.GetConnection()
   at TdEngine.Infrastructure.TDEngineHelper.Query(System.String)
   at Program+<>c.<<Main>$>b__0_0()
   at System.Threading.Tasks.Task`1[[System.__Canon, System.Private.CoreLib, Version=6.0.0.0, Culture=neutral, PublicKeyToken=7cec85d7bea7798e]].InnerInvoke()
   at System.Threading.Tasks.Task+<>c.<.cctor>b__272_0(System.Object)
   at System.Threading.ExecutionContext.RunFromThreadPoolDispatchLoop(System.Threading.Thread, System.Threading.ExecutionContext, System.Threading.ContextCallback, System.Object)
   at System.Threading.Tasks.Task.ExecuteWithThreadLocal(System.Threading.Tasks.Task ByRef, System.Threading.Thread)
   at System.Threading.Tasks.Task.ExecuteEntryUnsafe(System.Threading.Thread)
   at System.Threading.Tasks.Task.ExecuteFromThreadPool(System.Threading.Thread)
   at System.Threading.ThreadPoolWorkQueue.Dispatch()
   at System.Threading.PortableThreadPool+WorkerThread.WorkerThreadStart()
   at System.Threading.Thread.StartCallback()

环境描述

• TDengine Client Version: 2.4.0.33
• TDengine Server Version: 2.4.0.33
• TDengine.Connector: Nuget 1.0.7
• 表数据量: 619465
• 执行SQL: select ts from tbname order by ts desc limit 10;
• 测试代码：如下

// 开启多个线程频繁读数据
Task.Run(() =>
{
    while (true)
    {
        TDEngineHelper.Query("select ts from tbname order by ts desc limit 10");
        Console.WriteLine($"{Environment.CurrentManagedThreadId}-{DateTime.Now}");
        Thread.Sleep(200);
    }
});

Task.Run(() =>
{
    while (true)
    {
        TDEngineHelper.Query("select ts from tbname order by ts desc limit 10");
        Console.WriteLine($"{Environment.CurrentManagedThreadId}-{DateTime.Now}");
        Thread.Sleep(200);
    }
});

// 基于Github官方Demo封装的TDengine辅助类
public static class TDEngineHelper
{
    public static IntPtr GetConnection()
    {
        string host = "******";
        short port = 6030;
        string username = "******";
        string password = "******";
        string dbname = "******";
        var conn = TDengine.Connect(host, username, password, dbname, port);
        try
        {
            if (conn == IntPtr.Zero)
            {
                throw new Exception("Connect to TDengine failed");
            }
            return conn;
        }
        catch (Exception)
        {
            TDengine.Close(conn);
            TDengine.Cleanup();
            throw;
        }
    }

    public static DataTable Query(string sql)
    {
        var conn = GetConnection();

        try
        {
            IntPtr res = TDengine.Query(conn, sql);

            if (TDengine.ErrorNo(res) != 0)
            {
                TDengine.Close(conn);
                TDengine.Cleanup();
                throw new Exception("Failed to query since: " + TDengine.Error(res));
            }

            var dt = new DataTable();

            List<TDengineMeta> metas = TDengine.FetchFields(res);
            for (int i = 0; i < metas.Count; i++)
            {
                dt.Columns.Add(metas[i].name);
            }

            IntPtr row;
            while ((row = TDengine.FetchRows(res)) != IntPtr.Zero)
            {
                List<TDengineMeta> metaList = TDengine.FetchFields(res);
                int numOfFiled = TDengine.FieldCount(res);

                IntPtr colLengthPrt = TDengine.FetchLengths(res);
                int[] colLengthArr = new int[numOfFiled];
                Marshal.Copy(colLengthPrt, colLengthArr, 0, numOfFiled);

                var dr = dt.NewRow();

                for (int i = 0; i < numOfFiled; i++)
                {
                    TDengineMeta meta = metaList[i];
                    IntPtr data = Marshal.ReadIntPtr(row, IntPtr.Size * i);


                    if (data == IntPtr.Zero)
                    {
                        dr[i] = null;
                        continue;
                    }
                    dr[i] = (TDengineDataType)meta.type switch
                    {
                        TDengineDataType.TSDB_DATA_TYPE_BOOL => Marshal.ReadByte(data) == 0 ? false : true,
                        TDengineDataType.TSDB_DATA_TYPE_TINYINT => (sbyte)Marshal.ReadByte(data),
                        TDengineDataType.TSDB_DATA_TYPE_SMALLINT => Marshal.ReadInt16(data),
                        TDengineDataType.TSDB_DATA_TYPE_INT => Marshal.ReadInt32(data),
                        TDengineDataType.TSDB_DATA_TYPE_BIGINT => Marshal.ReadInt64(data),
                        TDengineDataType.TSDB_DATA_TYPE_FLOAT => (float)Marshal.PtrToStructure(data, typeof(float)),
                        TDengineDataType.TSDB_DATA_TYPE_DOUBLE => (double)Marshal.PtrToStructure(data, typeof(double)),
                        TDengineDataType.TSDB_DATA_TYPE_BINARY => Marshal.PtrToStringUTF8(data, colLengthArr[i]),
                        TDengineDataType.TSDB_DATA_TYPE_TIMESTAMP => Marshal.ReadInt64(data),
                        TDengineDataType.TSDB_DATA_TYPE_NCHAR => Marshal.PtrToStringUTF8(data, colLengthArr[i]),
                        TDengineDataType.TSDB_DATA_TYPE_UTINYINT => Marshal.ReadByte(data),
                        TDengineDataType.TSDB_DATA_TYPE_USMALLINT => (ushort)Marshal.ReadInt16(data),
                        TDengineDataType.TSDB_DATA_TYPE_UINT => (uint)Marshal.ReadInt32(data),
                        TDengineDataType.TSDB_DATA_TYPE_UBIGINT => (ulong)Marshal.ReadInt64(data),
                        TDengineDataType.TSDB_DATA_TYPE_JSONTAG => Marshal.PtrToStringUTF8(data, colLengthArr[i]),
                        _ => throw new Exception("nonsupport data type value"),
                    };


                }

                dt.Rows.Add(dr);
            }

            return dt;
        }
        catch (Exception)
        {
            TDengine.Close(conn);
            TDengine.Cleanup();
            throw;
        }
    }
}