🌦️ 🍄 🔑 निर्भरता वाले कार्यों को समानांतर करना

नमस्कार साथियों!

इस हफ्ते, हमने .NET में मैनिंग की पुस्तक कंजेंसी को दान कर दिया , जो जटिलता की एक महत्वाकांक्षी पुस्तक है:

लेखक ने कृपया माध्यम वेबसाइट पर 13 वें अध्याय का एक अंश पोस्ट किया है, जिसे हम प्रीमियर से बहुत पहले मूल्यांकन करने का प्रस्ताव देते हैं।
अच्छा पढ़ लो!

मान लें कि आपको एक उपकरण लिखने की आवश्यकता है जो आपको कई अतुल्यकालिक कार्यों को करने की अनुमति देता है, जिनमें से प्रत्येक की अपनी निर्भरताएं हैं जो संचालन के क्रम को प्रभावित करती हैं। ऐसी समस्याओं को सुसंगत और अनिवार्य निष्पादन के साथ हल किया जा सकता है, लेकिन यदि आप अधिकतम प्रदर्शन प्राप्त करना चाहते हैं, तो अनुक्रमिक संचालन आपके लिए काम नहीं करेगा। इसके बजाय, आपको कार्यों के समानांतर निष्पादन को व्यवस्थित करने की आवश्यकता है। कई प्रतिस्पर्धी समस्याओं की व्याख्या उनके इनपुट और आउटपुट डेटा के बीच निर्भरता के साथ परमाणु संचालन के स्थिर संग्रह के रूप में की जा सकती है। ऑपरेशन पूरा होने पर, इसका आउटपुट अन्य, आश्रित परिचालनों के लिए इनपुट के रूप में उपयोग किया जाता है। प्रदर्शन को अनुकूलित करने के लिए, कार्यों को उनकी निर्भरता के आधार पर असाइन किया जाना चाहिए, और एल्गोरिथ्म को कॉन्फ़िगर करना होगा ताकि निर्भर कार्य क्रमिक रूप से आवश्यक हो, और यथासंभव समानांतर हो।

आप एक पुन: प्रयोज्य घटक बनाना चाहते हैं जो समानांतर में कार्यों की एक श्रृंखला करता है, और यह सुनिश्चित करता है कि संचालन के क्रम को प्रभावित करने वाली सभी निर्भरता को ध्यान में रखा जाए। इस तरह के एक प्रोग्रामिंग मॉडल का निर्माण कैसे करें जो कि इस ऑपरेशन और अन्य लोगों के बीच निर्भरता के आधार पर कुशलता से, या तो समानांतर या क्रमिक रूप से किए गए संचालन के संग्रह के बुनियादी समानता को सुनिश्चित करेगा।

समाधान: हम F # से MailboxProcessor वर्ग का उपयोग करते हुए निर्भरता ग्राफ को लागू करते हैं और मानक कार्यों के रूप में तरीके प्रदान करते हैं ताकि वे C # से भस्म हो सकें

इस समाधान को ओरिएंटेड एसाइक्लिक ग्राफ (डीएजी) कहा जाता है और इसे अच्छी तरह से परिभाषित निर्भरताओं के साथ परमाणु समस्याओं के अनुक्रमों में संचालन को विभाजित करके एक ग्राफ बनाने के लिए डिज़ाइन किया गया है। इस मामले में, ग्राफ का एसाइक्लिक सार महत्वपूर्ण है, क्योंकि यह कार्यों के बीच गतिरोध की संभावना को समाप्त करता है, बशर्ते कि कार्य वास्तव में पूरी तरह से परमाणु हैं। ग्राफ सेट करते समय, कार्यों के बीच सभी निर्भरताओं को समझना महत्वपूर्ण है, विशेष रूप से अंतर्निहित निर्भरताएं जो गतिरोध या दौड़ की स्थिति पैदा कर सकती हैं। निम्नलिखित एक ग्राफ-जैसी डेटा संरचना का एक विशिष्ट उदाहरण है, जिसके साथ आप किसी सीमा पर संचालन के बीच बातचीत की योजना बनाते समय उत्पन्न होने वाली सीमाओं की कल्पना कर सकते हैं।

