Oracle connection issue while using tasks and blocking collection

Question

I have some tasks (nWorkers = 3):

var taskFactory = new TaskFactory(cancellationTokenSource.Token,
    TaskCreationOptions.LongRunning, TaskContinuationOptions.LongRunning,
    TaskScheduler.Default);

for (int i = 0; i < nWorkers; i++)
{
    var task = taskFactory.StartNew(() => this.WorkerMethod(parserItems,
        cancellationTokenSource));
    tasks[i] = task;
}    

And the following method called by the tasks:

protected override void WorkerMethod(BlockingCollection<ParserItem> parserItems,
    CancellationTokenSource cancellationTokenSource)
{
//...log-1...
using (var connection = new OracleConnection(connectionString))
{
    OracleTransaction transaction = null;
    try
    {
        cancellationTokenSource.Token.ThrowIfCancellationRequested();
        connection.Open();
        //...log-2...
        transaction = connection.BeginTransaction();
        //...log-3...
        using (var cmd = connection.CreateCommand())
        {                   
            foreach (var parserItem in parserItems.GetConsumingEnumerable(
                cancellationTokenSource.Token))
            {
                cancellationTokenSource.Token.ThrowIfCancellationRequested();
                try
                {
                    foreach (var statement in this.ProcessRecord(parserItem))
                    {                               
                        cmd.CommandText = statement;
                        try
                        {                                   
                            cmd.ExecuteNonQuery();                                  
                        }
                        catch (OracleException ex)
                        {
                            //...log-4...
                            if (!this.acceptedErrorCodes.Contains(ex.Number))
                            {
                                throw;
                            }
                        }
                    }
                }
                catch (FormatException ex)
                {
                    log.Warn(ex.Message);
                }
            }
            if (!cancellationTokenSource.Token.IsCancellationRequested)
            {                       
                transaction.Commit();                       
            }
            else
            {                       
                throw new Exception("DBComponent has been canceled");
            }
        }
    }
    catch (Exception ex)
    {
        //...log-5...
        cancellationTokenSource.Cancel();
        if (transaction != null)
        {
            try
            {
                transaction.Rollback();
                //...log-6...
            }
            catch (Exception rollbackException)
            {
                //...log-7...
            }
        }
        throw;
    }
    finally
    {
        if (transaction != null)
        {
            transaction.Dispose();
        }
        connection.Close();
        //...log-8...
    }
}
//...log-9...   
}

There is a producer of ParserItem objects and these are the consumers. Normally it works fine, there are sometimes that there is an Oracle connection timeout, but in these cases I can see the exception message and everything works as designed.

But sometimes the process get stuck. When it gets stuck, in the log file I can see log-1 message and after that (more or less 15 seconds later) I see log-8 message, but what is driving me nuts is why i cannot see neither the exception message log-5 nor the log-9 message. Since the cancellationTokenSource.Cancel() method is never called, the producer of items for the bounded collection is stuck until a timeout two hours later.

It is compiled for NET Framework 4 and I'm using Oracle.ManagedDataAccess libraries for the Oracle connection.

Any help would be greatly appreciated.

Answer 1

You should never dispose a transaction or connection when you use using scope. Second, you should rarely rely on exception based programming style. Your code rewritten below:

using (var connection = new OracleConnection(connectionString))
{
    using (var transaction = connection.BeginTransaction())
    {
        connection.Open();
        //...log-2...
        using (var cmd = connection.CreateCommand())
        {
            foreach (var parserItem in parserItems.GetConsumingEnumerable(cancellationTokenSource.Token))
            {
                if (!cancellationTokenSource.IsCancellationRequested)
                {
                    try
                    {
                        foreach (var statement in ProcessRecord(parserItem))
                        {
                            cmd.CommandText = statement;
                            try
                            {
                                cmd.ExecuteNonQuery();
                            }
                            catch (OracleException ex)
                            {
                                //...log-4...
                                if (!acceptedErrorCodes.Contains(ex.ErrorCode))
                                {
                                    log.Warn(ex.Message);
                                }
                            }
                        }
                    }
                    catch (FormatException ex)
                    {
                        log.Warn(ex.Message);
                    }
                }
            }
            if (!cancellationTokenSource.IsCancellationRequested)
            {
                transaction.Commit();
            }
            else
            {
                transaction.Rollback();
                throw new Exception("DBComponent has been canceled");
            }
        }
    }
}
//...log-9... 

Let me know if this helps.