See Part 3

In Part 3, I illustrated a simple strategy to re-run a “repeatable” transaction using a simple “while” loop. A better and more fault resilient way of handling such cases (we are talking about transactions, after all) is to employ a transactional resource manager such as MSMQ from which to retrieve the transaction input. If the transaction fails due to a catastrophic failure (power outage or worse), an in-memory recovery strategy like the one shown in “RunTx()” will not get us very far. Instead, we will put the transaction input into a transactional queue, read it from the queue using a specialized message queue listener and equip that listener with some handling logic that is similar to what is shown in the “RunTx()” example.

Before I will show that, however, I will give you a generic, multi-threaded message queue listener that is not transaction aware (in the desired way) and which I will specialize in Part 5 (the last part) of this series. The reasons that I post the message listener logic in two steps are that (a) you might find a generic listener useful, (b) it is quite a bit of code and I don’t want the “essentials” of the recovery code to be buried amongst the basic queue listener framework, and (c) I am simply too lazy today to explain it all ;-)

Using this queue listener is pretty straightforward and should be rather self-explanatory. To listen, create an instance of the listener using one of the constructors and supply a “queueName” argument that is compatible with the System.Messaging.MessageQueue class. The queue must exist. The listener will bubble messages to the MessageReceived delegate (“unicast event”), which you must implement in your class and assign to the listener in order to get at the messages.

MessageQueueListener queueListener = new MessageQueueListener(sourceQueuePath, 4 ); queueListener.MessageReceived = new MessageReceivedEventHandler(queueListener MessageReceived); queueListener.ProcessingError += new ProcessingErrorEventHandler(queueListener ProcessingError); queueListener.Start();

And here are 344 lines of pure developer joy:

