Python asyncio timeout lực

Question

Sử dụng asyncio một coroutine có thể được thực hiện với một thời gian chờ để nó bị hủy bỏ sau khi thời gian chờ:

@asyncio.coroutine 
def coro(): 
    yield from asyncio.sleep(10) 

loop = asyncio.get_event_loop() 
loop.run_until_complete(asyncio.wait_for(coro(), 5)) 

Ví dụ trên làm việc như mong đợi (nó lần ra sau 5 giây).

Tuy nhiên, khi coroutine không sử dụng asyncio.sleep() (hoặc khác asyncio coroutines) nó dường như không hết thời gian. Ví dụ:

@asyncio.coroutine 
def coro(): 
    import time 
    time.sleep(10) 

loop = asyncio.get_event_loop() 
loop.run_until_complete(asyncio.wait_for(coro(), 1)) 

Mất hơn 10 giây để chạy vì time.sleep(10) không bị hủy. Có thể thực thi việc hủy coroutine trong trường hợp này không?

Nếu sử dụng asyncio để giải quyết vấn đề này, tôi có thể làm như thế nào?

Answer 1

Không, bạn không thể ngắt một coroutine trừ khi nó kiểm soát trở lại vòng lặp sự kiện, có nghĩa là nó cần phải được bên trong một cuộc gọi yield from. asyncio là đơn luồng, vì vậy khi bạn đang chặn cuộc gọi time.sleep(10) trong ví dụ thứ hai của mình, không có cách nào để vòng lặp sự kiện chạy. Điều đó có nghĩa là khi hết thời gian chờ bạn đặt bằng cách sử dụng wait_for hết hạn, vòng lặp sự kiện sẽ không thể thực hiện hành động trên đó. Vòng lặp sự kiện không có cơ hội để chạy lại cho đến khi coro thoát, tại thời điểm quá muộn.

Đây là lý do tại sao, bạn nên luôn tránh mọi cuộc gọi chặn không đồng bộ; bất cứ lúc nào một khối cuộc gọi mà không có lãi suất cho vòng lặp sự kiện, không có gì khác trong chương trình của bạn có thể thực hiện, mà có lẽ không phải là những gì bạn muốn. Nếu bạn thực sự cần phải làm một dài, ngăn chặn hoạt động, bạn nên cố gắng sử dụng BaseEventLoop.run_in_executor để chạy nó trong một chủ đề hay quá trình bơi, mà sẽ tránh chặn vòng lặp sự kiện:

import asyncio 
import time 
from concurrent.futures import ProcessPoolExecutor 

@asyncio.coroutine 
def coro(loop): 
    ex = ProcessPoolExecutor(2) 
    yield from loop.run_in_executor(ex, time.sleep, 10) # This can be interrupted. 

loop = asyncio.get_event_loop() 
loop.run_until_complete(asyncio.wait_for(coro(loop), 1)) 

Answer 2

Thx @dano cho câu trả lời của bạn. Nếu chạy một coroutine không phải là một yêu cầu khó khăn, đây là một thiết kế lại, nhỏ gọn hơn phiên bản

import asyncio, time, concurrent 

timeout = 0.5 
loop = asyncio.get_event_loop() 
future = asyncio.wait_for(loop.run_in_executor(None, time.sleep, 2), timeout) 
try: 
    loop.run_until_complete(future) 
    print('Thx for letting me sleep') 
except concurrent.futures.TimeoutError: 
    print('I need more sleep !') 

Đối với tò mò, một chút gỡ lỗi trong Python 3.5.2 tôi cho thấy rằng đi qua None như một kết quả chấp hành viên trong việc tạo ra một _default_executor, như sau:

# _MAX_WORKERS = 5 
self._default_executor = concurrent.futures.ThreadPoolExecutor(_MAX_WORKERS) 

Answer 3

Ví dụ tôi đã xem để xử lý hết thời gian là rất nhỏ. Với thực tế, ứng dụng của tôi phức tạp hơn một chút. Trình tự là:

1. Khi một client kết nối đến máy chủ, có máy chủ tạo ra một kết nối đến máy chủ nội bộ
2. Khi kết nối máy chủ nội bộ là ok, chờ đợi cho khách hàng để gửi dữ liệu. Dựa trên dữ liệu này, chúng tôi có thể thực hiện một truy vấn đến máy chủ nội bộ.
3. Khi có dữ liệu để gửi đến máy chủ nội bộ, hãy gửi nó. Vì máy chủ nội bộ đôi khi không phản hồi đủ nhanh, hãy đưa yêu cầu này vào thời gian chờ.
4. Nếu lần hoạt động ra, sụp đổ tất cả các kết nối để báo hiệu cho khách hàng về lỗi

Để đạt được tất cả những điều trên, trong khi vẫn giữ sự kiện vòng lặp chạy, mã kết quả có chứa đoạn mã sau:

def connection_made(self, transport): 
    self.client_lock_coro = self.client_lock.acquire() 
    asyncio.ensure_future(self.client_lock_coro).add_done_callback(self._got_client_lock) 

def _got_client_lock(self, task): 
    task.result() # True at this point, but call there will trigger any exceptions 
    coro = self.loop.create_connection(lambda: ClientProtocol(self), 
              self.connect_info[0], self.connect_info[1]) 
    asyncio.ensure_future(asyncio.wait_for(coro, 
              self.client_connect_timeout 
              )).add_done_callback(self.connected_server) 

def connected_server(self, task): 
    transport, client_object = task.result() 
    self.client_transport = transport 
    self.client_lock.release() 

def data_received(self, data_in): 
    asyncio.ensure_future(self.send_to_real_server(message, self.client_send_timeout)) 

def send_to_real_server(self, message, timeout=5.0): 
    yield from self.client_lock.acquire() 
    asyncio.ensure_future(asyncio.wait_for(self._send_to_real_server(message), 
                timeout, loop=self.loop) 
           ).add_done_callback(self.sent_to_real_server) 

@asyncio.coroutine 
def _send_to_real_server(self, message): 
    self.client_transport.write(message) 

def sent_to_real_server(self, task): 
    task.result() 
    self.client_lock.release()