Skip to content

Commit 04fa8cb

Browse files
rojiroji
committed
Provide details on operation in progress
When a connection is already use, an InvalidOperationException was thrown, but many users were having a hard time tracking down concurrency issues causing the error in their program. We now throw a new NpgsqlOperationInProgressException which provides information about the command already in progress. Closes #1248
1 parent 366b4bf commit 04fa8cb

File tree

10 files changed

+104
-29
lines changed

10 files changed

+104
-29
lines changed

‎src/Npgsql/GeneratedAsync.cs‎

Lines changed: 27 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -80,6 +80,7 @@
8080
using JetBrains.Annotations;
8181
using Npgsql.BackendMessages;
8282
using Npgsql.Logging;
83+
using Npgsql.PostgresTypes;
8384
using Npgsql.Schema;
8485
using Npgsql.TypeHandlers;
8586
using NpgsqlTypes;
@@ -160,6 +161,7 @@
160161
using System.Threading;
161162
using System.Threading.Tasks;
162163
using JetBrains.Annotations;
164+
using Npgsql.PostgresTypes;
163165
using Npgsql.TypeHandlers;
164166
using System.Threading;
165167
using System.Threading.Tasks;
@@ -185,6 +187,8 @@
185187
using System.IO;
186188
using System.Runtime.InteropServices;
187189
using System.Text;
190+
using System.Threading;
191+
using System.Threading.Tasks;
188192
using JetBrains.Annotations;
189193
using System.Threading;
190194
using System.Threading.Tasks;
@@ -265,14 +269,14 @@ async Task SendAsync(PopulateMethod populateMethod, CancellationToken cancellati
265269
{
266270
var directBuf = new DirectBuffer();
267271
var completed = populateMethod(ref directBuf);
268-
await _connector.SendBufferAsync(cancellationToken).ConfigureAwait(false);
272+
await _connector.SendBufferAsync(cancellationToken);
269273
if (completed)
270274
break; // Sent all messages
271275
// The following is an optimization hack for writing large byte arrays without passing
272276
// through our buffer
273277
if (directBuf.Buffer != null)
274278
{
275-
await _connector.WriteBuffer.DirectWriteAsync(directBuf.Buffer, directBuf.Offset, directBuf.Size == 0 ? directBuf.Buffer.Length : directBuf.Size, cancellationToken).ConfigureAwait(false);
279+
await _connector.WriteBuffer.DirectWriteAsync(directBuf.Buffer, directBuf.Offset, directBuf.Size == 0 ? directBuf.Buffer.Length : directBuf.Size, cancellationToken);
276280
directBuf.Buffer = null;
277281
directBuf.Size = 0;
278282
}
@@ -282,9 +286,24 @@ async Task SendAsync(PopulateMethod populateMethod, CancellationToken cancellati
282286
// We've send all the messages for the first statement in a multistatement command.
283287
// If we continue blocking writes for the rest of the messages, we risk a deadlock where
284288
// PostgreSQL sends large results for the first statement, while we're sending large
285-
// parameter data for the second. To avoid this, switch to async sends.
286-
// See #641
287-
RemainingSendTask = SendRemaining(populateMethod, CancellationToken.None);
289+
// parameter data for the second. To avoid this, switch to async sends. See #641
290+
// When performing async sends here, our regular async code will do ConfigureAwait(false),
291+
// sending continuations to the thread pool. However, this is a synchronous operation -
292+
// so a deadlock may occur where TP threads synchronously block on database input which won't
293+
// become because async continuations can't run (TP is starved).
294+
// To work around this, we send all async continuations to a special synchronization context
295+
// which executes them on a special thread.
296+
var callerSyncContext = SynchronizationContext.Current;
297+
SynchronizationContext.SetSynchronizationContext(SingleThreadSynchronizationContext);
298+
try
299+
{
300+
RemainingSendTask = SendRemaining(populateMethod, CancellationToken.None);
301+
}
302+
finally
303+
{
304+
SynchronizationContext.SetSynchronizationContext(callerSyncContext);
305+
}
306+
288307
return;
289308
}
290309
}
@@ -293,7 +312,7 @@ async Task SendAsync(PopulateMethod populateMethod, CancellationToken cancellati
293312
async Task<int> ExecuteNonQueryInternalAsync(CancellationToken cancellationToken)
294313
{
295314
var connector = CheckReadyAndGetConnector();
296-
using (connector.StartUserAction())
315+
using (connector.StartUserAction(this))
297316
{
298317
Log.Trace("ExecuteNonQuery", connector.Id);
299318
// Optimization: unprepared unparameterized non-queries can go through the PostgreSQL
@@ -315,7 +334,7 @@ async Task<int> ExecuteNonQueryInternalAsync(CancellationToken cancellationToken
315334
async Task<object> ExecuteScalarInternalAsync(CancellationToken cancellationToken)
316335
{
317336
var connector = CheckReadyAndGetConnector();
318-
using (connector.StartUserAction())
337+
using (connector.StartUserAction(this))
319338
{
320339
Log.Trace("ExecuteNonScalar", connector.Id);
321340
using (var reader = Execute(CommandBehavior.SequentialAccess | CommandBehavior.SingleRow))
@@ -326,7 +345,7 @@ async Task<object> ExecuteScalarInternalAsync(CancellationToken cancellationToke
326345
async Task<NpgsqlDataReader> ExecuteDbDataReaderInternalAsync(CommandBehavior behavior, CancellationToken cancellationToken)
327346
{
328347
var connector = CheckReadyAndGetConnector();
329-
connector.StartUserAction();
348+
connector.StartUserAction(this);
330349
try
331350
{
332351
Log.Trace("ExecuteReader", connector.Id);

‎src/Npgsql/Npgsql.csproj‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -72,6 +72,7 @@
7272
<Compile Include="BackendMessages\ParseCompleteMessage.cs" />
7373
<Compile Include="BackendMessages\RowDescriptionMessage.cs" />
7474
<Compile Include="BackendMessages\DataRowSequentialMessage.cs" />
75+
<Compile Include="NpgsqlOperationInProgressException.cs" />
7576
<Compile Include="PostgresTypes\PostgresArrayType.cs" />
7677
<Compile Include="PostgresTypes\PostgresType.cs" />
7778
<Compile Include="FrontendMessages\CopyFailMessage.cs" />

‎src/Npgsql/NpgsqlCommand.cs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1085,7 +1085,7 @@ public override async Task<int> ExecuteNonQueryAsync(CancellationToken cancellat
10851085
int ExecuteNonQueryInternal()
10861086
{
10871087
var connector = CheckReadyAndGetConnector();
1088-
using (connector.StartUserAction())
1088+
using (connector.StartUserAction(this))
10891089
{
10901090
Log.Trace("ExecuteNonQuery", connector.Id);
10911091

@@ -1185,7 +1185,7 @@ public override async Task<object> ExecuteScalarAsync(CancellationToken cancella
11851185
object ExecuteScalarInternal()
11861186
{
11871187
var connector = CheckReadyAndGetConnector();
1188-
using (connector.StartUserAction())
1188+
using (connector.StartUserAction(this))
11891189
{
11901190
Log.Trace("ExecuteNonScalar", connector.Id);
11911191
using (var reader = Execute(CommandBehavior.SequentialAccess | CommandBehavior.SingleRow))
@@ -1243,7 +1243,7 @@ protected override async Task<DbDataReader> ExecuteDbDataReaderAsync(CommandBeha
12431243
NpgsqlDataReader ExecuteDbDataReaderInternal(CommandBehavior behavior)
12441244
{
12451245
var connector = CheckReadyAndGetConnector();
1246-
connector.StartUserAction();
1246+
connector.StartUserAction(this);
12471247
try
12481248
{
12491249
Log.Trace("ExecuteReader", connector.Id);

‎src/Npgsql/NpgsqlConnector.cs‎

Lines changed: 23 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -189,6 +189,7 @@ int ReceiveTimeout
189189
/// A lock that's taken while a connection keepalive is in progress. Used to make sure
190190
/// keepalives and user actions don't interfere with one another.
191191
/// </summary>
192+
[CanBeNull]
192193
readonly SemaphoreSlim _keepAliveLock;
193194

194195
/// <summary>
@@ -198,9 +199,15 @@ int ReceiveTimeout
198199
/// </summary>
199200
internal object CancelLock { get; }
200201

201-
readonly UserAction _userAction;
202202
readonly Timer _keepAliveTimer;
203203

204+
/// <summary>
205+
/// The command currently being executed by the connector, null otherwise.
206+
/// Used only for concurrent use error reporting purposes.
207+
/// </summary>
208+
[CanBeNull]
209+
NpgsqlCommand _currentCommand;
210+
204211
internal HashSet<string> PreparedStatements { get; } = new HashSet<string>();
205212

206213
/// <summary>
@@ -278,7 +285,6 @@ internal NpgsqlConnector(NpgsqlConnection connection)
278285
_preparedStatementIndex = 0;
279286

280287
_userLock = new SemaphoreSlim(1, 1);
281-
_userAction = new UserAction(this);
282288
CancelLock = new object();
283289

284290
if (IsKeepAliveEnabled) {
@@ -1435,6 +1441,7 @@ void Cleanup()
14351441
Connection = null;
14361442
BackendParams.Clear();
14371443
ServerVersion = null;
1444+
_currentCommand = null;
14381445

14391446
// Disposing SemaphoreSlim leaves CurrentCount at 0, so increment back to 1 if needed
14401447
if (_userLock.CurrentCount == 0)
@@ -1588,10 +1595,17 @@ void ClosePreparedStatements()
15881595

15891596
#region Locking
15901597

1591-
internal IDisposable StartUserAction(ConnectorState newState=ConnectorState.Executing)
1598+
internal UserAction StartUserAction(NpgsqlCommand command)
1599+
=> StartUserAction(ConnectorState.Executing, command);
1600+
1601+
internal UserAction StartUserAction(ConnectorState newState=ConnectorState.Executing, NpgsqlCommand command=null)
15921602
{
15931603
if (!_userLock.Wait(0))
1594-
throw new InvalidOperationException("An operation is already in progress.");
1604+
{
1605+
throw _currentCommand == null
1606+
? new NpgsqlOperationInProgressException(State)
1607+
: new NpgsqlOperationInProgressException(_currentCommand);
1608+
}
15951609

15961610
// We now have the user lock, no user operation may be in progress but a keepalive might be
15971611

@@ -1630,7 +1644,8 @@ internal IDisposable StartUserAction(ConnectorState newState=ConnectorState.Exec
16301644
Debug.Assert(IsReady);
16311645
Log.Trace("Start user action", Id);
16321646
State = newState;
1633-
return _userAction;
1647+
_currentCommand = command;
1648+
return new UserAction(this);
16341649
}
16351650

16361651
internal void EndUserAction()
@@ -1654,17 +1669,17 @@ internal void EndUserAction()
16541669

16551670
Log.Trace("End user action", Id);
16561671
State = ConnectorState.Ready;
1672+
_currentCommand = null;
16571673
_keepAliveLock?.Release();
16581674
_userLock.Release();
16591675
}
16601676

16611677
bool IsInUserAction => _userLock.CurrentCount == 0;
16621678

16631679
/// <summary>
1664-
/// An IDisposable wrapper around <see cref="StartUserAction"/> and
1665-
/// <see cref="EndUserAction"/>.
1680+
/// An IDisposable wrapper around <see cref="EndUserAction"/>.
16661681
/// </summary>
1667-
class UserAction : IDisposable
1682+
internal struct UserAction : IDisposable
16681683
{
16691684
readonly NpgsqlConnector _connector;
16701685

‎src/Npgsql/NpgsqlOperationInProgressException.cs‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
using System;
2+
3+
namespace Npgsql
4+
{
5+
/// <summary>
6+
/// Thrown when trying to use a connection that is already busy performing some other operation.
7+
/// Provides information on the already-executing operation to help with debugging.
8+
/// </summary>
9+
public sealed class NpgsqlOperationInProgressException : InvalidOperationException
10+
{
11+
internal NpgsqlOperationInProgressException(NpgsqlCommand command)
12+
: base("A command is already in progress: " + command.CommandText)
13+
{
14+
CommandInProgress = command;
15+
}
16+
17+
internal NpgsqlOperationInProgressException(ConnectorState state)
18+
: base($"The connection is already in state '{state}'")
19+
{
20+
}
21+
22+
/// <summary>
23+
/// If the connection is busy with another command, this will contain a reference to that command.
24+
/// Otherwise, if the connection if busy with another type of operation (e.g. COPY), contains null.
25+
/// </summary>
26+
public NpgsqlCommand CommandInProgress { get; }
27+
}
28+
}

‎test/Npgsql.Tests/CommandTests.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -577,7 +577,7 @@ public void DisposeCommandDoesNotCloseReader()
577577
cmd.ExecuteReader();
578578
cmd.Dispose();
579579
cmd = new NpgsqlCommand("SELECT 3", conn);
580-
Assert.That(() => cmd.ExecuteScalar(), Throws.Exception.TypeOf<InvalidOperationException>());
580+
Assert.That(() => cmd.ExecuteScalar(), Throws.Exception.TypeOf<NpgsqlOperationInProgressException>());
581581
}
582582
}
583583

‎test/Npgsql.Tests/ConnectionTests.cs‎

Lines changed: 15 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -841,9 +841,21 @@ CREATE OR REPLACE FUNCTION pg_temp.emit_notice() RETURNS VOID AS
841841
public void ConcurrentUse()
842842
{
843843
using (var conn = OpenConnection())
844-
using (var cmd = new NpgsqlCommand("SELECT 1", conn))
845-
using (cmd.ExecuteReader())
846-
Assert.That(() => conn.ExecuteScalar("SELECT 1"), Throws.Exception.TypeOf<InvalidOperationException>());
844+
{
845+
using (var cmd = new NpgsqlCommand("SELECT 1", conn))
846+
using (cmd.ExecuteReader())
847+
Assert.That(() => conn.ExecuteScalar("SELECT 2"),
848+
Throws.Exception.TypeOf<NpgsqlOperationInProgressException>()
849+
.With.Property(nameof(NpgsqlOperationInProgressException.CommandInProgress)).SameAs(cmd));
850+
851+
conn.ExecuteNonQuery("CREATE TEMP TABLE foo (bar INT)");
852+
using (conn.BeginBinaryImport("COPY foo (bar) FROM STDIN BINARY"))
853+
{
854+
Assert.That(() => conn.ExecuteScalar("SELECT 2"),
855+
Throws.Exception.TypeOf<NpgsqlOperationInProgressException>()
856+
.With.Message.Contains("Copy"));
857+
}
858+
}
847859
}
848860

849861
[Test]

‎test/Npgsql.Tests/CopyTests.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -642,7 +642,7 @@ void StateAssertions(NpgsqlConnection conn)
642642
Assert.That(conn.Connector.State, Is.EqualTo(ConnectorState.Copy));
643643
Assert.That(conn.State, Is.EqualTo(ConnectionState.Open));
644644
Assert.That(conn.FullState, Is.EqualTo(ConnectionState.Open | ConnectionState.Fetching));
645-
Assert.That(() => conn.ExecuteScalar("SELECT 1"), Throws.Exception.TypeOf<InvalidOperationException>());
645+
Assert.That(() => conn.ExecuteScalar("SELECT 1"), Throws.Exception.TypeOf<NpgsqlOperationInProgressException>());
646646
}
647647

648648
#endregion

‎test/Npgsql.Tests/PrepareTests.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -142,7 +142,7 @@ public void DisposeWithOpenReader()
142142
cmd2.Prepare();
143143
var reader = cmd2.ExecuteReader();
144144
reader.Read();
145-
Assert.That(() => cmd1.Dispose(), Throws.Exception.TypeOf<InvalidOperationException>());
145+
Assert.That(() => cmd1.Dispose(), Throws.Exception.TypeOf<NpgsqlOperationInProgressException>());
146146
reader.Close();
147147
cmd1.Dispose();
148148
cmd2.Dispose();

‎test/Npgsql.Tests/ReaderTests.cs‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -682,13 +682,13 @@ public void ReaderIsStillOpen()
682682
using (var cmd1 = new NpgsqlCommand("SELECT 1", conn))
683683
using (var reader1 = cmd1.ExecuteReader())
684684
{
685-
Assert.That(() => conn.ExecuteNonQuery("SELECT 1"), Throws.Exception.TypeOf<InvalidOperationException>());
686-
Assert.That(() => conn.ExecuteScalar("SELECT 1"), Throws.Exception.TypeOf<InvalidOperationException>());
685+
Assert.That(() => conn.ExecuteNonQuery("SELECT 1"), Throws.Exception.TypeOf<NpgsqlOperationInProgressException>());
686+
Assert.That(() => conn.ExecuteScalar("SELECT 1"), Throws.Exception.TypeOf<NpgsqlOperationInProgressException>());
687687

688688
using (var cmd2 = new NpgsqlCommand("SELECT 2", conn))
689689
{
690-
Assert.That(() => cmd2.ExecuteReader(), Throws.Exception.TypeOf<InvalidOperationException>());
691-
Assert.That(() => cmd2.Prepare(), Throws.Exception.TypeOf<InvalidOperationException>());
690+
Assert.That(() => cmd2.ExecuteReader(), Throws.Exception.TypeOf<NpgsqlOperationInProgressException>());
691+
Assert.That(() => cmd2.Prepare(), Throws.Exception.TypeOf<NpgsqlOperationInProgressException>());
692692
}
693693
}
694694
}

0 commit comments

Comments
 (0)