7

データベースに保存する必要があるデータを生成するシミュレーションがあります。

ParallelLoopResult res = Parallel.For(0, 1000000, options, (r, state) =>
{
    ComplexDataSet cds = GenerateData(r);

    SaveDataToDatabase(cds);

});

シミュレーションでは大量のデータが生成されるため、最初にデータを生成してからデータベースに保存するのは現実的ではなく (最大 1 GB のデータ)、データベースに 1 つずつ保存するのも意味がありません。 （実用的であるには小さすぎるトランザクション）。それらを制御されたサイズのバッチ挿入としてデータベースに挿入したい（1回のコミットで100など）。

ただし、並列コンピューティングに関する私の知識は理論的なものではないと思います。私はこれを思いつきました（ご覧のとおり、非常に欠陥があります）：

DataBuffer buffer = new DataBuffer(...);

ParallelLoopResult res = Parallel.For(0, 10000000, options, (r, state) =>
{
    ComplexDataSet cds = GenerateData(r);

    buffer.SaveDataToBuffer(cds, i == r - 1);

});

public class DataBuffer
{
    int count = 0;
    int limit = 100

    object _locker = new object();

    ConcurrentQueue<ConcurrentBag<ComplexDataSet>> ComplexDataBagQueue{ get; set; }

    public void SaveDataToBuffer(ComplexDataSet data, bool isfinalcycle)
    {
            lock (_locker)
            {
                if(count >= limit)
                {
                    ConcurrentBag<ComplexDataSet> dequeueRef;
                    if(ComplexDataBagQueue.TryDequeue(out dequeueRef))
                    {
                        Commit(dequeueRef);
                    }

                    _lastItemRef = new ConcurrentBag<ComplexDataSet>{data};
                    ComplexDataSetsQueue.Enqueue(_lastItemRef);
                    count = 1;
                }
                else
                {
                    // First time
                    if(_lastItemRef == null)
                    {
                        _lastItemRef = new ConcurrentBag<ComplexDataSet>{data};
                        ComplexDataSetsQueue.Enqueue(_lastItemRef);
                        count = 1;
                    }
                    // If buffer isn't full
                    else
                    {
                        _lastItemRef.Add(data);
                        count++;
                    }
                }

                if(isfinalcycle)
                {
                        // Commit everything that hasn't been committed yet
                        ConcurrentBag<ComplexDataSet> dequeueRef;    
                    while (ComplexDataSetsQueue.TryDequeue(out dequeueRef))
                    {
                        Commit(dequeueRef);
                    }
                }
            }
    }

    public void Commit(ConcurrentBag<ComplexDataSet> data)
    {
        // Commit data to database..should this be somehow in another thread or something ?
    }
}

ご覧のとおり、キューを使用してバッファーを作成し、コミットするタイミングを手動で決定しています。ただし、これは私の問題に対する解決策としてはあまり効果的ではないと強く感じています。まず、ロックを正しく行っているかどうかわかりません。第二に、これが完全にスレッドセーフであるかどうか（またはまったく）もわかりません。

ちょっと見て、どうすればいいのかコメントしてもらえますか？または、これを行うための完全に優れた方法がある場合 (ある種の生産者と消費者の手法または何かを使用) ?

感謝と最高の願い、D.

4

3 に答える 3

4

ロックや高価な同時実行セーフ データ構造を使用する必要はありません。データはすべて独立しているため、ロックと共有を導入しても、パフォーマンスとスケーラビリティが損なわれるだけです。

Parallel.Forには、スレッドごとのデータを指定できるオーバーロードがあります。これには、プライベート キューとプライベート データベース接続を格納できます。

またParallel.For、範囲を内部的に小さなチャンクに分割します。大きな範囲を渡すことは完全に効率的であるため、そこを変更する必要はありません。

Parallel.For(0, 10000000, () => new ThreadState(),
    (i, loopstate, threadstate) =>
{
    ComplexDataSet data = GenerateData(i);

    threadstate.Add(data);

    return threadstate;
}, threadstate => threadstate.Dispose());

sealed class ThreadState : IDisposable
{
    readonly IDisposable db;
    readonly Queue<ComplexDataSet> queue = new Queue<ComplexDataSet>();

