मैं जावा में एक सेमफोर कैसे लिख सकता हूं जो पिछले सफल आवेदकों को प्राथमिकता देता है?

Question

मुझे अपने जावा प्रोग्राम में एकल-परमिट सेमफोर ऑब्जेक्ट की आवश्यकता है, जहां एक अतिरिक्त अधिग्रहण विधि है जो इस तरह दिखती है:

boolean tryAcquire(int id) 

और निम्नानुसार व्यवहार करता है: यदि आईडी पहले नहीं सामना किया गया है, तो इसे याद रखें और फिर जो कुछ भी java.util.concurrent.Semaphore करता है। यदि id और से पहले सामना किया गया है तो मुठभेड़ के परिणामस्वरूप परमिट के पट्टे में परिणामस्वरूप अन्य थ्रेड पर यह थ्रेड प्राथमिकता दें जो परमिट की प्रतीक्षा कर रहे हैं। मैं एक अतिरिक्त रिलीज विधि भी चाहूंगा जैसे:

void release(int id) 

जो java.util.concurrent.Semaphore करता है, साथ ही आईडी के बारे में भी "भूल जाता है" करता है।

मुझे वास्तव में यह नहीं पता कि यह कैसे पहुंचा जाए, लेकिन यहां एक संभावित कार्यान्वयन की शुरुआत है, लेकिन मुझे डर है कि यह कहीं भी नहीं जा रहा है:

public final class SemaphoreWithMemory { 

    private final Semaphore semaphore = new Semaphore(1, true); 
    private final Set<Integer> favoured = new ConcurrentSkipListSet<Integer>(); 

    public boolean tryAcquire() { 
     return semaphore.tryAcquire(); 
    } 

    public synchronized boolean tryAcquire(int id) { 
     if (!favoured.contains(id)) { 
      boolean gotIt = tryAcquire(); 
      if (gotIt) { 
       favoured.add(id); 
       return true; 
      } 
      else { 
       return false; 
      } 
     } 
     else { 
      // what do I do here??? 
     } 
    } 

    public void release() { 
     semaphore.release(); 
    } 

    public synchronized void release(int id) { 
     favoured.remove(id); 
     semaphore.release(); 
    } 

} 

Answer 1

संपादित करें:
कुछ प्रयोग किया। परिणाम के लिए कृपया this answer देखें।

सिद्धांत रूप में, सेमफोर में आंतरिक रूप से धागे की कतार है, इसलिए एंड्रयू कहता है कि यदि आप इस कतार को प्राथमिकता कतार और परमिट देने के लिए इस कतार से मतदान करते हैं, तो संभवतया आप जिस तरह से चाहते हैं उससे व्यवहार करते हैं। ध्यान दें कि आप tryAcquire के साथ ऐसा नहीं कर सकते हैं क्योंकि इस तरह के धागे कतारबद्ध नहीं होते हैं। जो कुछ मैं देखता हूं उससे ऐसा लगता है कि आपको ऐसा करने के लिए AbstractQueuedSynchronizer कक्षा को हैक करना होगा।

मैं भी एक संभाव्य दृष्टिकोण के बारे में सोच सकता है, इस तरह: (!। मैं यह नहीं कह रहा हूँ कि कोड के नीचे एक अच्छा विचार होगा बस यहाँ बुद्धिशीलता)

public class SemaphoreWithMemory { 

    private final Semaphore semaphore = new Semaphore(1); 
    private final Set<Integer> favoured = new ConcurrentSkipListSet<Integer>(); 
    private final ThreadLocal<Random> rng = //some good rng 

    public boolean tryAcquire() { 
     for(int i=0; i<8; i++){ 
      Thread.yield(); 
      // Tend to waste more time than tryAcquire(int id) 
      // would waste. 
      if(rng.get().nextDouble() < 0.3){ 
       return semaphore.tryAcquire(); 
      } 
     } 
     return semaphore.tryAcquire(); 
    } 

