asyncio中的鎖定優化失敗

Question

更新：編輯標題以關注主要問題。查看我的答案以獲得完整更新。

在下面的代碼中，a()和b()是相同的。他們每個人同時計數從0到9，同時每2次獲得一個鎖併產生一個鎖。

import asyncio 

lock = asyncio.Lock() 

def a(): 
yield from lock.acquire() 
for i in range(10): 
    print('a: ' + str(i)) 
    if i % 2 == 0: 
    lock.release() 
    yield from lock.acquire() 
lock.release() 

def b(): 
yield from lock.acquire() 
for i in range(10): 
    print('b: ' + str(i)) 
    if i % 2 == 0: 
    lock.release() 
    yield from lock.acquire() 
lock.release() 

asyncio.get_event_loop().run_until_complete(asyncio.gather(a(), b())) 

print('done') 

我的預期交織輸出，而是我得到：

b: 0 
b: 1 
b: 2 
b: 3 
b: 4 
b: 5 
b: 6 
b: 7 
b: 8 
b: 9 
a: 0 
a: 1 
a: 2 
a: 3 
a: 4 
a: 5 
a: 6 
a: 7 
a: 8 
a: 9 
done 

看來，第二yield實際上並沒有屈服，而是立即重新獲取鎖並繼續。

這對我來說似乎是一個錯誤。我對嗎？還是有另一種解釋？

下面的代碼，用額外的初始「noop」yield修改，工作正常。這讓我相信鎖確實是公平的，可能是正確的。

import asyncio 

lock = asyncio.Lock() 

def a(): 
yield from lock.acquire() 
yield from asyncio.sleep(0) 
for i in range(10): 
    print('a: ' + str(i)) 
    if i % 2 == 0: 
    lock.release() 
    yield from lock.acquire() 
lock.release() 

def b(): 
yield from lock.acquire() 
yield from asyncio.sleep(0) 
for i in range(10): 
    print('b: ' + str(i)) 
    if i % 2 == 0: 
    lock.release() 
    yield from lock.acquire() 
lock.release() 

asyncio.get_event_loop().run_until_complete(asyncio.gather(a(), b())) 

print('done') 

輸出：

a: 0 
b: 0 
a: 1 
a: 2 
b: 1 
b: 2 
a: 3 
a: 4 
b: 3 
b: 4 
a: 5 
a: 6 
b: 5 
b: 6 
a: 7 
a: 8 
b: 7 
b: 8 
a: 9 
b: 9 
done 

注意，我在開始做無操作產量只有一次，而不是每2個字。然而，這樣做會導致在第一段代碼中按照預期每隔兩次進行交錯。

在調度程序中只有一些優化（我認爲是一個錯誤），當獲取一個沒有其他人正在等待的鎖時，並不真正yield。

如何解釋第一個輸出？

Answer 1

更新：下面是我對GitHub的問題評論（link）的光過時。該評論指出，您可以使用Lock.locked()來預測Lock.acquire()是否會收益。它還觀察到許多其他協程在快速情況下不會產生，因此即使考慮修復所有協程也是失敗的原因。最後，它將這個問題與一個不同的問題如何解決聯繫起來，並表明它可以得到更好的解決。那就是要求一個asyncio.nop()的方法，它只會屈服於調度器，而別的什麼都不做。而不是添加該方法的，他們決定過載asyncio.sleep(0)和「去優化」它（在這種dicussion的lock.acquire()的上下文中），以產生到所述調度時參數爲0以下。

原始的答案，但通過上述取代段：

的根本原因是asyncio.lock的實現嘗試在其first three lines太聰明，不給控制回調度如果沒有服務員：

if not self._locked and all(w.cancelled() for w in self._waiters): 
    self._locked = True 
    return True 

然而，正如我第一個例子表明，這可以防止其他協程成爲一名服務員。他們沒有機會跑到他們試圖獲得鎖定的地步。

的inefficent解決方法是總是立即yield from asyncio.sleep(0)之前獲取鎖。

這是低效率的，因爲在一般情況下會有其他服務員，並且獲取鎖也會使控制權回到調度程序。因此，在大多數情況下，你會屈服控制回調度兩次，這吮吸。

另請注意，鎖的文檔含糊地說：「此方法阻塞，直到鎖被解鎖，然後將其設置爲鎖定並返回True。」當然給人的印象是它在獲得鎖之前會對調度器產生控制。

在我看來，正確的做法是讓鎖執行總是屈服，而不是太聰明。 或者，鎖實施應該有告訴你，如果收購，使你的代碼可以手動生成如果鎖收購不會它是否會產生一種方法。 另一種選擇是讓acquire()調用返回，告訴你它是否真的產生了價值。這不太可取，但仍比現狀好。

有人可能會認爲更好的解決方法可能是在release()時手動生成。但是，如果你看一個緊密循環釋放和重新獲取作品的片斷之後，然後它相當於同樣的事情 - 在通常情況下它仍然會在獲得時間產生兩次，一次是在發佈時，再次增加效率。

Answer 2

目前還不清楚你想要達到什麼效果，但看起來好像Lock不是你需要的工具。交織Python代碼，你可以做簡單：

def foo(tag, n): 
    for i in range(n): 
     print("inside", tag, i) 
     yield (tag, i) 


print('start') 

for x in zip(foo('A', 10), foo('B', 10)): 
    print(x) 

print('done') 

無需asyncio或threading。無論如何，asyncio沒有IO沒有很大意義。

threading。 Lock用於同步程序的關鍵部分，否則以獨立線程運行。 asyncio.Lock將允許其他協程繼續與IO而一個協同程序等待：

import asyncio 
import random 

lock = asyncio.Lock() 


async def foo(tag): 
    print(tag, "Start") 
    for i in range(10): 
     print(tag, '>', i) 
     await asyncio.sleep(random.uniform(0.1, 1)) 
     print(tag, '<', i) 

    async with lock: 
     # only one coroutine can execute the critical section at once. 
     # other coroutines can still use IO. 
     print(tag, "CRITICAL START") 
     await asyncio.sleep(1) 
     print(tag, "STILL IN CRITICAL") 
     await asyncio.sleep(1) 
     print(tag, "CRITICAL END") 

    for i in range(10, 20): 
     print(tag, '>', i) 
     await asyncio.sleep(random.uniform(0.1, 1)) 
     print(tag, '<', i) 

    print(tag, "Done") 


print('start') 

loop = asyncio.get_event_loop() 
tasks = asyncio.gather(foo('A'), foo('B'), foo('C')) 
loop.run_until_complete(tasks) 
loop.close() 

print('done') 

請記住，關鍵字yield並不總是服從產量 :-)英文含義。

你可以看到，async with lock會更有意義，async with lock將立即獲得鎖定，而無需等待其他協同程序做更多的工作：到達關鍵部分的第一個couroutine應該開始運行它。 （即，在async with lock:之前添加await asyncio.sleep(0)只是沒有任何意義。）