    public ThreadState()
    {
        // initialize db with a private MongoDb connection.
    }

    public void Add(ComplexDataSet cds)
    {
        queue.Enqueue(cds);

        if(queue.Count == 100)
        {
            Commit();
        }
    }

    void Commit()
    {
        db.Write(queue);
        queue.Clear();
    }

    public void Dispose()
    {
        try
        {
            if(queue.Count > 0)
            {
                Commit();
            }
        }
        finally
        {
            db.Dispose();
        }
    }
}

現在、MongoDb は現在、真の同時挿入をサポートしていません。サーバーに高価なロックがいくつか保持されているため、並列コミットは (あったとしても) 速度を大幅に向上させません。彼らは将来これを修正したいと考えているので、いつの日か無料でスピードアップできるかもしれません。

保持されるデータベース接続の数を制限する必要がある場合は、プロデューサー/コンシューマーのセットアップが適切な代替手段です。BlockingCollectionキューを使用して、ロックを使用せずにこれを効率的に行うことができます。

// Specify a maximum of 1000 items in the collection so that we don't
// run out of memory if we get data faster than we can commit it.
// Add() will wait if it is full.

BlockingCollection<ComplexDataSet> commits =
    new BlockingCollection<ComplexDataSet>(1000);

Task consumer = Task.Factory.StartNew(() =>
    {
        // This is the consumer.  It processes the
        // "commits" queue until it signals completion.

        while(!commits.IsCompleted)
        {
            ComplexDataSet cds;

            // Timeout of -1 will wait for an item or IsCompleted == true.

            if(commits.TryTake(out cds, -1))
            {
                // Got at least one item, write it.
                db.Write(cds);

                // Continue dequeuing until the queue is empty, where it will
                // timeout instantly and return false, or until we've dequeued
                // 100 items.

                for(int i = 1; i < 100 && commits.TryTake(out cds, 0); ++i)
                {
                    db.Write(cds);
                }

                // Now that we're waiting for more items or have dequeued 100
                // of them, commit.  More can be continue to be added to the
                // queue by other threads while this commit is processing.

                db.Commit();
            }
        }
    }, TaskCreationOptions.LongRunning);

try
{
    // This is the producer.

    Parallel.For(0, 1000000, i =>
        {
            ComplexDataSet data = GenerateData(i);
            commits.Add(data);
        });
}
finally // put in a finally to ensure the task closes down.
{
    commits.CompleteAdding(); // set commits.IsFinished = true.
    consumer.Wait(); // wait for task to finish committing all the items.
}
于 2011-08-04T09:59:57.303 に答える
2

あなたの例では、10 000 000 個の作業パッケージがあります。これはそれぞれ、スレッドに分散する必要があります。本当に多数の CPU コアを持っていないと仮定すると、これは最適ではありません。また、すべてのbuffer.SaveDataToBuffer呼び出しで (ロックを使用して) スレッドを同期する必要があります。さらに、変数は年表ビューで必ずしも 1 ずつ増加しないことに注意するr必要があります (例: Thread1 は 1、2、3 で r を実行し、Thread2 は 4、5、6 で r を実行します) r。 1,4,2,5,3,6 SaveDataToBuffer(およそ))。

作業のパッケージを大きくしてから、各パッケージを一度にコミットします。これには、すべてを頻繁にロック/同期する必要がないという利点もあります。

次に例を示します。

int total = 10000000;
int step = 1000;

Parallel.For(0, total / step, (r, state) =>
{
    int start = r * start;
    int end = start + step;

    ComplexDataSet[] result = new ComplexDataSet[step];

    for (int i = start; i < end; i++)
    {
        result[i - start] = GenerateData(i);
    }

    Commit(result);
});

この例では、作業全体が 10,000 個のパッケージ (並列で実行される) に分割され、すべてのパッケージが 1000 個のデータ項目を生成してデータベースにコミットします。

このソリューションでは、Commit賢明に設計されていない場合、メソッドがボトルネックになる可能性があります。ロックを使用せずにスレッドセーフにするのが最善です。これは、同期が必要なスレッド間で共通のオブジェクトを使用しない場合に実現できます。