ग्राफ़ एक अत्यंत शक्तिशाली डेटा संरचना है, और कोई भी इसके आधार पर शक्तिशाली एल्गोरिदम लिख सकता है।

अंजीर। 1 एक ग्राफ किनारों द्वारा जुड़े कोने का एक संग्रह है। निर्देशित ग्राफ के इस प्रतिनिधित्व में, नोड 1 नोड 4 और 5 पर निर्भर करता है, नोड 2 नोड 5 पर निर्भर करता है, नोड 3 नोड 5 और 6 पर निर्भर करता है, आदि।

DAG संरचना निर्भरता के क्रम को ध्यान में रखते हुए कार्यों के समानांतर निष्पादन के लिए एक रणनीति के रूप में लागू होती है, जो उत्पादकता में सुधार करती है। ऐसे ग्राफ की संरचना F # से MailboxProcessor वर्ग का उपयोग करके निर्धारित की जा सकती है; इस वर्ग में, किनारों की निर्भरता के रूप में निष्पादन के लिए पंजीकृत कार्यों के लिए आंतरिक स्थिति संरक्षित है।

ओरिएंटेड एसाइक्लिक ग्राफ वैलिडेशन

किसी भी ग्राफ डेटा संरचना जैसे डीएजी के साथ काम करते समय, आपको किनारों के सही पंजीकरण का ध्यान रखना होगा। उदाहरण के लिए, चित्र 1 पर लौटना: यदि नोड 5 पर निर्भरता के साथ हमने नोड 2 पंजीकृत किया है तो क्या होता है, लेकिन नोड 5 मौजूद नहीं है? ऐसा भी हो सकता है कि कुछ किनारे एक-दूसरे पर निर्भर होते हैं, जिसके परिणामस्वरूप एक उन्मुख चक्र होता है। एक उन्मुख चक्र की उपस्थिति में, समानांतर में कुछ कार्यों को करना महत्वपूर्ण है; अन्यथा, कुछ कार्य हमेशा दूसरों के पूरा होने की प्रतीक्षा कर सकते हैं और एक गतिरोध उत्पन्न होगा।

समस्या को टोपोलॉजिकल सॉर्टिंग द्वारा हल किया जाता है: इसका मतलब है कि हम ग्राफ के सभी शीर्षों को आदेश दे सकते हैं ताकि कोई भी धार कम संख्या से उच्च संख्या के साथ शीर्ष से शिखर की ओर जाए। इसलिए, यदि कार्य A को कार्य B से पहले समाप्त करना चाहिए, और कार्य B से पहले, कार्य C, जो, बदले में, कार्य A से पहले पूरा होना चाहिए, एक परिपत्र लिंक दिखाई देता है, और सिस्टम आपको इस त्रुटि की सूचना देगा, अपवाद को फेंकते हुए। यदि क्रम प्रबंधन में एक उन्मुख चक्र उत्पन्न होता है, तो कोई समाधान नहीं है। इस तरह के एक चेक को "निर्देशित ग्राफ में एक चक्र ढूंढना" कहा जाता है। यदि एक निर्देशित ग्राफ वर्णित नियमों को संतुष्ट करता है, तो यह एक निर्देशित चक्रीय ग्राफ है जो समानांतर में कई कार्यों को चलाने के लिए पूरी तरह उपयुक्त है, जिसके बीच निर्भरताएं हैं।

सूची 2 का पूर्ण संस्करण, DAG सत्यापन कोड से युक्त है, जो ऑनलाइन स्रोत कोड में है।

निम्नलिखित सूची में, F # MailboxProccessor वर्ग का उपयोग एक DAG को लागू करने के लिए एक आदर्श उम्मीदवार के रूप में किया जाता है जो निर्भरता से संबंधित कार्यों को समानांतर में निष्पादित करने की अनुमति देता है। पहले, चलो एक लेबल यूनियन को परिभाषित करते हैं जिसके साथ हम कार्यों का प्रबंधन करेंगे और उनकी निर्भरता को पूरा करेंगे।

