जावा में कोरआउट

Question

मैं इस पृष्ठ को coroutines in Python और इस Wikipedia page. के बारे में पढ़ रहा हूं मैंने देखा कि कुछ libraries in Java कोरआउट को कार्यान्वित कर रहे हैं।

मेरा प्रश्न है: क्या कोई ज्ञात कारण है कि जावा डिजाइनरों अब तक coroutines को लागू नहीं करने का फैसला है और वहाँ किसी भी जावा के भविष्य के संस्करण में शामिल करने की योजना है?

धन्यवाद।

Answer 1

असल में सह-रूटीन की अवधारणा जावा 0 थ्रेडिंग सिस्टम के पहले डिज़ाइन थी। wait/notify तंत्र सह दिनचर्या का एक साधारण प्रपत्र जहां notify ज्यादा, किया गया है विशेष रूप से संरचनाओं धागा सुरक्षित नहीं बल्कि एल्गोरिथम की तुलना में बनाने के लिए तब से yield आदि

के बराबर है है। यह बोध कि यह कोड कि सिंक्रनाइज़ चाहिए/उपज लेकिन धागे कि धागा होने चाहिए के बीच संवाद के लिए इस्तेमाल किया डेटा संरचना नहीं है से निकला है।

Answer 2

पर सवाल की हिस्सा "क्या कोई योजना है ...", जवाब है:

इस स्तर पर

नहीं