たとえば、すべてのCommit()呼び出しのコンテキストで独自の SQL 接続を作成することを意味する SQL サーバー バックエンドの場合:

private void Commit(ComplexDataSet[] data)
{
    using (var connection = new SqlConnection("connection string..."))
    {
        connection.Open();

        // insert your data here...
    }
}
于 2011-08-03T18:55:44.083 に答える
1

ソフトウェアの複雑さを増す代わりに、単純化を検討してください。コードは次の3つの部分にリファクタリングできます。

  1. エンキューするワーカー

    これは、並列でのGenerateDataの並行実行です。そのため、大量の計算が行われ、 ComplexDataSetが生成されます。

  2. 実際のキュー

    [1]からの結果を格納する並行キュー-非常に多くのComplexDataSet。ここでは、ComplexDataSetの1つのインスタンスは実際にはリソースを消費せず、かなり軽量であると想定しました。キューが並行している限り、並列の「挿入」と「削除」をサポートします。

  3. デキューするワーカー

    ComplexDataSetの1つのインスタンスを処理キュー[2]から取得し、それを並行バッグ（または他のストレージ）に配置するコード。バッグにブロックするアイテムがN個ある場合は、デキューを停止し、バッグの内容をデータベースにフラッシュしてクリアします。最後に、ブロックを解除してデキューを再開します。

ここにいくつかのメタコードがあります（それはまだコンパイルされますが、改善が必要です）

[1]

// [1] - Class is responsible for generating complex data sets and 
// adding them to processing queue
class EnqueueWorker
{
    //generate data and add to queue
    internal void ParrallelEnqueue(ConcurrentQueue<ComplexDataSet> resultQueue)
    {
        Parallel.For(1, 10000, (i) =>
        {
            ComplexDataSet cds = GenerateData(i);
            resultQueue.Enqueue(cds);

        });
    }

    //generate data
    ComplexDataSet GenerateData(int i)
    {
        return new ComplexDataSet();
    }
}

[3]

//[3] This guy takes sets from the processing queue and flush results when 
// N items have been generated
class DequeueWorker
{
    //buffer that holds processed dequeued data
    private static ConcurrentBag<ComplexDataSet> buffer;

    //lock to flush the data to the db once in a while
    private static object syncRoot = new object();

    //take item from processing queue and add it to internal buffer storage
    //once buffer is full - flush it to the database
    internal void ParrallelDequeue(ConcurrentQueue<ComplexDataSet> resultQueue)
    {
        buffer = new ConcurrentBag<ComplexDataSet>();
        int N = 100;

        Parallel.For(1, 10000, (i) =>
        {
            //try dequeue
            ComplexDataSet cds = null;

            var spinWait = new SpinWait();

            while (cds == null)
            {
                resultQueue.TryDequeue(out cds);
                spinWait.SpinOnce();
            }

            //add to buffer
            buffer.Add(cds);

            //flush to database if needed
            if (buffer.Count == N)
            {
                lock (syncRoot)
                {
                    IEnumerable<ComplexDataSet> data = buffer.ToArray();

                    // flush data to database

                    buffer = new ConcurrentBag<ComplexDataSet>();
                }
            }

        });
    }        
}

[2]と使用法

class ComplexDataSet { }

class Program
{
    //processing queueu - [2]
    private static ConcurrentQueue<ComplexDataSet> processingQueue;

    static void Main(string[] args)
    {
        // create new processing queue - single instance for whole app
        processingQueue = new ConcurrentQueue<ComplexDataSet>();

        //enqueue worker
        Task enqueueTask = Task.Factory.StartNew(() =>
            {
                EnqueueWorker enqueueWorker = new EnqueueWorker();
                enqueueWorker.ParrallelEnqueue(processingQueue);
            });

        //dequeue worker
        Task dequeueTask = Task.Factory.StartNew(() =>
        {
            DequeueWorker dequeueWorker = new DequeueWorker();
            dequeueWorker.ParrallelDequeue(processingQueue);
        });            
    }
}
于 2011-08-03T18:15:44.877 に答える