भविष्य

फ्यूचर एक कंटेनर है जो या तो गणना के परिणाम या उस गणना के दौरान हुई त्रुटि को संग्रहीत कर सकता है। जब कोई भविष्य बनाया जाता है, तो वह PENDING स्थिति में शुरू होता है। संभवतः परीक्षण उद्देश्यों को छोड़कर, लाइब्रेरी इस ऑब्जेक्ट को मैन्युअल रूप से बनाने का इरादा नहीं रखती है।

import concurrent.futures as futures

f = futures.Future()
assert(f._result is None)
assert(f._exception is None)
assert(f._state == 'PENDING')

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

import concurrent.futures as futures

def should_cancel_pending_future():
    f = futures.Future()
    assert(f._state == 'PENDING')
    assert(f.cancel())
    assert(f._state == 'CANCELLED')

def should_not_cancel_running_future():
    f = futures.Future()
    f.set_running_or_notify_cancel()
    assert(f._state == 'RUNNING')
    assert(not f.cancel())

def cancel_is_idempotent():
    f = futures.Future()
    assert(f.cancel())
    assert(f.cancel())


should_cancel_pending_future()
should_not_cancel_running_future()
cancel_is_idempotent()

थ्रेड पूल में अनुरोधित ऑपरेशन या तो गणना मूल्य के साथ पूरा हो सकता है या परिणाम में त्रुटि हो सकती है। परिणाम चाहे जो भी हो, भविष्य समाप्त स्थिति में परिवर्तित हो जाता है। फिर परिणाम या त्रुटि को संबंधित फ़ील्ड में संग्रहीत किया जाता है।

import concurrent.futures as futures

def future_completed_with_result():
    f = futures.Future()
    f.set_result('foo')
    assert(f._state == 'FINISHED')
    assert(f._result == 'foo')
    assert(f._exception is None)

def future_completed_with_exception():
    f = futures.Future()
    f.set_exception(NameError())
    assert(f._state == 'FINISHED')
    assert(f._result is None)
    assert(isinstance(f._exception, NameError))

future_completed_with_result()
future_completed_with_exception()

गणना के परिणाम को पुनः प्राप्त करने के लिए, परिणाम विधि का उपयोग किया जाता है। यदि गणना अभी तक पूरी नहीं हुई है, तो यह विधि गणना समाप्त होने या प्रतीक्षा समय समाप्त होने तक वर्तमान थ्रेड (जिससे परिणाम बुलाया गया था) को अवरुद्ध कर देगी।

यदि गणना त्रुटियों के बिना सफलतापूर्वक पूरी हो जाती है, तो परिणाम विधि गणना की गई मान लौटा देती है।

import concurrent.futures as futures
import time
import threading

f = futures.Future()
def target():
    time.sleep(1)
    f.set_result('foo')
threading.Thread(target=target).start()
assert(f.result() == 'foo')

यदि गणना के दौरान कोई अपवाद होता है, तो परिणाम उस अपवाद को बढ़ा देगा।

import concurrent.futures as futures
import time
import threading

f = futures.Future()
def target():
    time.sleep(1)
    f.set_exception(NameError)
threading.Thread(target=target).start()
try:
    f.result()
    raise Exception()
except NameError:
    assert(True)

यदि प्रतीक्षा करते समय विधि का समय समाप्त हो जाता है, तो एक टाइमआउटत्रुटि उत्पन्न हो जाती है।

import concurrent.futures as futures

f = futures.Future()
try:
    f.result(1)
    raise Exception()
except TimeoutError:
    assert(f._result is None)
    assert(f._exception is None)

रद्द की गई गणना का परिणाम प्राप्त करने का प्रयास एक CanceledError उत्पन्न करेगा।

import concurrent.futures as futures

f = futures.Future()
assert(f.cancel())
try:
    f.result()
    raise Exception()
except futures.CancelledError:
    assert(True)

प्रतीक्षा की रणनीति

