parallel.forectach Spawning طريقة الكثير من المواضيع

Question

المشكلة

على الرغم من أن التعليمات البرمجية التي سأتحدث فيها هنا كتبت في F #، فإنها تعتمد على إطار .NET 4، وليس على وجه التحديد اعتمادا على أي خصوصية من F # (على الأقل يبدو ذلك!).

لدي بعض البيانات على القرص الذي يجب عليك تحديثه من الشبكة، أو حفظ أحدث إصدار إلى القرص:

type MyData =
    { field1 : int;
      field2 : float }

type MyDataGroup =
    { Data : MyData[];
      Id : int }

// load : int -> MyDataGroup
let load dataId =
    let data = ... // reads from disk
    { Data = data;
      Id = dataId }

// update : MyDataGroup -> MyDataGroup
let update dg =
    let newData = ... // reads from the network and process
                      // newData : MyData[]

    { dg with Data = dg.Data
                     |> Seq.ofArray
                     |> Seq.append newData
                     |> processDataSomehow
                     |> Seq.toArray }

// save : MyDataGroup -> unit
let save dg = ... // writes to the disk

let loadAndSaveAndUpdate = load >> update >> save

المشكلة هي ذلك ل loadAndSaveAndUpdate كل بياناتي، أود تنفيذ الوظيفة عديدة مرات:

{1 .. 5000} |> loadAndSaveAndUpdate

كل خطوة سوف تفعل

• بعض القرص IO،
• بعض البيانات الطفرة،
• بعض الشبكة IO (مع إمكانية الكثير من الكمون)،
• المزيد من البيانات الطحن،
• وبعض القرص IO.

لن يكون من الجيد أن يتم ذلك بالتوازي، إلى حد ما؟ لسوء الحظ، لا توجد أي من وظائف القراءة والتحليل الخاصة بي هي "Async-Workflows-ready".

الحلول الأولى (ليست جيدة جدا) وصلت

مهام

أول شيء قمت به هو إعداد Task[] وبدء كل منهم:

let createTask id = new Task(fun _ -> loadAndUpdateAndSave id)
let tasks = {1 .. 5000}
            |> Seq.map createTask
            |> Seq.toArray

tasks |> Array.iter (fun x -> x.Start())
Task.WaitAll(tasks)

ثم اضغط على CTRL + ESC فقط لمعرفة عدد المواضيع التي يستخدمها. 15، 17، ...، 35، ...، 170، ... حتى قتل الطلب! كان هناك خطأ ما يحدث.

موازي

فعلت تقريبا نفس الشيء ولكن باستخدام Parallel.ForEach(...) وكانت النتائج هي نفسها: الكثير والكثير والكثير من الخيوط.

حل يعمل ... نوع من

ثم قررت أن أبدأ فقط n الخيوط، Task.WaitAll(of them), ، إذن أخرى n, ، حتى لم تكن هناك المزيد من المهام المتاحة.

هذا يعمل، ولكن المشكلة هي أنه عندما انتهى من المعالجة، قل، n-1 المهام، سوف تنتظر، انتظر، انتظر المهمة الأخيرة لعنة التي تصر على الحظر بسبب الكثير من الكمون الشبكة. هذا ليس جيدا!

وبالتالي، كيف تهاجم هذه المشكلةب سأكون ممتنا لعرض حلول مختلفة، إما إما سير عمل ASYNC (وفي هذه الحالة كيفية تكييف وظائف غير ASYNC)، والملحقات الموازية، وأنماط موازية غريبة، إلخ.

شكرا.

Answer 1

هل أنت متأكد من أن مهامك الفردية تتكمل في الوقت المناسب؟ أعتقد أن كلاهما Parallel.ForEach و ال Task فئة تستخدم بالفعل threadpool .NET. يجب أن تكون المهام عموما عناصر عمل قصيرة الأجل، وفي هذه الحالة سوف تفرخ Threadpool فقط عددا صغيرا من المواضيع الفعلية، ولكن إذا كانت مهامك لا تحرز تقدما وهناك مهام أخرى في قائمة الانتظار، فإن عدد المواضيع المستخدمة ستزيد بشكل مطرد الحد الأقصى (الذي افتراضيا هو 250 / المعالج في .NET 2.0 SP1، ولكن يختلف في إطار إصدارات مختلفة من الإطار). تجدر الإشارة أيضا إلى أن إنشاء مؤشر الترابط الجديد (على الأقل في .NET 2.0 SP1) خلق مؤشر ترابط جديد محنك إلى 2 خيوط جديدة في الثانية الواحدة، لذا فإن الحصول على عدد المواضيع التي تراها تشير إلى أن المهام لا تستكمل في كمية قصيرة من الوقت (لذلك قد لا يكون دقيقا تماما لتعديل اللوم Parallel.ForEach).

أعتقد أن اقتراح براين للاستخدام async سير العمل هو فكرة جيدة، خاصة إذا كان مصدر المهام الطويلة العمر IO، منذ ذلك الحين async سيعود المواضيع الخاصة بك إلى ThreadPool حتى يكمل IO. خيار آخر هو قبول ببساطة أن مهامك ليست ملاءمة بسرعة والسماح بتخزين العديد من المواضيع (والتي يمكن السيطرة عليها إلى حد ما باستخدام System.Threading.ThreadPool.SetMaxThreads) - اعتمادا على الموقف الخاص بك قد لا يكون مشكلة كبيرة تستخدم الكثير من المواضيع.

Answer 2

التوازن. maxdegreeofparallism. يحد من عدد العمليات المتزامنة التي تديرها مكالمات الأسلوب الموازي

Answer 3

باستخدام "ASYNC's" ستمكنك من القيام بالعمل I / O مرتبط دون حرق خيوط بينما تتصل بمكالمات الإدخال / الإخراج المختلفة "في Sea"، حتى يكون ذلك اقتراحي الأول. يجب أن تكون واضحة لتحويل الكود إلى ASYNC، وعادة ما على طول خطوط

• التفاف كل وظيفة وظيفة في async{...}, ، يضيف return عند الضرورة
• قم بإنشاء إصدارات ASYNC من أيئبي I / O التي ليست موجودة بالفعل في المكتبة عبر Async.FromBeginEnd
• تبديل مكالمات النموذج let r = Foo() ل let! r = AsyncFoo()
• يستخدم Async.Parallel لتحويل كائنات ASYNC 5000 في ASYNC واحدة تعمل بالتوازي

هناك العديد من البرامج التعليمية للقيام بذلك؛ واحد مثل هذا البث الشبكي هنا.

Answer 4

يمكنك دائما استخدام ThreadPool.

http://msdn.microsoft.com/en-us/library/system.threading.threadpool.aspx.

في الأساس:

1. إنشاء تجمع موضوع
2. اضبط عدد أقصى من المواضيع
3. قائمة انتظار جميع المهام باستخدام QueueUserWorkItem(WaitCallback)