व्यवहार में बहुआयामी

किसी तरह मुझे स्टैक ओवरफ्लो पर एक प्रश्न ( लिंक ) मिला।
Need to create java CLI programm that searchs for specific files matched some pattern. Need to use multi-threading approach without using util.concurrent package and to provide good performance on parallel controllers.
अनुवाद
, - . , util.concurrent. .


सामान्य तौर पर, सिद्धांत में, विचार जटिल नहीं था। क्योंकि इस स्थिति में, आप use.concurrent का उपयोग नहीं कर सकते हैं, तो आपको अपने स्वयं के थ्रेड पूल को लागू करने की आवश्यकता है, साथ ही कुछ कार्य लिखें जो इस थ्रेड पूल में स्पिन करेंगे।
इसके अलावा, मुझे यकीन नहीं था कि IO के बहु-थ्रेडेड उपयोग के साथ उत्पादकता में वृद्धि होगी।

मुझे तुरंत कहना होगा कि लक्ष्य कार्य को पूरा करना नहीं था, लेकिन समस्या की जांच करना, इसलिए सभी कोड सुंदर नहीं होंगे।

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

  import java.io.File; public class MainEntry { public static void main(String[] args) { walkin(new File("/home/user")); //Replace this with a suitable directory } /** * Recursive function to descend into the directory tree and find all the files * that end with ".mp3" * @param dir A file object defining the top directory **/ public static void walkin(File dir) { String pattern = ".mp3"; File listFile[] = dir.listFiles(); if(listFile != null) { for(int i=0; i<listFile.length; i++) { if(listFile[i].isDirectory()) { walkin(listFile[i]); } else { if(listFile[i].getName().endsWith(pattern)) { System.out.println(listFile[i].getPath()); } } } } } } 

पहले, आइए देखें कि एकल-थ्रेडेड कार्यान्वयन कितनी तेज़ी से काम करता है, मैंने एकल-थ्रेडेड कार्यान्वयन का कोड नहीं दिया है, यह संग्रह में है।
परिणाम इस प्रकार हैं:
154531 एमएस

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

ऐसा करने के लिए, एक पुनरावर्ती कॉल के बजाय, हम एक टास्क बनाएंगे जिसे हम निष्पादन के लिए थ्रेड पूल को देंगे। यह भी आवश्यक है कि, कार्य के परिणाम के अनुसार, किसी तरह परिणामों को रिपोर्ट करने का अवसर था। साथ ही, आपको थ्रेड पूल (पुनरावृत्ति के बजाय) में एक नया कार्य जोड़ने की आवश्यकता है।
यहां यह तुरंत रोकना आवश्यक है कि मैंने नए धागे बनाने के बजाय सिर्फ छोटे कार्यों और एक थ्रेड पूल को क्यों चुना। कल्पना कीजिए कि हमारे पास निर्देशिकाओं का एक गुच्छा है, और हम प्रत्येक निर्देशिका के लिए एक नया धागा बनाएंगे? हम सिर्फ OOM (OutOfMemory) पर गिर सकते हैं या OS थ्रेड्स के बीच स्विच करने के कारण बस सब कुछ बहुत धीमा हो जाएगा (विशेषकर यदि यह सिंगल-कोर सिस्टम है)। हम हर बार एक नया धागा शुरू करने में भी समय बिताएंगे।

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

यहाँ कोड है:

 import java.util.ArrayList; import java.util.List; class BacklogWorker extends Thread { /*      */ private final LinkedList<Runnable> backlog = new LinkedList<Runnable>(); private static final int INITIAL_CAPACITY = 100; /*   ,    */ private final List<Runnable> workQueue = new ArrayList<Runnable>(INITIAL_CAPACITY); BacklogWorker(String name) { super(name); } /*   */ synchronized void enque(Runnable work) { if (work != null) { backlog.add(work); } notify(); } public void run() { while (!isInterrupted()) { /*      ,   */ synchronized (this) { if (backlog.isEmpty()) { try { wait(); } catch (InterruptedException e) { interrupt(); } } int size = backlog.size(); for (int i = 0; i < INITIAL_CAPACITY && i < size; i++) { workQueue.add(backlog.poll()); } backlog.clear(); } for (Runnable task : workQueue) { try { task.run(); } catch (RuntimeException e) { e.printStackTrace(); } } workQueue.clear(); } } } 

अब हमें एक थ्रेडपूल बनाने की आवश्यकता है जो इसके थ्रेड्स के बीच काम को वितरित करेगा।
कक्षा की आवश्यकताएँ इस प्रकार हैं:
- थ्रेडसेफ़
- स्केलेबल
- कार्य के काम कर रहे धागे के बीच एक समान वितरण
- अवरुद्ध नहीं

कोड निम्नानुसार है:

  import java.util.concurrent.Executor; public class BacklogThreadPool implements Executor/*i don't use anything from concurrent, just only one interface*/ { private final BacklogWorker workers[]; private final int mask; private static volatile int sequence; public BacklogThreadPool(int threadCount, String id) { int tc; for (tc = 1; tc < threadCount; tc <<= 1) ; mask = tc - 1; if (id == null) { id = Integer.toString(getSequence()); } workers = new BacklogWorker[tc]; for (int i = 0; i < tc; i++) { workers[i] = new BacklogWorker((new StringBuilder()).append("thead-pool-worker-").append(id).append(":").append(i).toString()); workers[i].start(); } } private synchronized int getSequence() { return sequence++; } public void shutdown() { for (int i = 0; i < workers.length; i++) workers[i].interrupt(); } @Override public void execute(Runnable command) { workers[command.hashCode() & mask].enque(command); } } 

सिद्धांत रूप में, यहां सब कुछ स्पष्ट है, और शायद इस पर टिप्पणी करने के लिए कुछ भी नहीं है।

अब हमें एक कार्य लिखने की आवश्यकता है जिसे थ्रेडपूल में निष्पादित किया जाएगा।
दुर्भाग्य से, मेरे लिए पहला संस्करण खो गया था, इसलिए मैं फिर से एक त्वरित लिखित संस्करण ला रहा हूं।

 import java.io.File; import java.util.ArrayList; import java.util.List; import java.util.regex.Matcher; import java.util.regex.Pattern; public class WalkinTask1 implements BacklogTask { private List<File> dirs; private ParseHandler parseHandler; public WalkinTask1(List<File> dirs, ParseHandler parseHandler) { this.parseHandler = parseHandler; //this.parseHandler.taskStart(); this.parseHandler.taskStartUnblock(); this.dirs = dirs; } @Override public void run() { try { List<String> filePaths = new ArrayList<String>(); List<File> dirPaths = new ArrayList<File>(); for (File dir : dirs) { if (!dirPaths.isEmpty()) { dirPaths = new ArrayList<File>(); } if (!filePaths.isEmpty()) { filePaths = new ArrayList<String>(); } File listFile[] = dir.listFiles(); if (listFile != null) { for (File file : listFile) { if (file.isDirectory()) { dirPaths.add(file); } else { filePaths.add(file.getPath()); } } } if (!dirPaths.isEmpty()) { parseHandler.schedule(TaskFactory.createWalking1Task(parseHandler, dirPaths)); } if (!filePaths.isEmpty()) { Pattern pattern = parseHandler.getPattern(); List<String> result = new ArrayList<String>(); for (String path : filePaths) { Matcher matcher = pattern.matcher(path); while (matcher.find()) { String str = matcher.group(); if (!"".equals(str)) { result.add(str); } } } parseHandler.taskComplete(result); } } } finally { //parseHandler.taskFinish(); parseHandler.taskFinishUnblock(); } } @Override public int getTaskType() { return 1; //TODO } } 

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

अब प्रोफाइलर में परिणाम देखें:
कार्यक्रम का निष्पादन समय:
कुल कार्य: 78687
55188ms

नतीजतन, काम की गति लगभग 3 गुना बढ़ गई।


यहां हम देखते हैं कि थ्रेड पूल के सभी धागे लगभग पूरे समय काम में व्यस्त थे। लगभग कोई थ्रेड लॉक नहीं हैं।

दूसरी तस्वीर में हम देखते हैं कि ज्यादातर समय CPU IO पर खर्च होता है।

यहाँ हम देखते हैं कि CPU उपयोग> 80% है।

यहां हमें केवल एक थ्रेड लॉक दिखाई देता है, जो कि 1ms से कम समय लेता है, 78 हजार कार्यों के साथ एक बहुत अच्छा परिणाम है।

जैसा कि हम देख सकते हैं, सिद्धांत रूप में, हम सीपीयू को लोड करते हैं, और हमारे पास कोई डाउनटाइम नहीं है, क्योंकि सभी धागे लगभग पूरी तरह से काम से भरे हुए हैं। तालों पर ताले नहीं हैं।

चित्र संख्या 2 को देखना दिलचस्प होगा। जैसा कि हम देख सकते हैं, सबसे "महंगा" ऑपरेशन java.io.File.isDirectory () है, इसमें कुल समय का लगभग 46% लगता है। इस समस्या के बारे में, मुझे जावा 7, अच्छी तरह से या निर्भरता ओएस सुविधाओं का उपयोग करने की क्षमता के अलावा कुछ भी नहीं मिला। इसलिए, इस भाग को अनुकूलित करने की क्षमता जैसा कि मैं देख रहा हूं यह अधिक नहीं है। इसके बाद पार्सर आता है - java.util.regex.Matcher.find (), लेकिन यहां आप पहले से ही इसे गति दे सकते हैं। आप एक और कार्य बना सकते हैं जो केवल पार्सिंग से निपटेगा। यानी हम दो सबसे कठिन कार्यों को अलग करेंगे:
1) फ़ाइल सिस्टम के साथ काम करते हैं
2) नाम पार्सिंग
तीसरा ऑपरेशन फिर से IO है, और इसे गति देना भी मुश्किल है।

इसलिए, हम पहले कार्य को थोड़ा संशोधित करते हैं और एक नया जोड़ते हैं:

  import java.io.File; import java.util.ArrayList; import java.util.List; public class WalkinTask implements BacklogTask { private List<File> dirs; private ParseHandler parseHandler; public WalkinTask(List<File> dirs, ParseHandler parseHandler) { this.parseHandler = parseHandler; //this.parseHandler.taskStart(); this.parseHandler.taskStartUnblock(); this.dirs = dirs; } @Override public void run() { try { List<String> filePaths = new ArrayList<String>(); List<File> dirPaths = new ArrayList<File>(); for (File dir : dirs) { if (!dirPaths.isEmpty()) { dirPaths = new ArrayList<File>(); } if (!filePaths.isEmpty()) { filePaths = new ArrayList<String>(); } File listFile[] = dir.listFiles(); if (listFile != null) { for (File file : listFile) { if (file.isDirectory()) { dirPaths.add(file); } else { filePaths.add(file.getPath()); } } } if (!dirPaths.isEmpty()) { parseHandler.schedule(TaskFactory.createWalkingTask(parseHandler, dirPaths)); } if (!filePaths.isEmpty()) { parseHandler.schedule(TaskFactory.createParseTask(parseHandler, filePaths)); } } } finally { //parseHandler.taskFinish(); parseHandler.taskFinishUnblock(); } } @Override public int getTaskType() { return 1; //TODO } } 

 import java.util.ArrayList; import java.util.List; import java.util.regex.Matcher; import java.util.regex.Pattern; public class ParseTask implements BacklogTask { private ParseHandler handler; private List<String> paths; public ParseTask(ParseHandler hander, List<String> paths) { this.handler = hander; this.paths = paths; handler.taskStartUnblock(); } @Override public void run() { try { Pattern pattern = handler.getPattern(); List<String> result = new ArrayList<String>(); for (String path : paths) { Matcher matcher = pattern.matcher(path); while (matcher.find()) { String str = matcher.group(); if (!"".equals(str)) { result.add(str); } } } handler.taskComplete(result); } finally { handler.taskFinishUnblock(); } } @Override public int getTaskType() { return 0; } } 

और फिर से दौड़ें:
कुल कार्य: 221560
52328


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

अब हमें यह सोचने की ज़रूरत है कि कार्यक्रम से बाहर कैसे निकलें, और परिणाम कैसे लौटाएं। हमने कार्यों का एक समूह बनाया है, और हम नहीं जानते कि वे कब पूरे होंगे। मैं कार्यों की कुल संख्या की गणना से बेहतर कुछ भी नहीं आया, और काउंटर शून्य होने तक प्रतीक्षा करें। इसके लिए हमें एक वैरिएबल चाहिए जिसमें हम वैल्यूज को संचित करेंगे। लेकिन यहाँ भी, इतना आसान नहीं है। उदाहरण के लिए, यदि हम एक नियमित चर लेते हैं और कार्य बनने पर इसे बढ़ाते हैं, और कार्य समाप्त होने पर इसे घटाते हैं। लेकिन इस दृष्टिकोण के साथ, परिणाम विनाशकारी होगा, क्योंकि जावा में, i ++ ऑपरेशन परमाणु नहीं है, भले ही हम प्रतिष्ठित वाष्पशील संशोधक की आपूर्ति करते हैं। यह AtomicIteger लेने के लिए आदर्श होगा, लेकिन शर्त के अनुसार हम use.concurrent पैकेज का उपयोग नहीं कर सकते। इसलिए, हमें अपना परमाणु बनाना होगा। यदि आप इस बात पर ध्यान केंद्रित करते हैं कि परमाणु जावा में कैसे काम करता है, तो हम एक देशी पद्धति में आते हैं। एक चर परिवर्तन की परमाणु प्रकृति को एक प्रोसेसर कमांड के रूप में लागू किया जाता है, इसलिए जावा देशी ओएस कमांड को कॉल करता है, जो पहले से ही प्रोसेसर कमांड को कॉल करता है।
असल में, हम नियमित सिंक्रनाइज़ का उपयोग कर सकते हैं। लेकिन बड़ी संख्या में कार्यों के साथ, लॉक रेस शुरू हो जाएगी, और उत्पादकता घट जाएगी (हालांकि महत्वपूर्ण नहीं)। यहाँ एक नमूना कोड है जो CAS एल्गोरिथ्म को लागू करता है (कोड ibm वेबसाइट पर पाया गया था):

  public class SimulatedCAS { private volatile int value; public synchronized int getValue() { return value; } public synchronized int compareAndSwap(int expectedValue, int newValue) { int oldValue = value; if (value == expectedValue) value = newValue; return oldValue; } } 

  public class CasCounter { private SimulatedCAS value = new SimulatedCAS(); public int getValue() { return value.getValue(); } public int increment() { int oldValue = value.getValue(); while (value.compareAndSwap(oldValue, oldValue + 1) != oldValue) oldValue = value.getValue(); return oldValue + 1; } public int decrement() { int oldValue = value.getValue(); while (value.compareAndSwap(oldValue, oldValue - 1) != oldValue) oldValue = value.getValue(); return oldValue - 1; } } 

बस इतना ही।
स्रोत के साथ पुरालेख: लिंक

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

PSS यह हब पर मेरा पहला बड़ा विषय है, इसलिए कृपया डिजाइन पर ज्यादा आलोचना न करें, यदि संभव हो तो मैं इसे सही करूंगा।