From 95b0e84aebbab7e8aeed7dc9ee526d8e1c116c37 Mon Sep 17 00:00:00 2001 From: Wraith2 Date: Sat, 14 Sep 2019 21:35:48 +0100 Subject: [PATCH 1/4] rework TdsParserStateObject snapshots --- .../Microsoft/Data/SqlClient/SqlDataReader.cs | 20 +- .../SqlClient/SqlInternalConnectionTds.cs | 6 +- .../src/Microsoft/Data/SqlClient/TdsParser.cs | 44 +- .../Data/SqlClient/TdsParserSessionPool.cs | 2 +- .../Data/SqlClient/TdsParserStateObject.cs | 421 ++++++++++++------ 5 files changed, 315 insertions(+), 178 deletions(-) diff --git a/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/SqlDataReader.cs b/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/SqlDataReader.cs index a898c02155..4d263728ea 100644 --- a/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/SqlDataReader.cs +++ b/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/SqlDataReader.cs @@ -803,7 +803,7 @@ private bool TryCleanPartialRead() } #if DEBUG - if (_stateObj._pendingData) + if (_stateObj.HasPendingData) { byte token; if (!_stateObj.TryPeekByte(out token)) @@ -936,7 +936,7 @@ private bool TryCloseInternal(bool closeReader) try { - if ((!_isClosed) && (parser != null) && (stateObj != null) && (stateObj._pendingData)) + if ((!_isClosed) && (parser != null) && (stateObj != null) && (stateObj.HasPendingData)) { // It is possible for this to be called during connection close on a // broken connection, so check state first. @@ -1118,7 +1118,7 @@ private bool TryConsumeMetaData() { // warning: Don't check the MetaData property within this function // warning: as it will be a reentrant call - while (_parser != null && _stateObj != null && _stateObj._pendingData && !_metaDataConsumed) + while (_parser != null && _stateObj != null && _stateObj.HasPendingData && !_metaDataConsumed) { if (_parser.State == TdsParserState.Broken || _parser.State == TdsParserState.Closed) { @@ -3053,7 +3053,7 @@ private bool TryHasMoreResults(out bool moreResults) Debug.Assert(null != _command, "unexpected null command from the data reader!"); - while (_stateObj._pendingData) + while (_stateObj.HasPendingData) { byte token; if (!_stateObj.TryPeekByte(out token)) @@ -3137,7 +3137,7 @@ private bool TryHasMoreRows(out bool moreRows) moreRows = false; return true; } - if (_stateObj._pendingData) + if (_stateObj.HasPendingData) { // Consume error's, info's, done's on HasMoreRows, so user obtains error on Read. byte b; @@ -3178,7 +3178,7 @@ private bool TryHasMoreRows(out bool moreRows) moreRows = false; return false; } - if (_stateObj._pendingData) + if (_stateObj.HasPendingData) { if (!_stateObj.TryPeekByte(out b)) { @@ -3451,7 +3451,7 @@ private bool TryReadInternal(bool setTimeout, out bool more) if (moreRows) { // read the row from the backend (unless it's an altrow were the marker is already inside the altrow ...) - while (_stateObj._pendingData) + while (_stateObj.HasPendingData) { if (_altRowStatus != ALTROWSTATUS.AltRow) { @@ -3483,7 +3483,7 @@ private bool TryReadInternal(bool setTimeout, out bool more) } } - if (!_stateObj._pendingData) + if (!_stateObj.HasPendingData) { if (!TryCloseInternal(false /*closeReader*/)) { @@ -3507,7 +3507,7 @@ private bool TryReadInternal(bool setTimeout, out bool more) { // if we are in SingleRow mode, and we've read the first row, // read the rest of the rows, if any - while (_stateObj._pendingData && !_sharedState._dataReady) + while (_stateObj.HasPendingData && !_sharedState._dataReady) { if (!_parser.TryRun(RunBehavior.ReturnImmediately, _command, this, null, _stateObj, out _sharedState._dataReady)) { @@ -3548,7 +3548,7 @@ private bool TryReadInternal(bool setTimeout, out bool more) more = false; #if DEBUG - if ((!_sharedState._dataReady) && (_stateObj._pendingData)) + if ((!_sharedState._dataReady) && (_stateObj.HasPendingData)) { byte token; if (!_stateObj.TryPeekByte(out token)) diff --git a/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/SqlInternalConnectionTds.cs b/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/SqlInternalConnectionTds.cs index 322c25e027..069e2b2d72 100644 --- a/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/SqlInternalConnectionTds.cs +++ b/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/SqlInternalConnectionTds.cs @@ -697,11 +697,11 @@ internal override void ValidateConnectionForExecute(SqlCommand command) // or if MARS is off, then a datareader exists throw ADP.OpenReaderExists(parser.MARSOn); } - else if (!parser.MARSOn && parser._physicalStateObj._pendingData) + else if (!parser.MARSOn && parser._physicalStateObj.HasPendingData) { parser.DrainData(parser._physicalStateObj); } - Debug.Assert(!parser._physicalStateObj._pendingData, "Should not have a busy physicalStateObject at this point!"); + Debug.Assert(!parser._physicalStateObj.HasPendingData, "Should not have a busy physicalStateObject at this point!"); parser.RollbackOrphanedAPITransactions(); } @@ -840,7 +840,7 @@ private void ResetConnection() // obtains a clone. Debug.Assert(!HasLocalTransactionFromAPI, "Upon ResetConnection SqlInternalConnectionTds has a currently ongoing local transaction."); - Debug.Assert(!_parser._physicalStateObj._pendingData, "Upon ResetConnection SqlInternalConnectionTds has pending data."); + Debug.Assert(!_parser._physicalStateObj.HasPendingData, "Upon ResetConnection SqlInternalConnectionTds has pending data."); if (_fResetConnection) { diff --git a/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/TdsParser.cs b/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/TdsParser.cs index 1fec3e066f..0b1a52f779 100644 --- a/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/TdsParser.cs +++ b/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/TdsParser.cs @@ -502,7 +502,7 @@ internal TdsParserStateObject GetSession(object owner) { session = _sessionPool.GetSession(owner); - Debug.Assert(!session._pendingData, "pending data on a pooled MARS session"); + Debug.Assert(!session.HasPendingData, "pending data on a pooled MARS session"); } else { @@ -943,7 +943,7 @@ internal void Deactivate(bool connectionIsDoomed) if (!connectionIsDoomed && null != _physicalStateObj) { - if (_physicalStateObj._pendingData) + if (_physicalStateObj.HasPendingData) { DrainData(_physicalStateObj); } @@ -1800,7 +1800,7 @@ internal bool TryRun(RunBehavior runBehavior, SqlCommand cmdHandler, SqlDataRead { if (token == TdsEnums.SQLERROR) { - stateObj._errorTokenReceived = true; // Keep track of the fact error token was received - for Done processing. + stateObj.HasReceivedError = true; // Keep track of the fact error token was received - for Done processing. } SqlError error; @@ -2320,18 +2320,18 @@ internal bool TryRun(RunBehavior runBehavior, SqlCommand cmdHandler, SqlDataRead break; } - Debug.Assert(stateObj._pendingData || !dataReady, "dataReady is set, but there is no pending data"); + Debug.Assert(stateObj.HasPendingData || !dataReady, "dataReady is set, but there is no pending data"); } // Loop while data pending & runbehavior not return immediately, OR // if in attention case, loop while no more pending data & attention has not yet been // received. - while ((stateObj._pendingData && + while ((stateObj.HasPendingData && (RunBehavior.ReturnImmediately != (RunBehavior.ReturnImmediately & runBehavior))) || - (!stateObj._pendingData && stateObj._attentionSent && !stateObj._attentionReceived)); + (!stateObj.HasPendingData && stateObj._attentionSent && !stateObj.HasReceivedAttention)); #if DEBUG - if ((stateObj._pendingData) && (!dataReady)) + if ((stateObj.HasPendingData) && (!dataReady)) { byte token; if (!stateObj.TryPeekByte(out token)) @@ -2342,7 +2342,7 @@ internal bool TryRun(RunBehavior runBehavior, SqlCommand cmdHandler, SqlDataRead } #endif - if (!stateObj._pendingData) + if (!stateObj.HasPendingData) { if (null != CurrentTransaction) { @@ -2352,7 +2352,7 @@ internal bool TryRun(RunBehavior runBehavior, SqlCommand cmdHandler, SqlDataRead // if we received an attention (but this thread didn't send it) then // we throw an Operation Cancelled error - if (stateObj._attentionReceived) + if (stateObj.HasReceivedAttention) { // Dev11 #344723: SqlClient stress hang System_Data!Tcp::ReadSync via a call to SqlDataReader::Close // Spin until SendAttention has cleared _attentionSending, this prevents a race condition between receiving the attention ACK and setting _attentionSent @@ -2363,7 +2363,7 @@ internal bool TryRun(RunBehavior runBehavior, SqlCommand cmdHandler, SqlDataRead { // Reset attention state. stateObj._attentionSent = false; - stateObj._attentionReceived = false; + stateObj.HasReceivedAttention = false; if (RunBehavior.Clean != (RunBehavior.Clean & runBehavior) && !stateObj._internalTimeout) { @@ -2781,7 +2781,7 @@ private bool TryProcessDone(SqlCommand cmd, SqlDataReader reader, ref RunBehavio { Debug.Assert(TdsEnums.DONE_MORE != (status & TdsEnums.DONE_MORE), "Not expecting DONE_MORE when receiving DONE_ATTN"); Debug.Assert(stateObj._attentionSent, "Received attention done without sending one!"); - stateObj._attentionReceived = true; + stateObj.HasReceivedAttention = true; Debug.Assert(stateObj._inBytesUsed == stateObj._inBytesRead && stateObj._inBytesPacket == 0, "DONE_ATTN received with more data left on wire"); } if ((null != cmd) && (TdsEnums.DONE_COUNT == (status & TdsEnums.DONE_COUNT))) @@ -2805,7 +2805,7 @@ private bool TryProcessDone(SqlCommand cmd, SqlDataReader reader, ref RunBehavio } } - stateObj._receivedColMetaData = false; + stateObj.HasReceivedColumnMetadata = false; // Surface exception for DONE_ERROR in the case we did not receive an error token // in the stream, but an error occurred. In these cases, we throw a general server error. The @@ -2814,7 +2814,7 @@ private bool TryProcessDone(SqlCommand cmd, SqlDataReader reader, ref RunBehavio // the server has reached its max connection limit. Bottom line, we need to throw general // error in the cases where we did not receive an error token along with the DONE_ERROR. if ((TdsEnums.DONE_ERROR == (TdsEnums.DONE_ERROR & status)) && stateObj.ErrorCount == 0 && - stateObj._errorTokenReceived == false && (RunBehavior.Clean != (RunBehavior.Clean & run))) + stateObj.HasReceivedError == false && (RunBehavior.Clean != (RunBehavior.Clean & run))) { stateObj.AddError(new SqlError(0, 0, TdsEnums.MIN_ERROR_CLASS, _server, SQLMessage.SevereError(), "", 0)); @@ -2848,17 +2848,17 @@ private bool TryProcessDone(SqlCommand cmd, SqlDataReader reader, ref RunBehavio // stop if the DONE_MORE bit isn't set (see above for attention handling) if (TdsEnums.DONE_MORE != (status & TdsEnums.DONE_MORE)) { - stateObj._errorTokenReceived = false; + stateObj.HasReceivedError = false; if (stateObj._inBytesUsed >= stateObj._inBytesRead) { - stateObj._pendingData = false; + stateObj.HasPendingData = false; } } // _pendingData set by e.g. 'TdsExecuteSQLBatch' // _hasOpenResult always set to true by 'WriteMarsHeader' // - if (!stateObj._pendingData && stateObj._hasOpenResult) + if (!stateObj.HasPendingData && stateObj.HasOpenResult) { /* Debug.Assert(!((sqlTransaction != null && _distributedTransaction != null) || @@ -4202,7 +4202,7 @@ internal void ThrowUnsupportedCollationEncountered(TdsParserStateObject stateObj { DrainData(stateObj); - stateObj._pendingData = false; + stateObj.HasPendingData = false; } ThrowExceptionAndWarning(stateObj); @@ -4714,7 +4714,7 @@ private bool TryCommonProcessMetaData(TdsParserStateObject stateObj, _SqlMetaDat // We get too many DONE COUNTs from the server, causing too many StatementCompleted event firings. // We only need to fire this event when we actually have a meta data stream with 0 or more rows. - stateObj._receivedColMetaData = true; + stateObj.HasReceivedColumnMetadata = true; return true; } @@ -8072,7 +8072,7 @@ internal void TdsLogin(SqlLogin rec, TdsEnums.FeatureExtension requestedFeatures _physicalStateObj.WritePacket(TdsEnums.HARDFLUSH); _physicalStateObj.ResetSecurePasswordsInformation(); - _physicalStateObj._pendingData = true; + _physicalStateObj.HasPendingData = true; _physicalStateObj._messageStatus = 0; }// tdsLogin @@ -8100,7 +8100,7 @@ internal void SendFedAuthToken(SqlFedAuthToken fedAuthToken) _physicalStateObj.WriteByteArray(accessToken, accessToken.Length, 0); _physicalStateObj.WritePacket(TdsEnums.HARDFLUSH); - _physicalStateObj._pendingData = true; + _physicalStateObj.HasPendingData = true; _physicalStateObj._messageStatus = 0; _connHandler._federatedAuthenticationRequested = true; @@ -8397,7 +8397,7 @@ bool isDelegateControlRequest Task writeTask = stateObj.WritePacket(TdsEnums.HARDFLUSH); Debug.Assert(writeTask == null, "Writes should not pend when writing sync"); - stateObj._pendingData = true; + stateObj.HasPendingData = true; stateObj._messageStatus = 0; SqlDataReader dtcReader = null; @@ -9828,7 +9828,7 @@ internal Task WriteBulkCopyDone(TdsParserStateObject stateObj) WriteShort(0, stateObj); WriteInt(0, stateObj); - stateObj._pendingData = true; + stateObj.HasPendingData = true; stateObj._messageStatus = 0; return stateObj.WritePacket(TdsEnums.HARDFLUSH); } diff --git a/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/TdsParserSessionPool.cs b/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/TdsParserSessionPool.cs index 71b9ae3936..789afde0ce 100644 --- a/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/TdsParserSessionPool.cs +++ b/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/TdsParserSessionPool.cs @@ -167,7 +167,7 @@ internal void PutSession(TdsParserStateObject session) else if ((okToReuse) && (_freeStateObjectCount < MaxInactiveCount)) { // Session is good to re-use and our cache has space - Debug.Assert(!session._pendingData, "pending data on a pooled session?"); + Debug.Assert(!session.HasPendingData, "pending data on a pooled session?"); _freeStateObjects[_freeStateObjectCount] = session; _freeStateObjectCount++; diff --git a/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/TdsParserStateObject.cs b/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/TdsParserStateObject.cs index 499b22de44..b1fa2336cd 100644 --- a/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/TdsParserStateObject.cs +++ b/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/TdsParserStateObject.cs @@ -22,6 +22,17 @@ sealed internal class LastIOTimer internal abstract class TdsParserStateObject { + [Flags] + internal enum SnapshottedStateFlags : byte + { + None = 0, + PendingData = 1 << 1, + OpenResult = 1 << 2, + ErrorTokenReceived = 1 << 3, // Keep track of whether an error was received for the result. This is reset upon each done token + ColMetaDataReceived = 1 << 4, // Used to keep track of when to fire StatementCompleted event. + AttentionReceived = 1 << 5 // NOTE: Received is not volatile as it is only ever accessed\modified by TryRun its callees (i.e. single threaded access) + } + private const int AttentionTimeoutSeconds = 5; // Ticks to consider a connection "good" after a successful I/O (10,000 ticks = 1 ms) @@ -54,41 +65,50 @@ internal abstract class TdsParserStateObject internal int _outBytesUsed = TdsEnums.HEADER_LEN; // number of bytes used in internal write buffer - // - initialize past header // In buffer variables - protected byte[] _inBuff; // internal read buffer - initialize on login - internal int _inBytesUsed = 0; // number of bytes used in internal read buffer - internal int _inBytesRead = 0; // number of bytes read into internal read buffer - internal int _inBytesPacket = 0; // number of bytes left in packet + /// + /// internal read buffer - initialize on login + /// + protected byte[] _inBuff; + /// + /// number of bytes used in internal read buffer + /// + internal int _inBytesUsed; + /// + /// number of bytes read into internal read buffer + /// + internal int _inBytesRead; + + /// + /// number of bytes left in packet + /// + internal int _inBytesPacket; // Packet state variables - internal byte _outputMessageType = 0; // tds header type - internal byte _messageStatus; // tds header status - internal byte _outputPacketNumber = 1; // number of packets sent to server in message - start at 1 per ramas + internal byte _outputMessageType; // tds header type + internal byte _messageStatus; // tds header status + internal byte _outputPacketNumber = 1; // number of packets sent to server in message - start at 1 per ramas internal uint _outputPacketCount; - internal bool _pendingData = false; - internal volatile bool _fResetEventOwned = false; // ResetEvent serializing call to sp_reset_connection - internal volatile bool _fResetConnectionSent = false; // For multiple packet execute - - internal bool _errorTokenReceived = false; // Keep track of whether an error was received for the result. - // This is reset upon each done token - there can be - - internal bool _bulkCopyOpperationInProgress = false; // Set to true during bulk copy and used to turn toggle write timeouts. - internal bool _bulkCopyWriteTimeout = false; // Set to trun when _bulkCopyOpeperationInProgress is trun and write timeout happens - - // SNI variables // multiple resultsets in one batch. + internal volatile bool _fResetEventOwned; // ResetEvent serializing call to sp_reset_connection + internal volatile bool _fResetConnectionSent; // For multiple packet execute + internal bool _bulkCopyOpperationInProgress; // Set to true during bulk copy and used to turn toggle write timeouts. + internal bool _bulkCopyWriteTimeout; // Set to trun when _bulkCopyOpeperationInProgress is trun and write timeout happens - protected readonly object _writePacketLockObject = new object(); // Used to synchronize access to _writePacketCache and _pendingWritePackets + // SNI variables + /// + /// Used to synchronize access to _writePacketCache and _pendingWritePackets + /// + protected readonly object _writePacketLockObject = new object(); // Async variables - private int _pendingCallbacks; // we increment this before each async read/write call and decrement it in the callback. We use this to determine when to release the GcHandle... + private int _pendingCallbacks; // we increment this before each async read/write call and decrement it in the callback. We use this to determine when to release the GcHandle... // Timeout variables private long _timeoutMilliseconds; - private long _timeoutTime; // variable used for timeout computations, holds the value of the hi-res performance counter at which this request should expire - internal volatile bool _attentionSent = false; // true if we sent an Attention to the server - internal bool _attentionReceived = false; // NOTE: Received is not volatile as it is only ever accessed\modified by TryRun its callees (i.e. single threaded access) - internal volatile bool _attentionSending = false; - internal bool _internalTimeout = false; // an internal timeout occurred + private long _timeoutTime; // variable used for timeout computations, holds the value of the hi-res performance counter at which this request should expire + internal volatile bool _attentionSent; // true if we sent an Attention to the server + internal volatile bool _attentionSending; + internal bool _internalTimeout; // an internal timeout occurred private readonly LastIOTimer _lastSuccessfulIOTimer; @@ -113,37 +133,33 @@ internal abstract class TdsParserStateObject private const int _waitForCancellationLockPollTimeout = 100; private WeakReference _cancellationOwner = new WeakReference(null); - internal bool _hasOpenResult = false; // Cache the transaction for which this command was executed so upon completion we can // decrement the appropriate result count. - internal SqlInternalTransaction _executedUnderTransaction = null; + internal SqlInternalTransaction _executedUnderTransaction; // TDS stream processing variables internal ulong _longlen; // plp data length indicator internal ulong _longlenleft; // Length of data left to read (64 bit lengths) - internal int[] _decimalBits = null; // scratch buffer for decimal/numeric data + internal int[] _decimalBits; // scratch buffer for decimal/numeric data internal byte[] _bTmp = new byte[TdsEnums.YUKON_HEADER_LEN]; // Scratch buffer for misc use - internal int _bTmpRead = 0; // Counter for number of temporary bytes read - internal Decoder _plpdecoder = null; // Decoder object to process plp character data - internal bool _accumulateInfoEvents = false; // TRUE - accumulate info messages during TdsParser.Run, FALSE - fire them - internal List _pendingInfoEvents = null; - internal byte[] _bLongBytes = null; // scratch buffer to serialize Long values (8 bytes). - internal byte[] _bIntBytes = null; // scratch buffer to serialize Int values (4 bytes). - internal byte[] _bShortBytes = null; // scratch buffer to serialize Short values (2 bytes). - internal byte[] _bDecimalBytes = null; // scratch buffer to serialize decimal values (17 bytes). + internal int _bTmpRead; // Counter for number of temporary bytes read + internal Decoder _plpdecoder; // Decoder object to process plp character data + internal bool _accumulateInfoEvents; // TRUE - accumulate info messages during TdsParser.Run, FALSE - fire them + internal List _pendingInfoEvents; + internal byte[] _bLongBytes; // scratch buffer to serialize Long values (8 bytes). + internal byte[] _bIntBytes; // scratch buffer to serialize Int values (4 bytes). + internal byte[] _bShortBytes; // scratch buffer to serialize Short values (2 bytes). + internal byte[] _bDecimalBytes; // scratch buffer to serialize decimal values (17 bytes). // // DO NOT USE THIS BUFFER FOR OTHER THINGS. // ProcessHeader can be called ANYTIME while doing network reads. private byte[] _partialHeaderBuffer = new byte[TdsEnums.HEADER_LEN]; // Scratch buffer for ProcessHeader - internal int _partialHeaderBytesRead = 0; - - internal _SqlMetaDataSet _cleanupMetaData = null; - internal _SqlMetaDataSetCollection _cleanupAltMetaDataSetArray = null; - + internal int _partialHeaderBytesRead; - internal bool _receivedColMetaData; // Used to keep track of when to fire StatementCompleted event. + internal _SqlMetaDataSet _cleanupMetaData; + internal _SqlMetaDataSetCollection _cleanupAltMetaDataSetArray; private SniContext _sniContext = SniContext.Undefined; #if DEBUG @@ -159,33 +175,35 @@ internal abstract class TdsParserStateObject internal TaskCompletionSource _networkPacketTaskSource; private Timer _networkPacketTimeout; internal bool _syncOverAsync = true; - private bool _snapshotReplay = false; + private bool _snapshotReplay; private StateSnapshot _snapshot; + private StateSnapshot _cachedSnapshot; + private SnapshottedStateFlags _snapshottedState; internal ExecutionContext _executionContext; - internal bool _asyncReadWithoutSnapshot = false; + internal bool _asyncReadWithoutSnapshot; #if DEBUG // Used to override the assert than ensures that the stacktraces on subsequent replays are the same // This is useful is you are purposefully running the replay from a different thread (e.g. during SqlDataReader.Close) - internal bool _permitReplayStackTraceToDiffer = false; + internal bool _permitReplayStackTraceToDiffer; // Used to indicate that the higher level object believes that this stateObj has enough data to complete an operation // If this stateObj has to read, then it will raise an assert - internal bool _shouldHaveEnoughData = false; + internal bool _shouldHaveEnoughData; #endif // local exceptions to cache warnings and errors internal SqlErrorCollection _errors; internal SqlErrorCollection _warnings; internal object _errorAndWarningsLock = new object(); - private bool _hasErrorOrWarning = false; + private bool _hasErrorOrWarning; // local exceptions to cache warnings and errors that occurred prior to sending attention internal SqlErrorCollection _preAttentionErrors; internal SqlErrorCollection _preAttentionWarnings; - private volatile TaskCompletionSource _writeCompletionSource = null; - protected volatile int _asyncWriteCount = 0; - private volatile Exception _delayedWriteAsyncCallbackException = null; // set by write async callback if completion source is not yet created + private volatile TaskCompletionSource _writeCompletionSource; + protected volatile int _asyncWriteCount; + private volatile Exception _delayedWriteAsyncCallbackException; // set by write async callback if completion source is not yet created // _readingcount is incremented when we are about to read. // We check the parser state afterwards. @@ -224,13 +242,13 @@ internal abstract class TdsParserStateObject // Prevents any pending read from completing until the user signals it using // CompletePendingReadWithSuccess() or CompletePendingReadWithFailure(int errorCode) in SqlCommand\SqlDataReader - internal static bool _forcePendingReadsToWaitForUser; - internal TaskCompletionSource _realNetworkPacketTaskSource = null; + internal static bool _forcePendingReadsToWaitForUser = false; + internal TaskCompletionSource _realNetworkPacketTaskSource; // Field is never assigned to, and will always have its default value #pragma warning disable 0649 // Set to true to enable checking the call stacks match when packet retry occurs. - internal static bool _checkNetworkPacketRetryStacks; + internal static bool _checkNetworkPacketRetryStacks = false; #pragma warning restore 0649 #endif @@ -311,17 +329,7 @@ internal SniContext DebugOnlyCopyOfSniContext return _debugOnlyCopyOfSniContext; } } -#endif - - internal bool HasOpenResult - { - get - { - return _hasOpenResult; - } - } -#if DEBUG internal void InvalidateDebugOnlyCopyOfSniContext() { _debugOnlyCopyOfSniContext = SniContext.Undefined; @@ -608,7 +616,7 @@ internal void Cancel(object caller) { _cancelled = true; - if (_pendingData && !_attentionSent) + if (HasPendingData && !_attentionSent) { bool hasParserLock = false; // Keep looping until we have the parser lock (and so are allowed to write), or the connection closes\breaks @@ -799,7 +807,7 @@ internal bool Deactivate() TdsParserState state = Parser.State; if (state != TdsParserState.Broken && state != TdsParserState.Closed) { - if (_pendingData) + if (HasPendingData) { Parser.DrainData(this); // This may throw - taking us to catch block.c } @@ -849,7 +857,7 @@ internal void DecrementOpenResultCount() _executedUnderTransaction.DecrementAndObtainOpenResultCount(); _executedUnderTransaction = null; } - _hasOpenResult = false; + HasOpenResult = false; } internal int DecrementPendingCallbacks(bool release) @@ -889,7 +897,7 @@ internal void DisposeCounters() internal int IncrementAndObtainOpenResultCount(SqlInternalTransaction transaction) { - _hasOpenResult = true; + HasOpenResult = true; if (transaction == null) { @@ -962,13 +970,13 @@ internal Task ExecuteFlush() Task writePacketTask = WritePacket(TdsEnums.HARDFLUSH); if (writePacketTask == null) { - _pendingData = true; + HasPendingData = true; _messageStatus = 0; return null; } else { - return AsyncHelper.CreateContinuationTask(writePacketTask, () => { _pendingData = true; _messageStatus = 0; }); + return AsyncHelper.CreateContinuationTask(writePacketTask, () => { HasPendingData = true; _messageStatus = 0; }); } } } @@ -1128,6 +1136,49 @@ internal void ResetPacketCounters() _outputPacketCount = 0; } + private void SetSnapshottedState(SnapshottedStateFlags flag, bool value) + { + if (value) + { + _snapshottedState |= flag; + } + else + { + _snapshottedState &= ~flag; + } + } + + internal bool HasOpenResult + { + get => _snapshottedState.HasFlag(SnapshottedStateFlags.OpenResult); + set => SetSnapshottedState(SnapshottedStateFlags.OpenResult, value); + } + + internal bool HasPendingData + { + get => _snapshottedState.HasFlag(SnapshottedStateFlags.PendingData); + set => SetSnapshottedState(SnapshottedStateFlags.PendingData, value); + } + + internal bool HasReceivedError + { + get => _snapshottedState.HasFlag(SnapshottedStateFlags.ErrorTokenReceived); + set => SetSnapshottedState(SnapshottedStateFlags.ErrorTokenReceived, value); + } + + internal bool HasReceivedAttention + { + get => _snapshottedState.HasFlag(SnapshottedStateFlags.AttentionReceived); + set => SetSnapshottedState(SnapshottedStateFlags.AttentionReceived, value); + } + + internal bool HasReceivedColumnMetadata + { + get => _snapshottedState.HasFlag(SnapshottedStateFlags.ColMetaDataReceived); + set => SetSnapshottedState(SnapshottedStateFlags.ColMetaDataReceived, value); + } + + internal bool SetPacketSize(int size) { if (size > TdsEnums.MAX_PACKET_SIZE) @@ -1991,14 +2042,29 @@ internal bool TrySkipBytes(int num) internal void SetSnapshot() { - _snapshot = new StateSnapshot(this); - _snapshot.Snap(); + StateSnapshot snapshot = _snapshot; + if (snapshot is null) + { + snapshot = Interlocked.Exchange(ref _cachedSnapshot, null) ?? new StateSnapshot(); + } + else + { + snapshot.Clear(); + } + _snapshot = snapshot; + _snapshot.Snap(this); _snapshotReplay = false; } internal void ResetSnapshot() { - _snapshot = null; + if (_snapshot != null) + { + StateSnapshot snapshot = _snapshot; + _snapshot = null; + snapshot.Clear(); + Interlocked.CompareExchange(ref _cachedSnapshot, snapshot, null); + } _snapshotReplay = false; } @@ -3735,7 +3801,7 @@ internal void AssertStateIsClean() Debug.Assert(_asyncWriteCount == 0, "StateObj still has outstanding async writes"); Debug.Assert(_delayedWriteAsyncCallbackException == null, "StateObj has an unobserved exceptions from an async write"); // Attention\Cancellation\Timeouts - Debug.Assert(!_attentionReceived && !_attentionSent && !_attentionSending, $"StateObj is still dealing with attention: Sent: {_attentionSent}, Received: {_attentionReceived}, Sending: {_attentionSending}"); + Debug.Assert(!HasReceivedAttention && !_attentionSent && !_attentionSending, $"StateObj is still dealing with attention: Sent: {_attentionSent}, Received: {HasReceivedAttention}, Sending: {_attentionSending}"); Debug.Assert(!_cancelled, "StateObj still has cancellation set"); Debug.Assert(!_internalTimeout, "StateObj still has internal timeout set"); // Errors and Warnings @@ -3821,46 +3887,71 @@ internal void CloneCleanupAltMetaDataSetArray() } } - private class PacketData + private sealed class StateSnapshot { - public byte[] Buffer; - public int Read; -#if DEBUG - public string Stack; -#endif - } + private sealed class PLPData + { + public readonly ulong SnapshotLongLen; + public readonly ulong SnapshotLongLenLeft; - private class StateSnapshot - { - private List _snapshotInBuffs; - private int _snapshotInBuffCurrent = 0; - private int _snapshotInBytesUsed = 0; - private int _snapshotInBytesPacket = 0; - private bool _snapshotPendingData = false; - private bool _snapshotErrorTokenReceived = false; - private bool _snapshotHasOpenResult = false; - private bool _snapshotReceivedColumnMetadata = false; - private bool _snapshotAttentionReceived; - private byte _snapshotMessageStatus; + public PLPData(ulong snapshotLongLen, ulong snapshotLongLenLeft) + { + SnapshotLongLen = snapshotLongLen; + SnapshotLongLenLeft = snapshotLongLenLeft; + } + } - private NullBitmap _snapshotNullBitmapInfo; - private ulong _snapshotLongLen; - private ulong _snapshotLongLenLeft; - private _SqlMetaDataSet _snapshotCleanupMetaData; - private _SqlMetaDataSetCollection _snapshotCleanupAltMetaDataSetArray; + private sealed partial class PacketData + { + public byte[] Buffer; + public int Read; + public PacketData Prev; + + public void SetStack(string value) + { + SetStackInternal(value); + } + partial void SetStackInternal(string value); - private readonly TdsParserStateObject _stateObj; + public void Clear() + { + Buffer = null; + Read = 0; + Prev = null; + SetStackInternal(null); + } + } - public StateSnapshot(TdsParserStateObject state) +#if DEBUG + private sealed partial class PacketData { - _snapshotInBuffs = new List(); - _stateObj = state; + public string Stack; + + partial void SetStackInternal(string value) + { + Stack = value; + } } -#if DEBUG private int _rollingPend = 0; private int _rollingPendCount = 0; +#endif + private PacketData _snapshotInBuffList; + private PacketData _sparePacket; + private NullBitmap _snapshotNullBitmapInfo; + private _SqlMetaDataSet _snapshotCleanupMetaData; + private _SqlMetaDataSetCollection _snapshotCleanupAltMetaDataSetArray; + private PLPData _plpData; + private TdsParserStateObject _stateObj; + + private int _snapshotInBuffCount; + private int _snapshotInBuffCurrent; + private int _snapshotInBytesUsed; + private int _snapshotInBytesPacket; + private SnapshottedStateFlags _state; + private byte _snapshotMessageStatus; +#if DEBUG internal bool DoPend() { if (_failAsyncPends || !_forceAllPends) @@ -3878,8 +3969,25 @@ internal bool DoPend() _rollingPendCount++; return false; } -#endif + internal void AssertCurrent() + { + Debug.Assert(_snapshotInBuffCurrent == _snapshotInBuffCount, "Should not be reading new packets when not replaying last packet"); + } + + internal void CheckStack(string trace) + { + PacketData prev = _snapshotInBuffList?.Prev; + if (prev.Stack == null) + { + prev.Stack = trace; + } + else + { + Debug.Assert(_stateObj._permitReplayStackTraceToDiffer || prev.Stack.ToString() == trace.ToString(), "The stack trace on subsequent replays should be the same"); + } + } +#endif internal void CloneNullBitmapInfo() { if (_stateObj._nullBitmapInfo.ReferenceEquals(_snapshotNullBitmapInfo)) @@ -3898,42 +4006,45 @@ internal void CloneCleanupAltMetaDataSetArray() internal void PushBuffer(byte[] buffer, int read) { - Debug.Assert(!_snapshotInBuffs.Any(b => object.ReferenceEquals(b.Buffer, buffer))); - - PacketData packetData = new PacketData(); - packetData.Buffer = buffer; - packetData.Read = read; #if DEBUG - packetData.Stack = _stateObj._lastStack; + for (PacketData current = _snapshotInBuffList; current != null; current = current.Prev) + { + Debug.Assert(!object.ReferenceEquals(current.Buffer, buffer)); + } #endif - _snapshotInBuffs.Add(packetData); - } - -#if DEBUG - internal void AssertCurrent() - { - Debug.Assert(_snapshotInBuffCurrent == _snapshotInBuffs.Count, "Should not be reading new packets when not replaying last packet"); - } - internal void CheckStack(string trace) - { - PacketData prev = _snapshotInBuffs[_snapshotInBuffCurrent - 1]; - if (prev.Stack == null) + PacketData packetData = _sparePacket; + if (packetData is null) { - prev.Stack = trace; + packetData = new PacketData(); } else { - Debug.Assert(_stateObj._permitReplayStackTraceToDiffer || prev.Stack.ToString() == trace.ToString(), "The stack trace on subsequent replays should be the same"); + _sparePacket = null; } - } + packetData.Buffer = buffer; + packetData.Read = read; + packetData.Prev = _snapshotInBuffList; +#if DEBUG + packetData.SetStack(_stateObj._lastStack); #endif + _snapshotInBuffList = packetData; + _snapshotInBuffCount++; + } internal bool Replay() { - if (_snapshotInBuffCurrent < _snapshotInBuffs.Count) + if (_snapshotInBuffCurrent < _snapshotInBuffCount) { - PacketData next = _snapshotInBuffs[_snapshotInBuffCurrent]; + PacketData next = _snapshotInBuffList; + for ( + int position = (_snapshotInBuffCount - 1); + position != _snapshotInBuffCurrent; + position -= 1 + ) + { + next = next.Prev; + } _stateObj._inBuff = next.Buffer; _stateObj._inBytesUsed = 0; _stateObj._inBytesRead = next.Read; @@ -3944,25 +4055,27 @@ internal bool Replay() return false; } - internal void Snap() + internal void Snap(TdsParserStateObject state) { - _snapshotInBuffs.Clear(); + _stateObj = state; + _snapshotInBuffList = null; + _snapshotInBuffCount = 0; _snapshotInBuffCurrent = 0; _snapshotInBytesUsed = _stateObj._inBytesUsed; _snapshotInBytesPacket = _stateObj._inBytesPacket; - _snapshotPendingData = _stateObj._pendingData; - _snapshotErrorTokenReceived = _stateObj._errorTokenReceived; + _snapshotMessageStatus = _stateObj._messageStatus; // _nullBitmapInfo must be cloned before it is updated _snapshotNullBitmapInfo = _stateObj._nullBitmapInfo; - _snapshotLongLen = _stateObj._longlen; - _snapshotLongLenLeft = _stateObj._longlenleft; + if (_stateObj._longlen != 0 || _stateObj._longlenleft != 0) + { + _plpData = new PLPData(_stateObj._longlen, _stateObj._longlenleft); + } _snapshotCleanupMetaData = _stateObj._cleanupMetaData; // _cleanupAltMetaDataSetArray must be cloned before it is updated _snapshotCleanupAltMetaDataSetArray = _stateObj._cleanupAltMetaDataSetArray; - _snapshotHasOpenResult = _stateObj._hasOpenResult; - _snapshotReceivedColumnMetadata = _stateObj._receivedColMetaData; - _snapshotAttentionReceived = _stateObj._attentionReceived; + + _state = _stateObj._snapshottedState; #if DEBUG _rollingPend = 0; _rollingPendCount = 0; @@ -3983,23 +4096,21 @@ internal void ResetSnapshotState() _stateObj._inBytesUsed = _snapshotInBytesUsed; _stateObj._inBytesPacket = _snapshotInBytesPacket; - _stateObj._pendingData = _snapshotPendingData; - _stateObj._errorTokenReceived = _snapshotErrorTokenReceived; + _stateObj._messageStatus = _snapshotMessageStatus; _stateObj._nullBitmapInfo = _snapshotNullBitmapInfo; _stateObj._cleanupMetaData = _snapshotCleanupMetaData; _stateObj._cleanupAltMetaDataSetArray = _snapshotCleanupAltMetaDataSetArray; - _stateObj._hasOpenResult = _snapshotHasOpenResult; - _stateObj._receivedColMetaData = _snapshotReceivedColumnMetadata; - _stateObj._attentionReceived = _snapshotAttentionReceived; + + _stateObj._snapshottedState = _state; // Reset partially read state (these only need to be maintained if doing async without snapshot) _stateObj._bTmpRead = 0; _stateObj._partialHeaderBytesRead = 0; // reset plp state - _stateObj._longlen = _snapshotLongLen; - _stateObj._longlenleft = _snapshotLongLenLeft; + _stateObj._longlen = _plpData?.SnapshotLongLen ?? 0; + _stateObj._longlenleft = _plpData?.SnapshotLongLenLeft ?? 0; _stateObj._snapshotReplay = true; @@ -4010,6 +4121,32 @@ internal void PrepareReplay() { ResetSnapshotState(); } + + internal void Clear() + { + PacketData packet = _snapshotInBuffList; + _snapshotInBuffList = null; + _snapshotInBuffCount = 0; + _snapshotInBuffCurrent = 0; + _snapshotInBytesUsed = 0; + _snapshotInBytesPacket = 0; + _snapshotMessageStatus = 0; + _snapshotNullBitmapInfo = default; + _plpData = null; + _snapshotCleanupMetaData = null; + _snapshotCleanupAltMetaDataSetArray = null; + _state = SnapshottedStateFlags.None; +#if DEBUG + _rollingPend = 0; + _rollingPendCount = 0; + _stateObj._lastStack = null; +#endif + _stateObj = null; + + packet.Clear(); + _sparePacket = packet; + } + } /* From 078e363d338776f655b6cf1b0736f43adeae9873 Mon Sep 17 00:00:00 2001 From: Wraith2 Date: Sat, 14 Sep 2019 23:10:56 +0100 Subject: [PATCH 2/4] cache SqlDataReader snapshots --- .../Microsoft/Data/SqlClient/SqlDataReader.cs | 60 ++++++++++++------- 1 file changed, 38 insertions(+), 22 deletions(-) diff --git a/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/SqlDataReader.cs b/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/SqlDataReader.cs index 4d263728ea..5fadbae100 100644 --- a/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/SqlDataReader.cs +++ b/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/SqlDataReader.cs @@ -87,6 +87,7 @@ internal class SharedState private Task _currentTask; private Snapshot _snapshot; + private Snapshot _cachedSnapshot; private CancellationTokenSource _cancelAsyncOnCloseTokenSource; private CancellationToken _cancelAsyncOnCloseToken; @@ -3770,6 +3771,10 @@ private bool TryReadColumnInternal(int i, bool readHeaderOnly = false) { // reset snapshot to save memory use. We can safely do that here because all SqlDataReader values are stable. // The retry logic can use the current values to get back to the right state. + if (_cachedSnapshot is null) + { + _cachedSnapshot = _snapshot; + } _snapshot = null; PrepareAsyncInvocation(useSnapshot: true); } @@ -4659,6 +4664,10 @@ public override Task ReadAsync(CancellationToken cancellationToken) if (!rowTokenRead) { rowTokenRead = true; + if (_cachedSnapshot is null) + { + _cachedSnapshot = _snapshot; + } _snapshot = null; PrepareAsyncInvocation(useSnapshot: true); } @@ -5112,28 +5121,27 @@ private void PrepareAsyncInvocation(bool useSnapshot) if (_snapshot == null) { - _snapshot = new Snapshot - { - _dataReady = _sharedState._dataReady, - _haltRead = _haltRead, - _metaDataConsumed = _metaDataConsumed, - _browseModeInfoConsumed = _browseModeInfoConsumed, - _hasRows = _hasRows, - _altRowStatus = _altRowStatus, - _nextColumnDataToRead = _sharedState._nextColumnDataToRead, - _nextColumnHeaderToRead = _sharedState._nextColumnHeaderToRead, - _columnDataBytesRead = _columnDataBytesRead, - _columnDataBytesRemaining = _sharedState._columnDataBytesRemaining, - - // _metadata and _altaMetaDataSetCollection must be Cloned - // before they are updated - _metadata = _metaData, - _altMetaDataSetCollection = _altMetaDataSetCollection, - _tableNames = _tableNames, - - _currentStream = _currentStream, - _currentTextReader = _currentTextReader, - }; + _snapshot = Interlocked.Exchange(ref _cachedSnapshot, null) ?? new Snapshot(); + + _snapshot._dataReady = _sharedState._dataReady; + _snapshot._haltRead = _haltRead; + _snapshot._metaDataConsumed = _metaDataConsumed; + _snapshot._browseModeInfoConsumed = _browseModeInfoConsumed; + _snapshot._hasRows = _hasRows; + _snapshot._altRowStatus = _altRowStatus; + _snapshot._nextColumnDataToRead = _sharedState._nextColumnDataToRead; + _snapshot._nextColumnHeaderToRead = _sharedState._nextColumnHeaderToRead; + _snapshot._columnDataBytesRead = _columnDataBytesRead; + _snapshot._columnDataBytesRemaining = _sharedState._columnDataBytesRemaining; + + // _metadata and _altaMetaDataSetCollection must be Cloned + // before they are updated + _snapshot._metadata = _metaData; + _snapshot._altMetaDataSetCollection = _altMetaDataSetCollection; + _snapshot._tableNames = _tableNames; + + _snapshot._currentStream = _currentStream; + _snapshot._currentTextReader = _currentTextReader; _stateObj.SetSnapshot(); } @@ -5186,6 +5194,10 @@ private void CleanupAfterAsyncInvocationInternal(TdsParserStateObject stateObj, stateObj._permitReplayStackTraceToDiffer = false; #endif + if (_cachedSnapshot is null) + { + _cachedSnapshot = _snapshot; + } // We are setting this to null inside the if-statement because stateObj==null means that the reader hasn't been initialized or has been closed (either way _snapshot should already be null) _snapshot = null; } @@ -5224,6 +5236,10 @@ private void SwitchToAsyncWithoutSnapshot() Debug.Assert(_snapshot != null, "Should currently have a snapshot"); Debug.Assert(_stateObj != null && !_stateObj._asyncReadWithoutSnapshot, "Already in async without snapshot"); + if (_cachedSnapshot is null) + { + _cachedSnapshot = _snapshot; + } _snapshot = null; _stateObj.ResetSnapshot(); _stateObj._asyncReadWithoutSnapshot = true; From ba9e55f6780f50c70cc3e7ddba5d9c00d4f8a2d7 Mon Sep 17 00:00:00 2001 From: Wraith2 Date: Wed, 16 Oct 2019 00:18:13 +0100 Subject: [PATCH 3/4] address feedback --- .../netcore/src/Microsoft/Data/SqlClient/TdsParserStateObject.cs | 1 - 1 file changed, 1 deletion(-) diff --git a/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/TdsParserStateObject.cs b/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/TdsParserStateObject.cs index b1fa2336cd..f46eee23a8 100644 --- a/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/TdsParserStateObject.cs +++ b/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/TdsParserStateObject.cs @@ -4146,7 +4146,6 @@ internal void Clear() packet.Clear(); _sparePacket = packet; } - } /* From b754d23750a23209a1556823934edec0f53cc56b Mon Sep 17 00:00:00 2001 From: Wraith2 Date: Wed, 6 Nov 2019 19:52:59 +0000 Subject: [PATCH 4/4] amend merge conflict --- .../netcore/src/Microsoft/Data/SqlClient/TdsParser.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/TdsParser.cs b/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/TdsParser.cs index 0b1a52f779..e530dacc6a 100644 --- a/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/TdsParser.cs +++ b/src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClient/TdsParser.cs @@ -2799,7 +2799,7 @@ private bool TryProcessDone(SqlCommand cmd, SqlDataReader reader, ref RunBehavio } } // Skip the bogus DONE counts sent by the server - if (stateObj._receivedColMetaData || (curCmd != TdsEnums.SELECT)) + if (stateObj.HasReceivedColumnMetadata || (curCmd != TdsEnums.SELECT)) { cmd.OnStatementCompleted(count); }