आंतरिक सिंक्रनाइज़ेशन

Question

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

#1: J1, J2, J3 
<wait for J1, J2, run other lists if possible> 
#2: J4 

हालांकि, इस के रूप में ऐसा लगता है कि आसान नहीं है के रूप में सूचियों का एक सेट दिया:

J1, J2, S, J3, W, J4 

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

क्या यह कुशलतापूर्वक कार्यान्वित करने का कोई "मानक" तरीका है? मुझे आश्चर्य है कि नौकरी सूची निष्पादन को सर्वोत्तम तरीके से कैसे निर्धारित किया जाए, अभी, प्रत्येक कोर नौकरी सूची पकड़ लेता है, और इसमें सभी नौकरियों को शेड्यूल करता है, जो बहुत अच्छी स्केलिंग देता है (32k नौकरियों के लिए 0.7 एमएस, मुझे 101% मिलता है, जो मुझे लगता है आंशिक रूप से इस तथ्य के कारण है कि सिंगल थ्रेडेड संस्करण को कुछ बार विभिन्न कोरों पर निर्धारित किया जा रहा है।)

Answer 1

यह अपेक्षाकृत कठोर शेड्यूलिंग एल्गोरिदम है। कुछ मुद्दे पहले मुश्किल लगते हैं लेकिन वास्तव में नहीं हैं (संकेत/प्रतीक्षा और कैश इलाके)। मैं तकनीकों की व्याख्या करूंगा, फिर अवधारणाओं को चित्रित करने के लिए मैंने लिखा कुछ कोड दें, फिर ट्यूनिंग पर कुछ अंतिम नोट दें।

एल्गोरिदम

उपयोग करने के लिए संकेत हैंडलिंग/इंतजार कुशलता है पहली बार में मुश्किल लगता है, लेकिन वास्तव में बेहद आसान होने के लिए पता चला है। चूंकि सिग्नल/प्रतीक्षा जोड़े घोंसला या ओवरलैप नहीं कर सकते हैं, वास्तव में केवल दो ही संतुष्ट हो सकते हैं और किसी भी समय किसी पर इंतजार किया जा रहा है। बस हाल ही में असंतुष्ट सिग्नल के लिए "CurrentSignal" पॉइंटर रखना, बहीखाता करने के लिए आवश्यक है।

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

लॉकिंग केवल शेड्यूलर या जिस सूची में आप काम कर रहे हैं उसे लॉक करके सरल रखा जा सकता है, कभी भी दोनों नहीं।

आपने सूची को पहले ही निष्पादित करना शुरू करने के बाद सूची में नौकरियां जोड़ने के बारे में कुछ चिंता व्यक्त की है। यह पता चला है कि इसका समर्थन करना लगभग मामूली है: वर्तमान में पूरा होने वाली सूची में नौकरी जोड़ने पर शेड्यूलर को सूची से कॉल करने की ज़रूरत है, इसलिए शेड्यूलर नई नौकरी निर्धारित कर सकता है।

डाटा संरचनाओं

यहाँ आप की आवश्यकता होगी बुनियादी डेटा संरचनाओं हैं:

class Scheduler 
{ 
    LinkedList<JobList>[] Ready; // Indexed by number of cores working on list 
    LinkedList<JobList> Blocked; 
    int ReadyCount; 
    bool Exit; 

    public: 
    void AddList(JobList* joblist); 
    void DoWork(); 

    internal: 
    void UpdateQueues(JobList* joblist); 

    void NotifyBlockedCores(); 
    void WaitForNotifyBlockedCores(); 
} 

class JobList 
{ 
    Scheduler Scheduler; 
    LinkedList<JobList> CurrentQueue; 

    LinkedList<Job> Jobs;   // All jobs in the job list 
    LinkedList<SignalPoint> Signals; // All signal/wait pairs in the job list, 
             plus a dummy 

    Job* NextJob;     // The next job to schedule, if any 
    int NextJobIndex;    // The index of NextJob 

    SignalPoint* CurrentSignal;  // First signal not fully satisfied 

    int WorkerCount;     // # of cores executing in this list 

    public: 
    void AddJob(Job* job); 
    void AddSignal(); 
    void AddWait(); 

    internal: 
    void Ready { get; } 
    void GetNextReadyJob(Job& job, int& jobIndex); 
    void MarkJobCompleted(Job job, int jobIndex); 
} 
class SignalPoint 
{ 
    int SignalJobIndex = int.MaxValue; 
    int WaitJobIndex = int.MaxValue; 
    int IncompleteCount = 0; 
} 

ध्यान दें कि किसी दिए गए joblist के लिए संकेत अंक सबसे सुविधाजनक नौकरियों की वास्तविक सूची से अलग से जमा हो जाती है ।

समयबद्धक कार्यान्वयन

अनुसूचक काम सूचियों का ट्रैक रखता है, उन्हें कोर करने के लिए देता है, और काम सूचियों से नौकरियों निष्पादित करता है।

AddList शेड्यूलर को नौकरी जोड़ता है। इसे तैयार या अवरुद्ध कतार पर रखा जाना चाहिए, इस पर निर्भर करता है कि उसके पास कोई काम है या नहीं (यानी। क्या कोई नौकरियां अभी तक इसमें शामिल की गई हैं), तो बस UpdateQueues को कॉल करें।

void Scheduler.AddList(JobList* joblist) 
{ 
    joblist.Scheduler = this; 
    UpdateQueues(joblist); 
} 

अद्यतनक्यू कतार अद्यतन तर्क को केंद्रीकृत करता है। सूचना एल्गोरिथ्म एक नई पंक्ति के चयन के लिए, तथा उन्हें सूचना कोर निष्क्रिय करने के लिए जब काम उपलब्ध हो जाता है:

void Scheduler.UpdateQueues(JobList* joblist) 
{ 
    lock(this) 
    { 
    // Remove from prior queue, if any 
    if(joblist.CurrentQueue!=null) 
    { 
     if(joblist.CurrentQueue!=Blocked) ReadyCount--; 
     joblist.CurrentQueue.Remove(joblist); 
    } 

    // Select new queue 
    joblist.CurrentQueue = joblist.Ready ? Ready[joblist.WorkerCount] : Blocked; 

    // Add to new queue 
    joblist.CurrentQueue.Add(joblist); 
    if(joblist.CurrentQueue!=Blocked) 
     if(++ReadyCount==1) NotifyBlockedCores(); 
    } 
} 

DoWork है, को छोड़कर एक सामान्य अनुसूचक काम: 1. यह सबसे कम कार्यकर्ताओं के साथ JobList चयन करता है, 2. यह किसी दिए गए जॉबलिस्ट से जॉब्स काम करता है जब तक कि यह और नहीं हो सकता है, और 3. यह जॉब इंडेक्स के साथ-साथ नौकरी भी संग्रहीत करता है ताकि जॉबलिस्ट आसानी से समापन स्थिति (कार्यान्वयन विवरण) अपडेट कर सके।

void Scheduler.DoWork() 
{ 
    while(!Exit) 
    { 
    // Get a job list to work on 
    JobList *list = null; 
    lock(this) 
    { 
     for(int i=0; i<Ready.Length; i++) 
     if(!Ready[i].Empty) 
     { 
      list = Ready[i].First; 
      break; 
     } 
     if(list==null) // No work to do 
     { 
     WaitForNotifyBlockedCores(); 
     continue; 
     } 
     list.WorkerCount++; 
     UpdateQueues(list); 
    } 

    // Execute jobs in the list as long as possible 
    while(true) 
    { 
     int jobIndex; 
     Job job; 
     if(!GetNextReadyJob(&job, &jobIndex)) break; 

     job.Execute(); 

     list.MarkJobCompleted(job, jobIndex); 
    } 

    // Release the job list 
    lock(this) 
    { 
     list.WorkerCount--; 
     UpdateQueues(list); 
    } 
    } 
} 

JobList कार्यान्वयन

JobList कैसे संकेत/प्रतीक्षा नौकरियों के साथ बीच-बीच में कर रहे हैं का ट्रैक रखता है और जो संकेत का ट्रैक रखता है/इंतजार जोड़े पहले से ही अपने संकेत बिंदु से पहले सब कुछ पूरा कर लिया है।

कन्स्ट्रक्टर नौकरियों को जोड़ने के लिए एक डमी सिग्नल पॉइंट बनाता है। जब भी कोई नया "सिग्नल" जोड़ा जाता है, तो यह सिग्नल पॉइंट एक वास्तविक संकेत बिंदु बन जाता है (और एक नया डमी जोड़ा जाता है)।

JobList.JobList() 
{ 
    // Always have a dummy signal point at the end 
    Signals.Add(CurrentSignal = new SignalPoint()); 
} 

AddJob सूची में नौकरी जोड़ता है। इसे सिग्नलपॉइंट में अपूर्ण के रूप में चिह्नित किया गया है। जब नौकरी वास्तव में निष्पादित की जाती है, तो उसी सिग्नलपॉइंट की अधूरी गणना कम हो जाती है। शेड्यूलर को यह बताने के लिए भी जरूरी है कि चीजें बदली हों, क्योंकि नई नौकरी तुरंत निष्पादन योग्य हो सकती है। ध्यान दें कि डेडलॉक से बचने के लिए "यह" रिलीज़ होने पर लॉक के बाद शेड्यूलर को कॉल किया जाता है।

void JobList.AddJob(Job job) 
{ 
    lock(this) 
    { 
    Jobs.Add(job); 
    Signals.Last.IncompleteCount++; 
    if(NextJob == null) 
     NextJob = job; 
    } 
    if(Scheduler!=null) 
    Scheduler.UpdateQueues(this); 
} 

AddSignal और AddWait सिग्नल जोड़ते हैं और नौकरी सूची में प्रतीक्षा करते हैं। ध्यान दें कि AddSignal वास्तव में एक नया सिग्नलपॉइंट बनाता है, और AddWait पहले बनाए गए सिग्नलपॉइंट में प्रतीक्षा बिंदु जानकारी में भर जाता है।

void JobList.AddSignal() 
{ 
    lock(this) 
    { 
    Signals.Last.SignalJobIndex = Jobs.Count; // Reify dummy signal point 
    Signals.Add(new SignalPoint());   // Create new dummy signal point 
    } 
} 


void JobList.AddWait() 
{ 
    lock(this) 
    { 
    Signals.Last.Previous.WaitJobIndex = Jobs.Count; 
    } 
} 

तैयार संपत्ति यह निर्धारित करती है कि सूची अतिरिक्त कोर के लिए तैयार है या नहीं। सूची में काम करने के दो या तीन कोर सूची के बिना काम कर रहे हैं, यदि अगली नौकरी शुरू होने से पहले सिग्नल की प्रतीक्षा कर रही है।

bool JobList.Ready 
{ 
    get 
    { 
    lock(this) 
    { 
     return NextJob!=null && 
     (CurrentSignal==Signals.Last || 
     NextJobIndex < CurrentSignal.WaitJobIndex); 
    } 
    } 
} 

GetNextReadyJob बहुत सरल है: अगर हम तैयार हैं, बस सूची में अगले नौकरी वापस जाएँ।

void JobList.GetNextReadyJob(Job& job, int& jobIndex) 
{ 
    lock(this) 
    { 
    if(!Ready) return false; 
    jobIndex = list.NextJobIndex++; 
    job = list.NextJob; list.NextJob = job.Next; 
    return true; 

    } 
} 

MarkJob पूर्णतः सभी का सबसे दिलचस्प है। सिग्नल की संरचना और प्रतीक्षा के कारण, वर्तमान नौकरी या तो CurrentSignal से पहले है या CurrentSignal और CurrentSignal.Next के बीच है (यदि यह अंतिम वास्तविक सिग्नल के बाद है, तो इसे अंत में CurrentSignal और डमी सिग्नलपॉइंट के बीच के रूप में गिना जाएगा)। हमें अपूर्ण नौकरियों की गिनती को कम करने की जरूरत है। यदि यह गिनती शून्य हो जाती है तो हमें अगले सिग्नल पर भी जाना पड़ सकता है। बेशक हम कभी अंत में डमी सिग्नलपॉइंट पास नहीं करते हैं।

ध्यान दें कि इस कोड में शेड्यूलर के लिए कोई कॉल नहीं है। अद्यतन करें क्योंकि हम जानते हैं कि शेड्यूलर केवल एक सेकंड में GetNextReadyJob को कॉल करेगा और यदि यह गलत लौटाएगा तो यह UpdateQueue को वैसे भी कॉल करेगा। सूची की लंबाई, काम लंबाई का अनुमान है, आदि

ऊपर कोड कितनी देर तक काम सूचियों, देखते हैं, इसलिए यदि एक सौ छोटे काम सूचियों के लिए किसी भी ध्यान देना नहीं है के आधार पर

void JobList.MarkJobCompleted(Job job, int jobIndex) 
{ 
    lock(this) 
    { 
    if(jobIndex >= CurrentSignal.SignalJobIndex) 
     CurrentSignal.Next.IncompleteCount--; 
    else 
    { 
     CurrentSignal.IncompleteCount--; 
     if(CurrentSignal.IncompleteCount==0) 
     if(CurrentSignal.WaitJobIndex < int.MaxValue) 
      CurrentSignal = CurrentSignal.Next; 
    } 
    } 
} 

ट्यूनिंग और एक विशाल व्यक्ति प्रत्येक कोर के लिए एक अलग छोटी नौकरी सूची लेना संभव है और फिर सभी बड़े पैमाने पर एकत्र होते हैं, जिससे अक्षमता होती है। इसे (joblist.Jobs.Count - joblist.NextJobIndex) पर प्राथमिकता प्राप्त प्राथमिकता कतारों की एक सरणी तैयार करके हल किया जा सकता है, लेकिन प्राथमिकता के साथ केवल दक्षता के लिए सामान्य अद्यतनक्यूयू स्थितियों में वास्तव में अद्यतन किया जाता है।

यह एक ह्युरिस्टिक बनाकर और भी परिष्कृत हो सकता है जो प्राथमिकता निर्धारित करने के लिए सिग्नल/प्रतीक्षा संयोजनों की संख्या और अंतर को ध्यान में रखता है। नौकरी अवधि और संसाधन उपयोग के वितरण का उपयोग करके यह ह्युरिस्टिक सबसे अच्छा होगा।

यदि व्यक्तिगत नौकरी अवधि ज्ञात हैं, या यदि उनके लिए अच्छे अनुमान उपलब्ध हैं, तो हेरिस्टिक केवल सूची की लंबाई के बजाय अनुमानित शेष अवधि का उपयोग कर सकता है।

अंतिम नोटों

यह समस्या आप वर्तमान तक एक नहीं बल्कि मानक समाधान है। आप एल्गोरिदम मैं दिया उपयोग कर सकते हैं और वे ताला सहित, काम करेंगे, लेकिन आप कोड मैं कई कारणों से ऊपर लिखा संकलित करने के लिए सक्षम नहीं होगा:

1. यह सी के एक पागल मिश्रण है ++ और सी # वाक्य - विन्यास। मैंने मूल रूप से सी # में लिखना शुरू किया, फिर सी ++ शैली में सिंटैक्स का एक गुच्छा बदल दिया क्योंकि मैंने सोचा था कि इस तरह की एक परियोजना के लिए आप क्या उपयोग करेंगे। लेकिन मैंने कुछ सी # -इम्स में छोड़ा। सौभाग्य से कोई LINQ ;-)।

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

3. मैंने संकलित या परीक्षण नहीं किया। मैं गारंटी देता हूं कि वहां कहीं भी एक बग या दो है।

निष्कर्ष: किसी भी आप स्यूडोकोड भले ही यह असली McCoy तरह लग रहा है के रूप में ऊपर कोड व्यवहार करना चाहिए।

आनंद लें!

Answer 2

आप अपने परिवेश में work stealing framework की पहुंच है (उदाहरण के लिए, यदि आप Cilk सी में कर रहे हैं, या जावा में डौग ली की fork/join framework), आप आसानी से एक सरल और साफ समाधान प्राप्त कर सकते हैं (निम्न स्तर की तुलना में विज्ञापन-प्रयास प्रयास, जो आपको शायद ऐसा करना होगा यदि आप ऐसा कुछ नहीं कर सकते हैं), जो आपको स्वत: लोड संतुलन और अच्छी डेटा इलाके देता है।

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

"कांटा" एक धागे की स्थानीय कतार में काम डाल का मतलब है, जो या तो बाद में धागे द्वारा निष्पादित किया जाएगा, या इसे किसी अन्य थ्रेड द्वारा चुराया जा सकता है जो कुछ काम करने के लिए दिखता है।

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

import java.util.*; 
import java.util.concurrent.*; 
import jsr166y.ForkJoinPool; 
import jsr166y.ForkJoinTask; 
import jsr166y.RecursiveTask; 

public class FJTest { 
    public static void main(String[] args) throws Exception { 
     Iterable<List<TaskType>> lists = createLists(10); 

     ForkJoinPool pool = new ForkJoinPool(); 

     for (final List<TaskType> list : lists) { 
      pool.submit(new Runnable() { 
       public void run() { 
        List<ForkJoinTask> beforeSignal = new ArrayList<ForkJoinTask>(); 
        List<ForkJoinTask> afterSignal = new ArrayList<ForkJoinTask>(); 
        boolean signaled = false; 
        for (TaskType task : list) { 
         switch (task) { 
          case JOB: 
           ForkJoinTask job = new Job(); 
           if (signaled == false) 
            beforeSignal.add(job); 
           else 
            afterSignal.add(job); 
           job.fork(); 
           break; 
          case SIGNAL: 
           signaled = true; 
           break; 
          case WAIT: 
           signaled = false; 
           for (ForkJoinTask t : beforeSignal) { 
            t.helpJoin(); 
           } 
           beforeSignal = afterSignal; 
           afterSignal = new ArrayList<ForkJoinTask>(); 
         } 
        } 
       } 
      }); 
     } 

     pool.shutdown(); 
     pool.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS); 
    } 

    private static Iterable<List<TaskType>> createLists(int size) { 
     List<List<TaskType>> tasks = new ArrayList<List<TaskType>>(); 
     for (int i = 0; i < size; i++) { 
      tasks.add(createSomeList()); 
     } 
     return tasks; 
    } 

    private static List<TaskType> createSomeList() { 
     return Arrays.asList(
       TaskType.JOB, 
       TaskType.JOB, 
       TaskType.SIGNAL, 
       TaskType.JOB, 
       TaskType.WAIT, 
       TaskType.JOB); 
    } 

} 

enum TaskType { 
    JOB, SIGNAL, WAIT; 
} 
class Job extends RecursiveTask<Void> { 
    @Override 
    protected Void compute() { 
     long x = 1; 
     for (long i = 1; i < 200000001; i++) { 
      x = i * x; 
     } 
     System.out.println(x); //just to use x 
     return null; 
    } 
}