using System; using System.Collections; using System.Messaging; using System.Threading; namespace newtelligence.EnterpriseTools.Msmq {     ///<summary>     /// Event argument class for the MessageQueueListener.MessageReceived unicast event     ///</summary>     publicclass MessageReceivedEventArgs : EventArgs     {         private Message receivedMessage;         ///<summary>         /// Constructor         ///</summary>         ///<param name="receivedMessage">A message</param>         public MessageReceivedEventArgs( Message receivedMessage )         {            this.receivedMessage = receivedMessage;         }         ///<summary>         /// Received message         ///</summary>         public Message ReceivedMessage         {             get             {                 return receivedMessage;             }         }     }     ///<summary>     /// Event argument class for the MessageQueueListener.ProcessingError event     ///</summary>     publicclass ProcessingErrorEventArgs : EventArgs     {         ///<summary>         ///         ///</summary>         private Exception exception;         ///<summary>         /// Constructor         ///</summary>         ///<param name="exception">An exception</param>         public ProcessingErrorEventArgs( Exception exception )         {             this.exception = exception;         }         ///<summary>         /// Intercepted exception         ///</summary>         public Exception Exception         {             get             {                 return exception;             }         }     }     ///<summary>     /// Delegate for the MessageQueueListener.MessageReceived unicast event     ///</summary>     publicdelegatevoid MessageReceivedEventHandler( object sender, MessageReceivedEventArgs ea );     ///<summary>     /// Delegate for the MessageQueueListener.ProcessingError event     ///</summary>     publicdelegatevoid ProcessingErrorEventHandler( object sender, ProcessingErrorEventArgs ea );         ///<summary>     /// This class implements a multi-threaded message queue listener     ///</summary>     publicclass MessageQueueListener     {         protected Guid instanceId = Guid.NewGuid();         protected Thread[] listenerThreads;         protectedstring queueName=null;         protectedint numThreads=1;         protected ThreadPriority priority=ThreadPriority.Normal;         protected ManualResetEvent continueInterrupt = new ManualResetEvent(true);         protected ManualResetEvent pauseInterrupt = new ManualResetEvent(false);         protected ManualResetEvent stopInterrupt = new ManualResetEvent(false);         protectedint peekTimeout = 5;         protected MessageReceivedEventHandler messageReceived;         protectedint refCount = 0;                 ///<summary>         /// Initializes a new instance of the MessageQueueListener class         ///</summary>         ///<param name="queueName">Message name where listener waits messages</param>         public MessageQueueListener(string queueName)         {             queueName = queueName;             Initialize();         }         ///<summary>         /// Initializes a new instance of the MessageQueueListener class         ///</summary>         ///<param name="queueName">Name of the queue to listen on</param>         ///<param name="numThreads">Number of threads to run the listener on</param>         public MessageQueueListener(string queueName, int numThreads)         {             queueName = queueName;             numThreads = numThreads;             Initialize();         }         ///<summary>         /// Initializes a new instance of the MessageQueueListener class         ///</summary>         ///<param name="queueName">Name of the queue to listen on</param>         ///<param name="numThreads">Number of threads run the listener on</param>         ///<param name="priority">Thread priority</param>         public MessageQueueListener(string queueName, int numThreads, ThreadPriority priority)         {             queueName = queueName;             numThreads = numThreads;             priority = priority;             Initialize();         }         ///<summary>         /// Adds a reference to the reference counter.         ///</summary>         ///<returns>Current reference count</returns>         publicint AddReference()         {             return ++ refCount;         }         ///<summary>         /// Releases a reference from the reference counter.         ///</summary>         ///<returns>Current reference count</returns>         ///<remarks>If the reference counter drops to or below zero, the listener is stopped.</remarks>         publicint ReleaseReference()         {             if ( -- refCount <= 0 )             {                 Stop();             }             return refCount;         }         ///<summary>         /// Initializes the MessageQueueListener         ///</summary>         protectedvirtualvoid Initialize()         {             listenerThreads = new Thread[ numThreads];         }         ///<summary>         /// Unicast event that is raised when a message has been received         ///</summary>         public MessageReceivedEventHandler MessageReceived         {             get             {                 return messageReceived;             }             set             {                 if ( messageReceived == value)                     return;                  messageReceived = value;             }         }         ///<summary>         /// Multicast event that is raised when an exception occurs during processing.         ///</summary>         publicevent ProcessingErrorEventHandler ProcessingError;         ///<summary>         /// Raises the ProcessingError event         ///</summary>         ///<param name="pea"></param>         protectedvoid OnProcessingError( ProcessingErrorEventArgs pea )         {             if (ProcessingError != null )             {                 ProcessingError( this, pea );             }         }         ///<summary>         /// Starts the message queue listener. If the threads are already running,         /// Start() does nothing.         ///</summary>         publicvoid Start()         {             for(int i=0;i< numThreads;i++)             {                 if ( listenerThreads[i] == null )                 {                     Thread thread = new Thread(new ThreadStart(this.ReceiveLoop));                     thread.Name = String.Format("{0}:{1}:{2}",i, instanceId,GetType().Name);                     thread.IsBackground = true;                     thread.Priority = priority;                     listenerThreads[i] = thread;                     thread.Start();                 }             }         }         ///<summary>         /// Set listener state in stop         ///</summary>         publicvoid Stop()         {             Resume();             stopInterrupt.Set();             for(int i=0;i< numThreads;i++)             {                 listenerThreads[i].Join();                 listenerThreads[i] = null;             }         }         ///<summary>         /// Set listener state to "paused".         ///</summary>         ///<remarks>         /// Setting the listener ito the "paused" state will not suspend processing         /// immediately. All messages that are currently being processed will be         /// processed, but no new messages are being dequeued.         ///</remarks>         publicvoid Pause()         {             continueInterrupt.Reset();             pauseInterrupt.Set();                                     }         ///<summary>         /// Set listener state to "resume". This will resume the listener from         /// the "paused" state.         ///</summary>         publicvoid Resume()         {             pauseInterrupt.Reset();             continueInterrupt.Set();                    }         ///<summary>         /// Gets a value indicating whether the listener is paused.         ///</summary>         publicbool IsPaused         {             get             {                 return pauseInterrupt.WaitOne(TimeSpan.Zero,false);             }         }         ///<summary>         /// Gets a value indicating whether the listener is paused.         ///</summary>         publicbool IsStopping         {             get             {                 return stopInterrupt.WaitOne(TimeSpan.Zero,false);             }         }         ///<summary>         /// This is the thread entry point for the main receive loop.         ///</summary>         protectedvirtualvoid ReceiveLoop()         {             using (MessageQueue sourceQueue = new MessageQueue( queueName, false ))             {                 bool isTransactional = sourceQueue.Transactional;                 sourceQueue.MessageReadPropertyFilter.SetAll();                 while ( !IsStopping )                 {                     // if the continue interrupt is not set, we will hang here                     // until it is set again (resuming)                     WaitHandle.WaitAny(new WaitHandle[]{ continueInterrupt, stopInterrupt});                     try                     {                         //peek on the queue and wait for a message becoming available                         IAsyncResult asynchResult = sourceQueue.BeginPeek(TimeSpan.FromSeconds( peekTimeout),null,null);                         int firedWaitHandle = WaitHandle.WaitAny(new WaitHandle[]{asynchResult.AsyncWaitHandle, pauseInterrupt, stopInterrupt});                         if ( firedWaitHandle == 0 )                         {                             sourceQueue.EndPeek(asynchResult);                             Message receivedMessage = sourceQueue.Receive( TimeSpan.FromSeconds(0),                                 isTransactional?MessageQueueTransactionType.Single:MessageQueueTransactionType.None );                                                        try                             {                                 messageReceived(this,new MessageReceivedEventArgs(receivedMessage));                                                    }                             finally                             {                                 receivedMessage.Dispose();                             }                         }                         elseif ( firedWaitHandle == 1 )                         {                             // Pausing. Clean up the asynchronous Peek operation                             // (we'll assume that we have time for that) and then go                             // back to the top of the loop where we'll hang on the                             // continueInterrupt.                             sourceQueue.EndPeek(asynchResult);                         }                         else                         {                             // stopping. Skip the EndPeek() and exit the thread immediately.                             return;                         }                     }                     catch(MessageQueueException ex)                     {                         if ( ex.MessageQueueErrorCode != MessageQueueErrorCode.IOTimeout  )                         {                             break;                         }                     }                     catch( Exception ex )                     {                         // absorb all other exceptions and surface them from he listener                         // using the ProcessingError event.                         OnProcessingError( new ProcessingErrorEventArgs(ex));                     }                 }             }         }     } }

 

See Part 2

Now that we have a way to detect deadlocks and similar error conditions we might encounter during a transaction, we need to find a mechanism that will allow us to back out of the doomed transaction and retry the failed operation. The good thing about transaction failures of that sort is that you can essentially retry as often as you want without doing anything wrong – as long as all the resources you are reading from and writing to are guarded by transactional resource managers that are enlisted in the transaction. And that’s the hint at the trick we need to use.

If we were executing the code below through a remote call, it might be rather difficult to recover from a failure. The local transaction inside the component will roll back – ok – but that punts the problem back out to the caller, who will have to implement the appropriate logic to single out transaction failure causes that can be fixed by a retry and to actually resubmit the request. That can be rather tricky.

[Transaction(TransactionOption.Required)] publicclass DoesStuff : ServicedComponent {     /* things omitted */       [AutoComplete]     publicvoid DoStuff( string argument )     {         try         {            dbConnection.Open();            sprocUpdateAndQueryStuff.Parameters["@StuffArgument"].Value = argument;            result = this.GetResultFromReader( sprocUpdateAndQueryStuff.ExecuteReader() );         }         catch( Exception exception )         {             throw RepeatableOperationExceptionMapper.MapException( exception );                                    }         finally         {             dbConnection.Close();         }     } }

Even worse, we might not even learn about a failure depending on when and why the transaction fails. In distributed transaction scenarios, it is absolutely possible (even though rare) that a transaction fails long after the application code is done with all of its processing. Let’s assume everything in the call to DoStuff() above works just fine. The code calls a stored procedure on a remote SQL Server, SQL Server is completely happy and so is the component. Hence, neither has any good reason to complain. Let’s further assume there is a component DoesOtherStuff just like the one above and that component is subsequently called by the remote client and updates some other SQL Server database from within the same distributed transaction. That SQL Server and that component are both happy as well. Everybody is happy. The client commits the transaction, maybe by calling ServiceDomain.Leave() or leaving from a method of a ServicedComponent that serves as the transaction root. Now DTC (the transaction coordinator) wants to go around and ask everyone for their opinion about the transaction outcome. At this point, all user code you’ve contributed to this story is done executing. And now (we are talking about transactions, so this is where the fun is) disaster strikes and one of the SQL Servers you were talking to suddenly gets cut off from the network in some way. Hardware issue. Crash. Something. If you are lucky, the problem occurs while DTC is still walking around asking for votes (Phase 1). In this case, the transaction will just abort. If you are horribly unlucky, disaster strikes while DTC is already issuing commit commands (Phase 2). In that case, the transaction may (and will) just hang forever until either the network connection is restored or some operator throws in the hammer and resolves the transaction manually (and may deliberately cause inconsistency). Since all this might happen asynchronously, none of your code might ever end up knowing that the transaction failed and the (transient) transaction results just end up being quietly rolled back in the background.

Before you get all nervous about the asynchronous case, let’s look at a strategy to deal with the scenario from the client side, assuming that the transaction is resolved synchronously. We will take into account that there are transaction failures that can be fixed by rerunning the transaction.

public void RunTx(string argument) {     TransactionStatus status;     bool done = false;     while( !done )     {         ServiceConfig cfg = new ServiceConfig();         cfg.Transaction = TransactionOption.Required;         ServiceDomain.Enter(cfg);         try         {             DoesStuff doesStuff = new DoesStuff();             doesStuff.DoStuff(argument);             ContextUtil.SetComplete();             done = true;         }         catch (RepeatableOperationException re)         {             ContextUtil.SetAbort();         }         catch         {             ContextUtil.SetAbort();             throw;         }         finally         {             status = ServiceDomain.Leave();         }     }     return status; }

If a component throws an exception indicating that running the transaction again might yield a success, we’ll catch that condition, abort the local transaction and since we don’t set done to true, we will get back to the top of the loop and rerun the transaction. If we find another exception, we will abort the transaction and re-throw the exception we caught. The loop might look a bit awkward, but it does the job of re-running the transaction whenever that’s desired.  So all you need to do is to use that sort of a loop everywhere in your code and you are all set. Well … almost.

Assume that the argument we pass into RunTx() is something as complex and valuable (!) as a purchase order that you just got into your system. A purchase order is as good as cash. The longer you keep that purchase order around in memory, the greater is the risk that your local process (or machine) will fail catastrophically for some arbitrary reason. So if you need to re-run the transaction one or two times because the backend system is struggling (again, this all technology for the worst imaginable case) you are making yourself increasingly vulnerable to a local failure until the transaction finally succeeds. To reduce that risk, it therefore makes sense to add another step here.

Before we even start the transaction that does the processing, we grab the input data and guard it using a transactional resource manager – such as MSMQ. If we throw the input data into a private, transactional MSMQ queue as the first step and before we do anything else, the data is safely guarded on disk. If the machine fails once the data is in the queue, the data – encapsulated in a message – will still be available once the machine comes back up. Using a message queue listener, the messages in the queue are then each read from within individual transactions and the data is submitted into the processing component. Because the messages are simply put back into the queue if the transaction fails, re-running the transaction is practically automatic once the message queue listener restarts. Which leaves a problem: There are failing transactions that we want to repeat and there are failing transactions of which we know that they will not succeed no matter how often we try them. If a transactional component throws an exception because it has a bug, the transaction will abort and no number of re-runs will have that transaction succeed. Messages (data) that cause this sort of behavior are “poisonous” and we need to sort them out. Stay tuned; the message queue listener that takes care of all of that is up next.

See Part 1

Before we can do anything about deadlocks or deal with similar troubles, we first need to be able to tell that we indeed have a deadlock situation. Finding this out is a matter of knowing the respective error codes that your database gives you and a mechanism to bubble that information up to some code that will handle the situation. So before we can think about and write the handling logic for failed/failing but safely repeatable transactions, we need to build a few little things. The first thing we’ll need is an exception class that will wrap the original exception indicating the reason for the transaction failure. The new exception class’s identity will later serve to filter out exceptions in a “catch” statement and take the appropriate actions.

using System; using System.Runtime.Serialization; namespace newtelligence.EnterpriseTools.Data {    [Serializable]    publicclass RepeatableOperationException : Exception    {        public RepeatableOperationException():base()        {        }        public RepeatableOperationException(Exception innerException)            :base(null,innerException)        {        }        public RepeatableOperationException(string message, Exception innerException)            :base(message,innerException)        {        }        public RepeatableOperationException(string message):base(message)        {        }         public RepeatableOperationException(           SerializationInfo serializationInfo,           StreamingContext streamingContext)             :base(serializationInfo,streamingContext)         {         }         publicoverridevoid GetObjectData(            System.Runtime.Serialization.SerializationInfo info,            System.Runtime.Serialization.StreamingContext context)         {             base.GetObjectData (info, context);         }    } }

Having an exception wrapper with the desired semantics, we know need to be able to figure out when to replace the original exception with this wrapper and re-throw it up on the call stack. The idea is that whenever you execute a database operation – or, more generally, any operation that might be repeatable on failure – you will catch the resulting exception and run it through a factory, which will analyze the exception and wrap it with the RepeatableOperationException if the issue at hand can be resolved by re-running the transaction. The (still a little naïve) code below illustrates how to such a factory in the application code. Later we will flesh out the catch block a little more, since we will lose the original call stack if we end up re-throwing the original exception like shown here:

Try {    dbConnection.Open();    sprocUpdateAndQueryStuff.Parameters["@StuffArgument"].Value = argument;    result = this.GetResultFromReader( sprocUpdateAndQueryStuff.ExecuteReader() ); } catch( Exception exception ) {    throw RepeatableOperationExceptionMapper.MapException( exception );                            } finally {    dbConnection.Close(); }

The factory class itself is rather simple in structure, but a bit tricky to put together, because you have to know the right error codes for all resource managers you will ever run into. In the example below I put in what I believe to be the appropriate codes for SQL Server and Oracle (corrections are welcome) and left the ODBC and OLE DB factories (for which would have to inspect the driver type and the respective driver-specific error codes) blank. The factory will check out the exception data type and delegate mapping to a private method that is specialized for a specific managed provider.

using System; using System.Data.SqlClient; using System.Data.OleDb; using System.Data.Odbc; using System.Data.OracleClient; namespace newtelligence.EnterpriseTools.Data {    publicclass RepeatableOperationExceptionMapper    {         ///<summary>         /// Maps the exception to a Repeatable exception, if the error code         /// indicates that the transaction is repeatable.         ///</summary>         ///<param name="sqlException"></param>         ///<returns></returns>         privatestatic Exception MapSqlException( SqlException sqlException )         {             switch ( sqlException.Number )             {                 case -2: /* Client Timeout */                 case 701: /* Out of Memory */                 case 1204: /* Lock Issue */                 case 1205: /* Deadlock Victim */                 case 1222: /* Lock Request Timeout */                 case 8645: /* Timeout waiting for memory resource */                 case 8651: /* Low memory condition */                     returnnew RepeatableOperationException(sqlException);                 default:                     return sqlException;             }         }         privatestatic Exception MapOleDbException( OleDbException oledbException )         {             switch ( oledbException.ErrorCode )             {                 default:                     return oledbException;             }         }         privatestatic Exception MapOdbcException( OdbcException odbcException )         {             return odbcException;                    }         privatestatic Exception MapOracleException( OracleException oracleException )         {             switch ( oracleException.Code )             {                 case 104:  /* ORA-00104: Deadlock detected; all public servers blocked waiting for resources */                 case 1013: /* ORA-01013: User requested cancel of current operation */                 case 2087: /* ORA-02087: Object locked by another process in same transaction */                 case 60:   /* ORA-00060: Deadlock detected while waiting for resource */                     returnnew RepeatableOperationException( oracleException );                 default:                     return oracleException;             }         }         publicstatic Exception MapException( Exception exception )         {             if ( exception is SqlException )             {                 return MapSqlException( exception as SqlException );             }             elseif ( exception is OleDbException )             {                 return MapOleDbException( exception as OleDbException );             }             elseif (exception is OdbcException )             {                 return MapOdbcException( exception as OdbcException );             }             elseif (exception is OracleException )             {                 return MapOracleException( exception as OracleException );             }             else             {                 return exception;             }         }    } }

With that little framework of two classes, we can now selectively throw exceptions that convey whether a failed/failing transaction is worth repeating. Next step: How do we do actually run such repeats and make sure we neither lose data nor make the user unhappy in the process? Stay tuned.

Deadlocks and other locking conflicts that cause transactional database operations to fail are things that puzzle many application developers. Sure, proper database design and careful implementation of database access (and appropriate support by the database engine) should take care of that problem, but it cannot do so in all cases. Sometimes, especially under stress and other situations with high lock contention, a database just has not much of a choice but picking at least one of the transactions competing for the same locks as the victim in resolving the deadlock situation and then aborts the chosen transaction. Generally speaking, transactions that abort and roll back are a good thing, because this behavior guarantees data integrity. In the end, we use transaction technology for those cases where data integrity is at risk. What’s interesting is that even though transactions are a technology that is explicitly about things going wrong, the strategy for dealing with failing transaction is often not much more than to bubble the problem up to the user and say “We apologize for the inconvenience. Please press OK”.

The appropriate strategy for handling a deadlock or some other recoverable reason for a transaction abort on the application level is to back out of the entire operation and to retry the transaction. Retrying is a gamble that the next time the transaction runs, it won’t run into the same deadlock situation again or that it will at least come out victorious when the database picks its victims. Eventually, it’ll work. Even if it takes a few attempts. That’s the idea. It’s quite simple.

What is not really all that simple is the implementation. Whenever you are using transactions, you must make your code aware that such “good errors” may occur at any time. Wrapping your transactional ODBC/OLEDB/ADO/ADO.NET code or calls to transactional Enterprise Services or COM+ components with a try/catch block, writing errors to log-files and showing message boxes to users just isn’t the right thing to do. The right thing is to simply do the same batch of work again and until it succeeds.

The problem that some developers seem to have with “just retry” is that it’s not so clear what should be retried. It’s a problem of finding and defining the proper transaction scope. Especially when user interaction is in the picture, things easily get very confusing. If a user has filled in a form on a web page or some dialog window and all of his/her input is complete and correct, should the user be bothered with a message that the update transaction failed due to a locking issue? Certainly not. Should the user know when the transaction fails because the database is currently unavailable? Maybe, but not necessarily. Should the user be made aware that the application he/she is using is for some sudden reason incompatible with the database schema of the backend database? Maybe, but what does Joe in the sales department do with that valuable piece of information?

If stuff fails, should we just forget about Joe’s input and tell him to come back when the system is happier to serve him? So, in other words, do we have Joe retry the job? That’s easy to program, but that sort of strategy doesn’t really make Joe happy, does it?

So what’s the right thing to do? One part of the solution is a proper separation between the things the user (or a program) does and the things that the transaction does. This will give us two layers and “a job” that can be handed down from the presentation layer down to the “transaction layer”. Once this separation is in place, we can come up with a mechanism that will run those jobs in transactions and will automate how and when transactions are to be retried. Transactional MSMQ queues turn out to be a brilliant tool to make this very easy to implement. More tomorrow. Stay tuned.