    public boolean tryAcquire(int id) { 
     if (!favoured.contains(id)) { 
      boolean gotIt = semaphore.tryAcquire(); 
      if (gotIt) { 
       favoured.add(id); 
       return true; 
      } else { 
       return false; 
      } 
     } else { 
      return tryAquire(); 
    } 
} 

या है " इष्ट "धागे लंबे समय तक इस तरह एक छोटा सा बाहर लटका:
संपादित करें: बाहर कर देता है यह एक बहुत ही बुरा विचार था (दोनों निष्पक्ष और गैर निष्पक्ष सेमाफोर के साथ) (विवरण के लिए देख मेरी experiment

public boolean tryAcquire(int id) { 
     if (!favoured.contains(id)) { 
      boolean gotIt = semaphore.tryAcquire(5,TimeUnit.MILLISECONDS); 
      if (gotIt) { 
       favoured.add(id); 
       return true; 
      } else { 
       return false; 
      } 
     } else { 
      return tryAquire(); 
    } 

।

मुझे लगता है कि इस तरह से आप परमिट जारी किए जा सकते हैं, जबकि यह उचित नहीं होगा। हालांकि इस कोड के साथ आप शायद बहुत समय के प्रदर्शन के साथ बर्बाद कर रहे होंगे ...

Answer 2

मान लीजिए कि आप धागे इंतजार करना चाहते हैं, मैंने एक ऐसा समाधान हैक किया जो सही नहीं है, लेकिन करना चाहिए।

विचार दो सेमेफोर और "पसंदीदा इंतज़ार कर रहा है" ध्वज है।

प्रत्येक थ्रेड जो सेमफोरविथमेमरी हासिल करने का प्रयास करता है, पहले "favouredSemaphore" प्राप्त करने का प्रयास करता है। एक "पसंदीदा" धागा सेमाफोर रखता है और एक गैर-पसंदीदा इसे तुरंत रिलीज़ करता है। एक बार जब उन्होंने इस सेमफोर को अधिग्रहण कर लिया है तो वांछित थ्रेड अन्य सभी आने वाले धागे को अवरुद्ध करता है।

फिर दूसरा "सामान्य सैमफोर" को पूरा करने के लिए अधिग्रहण किया जाना है। लेकिन गैर-पसंदीदा धागा फिर जांच करता है कि अस्थिर चर का उपयोग करके प्रतीक्षा करने वाला कोई पसंदीदा धागा नहीं है)।अगर कोई इंतजार नहीं कर रहा है तो वह बस जारी रहता है; अगर कोई इंतजार कर रहा है, तो वह सामान्य सेमफोर जारी करता है और फिर से कॉल हासिल करता है।

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

public final class SemaphoreWithMemory { 

private volatile boolean favouredAquired = false; 
private final Semaphore favouredSemaphore = new Semaphore(1, true); 
private final Semaphore normalSemaphore = new Semaphore(1, true); 
private final Set<Integer> favoured = new ConcurrentSkipListSet<Integer>(); 

public void acquire() throws InterruptedException { 
    normalSemaphore.acquire(); 
} 

public void acquire(int id) throws InterruptedException { 
    boolean idIsFavoured = favoured.contains(id); 
    favouredSemaphore.acquire(); 
    if (!idIsFavoured) { 
     favouredSemaphore.release(); 
    } else { 
     favouredAquired = true; 
    } 
    normalSemaphore.acquire(); 

    // check again that there is no favoured thread waiting 
    if (!idIsFavoured) { 
     if (favouredAquired) { 
      normalSemaphore.release(); 
      acquire(); // starving probability! 
     } else { 
      favoured.add(id); 
     } 
    } 

} 

public void release() { 
    normalSemaphore.release(); 
    if (favouredAquired) { 
     favouredAquired = false; 
     favouredSemaphore.release(); 
    } 
} 

public void release(int id) { 
    favoured.remove(id); 
    release(); 
} 

} 

Answer 3

मैं Ceki द्वारा इस लेख पढ़ सकते हैं और कैसे पक्षपातपूर्ण सेमाफोर अधिग्रहण हो सकता है रुचि थी (के बाद से मुझे लगा कि "पक्षपातपूर्ण ताला" व्यवहार के रूप में अच्छी तरह से सेमाफोर में मतलब होता है ..)। मेरे प्रोसेसर और एक सन जेवीएम 1.6 के साथ मेरे हार्डवेयर पर, यह वास्तव में बहुत समान पट्टा में परिणाम देता है।

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

नोट कोड नीचे Ceki के कोड पर आधारित है here

कोड:

import java.util.concurrent.*; 

public class BiasedSemaphore implements Runnable { 
    static ThreadLocal<Boolean> favored = new ThreadLocal<Boolean>(){ 
     private boolean gaveOut = false; 
     public synchronized Boolean initialValue(){ 
      if(!gaveOut){ 
       System.out.println("Favored " + Thread.currentThread().getName()); 
       gaveOut = true; 
       return true; 
      } 
      return false; 
     } 
    }; 

    static int THREAD_COUNT = Runtime.getRuntime().availableProcessors(); 
    static Semaphore SEM = new Semaphore(1); 
    static Runnable[] RUNNABLE_ARRAY = new Runnable[THREAD_COUNT]; 
    static Thread[] THREAD_ARRAY = new Thread[THREAD_COUNT]; 

    private int counter = 0; 

    public static void main(String args[]) throws InterruptedException { 
     printEnvironmentInfo(); 
     execute(); 
     printResults(); 
    } 

    public static void printEnvironmentInfo() { 
     System.out.println("java.runtime.version = " 
       + System.getProperty("java.runtime.version")); 
     System.out.println("java.vendor   = " 
       + System.getProperty("java.vendor")); 
     System.out.println("java.version   = " 
       + System.getProperty("java.version")); 
     System.out.println("os.name    = " 
       + System.getProperty("os.name")); 
     System.out.println("os.version   = " 
       + System.getProperty("os.version")); 
    } 

    public static void execute() throws InterruptedException { 
     for (int i = 0; i < THREAD_COUNT; i++) { 
      RUNNABLE_ARRAY[i] = new BiasedSemaphore(); 
      THREAD_ARRAY[i] = new Thread(RUNNABLE_ARRAY[i]); 
      System.out.println("Runnable at "+i + " operated with "+THREAD_ARRAY[i]); 
     } 

     for (Thread t : THREAD_ARRAY) { 
      t.start(); 
     } 
     // let the threads run for a while 
     Thread.sleep(10000); 

     for (int i = 0; i< THREAD_COUNT; i++) { 
      THREAD_ARRAY[i].interrupt(); 
     } 

     for (Thread t : THREAD_ARRAY) { 
      t.join(); 
     } 
    } 

    public static void printResults() { 
     System.out.println("Ran with " + THREAD_COUNT + " threads"); 
     for (int i = 0; i < RUNNABLE_ARRAY.length; i++) { 
      System.out.println("runnable[" + i + "]: " + RUNNABLE_ARRAY[i]); 
     } 
    } 


    public void run() { 
     while (!Thread.currentThread().isInterrupted()) { 
      if (favored.get()) { 
       stuff(); 
      } else { 
       Thread.yield(); 
//    try { 
//     Thread.sleep(1); 
//    } catch (InterruptedException e) { 
//     Thread.currentThread().interrupt(); 
//    } 
       stuff(); 
      } 
     } 
    } 

    private void stuff() { 
     if (SEM.tryAcquire()) { 
      //favored.set(true); 
      counter++; 
      try { 
       Thread.sleep(10); 
      } catch (InterruptedException ex) { 
       Thread.currentThread().interrupt(); 
      } 
      SEM.release(); 
     } else { 
      //favored.set(false); 
     } 
    } 

    public String toString() { 
     return "counter=" + counter; 
    } 
} 

परिणाम:

java.runtime.version = 1.6.0_21-b07 
java.vendor   = Sun Microsystems Inc. 
java.version   = 1.6.0_21 
os.name    = Windows Vista 
os.version   = 6.0 
Runnable at 0 operated with Thread[Thread-0,5,main] 
Runnable at 1 operated with Thread[Thread-1,5,main] 
Favored Thread-0 
Ran with 2 threads 
runnable[0]: counter=503 
runnable[1]: counter=425 

10 के बजाय 30 सेकंड के साथ की कोशिश की:

java.runtime.version = 1.6.0_21-b07 
java.vendor   = Sun Microsystems Inc. 
java.version   = 1.6.0_21 
os.name    = Windows Vista 
os.version   = 6.0 
Runnable at 0 operated with Thread[Thread-0,5,main] 
Runnable at 1 operated with Thread[Thread-1,5,main] 
Favored Thread-1 
Ran with 2 threads 
runnable[0]: counter=1274 
runnable[1]: counter=1496 

पी.एस .: की तरह एक बहुत बुरा विचार "बाहर लटक" था लग रहा है। जब मैंने गैर-पसंदीदा धागे के लिए SEM.tryAcquire(1,TimeUnit.MILLISECONDS); और SEM.tryAcquire() को गैर-पसंदीदा धागे के लिए कॉल करने का प्रयास किया, तो गैर-पसंदीदा धागे को पसंदीदा धागे से लगभग 5 गुना अधिक परमिट मिला!

इसके अलावा, मैं यह भी जोड़ना चाहता हूं कि ये परिणाम केवल 1 विशेष स्थिति के तहत मापा जाता है, इसलिए यह स्पष्ट नहीं है कि ये उपाय अन्य स्थितियों में कैसे व्यवहार करते हैं।

Answer 4

यह मुझे मारता है कि ऐसा करने का सबसे आसान तरीका सेमफोरस को आजमाने और गठबंधन करने के लिए नहीं है, बल्कि इसे मॉनीटर के शीर्ष पर खरोंच से बनाने के लिए है। यह आम तौर पर जोखिम भरा होता है, लेकिन इस मामले में, क्योंकि java.util.concurrent में कोई अच्छा बिल्डिंग ब्लॉक नहीं है, यह करने का सबसे स्पष्ट तरीका है।

public class SemaphoreWithMemory { 

    private final Set<Integer> favouredIDs = new HashSet<Integer>(); 
    private final Object favouredLock = new Object(); 
    private final Object ordinaryLock = new Object(); 
    private boolean available = true; 
    private int favouredWaiting = 0; 

    /** 
    Acquires the permit. Blocks until the permit is acquired. 
    */ 
    public void acquire(int id) throws InterruptedException { 
     Object lock; 
     boolean favoured = false; 

     synchronized (this) { 
      // fast exit for uncontended lock 
      if (available) { 
       doAcquire(favoured, id); 
       return; 
      } 
      favoured = favouredIDs.contains(id); 
      if (favoured) { 
       lock = favouredLock; 
       ++favouredWaiting; 
      } 
      else { 
       lock = ordinaryLock; 
      } 
     } 

     while (true) { 
      synchronized (this) { 
       if (available) { 
        doAcquire(favoured, id); 
        return; 
       } 
      } 
      synchronized (lock) { 
       lock.wait(); 
      } 
     } 
    } 

    private void doAcquire(boolean favoured, int id) { 
     available = false; 
     if (favoured) --favouredWaiting; 
     else favouredIDs.add(id); 
    } 

    /** 
    Releases the permit. 
    */ 
    public synchronized void release() { 
     available = true; 
     Object lock = (favouredWaiting > 0) ? favouredLock : ordinaryLock; 
     synchronized (lock) { 
      lock.notify(); 
     } 
    } 

} 

Answer 5

अधिग्रहण मॉडल अवरुद्ध के लिए, क्या इस बारे में:

यहाँ क्या मैं के साथ आया है

public class SemWithPreferred { 
    int max; 
    int avail; 
    int preferredThreads; 

    public SemWithPreferred(int max, int avail) { 
     this.max = max; 
     this.avail = avail; 
    } 

    synchronized public void get(int id) throws InterruptedException { 
     boolean thisThreadIsPreferred = idHasBeenServedSuccessfullyBefore(id); 
     if (thisThreadIsPreferred) { 
      preferredThreads++; 
     } 
     while (! (avail > 0 && (preferredThreads == 0 || thisThreadIsPreferred))) { 
      wait(); 
     } 
     System.out.println(String.format("granted, id = %d, preferredThreads = %d", id, preferredThreads)); 
     avail -= 1; 
     if (thisThreadIsPreferred) { 
      preferredThreads--; 
      notifyAll(); // removal of preferred thread could affect other threads' wait predicate 
     } 
    } 

    synchronized public void put() { 
     if (avail < max) { 
      avail += 1; 
      notifyAll(); 
     } 
    } 

    boolean idHasBeenServedSuccessfullyBefore(int id) { 
     // stubbed out, this just treats any id that is a 
     // multiple of 5 as having been served successfully before 
     return id % 5 == 0; 
    } 
}