JEP सूची (http://openjdk.java.net/jeps/0) नहीं है coroutines का कोई उल्लेख करें। सूची जावा 8 में जोड़े गए फीचर्स को कवर करती है, जावा 9 के लिए जोड़ा या लक्षित है, या भविष्य के रिलीज के लिए प्रस्तावित है।

दिलचस्प है, वहाँ एक RFE मार्च 2013 (https://bugs.openjdk.java.net/browse/JDK-8029988) में प्रस्तुत किया गया था। आरएफई को केवल एक वोट मिला, और यह एक जेईपी जमा करने के सुझाव के साथ 9 महीने बंद कर दिया गया था। इस विचार को आगे बढ़ाने के लिए किसी ने भी परेशान नहीं किया है, जो मुझे बता रहा है।

Answer 3

वहाँ एक एक और विकल्प यहाँ Java6 के लिए है +

एक pythonic coroutine कार्यान्वयन: (जैसे फिबोनैकी संख्या)

थ्रेड संस्करण

import java.lang.ref.WeakReference; 
import java.util.ArrayList; 
import java.util.List; 
import java.util.concurrent.*; 
import java.util.concurrent.atomic.AtomicBoolean; 
import java.util.concurrent.atomic.AtomicReference; 

class CorRunRAII { 
    private final List<WeakReference<? extends CorRun>> resources = new ArrayList<>(); 

    public CorRunRAII add(CorRun resource) { 
     if (resource == null) { 
      return this; 
     } 
     resources.add(new WeakReference<>(resource)); 

     return this; 
    } 

    public CorRunRAII addAll(List<? extends CorRun> arrayList) { 
     if (arrayList == null) { 
      return this; 
     } 
     for (CorRun corRun : arrayList) { 
      add(corRun); 
     } 

     return this; 
    } 

    @Override 
    protected void finalize() throws Throwable { 
     super.finalize(); 

     for (WeakReference<? extends CorRun> corRunWeakReference : resources) { 
      CorRun corRun = corRunWeakReference.get(); 
      if (corRun != null) { 
       corRun.stop(); 
      } 
     } 
    } 
} 

class CorRunYieldReturn<ReceiveType, YieldReturnType> { 
    public final AtomicReference<ReceiveType> receiveValue; 
    public final LinkedBlockingDeque<AtomicReference<YieldReturnType>> yieldReturnValue; 

    CorRunYieldReturn(AtomicReference<ReceiveType> receiveValue, LinkedBlockingDeque<AtomicReference<YieldReturnType>> yieldReturnValue) { 
     this.receiveValue = receiveValue; 
     this.yieldReturnValue = yieldReturnValue; 
    } 
} 

interface CorRun<ReceiveType, YieldReturnType> extends Runnable, Callable<YieldReturnType> { 
    boolean start(); 
    void stop(); 
    void stop(final Throwable throwable); 
    boolean isStarted(); 
    boolean isEnded(); 
    Throwable getError(); 

    ReceiveType getReceiveValue(); 
    void setResultForOuter(YieldReturnType resultForOuter); 
    YieldReturnType getResultForOuter(); 

    YieldReturnType receive(ReceiveType value); 
    ReceiveType yield(); 
    ReceiveType yield(YieldReturnType value); 
    <TargetReceiveType, TargetYieldReturnType> TargetYieldReturnType yieldFrom(final CorRun<TargetReceiveType, TargetYieldReturnType> another); 
    <TargetReceiveType, TargetYieldReturnType> TargetYieldReturnType yieldFrom(final CorRun<TargetReceiveType, TargetYieldReturnType> another, final TargetReceiveType value); 
} 

abstract class CorRunSync<ReceiveType, YieldReturnType> implements CorRun<ReceiveType, YieldReturnType> { 

    private ReceiveType receiveValue; 
    public final List<WeakReference<CorRun>> potentialChildrenCoroutineList = new ArrayList<>(); 

    // Outside 

    private AtomicBoolean isStarted = new AtomicBoolean(false); 
    private AtomicBoolean isEnded = new AtomicBoolean(false); 
    private Throwable error; 

    private YieldReturnType resultForOuter; 

    @Override 
    public boolean start() { 

     boolean isStarted = this.isStarted.getAndSet(true); 
     if ((! isStarted) 
       && (! isEnded())) { 
      receive(null); 
     } 

     return isStarted; 
    } 

    @Override 
    public void stop() { 
     stop(null); 
    } 

    @Override 
    public void stop(Throwable throwable) { 
     isEnded.set(true); 
     if (throwable != null) { 
      error = throwable; 
     } 

     for (WeakReference<CorRun> weakReference : potentialChildrenCoroutineList) { 
      CorRun child = weakReference.get(); 
      if (child != null) { 
       child.stop(); 
      } 
     } 
    } 

    @Override 
    public boolean isStarted() { 
     return isStarted.get(); 
    } 

    @Override 
    public boolean isEnded() { 
     return isEnded.get(); 
    } 

    @Override 
    public Throwable getError() { 
     return error; 
    } 

    @Override 
    public ReceiveType getReceiveValue() { 
     return receiveValue; 
    } 

    @Override 
    public void setResultForOuter(YieldReturnType resultForOuter) { 
     this.resultForOuter = resultForOuter; 
    } 

    @Override 
    public YieldReturnType getResultForOuter() { 
     return resultForOuter; 
    } 

    @Override 
    public synchronized YieldReturnType receive(ReceiveType value) { 
     receiveValue = value; 

     run(); 

     return getResultForOuter(); 
    } 

    @Override 
    public ReceiveType yield() { 
     return yield(null); 
    } 

    @Override 
    public ReceiveType yield(YieldReturnType value) { 
     resultForOuter = value; 
     return receiveValue; 
    } 

    @Override 
    public <TargetReceiveType, TargetYieldReturnType> TargetYieldReturnType yieldFrom(CorRun<TargetReceiveType, TargetYieldReturnType> another) { 
     return yieldFrom(another, null); 
    } 

    @Override 
    public <TargetReceiveType, TargetYieldReturnType> TargetYieldReturnType yieldFrom(CorRun<TargetReceiveType, TargetYieldReturnType> another, TargetReceiveType value) { 
     if (another == null || another.isEnded()) { 
      throw new RuntimeException("Call null or isEnded coroutine"); 
     } 

     potentialChildrenCoroutineList.add(new WeakReference<CorRun>(another)); 

     synchronized (another) { 
      boolean isStarted = another.start(); 
      boolean isJustStarting = ! isStarted; 
      if (isJustStarting && another instanceof CorRunSync) { 
       return another.getResultForOuter(); 
      } 

      return another.receive(value); 
     } 
    } 

    @Override 
    public void run() { 
     try { 
      this.call(); 
     } 
     catch (Exception e) { 
      e.printStackTrace(); 

      stop(e); 
      return; 
     } 
    } 
} 

abstract class CorRunThread<ReceiveType, YieldReturnType> implements CorRun<ReceiveType, YieldReturnType> { 

    private final ExecutorService childExecutorService = newExecutorService(); 
    private ExecutorService executingOnExecutorService; 

    private static final CorRunYieldReturn DUMMY_COR_RUN_YIELD_RETURN = new CorRunYieldReturn(new AtomicReference<>(null), new LinkedBlockingDeque<AtomicReference>()); 

    private final CorRun<ReceiveType, YieldReturnType> self; 
    public final List<WeakReference<CorRun>> potentialChildrenCoroutineList; 
    private CorRunYieldReturn<ReceiveType, YieldReturnType> lastCorRunYieldReturn; 

    private final LinkedBlockingDeque<CorRunYieldReturn<ReceiveType, YieldReturnType>> receiveQueue; 

    // Outside 

    private AtomicBoolean isStarted = new AtomicBoolean(false); 
    private AtomicBoolean isEnded = new AtomicBoolean(false); 
    private Future<YieldReturnType> future; 
    private Throwable error; 

    private final AtomicReference<YieldReturnType> resultForOuter = new AtomicReference<>(); 

    CorRunThread() { 
     executingOnExecutorService = childExecutorService; 

     receiveQueue = new LinkedBlockingDeque<>(); 
     potentialChildrenCoroutineList = new ArrayList<>(); 

     self = this; 
    } 

    @Override 
    public void run() { 
     try { 
      self.call(); 
     } 
     catch (Exception e) { 
      stop(e); 
      return; 
     } 

     stop(); 
    } 

    @Override 
    public abstract YieldReturnType call(); 

    @Override 
    public boolean start() { 
     return start(childExecutorService); 
    } 

    protected boolean start(ExecutorService executorService) { 
     boolean isStarted = this.isStarted.getAndSet(true); 
     if (!isStarted) { 
      executingOnExecutorService = executorService; 
      future = (Future<YieldReturnType>) executingOnExecutorService.submit((Runnable) self); 
     } 
     return isStarted; 
    } 

    @Override 
    public void stop() { 
     stop(null); 
    } 

    @Override 
    public void stop(final Throwable throwable) { 
     if (throwable != null) { 
      error = throwable; 
     } 
     isEnded.set(true); 

     returnYieldValue(null); 
     // Do this for making sure the coroutine has checked isEnd() after getting a dummy value 
     receiveQueue.offer(DUMMY_COR_RUN_YIELD_RETURN); 

     for (WeakReference<CorRun> weakReference : potentialChildrenCoroutineList) { 
      CorRun child = weakReference.get(); 
      if (child != null) { 
       if (child instanceof CorRunThread) { 
        ((CorRunThread)child).tryStop(childExecutorService); 
       } 
      } 
     } 

     childExecutorService.shutdownNow(); 
    } 

    protected void tryStop(ExecutorService executorService) { 
     if (this.executingOnExecutorService == executorService) { 
      stop(); 
     } 
    } 

    @Override 
    public boolean isEnded() { 
     return isEnded.get() || (
       future != null && (future.isCancelled() || future.isDone()) 
       ); 
    } 

    @Override 
    public boolean isStarted() { 
     return isStarted.get(); 
    } 

    public Future<YieldReturnType> getFuture() { 
     return future; 
    } 

    @Override 
    public Throwable getError() { 
     return error; 
    } 

    @Override 
    public void setResultForOuter(YieldReturnType resultForOuter) { 
     this.resultForOuter.set(resultForOuter); 
    } 

    @Override 
    public YieldReturnType getResultForOuter() { 
     return this.resultForOuter.get(); 
    } 

    @Override 
    public YieldReturnType receive(ReceiveType value) { 

     LinkedBlockingDeque<AtomicReference<YieldReturnType>> yieldReturnValue = new LinkedBlockingDeque<>(); 

     offerReceiveValue(value, yieldReturnValue); 

     try { 
      AtomicReference<YieldReturnType> takeValue = yieldReturnValue.take(); 
      return takeValue == null ? null : takeValue.get(); 
     } catch (InterruptedException e) { 
      e.printStackTrace(); 
     } 

     return null; 
    } 

    @Override 
    public ReceiveType yield() { 
     return yield(null); 
    } 

    @Override 
    public ReceiveType yield(final YieldReturnType value) { 
     returnYieldValue(value); 

     return getReceiveValue(); 
    } 

    @Override 
    public <TargetReceiveType, TargetYieldReturnType> TargetYieldReturnType yieldFrom(final CorRun<TargetReceiveType, TargetYieldReturnType> another) { 
     return yieldFrom(another, null); 
    } 

    @Override 
    public <TargetReceiveType, TargetYieldReturnType> TargetYieldReturnType yieldFrom(final CorRun<TargetReceiveType, TargetYieldReturnType> another, final TargetReceiveType value) { 
     if (another == null || another.isEnded()) { 
      throw new RuntimeException("Call null or isEnded coroutine"); 
     } 

     boolean isStarted = false; 
     potentialChildrenCoroutineList.add(new WeakReference<CorRun>(another)); 

     synchronized (another) { 
      if (another instanceof CorRunThread) { 
       isStarted = ((CorRunThread)another).start(childExecutorService); 
      } 
      else { 
       isStarted = another.start(); 
      } 

      boolean isJustStarting = ! isStarted; 
      if (isJustStarting && another instanceof CorRunSync) { 
       return another.getResultForOuter(); 
      } 

      TargetYieldReturnType send = another.receive(value); 
      return send; 
     } 
    } 

    @Override 
    public ReceiveType getReceiveValue() { 

     setLastCorRunYieldReturn(takeLastCorRunYieldReturn()); 

     return lastCorRunYieldReturn.receiveValue.get(); 
    } 

    protected void returnYieldValue(final YieldReturnType value) { 
     CorRunYieldReturn<ReceiveType, YieldReturnType> corRunYieldReturn = lastCorRunYieldReturn; 
     if (corRunYieldReturn != null) { 
      corRunYieldReturn.yieldReturnValue.offer(new AtomicReference<>(value)); 
     } 
    } 

    protected void offerReceiveValue(final ReceiveType value, LinkedBlockingDeque<AtomicReference<YieldReturnType>> yieldReturnValue) { 
     receiveQueue.offer(new CorRunYieldReturn(new AtomicReference<>(value), yieldReturnValue)); 
    } 

    protected CorRunYieldReturn<ReceiveType, YieldReturnType> takeLastCorRunYieldReturn() { 
     try { 
      return receiveQueue.take(); 
     } catch (InterruptedException e) { 
      e.printStackTrace(); 
     } 

     return null; 
    } 

    protected void setLastCorRunYieldReturn(CorRunYieldReturn<ReceiveType,YieldReturnType> lastCorRunYieldReturn) { 
     this.lastCorRunYieldReturn = lastCorRunYieldReturn; 
    } 

    protected ExecutorService newExecutorService() { 
     return Executors.newCachedThreadPool(getThreadFactory()); 
    } 

    protected ThreadFactory getThreadFactory() { 
     return new ThreadFactory() { 
      @Override 
      public Thread newThread(final Runnable runnable) { 
       Thread thread = new Thread(runnable); 
       thread.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() { 
        @Override 
        public void uncaughtException(Thread thread, Throwable throwable) { 
         throwable.printStackTrace(); 
         if (runnable instanceof CorRun) { 
          CorRun self = (CorRun) runnable; 
          self.stop(throwable); 
          thread.interrupt(); 
         } 
        } 
       }); 
       return thread; 
      } 
     }; 
    } 
} 

अब आप इस तरह से में pythonic coroutines उपयोग कर सकते हैं:

class Fib extends CorRunThread<Integer, Integer> { 

    @Override 
    public Integer call() { 
     Integer times = getReceiveValue(); 
     do { 
      int a = 1, b = 1; 
      for (int i = 0; times != null && i < times; i++) { 
       int temp = a + b; 
       a = b; 
       b = temp; 
      } 
      // A pythonic "yield", i.e., it returns `a` to the caller and waits `times` value from the next caller 
      times = yield(a); 
     } while (! isEnded()); 

     setResultForOuter(Integer.MAX_VALUE); 
     return getResultForOuter(); 
    } 
} 

class MainRun extends CorRunThread<String, String> { 

    @Override 
    public String call() { 

     // The fib coroutine would be recycled by its parent 
     // (no requirement to call its start() and stop() manually) 
     // Otherwise, if you want to share its instance and start/stop it manually, 
     // please start it before being called by yieldFrom() and stop it in the end. 
     Fib fib = new Fib(); 
     String result = ""; 
     Integer current; 
     int times = 10; 
     for (int i = 0; i < times; i++) { 

      // A pythonic "yield from", i.e., it calls fib with `i` parameter and waits for returned value as `current` 
      current = yieldFrom(fib, i); 

      if (fib.getError() != null) { 
       throw new RuntimeException(fib.getError()); 
      } 

      if (current == null) { 
       continue; 
      } 

      if (i > 0) { 
       result += ","; 
      } 
      result += current; 

     } 

     setResultForOuter(result); 

     return result; 
    } 
} 

सिंक (गैर धागा) संस्करण:

class Fib extends CorRunSync<Integer, Integer> { 

    @Override 
    public Integer call() { 
     Integer times = getReceiveValue(); 

     int a = 1, b = 1; 
     for (int i = 0; times != null && i < times; i++) { 
      int temp = a + b; 
      a = b; 
      b = temp; 
     } 
     yield(a); 

     return getResultForOuter(); 
    } 
} 

class MainRun extends CorRunSync<String, String> { 

    @Override 
    public String call() { 

     CorRun<Integer, Integer> fib = null; 
     try { 
      fib = new Fib(); 
     } catch (Exception e) { 
      e.printStackTrace(); 
     } 

     String result = ""; 
     Integer current; 
     int times = 10; 
     for (int i = 0; i < times; i++) { 

      current = yieldFrom(fib, i); 

      if (fib.getError() != null) { 
       throw new RuntimeException(fib.getError()); 
      } 

      if (current == null) { 
       continue; 
      } 

      if (i > 0) { 
       result += ","; 
      } 
      result += current; 
     } 

     stop(); 
     setResultForOuter(result); 

     if (Utils.isEmpty(result)) { 
      throw new RuntimeException("Error"); 
     } 

     return result; 
    } 
} 

---

निष्पादन (दोनों संस्करणों में काम करेंगे):

// Run the entry coroutine 
MainRun mainRun = new MainRun(); 
mainRun.start(); 

// Wait for mainRun ending for 5 seconds 
long startTimestamp = System.currentTimeMillis(); 
while(!mainRun.isEnded()) { 
    if (System.currentTimeMillis() - startTimestamp > TimeUnit.SECONDS.toMillis(5)) { 
     throw new RuntimeException("Wait too much time"); 
    } 
} 
// The result should be "1,1,2,3,5,8,13,21,34,55" 
System.out.println(mainRun.getResultForOuter());