`DummyExecutor` cho gói` futures`

Question

của Python futures cho phép chúng ta thưởng thức ThreadPoolExecutor và ProcessPoolExecutor để thực hiện các tác vụ song song.

Tuy nhiên, để gỡ lỗi đôi khi rất hữu ích để tạm thời thay thế song song thực sự bằng một giả, thực hiện nhiệm vụ theo cách nối tiếp trong chuỗi chính mà không sinh ra bất kỳ luồng hoặc quy trình nào.

Có nơi nào triển khai DummyExecutor không?

Answer 1

Something như thế này nên làm điều đó:

from concurrent.futures import Future, Executor 
from threading import Lock 


class DummyExecutor(Executor): 

    def __init__(self): 
     self._shutdown = False 
     self._shutdownLock = Lock() 

    def submit(self, fn, *args, **kwargs): 
     with self._shutdownLock: 
      if self._shutdown: 
       raise RuntimeError('cannot schedule new futures after shutdown') 

      f = Future() 
      try: 
       result = fn(*args, **kwargs) 
      except BaseException as e: 
       f.set_exception(e) 
      else: 
       f.set_result(result) 

      return f 

    def shutdown(self, wait=True): 
     with self._shutdownLock: 
      self._shutdown = True 


if __name__ == '__main__': 

    def fnc(err): 
     if err: 
      raise Exception("test") 
     else: 
      return "ok" 

    ex = DummyExecutor() 
    print(ex.submit(fnc, True)) 
    print(ex.submit(fnc, False)) 
    ex.shutdown() 
    ex.submit(fnc, True) # raises exception 

khóa có lẽ không cần thiết trong trường hợp này, nhưng không thể làm tổn thương để có nó.