Mogę odtworzyć ten problem w dość małym programie w języku C #, który wykorzystuje połączenia asynchroniczne, ale nie jestem w 100% pewien, dlaczego. Oto mój program repro, na wypadek gdyby inni chcieli go wypróbować - podejrzewam, że istnieje szereg rzeczy, które muszą być idealnie dopasowane, aby tak się stało:
- Pula połączeń włączona
- Używanie personifikacji w puli połączeń, uniemożliwianie cofania tego kontekstu personifikacji w puli połączeń
void Main()
{
var impersonateMyself = System.Security.Principal.WindowsIdentity.GetCurrent().Name;
var testCommand = "SELECT TOP 1 * FROM sys.objects";
var calls = Enumerable.Repeat(
$@"{testCommand};",
10
);
var impersonatedCalls = Enumerable.Repeat(
$@"EXECUTE AS LOGIN = '{impersonateMyself} WITH NO REVERT'; {testCommand}; REVERT;",
10
);
Dictionary<string, object> dict = new Dictionary<string, object>()
{
};
// Scenario 1: Impersonated Calls, With connection pooling -- will randomly fail
Parallel.ForEach(
impersonatedCalls,
c => new SqlAsync("Data Source=devsql2;Initial Catalog=Test;Integrated Security=true;Max Pool Size=2;").AsyncSqlCall<List<A>>(c, CommandType.Text, handleResultAsync, dict).Dump());
Parallel.ForEach(
impersonatedCalls,
c => new SqlSync("Data Source=devsql2;Initial Catalog=Test;Integrated Security=true;Max Pool Size=2;").SyncSqlCall<List<A>>(c, CommandType.Text, handleResultSync, dict).Dump());
// Scenario 2: Normal calls, with connection pooling -- should succeed every time
Parallel.ForEach(
calls,
c => new SqlAsync("Data Source=devsql2;Initial Catalog=Test;Integrated Security=true;Max Pool Size=2;").AsyncSqlCall<List<A>>(c, CommandType.Text, handleResultAsync, dict).Dump());
Parallel.ForEach(
calls,
c => new SqlSync("Data Source=devsql2;Initial Catalog=Test;Integrated Security=true;Max Pool Size=2;").SyncSqlCall<List<A>>(c, CommandType.Text, handleResultSync, dict).Dump());
// Scenario 3: Impersonated Calls, WITHOUT connection pooling -- should succeed every time
Parallel.ForEach(
impersonatedCalls,
c => new SqlAsync("Data Source=devsql2;Initial Catalog=Test;Integrated Security=true;Max Pool Size=200;").AsyncSqlCall<List<A>>(c, CommandType.Text, handleResultAsync, dict).Dump());
Parallel.ForEach(
impersonatedCalls,
c => new SqlSync("Data Source=devsql2;Initial Catalog=Test;Integrated Security=true;Max Pool Size=200;").SyncSqlCall<List<A>>(c, CommandType.Text, handleResultSync, dict).Dump());
}
public class SqlSync
{
private readonly string _connectionString;
public int Timeout {get; set;}
public SqlSync(string connString)
{
_connectionString = connString;
Timeout = 30;
}
public T SyncSqlCall<T>(string commandText, CommandType type, Func<SqlDataReader, T> handleResult, Dictionary<string, object> parameters = null)
{
using (SqlConnection conn = new SqlConnection(_connectionString))
using (SqlCommand cmd = new SqlCommand(commandText, conn))
{
cmd.CommandTimeout = Timeout;
cmd.CommandType = CommandType.Text;
if (parameters != null)
{
foreach (KeyValuePair<string, object> kvp in parameters)
cmd.Parameters.AddWithValue(kvp.Key, kvp.Value ?? DBNull.Value);
}
conn.Open();
using (var rdr = cmd.ExecuteReader(CommandBehavior.CloseConnection))
return handleResult(rdr);
}
}
}
public class SqlAsync
{
private readonly string _connectionString;
public int Timeout { get; set; }
public SqlAsync(string connString)
{
_connectionString = connString;
Timeout = 30;
}
public Task<T> AsyncSqlCall<T>(string sp, CommandType commandType, Func<SqlDataReader, Task<T>> handleResult, Dictionary<string, object> parameters = null)
{
return AsyncSqlCall<T>(sp, commandType, (reader, token) => handleResult(reader), CancellationToken.None, parameters);
}
public async Task<T> AsyncSqlCall<T>(string commandText, CommandType type, Func<SqlDataReader, CancellationToken, Task<T>> handleResult, CancellationToken cancellationToken, Dictionary<string, object> parameters = null)
{
using (SqlConnection conn = new SqlConnection(_connectionString))
using (SqlCommand cmd = new SqlCommand(commandText, conn))
{
cmd.CommandTimeout = Timeout;
cmd.CommandType = CommandType.Text;
if (parameters != null)
{
foreach (KeyValuePair<string, object> kvp in parameters)
cmd.Parameters.AddWithValue(kvp.Key, kvp.Value ?? DBNull.Value);
}
await conn.OpenAsync(cancellationToken);
// if (conn.State != ConnectionState.Open)
// await Task.Delay(TimeSpan.FromMilliseconds(10));
using (var rdr = await cmd.ExecuteReaderAsync(CommandBehavior.CloseConnection, cancellationToken))
return await handleResult(rdr, cancellationToken);
}
}
}
public class A
{
public string object_name { get; set; }
}
public static Func<SqlDataReader, Task<List<A>>> handleResultAsync = (SqlDataReader sdr) =>
{
var result = new List<A>();
while (sdr.Read())
{
result.Add(new A { object_name = sdr.GetFieldValue<string>(0) });
}
return Task.FromResult(result);
};
public static Func<SqlDataReader, List<A>> handleResultSync = (SqlDataReader sdr) =>
{
var result = new List<A>();
while (sdr.Read())
{
result.Add(new A { object_name = sdr.GetFieldValue<string>(0) });
}
return result;
};