उनकी निर्भरता के अनुसार कार्यों के समन्वय के लिए 1 संदेश प्रकार और डेटा संरचना सूचीबद्ध करना

 type TaskMessage = // #A | AddTask of int * TaskInfo | QueueTask of TaskInfo | ExecuteTasks and TaskInfo = // #B { Context : System.Threading.ExecutionContext Edges : int array; Id : int; Task : Func<Task> EdgesLeft : int option; Start : DateTimeOffset option End : DateTimeOffset option }

#A आधार dagAgent को ParallelTasksDAG भेजता है, जो समन्वय कार्यों के लिए जिम्मेदार है

# बी पूरा करने के लिए प्रत्येक कार्य का विवरण लपेटता है

TaskMessage प्रकार TaskMessage प्रकार के बेस एजेंट को भेजे गए संदेश आवरणों का प्रतिनिधित्व करता है। इन संदेशों का उपयोग कार्यों के समन्वय और निर्भरता को सिंक्रनाइज़ करने के लिए किया जाता है। TaskInfo प्रकार में DAG के निष्पादन के दौरान पंजीकृत कार्यों का विवरण होता है और निर्भरता किनारों सहित ट्रैक करता है। निष्पादन का संदर्भ (https://msdn.microsoft.com/en-us/library/system.threading.executioncontext(v=vs.110).aspx) विलंबित निष्पादन में सूचना तक पहुँचने के उद्देश्य से पकड़ा जाता है, उदाहरण के लिए, ऐसी जानकारी: वर्तमान उपयोगकर्ता, निष्पादन के तार्किक धागे से संबंधित कोई भी राज्य, कोड तक सुरक्षित पहुंच के बारे में जानकारी आदि। घटना शुरू होने के बाद, प्रारंभ और समाप्ति समय प्रकाशित होते हैं।

निर्भरता संबंधित कार्यों को समानांतर करने के लिए एफ # पर 2 डीएजी एजेंट की सूची बनाना

 type ParallelTasksDAG() = let onTaskCompleted = new Event<TaskInfo>() // #A let dagAgent = new MailboxProcessor<TaskMessage>(fun inbox -> let rec loop (tasks : Dictionary<int, TaskInfo>) // #B (edges : Dictionary<int, int list>) = async { // #B let! msg = inbox.Receive() // #C match msg with | ExecuteTasks -> // #D let fromTo = new Dictionary<int, int list>() let ops = new Dictionary<int, TaskInfo>() // #E for KeyValue(key, value) in tasks do // #F let operation = { value with EdgesLeft = Some(value.Edges.Length) } for from in operation.Edges do let exists, lstDependencies = fromTo.TryGetValue(from) if not <| exists then fromTo.Add(from, [ operation.Id ]) else fromTo.[from] <- (operation.Id :: lstDependencies) ops.Add(key, operation) ops |> Seq.iter (fun kv -> // #F match kv.Value.EdgesLeft with | Some(n) when n = 0 -> inbox.Post(QueueTask(kv.Value)) | _ -> ()) return! loop ops fromTo | QueueTask(op) -> // #G Async.Start <| async { // #G let start = DateTimeOffset.Now match op.Context with // #H | null -> op.Task.Invoke() |> Async.AwaitATsk | ctx -> ExecutionContext.Run(ctx.CreateCopy(), // #I (fun op -> let opCtx = (op :?> TaskInfo) opCtx.Task.Invoke().ConfigureAwait(false)), taskInfo) let end' = DateTimeOffset.Now onTaskCompleted.Trigger { op with Start = Some(start) End = Some(end') } // #L let exists, deps = edges.TryGetValue(op.Id) if exists && deps.Length > 0 then let depOps = getDependentOperation deps tasks [] edges.Remove(op.Id) |> ignore depOps |> Seq.iter (fun nestedOp -> inbox.Post(QueueTask(nestedOp))) } return! loop tasks edges | AddTask(id, op) -> tasks.Add(id, op) // #M return! loop tasks edges } loop (new Dictionary<int, TaskInfo>(HashIdentity.Structural)) (new Dictionary<int, int list>(HashIdentity.Structural))) [<CLIEventAttribute>] member this.OnTaskCompleted = onTaskCompleted.Publish // #L member this.ExecuteTasks() = dagAgent.Post ExecuteTasks // #N member this.AddTask(id, task, [<ParamArray>] edges : int array) = let data = { Context = ExecutionContext.Capture() Edges = edges; Id = id; Task = task NumRemainingEdges = None; Start = None; End = None } dagAgent.Post(AddTask(id, data)) // #O

# onTaskCompletedEvent क्लास का एक उदाहरण, जो किसी कार्य को पूरा करने के लिए सूचित किया जाता है

#B कार्य रजिस्टर और उनकी निर्भरता पर नज़र रखने के लिए एजेंट की आंतरिक स्थिति। कलेक्शन परस्पर भिन्न होते हैं क्योंकि ParallelTasksDAG के निष्पादन के दौरान स्थिति बदल जाती है, और चूंकि उन्हें थ्रेड सुरक्षा विरासत में मिली है क्योंकि वे एजेंट में स्थित हैं

#C अतुल्यकालिक रूप से निष्पादन की प्रतीक्षा कर रहा है

#D एक संदेश जो ParallelTasksDAG लॉन्च करता है

# ई संग्रह चलाने के लिए एक कार्य के साथ एक नीरस वृद्धि सूचकांक पर मैप किया गया

#F कार्य की सूची पर प्रक्रिया की पुनरावृत्ति होती है, एक सामयिक संरचना बनाने के लिए अन्य कार्यों के साथ निर्भरता का विश्लेषण करना जिसमें कार्यों का क्रम प्रस्तुत किया गया है

#G संदेश आवरण एक कार्य को पंक्तिबद्ध करने के लिए, उसे निष्पादित करता है, और, अंततः कार्य समाप्त होने के बाद एक सक्रिय निर्भरता के रूप में एजेंट राज्य से इस कार्य को हटाने के लिए।

#H यदि एक्ज़ीक्यूशन कोनटेक्स्ट null , तो कार्य को वर्तमान संदर्भ में चलाएं, अन्यथा #I पर जाएं

# मैं कार्य को एक्सेप्टेड एक्सेप्शनियनटेक्स्ट में चलाएं

onTaskCompleted और कार्य के पूरा होने की सूचना देने के लिए onTaskCompleted घटना को प्रकाशित करें। घटना में कार्य के बारे में जानकारी है।

#M संदेश आवरण अपनी निर्भरता के अनुसार निष्पादित करने के लिए एक कार्य जोड़ने के लिए, यदि कोई हो

#N पंजीकृत कार्यों का निष्पादन शुरू करता है

#O DAG को निष्पादित करने के लिए किसी कार्य, उसकी निर्भरता और वर्तमान ExecutionContext संदर्भ में जोड़ना।

AddTask फ़ंक्शन का उद्देश्य कार्य को मनमाने ढंग से निर्भरता किनारों के साथ पंजीकृत करना है। यह फ़ंक्शन एक अद्वितीय आईडी, एक कार्य करने के लिए स्वीकार करता है, और कई किनारों को अन्य सभी पंजीकृत कार्यों की आईडी का प्रतिनिधित्व करता है जिन्हें इस कार्य को पूरा करने से पहले पूरा किया जाना चाहिए। यदि सरणी खाली है, तो इसका मतलब है कि कोई निर्भरता नहीं हैं। MailboxProcessor नामक एक उदाहरण dagAgent वर्तमान स्थिति में पंजीकृत कार्यों "कार्यों" को संग्रहीत करता है, जो एक शब्दकोश ( tasks : Dictionary<int, TaskInfo> ) है जो प्रत्येक कार्य की आईडी और उसके विवरण को सहसंबंधित करता है। इसके अलावा, एजेंट प्रत्येक कार्य की आईडी द्वारा धार निर्भरता की स्थिति को भी संग्रहीत करता है ( edges : Dictionary<int, int list> )। जब एक एजेंट को निष्पादन शुरू करने की आवश्यकता के बारे में एक सूचना मिलती है, तो इस प्रक्रिया के हिस्से के रूप में, यह जांचा जाता है कि किनारों की सभी निर्भरताएं पंजीकृत हैं, और ग्राफ़ में कोई चक्र नहीं हैं। यह सत्यापन चरण ऑनलाइन कोड में प्रदान किए गए ParallelTasksDAG के पूर्ण कार्यान्वयन में उपलब्ध है। अगला, मैं C # में एक उदाहरण प्रस्तुत करता हूं, जहां मैं लाइब्रेरी का उल्लेख करता हूं कि F # ParallelTasksDAG को चलाने के लिए (और इसका उपभोग करें)। पंजीकृत कार्य अंजीर में ऊपर दी गई निर्भरता को दर्शाते हैं। 1।

 Func<int, int, Func<Task>> action = (id, delay) => async () => { Console.WriteLine($”Starting operation{id} in Thread Id {Thread.CurrentThread.ManagedThreadId}…”); await Task.Delay(delay); }; var dagAsync = new DAG.ParallelTasksDAG(); dagAsync.OnTaskCompleted.Subscribe(op => Console.WriteLine($”Operation {op.Id} completed in Thread Id { Thread.CurrentThread.ManagedThreadId}”)); dagAsync.AddTask(1, action(1, 600), 4, 5); dagAsync.AddTask(2, action(2, 200), 5); dagAsync.AddTask(3, action(3, 800), 6, 5); dagAsync.AddTask(4, action(4, 500), 6); dagAsync.AddTask(5, action(5, 450), 7, 8); dagAsync.AddTask(6, action(6, 100), 7); dagAsync.AddTask(7, action(7, 900)); dagAsync.AddTask(8, action(8, 700)); dagAsync.ExecuteTasks();

सहायक फ़ंक्शन का उद्देश्य एक संदेश प्रदर्शित करना है जो कार्य शुरू हो गया है, मल्टीथ्रेडिंग की पुष्टि करने Id वर्तमान थ्रेड की Id उल्लेख करता है। दूसरी ओर OnTaskCompleted , कंसोल Id वर्तमान थ्रेड के कार्य ID और Id आउटपुट के साथ प्रत्येक कार्य के पूरा होने के बारे में एक अधिसूचना देने के लिए लॉग ऑन है। यहाँ ExecuteTasks पद्धति को कॉल करने पर प्राप्त होने वाला आउटपुट है।

 Starting operation 8 in Thread Id 23… Starting operation 7 in Thread Id 24… Operation 8 Completed in Thread Id 23 Operation 7 Completed in Thread Id 24 Starting operation 5 in Thread Id 23… Starting operation 6 in Thread Id 25… Operation 6 Completed in Thread Id 25 Starting operation 4 in Thread Id 24… Operation 5 Completed in Thread Id 23 Starting operation 2 in Thread Id 27… Starting operation 3 in Thread Id 30… Operation 4 Completed in Thread Id 24 Starting operation 1 in Thread Id 28… Operation 2 Completed in Thread Id 27 Operation 1 Completed in Thread Id 28 Operation 3 Completed in Thread Id 30

जैसा कि आप देख सकते हैं, कार्यों को विभिन्न थ्रेड्स (उनके लिए थ्रेड ID अलग-अलग हैं) में समानांतर में निष्पादित किया जाता है, और निर्भरता के क्रम को संरक्षित किया जाता है।

संक्षेप में, यह है कि निर्भरता वाले कार्यों को कैसे समानांतर किया जाता है। .NET में किताब में और अधिक पढ़ें।

निर्भरता वाले कार्यों को समानांतर करना - .NET उदाहरण

More articles: