रेयन

Question

में प्रति-थ्रेड प्रारंभिकरण मैं रेयन के par_iter() का उपयोग करके अपने फ़ंक्शन को अनुकूलित करने का प्रयास कर रहा हूं।

एकल थ्रेड संस्करण कुछ इस प्रकार है:

fn verify_and_store(store: &mut Store, txs: Vec<Tx>) { 

    let result = txs.iter().map(|tx| { 

     tx.verify_and_store(store) 

    }).collect(); 

    ... 
} 

प्रत्येक Store उदाहरण एक थ्रेड द्वारा केवल इस्तेमाल किया जाना चाहिए, लेकिन Store के कई उदाहरण समवर्ती इस्तेमाल किया जा सकता है, इसलिए मैं इस clone -ing द्वारा थ्रेड बना सकते हैं store:

fn verify_and_store(store: &mut Store, txs: Vec<Tx>) { 

    let result = txs.par_iter().map(|tx| { 

     let mut local_store = store.clone(); 

     tx.verify_and_store(&mut local_store) 

    }).collect(); 

    ... 
} 

हालांकि, इस क्लोन हर यात्रा storeपर है, जो वा है वाई बहुत धीमी है। मैं प्रति थ्रेड एक स्टोर उदाहरण का उपयोग करना चाहता हूं।

क्या यह रेयन के साथ संभव है? या मुझे मैन्युअल थ्रेडिंग और वर्क-कतार का सहारा लेना चाहिए?

Answer 1

यह सुनिश्चित करने के लिए कि local_store किसी दिए गए थ्रेड में एक से अधिक बार नहीं बनाया गया है, थ्रेड-स्थानीय चर का उपयोग करना संभव है।

उदाहरण के लिए, इस संकलित (full source):

fn verify_and_store(store: &mut Store, txs: Vec<Tx>) { 
    use std::cell::RefCell; 
    thread_local!(static STORE: RefCell<Option<Store>> = RefCell::new(None)); 

    let mut result = Vec::new(); 

    txs.par_iter().map(|tx| { 
     STORE.with(|cell| { 
      let mut local_store = cell.borrow_mut(); 
      if local_store.is_none() { 
       *local_store = Some(store.clone()); 
      } 
      tx.verify_and_store(local_store.as_mut().unwrap()) 
     }) 
    }).collect_into(&mut result); 
} 

इस कोड के साथ दो समस्याएं हैं लेकिन,। एक, यदि store के क्लोन par_iter() किए जाने पर कुछ करने की आवश्यकता है, जैसे कि उनके बफर फ्लश करें, तो यह नहीं होगा - उनके Drop केवल तभी कॉल किए जाएंगे जब रेयन के कार्यकर्ता थ्रेड से बाहर निकलेंगे, और यहां तक ​​कि is not guaranteed भी।

दूसरा, और अधिक गंभीर समस्या यह है कि store के क्लोन प्रति कार्यकर्ता धागे में बिल्कुल एक बार बनाए जाते हैं। यदि रेयन अपने थ्रेड पूल को कैश करता है (और मुझे विश्वास है कि यह करता है), तो इसका मतलब है कि verify_and_store पर एक असंबंधित बाद की कॉल store के अंतिम ज्ञात क्लोन के साथ काम करना जारी रखेगी, जिसका संभवतः वर्तमान स्टोर से कोई लेना देना नहीं है।

यह कुछ हद तक कोड उलझी द्वारा सुधारा जा सकता है:

• स्टोर Option के बजाय एक Mutex<Option<...>> में क्लोन चर, ताकि वे धागा कि par_iter() लागू द्वारा पहुँचा जा सकता। इसमें प्रत्येक पहुंच पर एक म्यूटेक्स लॉक लगेगा, लेकिन लॉक अनचाहे और इसलिए सस्ता होगा।

• एक वेक्टर में बनाए गए स्टोर क्लोन के संदर्भ एकत्र करने के लिए mutex के आस-पास Arc का उपयोग करें। पुनरावृत्ति समाप्त होने के बाद इन वेक्टर को None पर रीसेट करके स्टोर को साफ करने के लिए उपयोग किया जाता है।

• एक असंबद्ध म्यूटेक्स में पूरी कॉल लपेटें, ताकि verify_and_store पर दो समांतर कॉल एक-दूसरे के स्टोर क्लोन को देख न सकें। (यह तब से रोका जा सकता है जब एक नया धागा पूल बनाया गया था और पुनरावृत्ति से पहले स्थापित किया गया था।) उम्मीद है कि यह क्रमिकरण verify_and_store के प्रदर्शन को प्रभावित नहीं करेगा, क्योंकि प्रत्येक कॉल पूरे थ्रेड पूल का उपयोग करेगी।

परिणाम सुंदर नहीं है, लेकिन यह, संकलित केवल सुरक्षित कोड का उपयोग करता है, और काम करने के लिए प्रकट होता है:

fn verify_and_store(store: &mut Store, txs: Vec<Tx>) { 
    use std::sync::{Arc, Mutex}; 
    type SharedStore = Arc<Mutex<Option<Store>>>; 

    lazy_static! { 
     static ref STORE_CLONES: Mutex<Vec<SharedStore>> = Mutex::new(Vec::new()); 
     static ref NO_REENTRY: Mutex<()> = Mutex::new(()); 
    } 
    thread_local!(static STORE: SharedStore = Arc::new(Mutex::new(None))); 

    let mut result = Vec::new(); 
    let _no_reentry = NO_REENTRY.lock(); 

    txs.par_iter().map({ 
     |tx| { 
      STORE.with(|arc_mtx| { 
       let mut local_store = arc_mtx.lock().unwrap(); 
       if local_store.is_none() { 
        *local_store = Some(store.clone()); 
        STORE_CLONES.lock().unwrap().push(arc_mtx.clone()); 
       } 
       tx.verify_and_store(local_store.as_mut().unwrap()) 
      }) 
     } 
    }).collect_into(&mut result); 

    let mut store_clones = STORE_CLONES.lock().unwrap(); 
    for store in store_clones.drain(..) { 
     store.lock().unwrap().take(); 
    } 
}