विकास प्रक्रिया में, थ्रेड पूल पर एन गणना चलाने और उनके पूरा होने की प्रतीक्षा करने की आवश्यकता काफी आम है। इसे प्राप्त करने के लिए, लाइब्रेरी प्रतीक्षा फ़ंक्शन प्रदान करती है। प्रतीक्षा की कई रणनीतियाँ हैं: FIRST_COMPLETED, FIRST_EXCEPTION, ALL_COMPLETED.

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

import concurrent.futures as futures

def test(return_when):
    f1, f2, f3 = futures.Future(), futures.Future(), futures.Future()
    f1.cancel()
    f1.set_running_or_notify_cancel() # required
    f2.set_result('foo')
    f3.set_exception(NameError)

    r = futures.wait([f1, f2, f3], return_when=return_when)
    assert(len(r.done) == 3)
    assert(len(r.not_done) == 0)

for return_when in [futures.ALL_COMPLETED, futures.FIRST_EXCEPTION, futures.FIRST_COMPLETED]:
    test(return_when)

ALL_COMPLETED रणनीति

ALL_COMPLETED रणनीति सभी पारित वायदा के पूरा होने की प्रतीक्षा करने की गारंटी देती है, या उस क्षण तक पूरे किए गए वायदा के संग्रह के साथ एक समय समाप्ति के बाद बाहर निकलने की गारंटी देती है, जो अधूरा हो सकता है।

import concurrent.futures as futures
import threading
import time

def should_wait_for_all_futures_to_complete():
    f1 = futures.Future()
    f1.set_result('foo')
    f2 = futures.Future()

    def target():
        time.sleep(1)
        f2.set_result('bar')

    threading.Thread(target=target).start()
    r = futures.wait([f1, f2], return_when=futures.ALL_COMPLETED)
    assert(len(r.done) == 2)

def should_exit_on_timeout():
    f1 = futures.Future()
    f1.set_result('foo')
    f2 = futures.Future()
    r = futures.wait(fs=[f1, f2], timeout=1, return_when=futures.ALL_COMPLETED)
    assert(len(r.done) == 1)


should_wait_for_all_futures_to_complete()
should_exit_on_timeout()

FIRST_पूरा

FIRST_COMPLETED रणनीति कम से कम एक पूर्ण भविष्य या टाइमआउट के मामले में एक खाली संग्रह के साथ संग्रह की वापसी की गारंटी देती है। इस रणनीति का अर्थ यह नहीं है कि लौटाए गए संग्रह में एकाधिक तत्व नहीं हो सकते हैं।

import concurrent.futures as futures
import threading
import time

f1 = futures.Future()
f2 = futures.Future()

def target():
    time.sleep(1)
    f1.set_result(True)

threading.Thread(target=target).start()

r = futures.wait([f1, f2], return_when=futures.FIRST_COMPLETED)
assert(len(r.done) == 1)
assert(len(r.not_done) == 1)

FIRST_अपवाद

यदि कोई गणना किसी त्रुटि के साथ समाप्त होती है तो FIRST_EXCEPTION रणनीति प्रतीक्षा को बाधित करती है। यदि कोई अपवाद नहीं होता है, तो व्यवहार ALL_COMPLETED भविष्य के समान है।

import concurrent.futures as futures
import threading
import time

f1 = futures.Future()
f1.set_result('foo')
f2, f3 = futures.Future(), futures.Future()

def target():
    time.sleep(1)
    f2.set_exception(NameError())

threading.Thread(target=target).start()

r = futures.wait(fs=[f1, f2, f3], return_when=futures.FIRST_EXCEPTION)
assert(len(r.done) == 2)

थ्रेडपूल निष्पादक

ऑब्जेक्ट थ्रेड पूल बनाने के लिए ज़िम्मेदार है। इस ऑब्जेक्ट के साथ इंटरैक्ट करने की मुख्य विधि सबमिट विधि है। यह थ्रेड पूल में गणना पंजीकृत करने की अनुमति देता है। प्रतिक्रिया में, एक Future ऑब्जेक्ट लौटाया जाता है, जिसका उपयोग गणना की स्थिति की निगरानी करने और अंतिम परिणाम प्राप्त करने के लिए किया जाता है।

गुण

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