Is filling different DataTables of the same DataSet a thread-safe operation?

user_4499267 0
Viewed 67

I have many methods filling different DataTables in the same DataSet:

public override void FillMethod1(MyDataSet ds)
{
    try
    {
        using (SqlDataAdapter adapter = new SqlDataAdapter(query, connectionObj))
        {
            using (SqlCommandBuilder adapterSCB = new SqlCommandBuilder(adapter))
            {
                adapter.Fill(ds.MyTable);
            }
        }
    }
    catch (Exception e)
    {
        //Exception handling
    }
}

public override void FillMethod2(MyDataSet ds)
{
    try
    {
        using (SqlDataAdapter adapter = new SqlDataAdapter(query, connectionObj))
        {
            using (SqlCommandBuilder adapterSCB = new SqlCommandBuilder(adapter))
            {
                adapter.Fill(ds.MyTable2);
            }
        }
    }
    catch (Exception e)
    {
        //Exception handling
    }
}

[....]

using(MyDataSet ds = new MyDataSet())
{
    FillMethod1(ds);
    FillMethod2(ds);
    [...]
}

And I'd like to parallelize these operation using Task:

using(MyDataSet ds = new MyDataSet())
{
    Task fill1 = Task.Run(() => FillMethod1(ds));
    Task fill2 = Task.Run(() => FillMethod2(ds));
    [...]

    Task.WaitAll(fill1, fill2, [...]);
}

After some research I found that DataSet is not thread-safe, but is it safe when working on different DataTable?

dataset multithreading sql-server c#
1 Answers

I created a class handling parallel fills on the same DataSet. Each fill has its own DataSet and it will be merged into the original one at the end of the execution. To handle data table relations, it is possible to create a sequential list of fills that will be run in parallel with other fills. The merge method will wait for all operations to finish and merge results into the original DataSet (passed in the constructor)

public class FillParallelizer<TDataSet> : IDisposable
    where TDataSet : DataSet, new()
{
    private bool _seqFillStarted = false;
    private IList<QueryObject> _queryObjList = new List<QueryObject>();
    private Queue<FillOperation> _seqOperationBufferQueue = new Queue<FillOperation>();

    public DataSet MasterDS { get; }

    public FillParallelizer(DataSet masterDS)
    {
        MasterDS = masterDS ?? throw new ArgumentNullException(nameof(masterDS));
    }

    public FillParallelizer<TDataSet> ParallelFill(Action<TDataSet> methodAction)
    {
        if (methodAction == null) throw new ArgumentNullException(nameof(methodAction));

        QueryObject queryObj = new QueryObject(methodAction);
        _queryObjList.Add(queryObj);

        return this;
    }

    public void Merge()
    {
        var taskList = _queryObjList.Select(x => x.Task).ToArray();
        Task.WaitAll(taskList);

        foreach (var item in _queryObjList)
        {
            MasterDS.Merge(item.TempDS);
        }
    }

    public FillParallelizer<TDataSet> SequentialFill(Action<TDataSet> action)
    {
        if(!_seqFillStarted)
        {
            _seqFillStarted = true;
        }

        if (!_seqFillStarted) throw new ArgumentException("Sequential fill not started");

        _seqOperationBufferQueue.Enqueue(new FillOperation(action));
        return this;
    }

    public FillParallelizer<TDataSet> RunSequentialFill()
    {
        _seqFillStarted = false;
        QueryObject queryObj = new QueryObject(new Queue<FillOperation>(_seqOperationBufferQueue));
        _seqOperationBufferQueue.Clear();
        _queryObjList.Add(queryObj);

        return this;
    }

    public void Dispose()
    {
        if (_queryObjList == null || !_queryObjList.Any())
        {
            return;
        }

        foreach (var item in _queryObjList) item.Dispose();
    }

    class FillOperation
    {
        public Action<TDataSet> Action { get; }

        public FillOperation(Action<TDataSet> action)
        {
            Action = action ?? throw new ArgumentNullException(nameof(action));
        }

        public void Run(TDataSet tempDS)
        {
            if (tempDS == null) throw new ArgumentNullException(nameof(tempDS));
            Action(tempDS);
        }
    }

    class QueryObject : IDisposable
    {
        public TDataSet TempDS { get; }
        public Queue<FillOperation> FillOperationList { get; }
        public Task Task { get; }

        public QueryObject(Queue<FillOperation> fillOperationList)
        {
            TempDS = new TDataSet();
            FillOperationList = fillOperationList ?? throw new ArgumentNullException(nameof(fillOperationList));
            Task =
                Task
                    .Run
                    (() =>
                    {
                        foreach (FillOperation op in FillOperationList) op.Run(TempDS);
                    }
                    );
        }

        public QueryObject(Action<TDataSet> action)
            : this(new Queue<FillOperation>(new FillOperation[] { new FillOperation(action) }))
        {
        }

        public void Dispose()
        {
            Task.Dispose();
        }
    }
}

An example of usage:

using(MyDataSet myDS = new MyDataSet())
using (FillParallelizer<MyDataSet> parallelizer = new FillParallelizer<MyDataSet>(myDS))
{
    parallelizer
        .ParallelFill(ds => MyDAO.FillTable1(ds))
        .ParallelFill(ds => MyDAO.FillTable2(ds))

        .SequentialFill(ds => MyDAO.FillTable3(ds))
        .SequentialFill(ds => MyDAO.FillTable4(ds))
        .SequentialFill(ds => MyDAO.FillTable5(ds))
        .RunSequentialFill()

        .ParallelFill(ds => MyDAO.FillTable6(ds))
        .ParallelFill(ds => MyDAO.FillTable7(ds))

        .Merge();

    //All fills completed
}

Fill methods 3, 4, 5 are done sequentially but in parallel with other fills, to handle relations between tables

user_44992670
Related