Python ThreadPoolExecutor - 是否保证在提交的func在同一个线程中运行的回调？ [英] Python ThreadPoolExecutor - is the callback guaranteed to run in the same thread as submitted func?

问题描述

在ThreadPoolExecutor（TPE）中，回调总是保证在与提交的函数相同的线程中运行？

例如，我用下面的代码测试了这个。我跑了很多次，似乎 func 和 callback 总是在同一个线程中运行。

  import concurrent.futures 
 import random 
 import threading 
 import time 
 
 executor = concurrent.futures.ThreadPoolExecutor（max_workers = 3）
 
 def func（x）：
 time.sleep（random.random（））
 .current_thread（）。name 
 
 def callback（future）：
 time.sleep（random.random（））
x = future.result（）
 cur_thread = threading.current_thread（）。name 
 if（cur_thread！= x）：
 print（cur_thread，x）
 
 print（'main thread：％s'％threading.current_thread （））
 for i in range（10000）：
 future = executor.submit（func，i）
 future.add_done_callback（callback）
 然而，当我删除 time.sleep（random.random（）） 
 $ b >语句，即至少有几个 func 函数和回调 没有同一个线程。 
 
 
 
对于我正在进行的项目，回调必须始终在与提交的函数相同的线程上运行，因此我想确保这是由TPE 。 （也没有随机睡眠的测试结果似乎令人困惑）。 
 
 
 
我查看了执行器的源代码，它似乎不像我们在我们运行回调之前将线程切换到主线程。但只是想确定。 
解决方案
 未来的回调提交给 ThreadPoolExecutor 将在任务本身运行的同一个线程中运行，但只有在任务之前将回调添加到 Future 完成。如果在 Future 完成后添加回调，回调将在您调用 add_done_callback 的任何线程中执行。您可以请参阅 add_done_callback 来源：
  def add_done_callback ，fn）：
附加一个callable，当未来结束时被调用
 
 Args：
 fn：一个可调用，只有
参数，当未来完成或被取消时，可调用的
总是被同一进程中的线程调用
它被添加。如果未来已经完成或者
已取消，则callable将被立即调用。
 callables按照它们被添加的顺序被调用

 with self._condition：
 if self ._state不在[CANCELLED，CANCELLED_AND_NOTIFIED，FINISHED]：
 self._done_callbacks.append（fn）
 return 
 fn（self）
  pre> 
 
 
如果未来的状态表明已取消或完成， fn 只是立即在当前执行线程中调用。否则，当未来完成时，它会添加到内部回调列表中运行。
 
 
 
例如： / p> 
 
 
 >>> def func（* args）：
 ... time.sleep（5）
 ... print（func {}。format（threading.current_thread（）））
> >> def cb（a）：print（cb {}。format（threading.current_thread（）））
 ... 
>> fut = ex.submit（func）
>>>> func< Thread（Thread-1，started daemon 140084551563008）> 
>>>> fut = e.add_done_callback（cb）
 cb <_MainThread（MainThread，started 140084622018368）> 
  
 
In the ThreadPoolExecutor (TPE), is the callback always guaranteed to run in the same thread as the submitted function? 

For example, I tested this with the following code. I ran it many times and it seemed like func and callback always ran in the same thread. 
import concurrent.futures 
import random 
import threading 
import time 

executor = concurrent.futures.ThreadPoolExecutor(max_workers=3) 

def func(x): 
    time.sleep(random.random()) 
    return threading.current_thread().name 

def callback(future): 
    time.sleep(random.random()) 
    x = future.result() 
    cur_thread = threading.current_thread().name 
    if (cur_thread != x): 
        print(cur_thread, x) 

print('main thread: %s' % threading.current_thread()) 
for i in range(10000): 
    future = executor.submit(func, i) 
    future.add_done_callback(callback) 
However, it seemed to fail when I removed the time.sleep(random.random()) statements, i.e. at least a few func functions and callbacks did not run in the same thread. 

For a project that I am working on, the callback must always run on the same thread as the submitted function, so I wanted to be sure that this is guaranteed by TPE. (And also the results of the test without the random sleep seemed puzzling). 

I looked at the source code for executors and it does not seem like we switch the thread to the main thread before we run the callback. But just wanted to be sure. 
解决方案
The callback for a Future submitted to a ThreadPoolExecutor will be run in the same thread as the task itself is running, but only if the callback is added to the Future before the task completes. If you add the callback after the Future completes, the callback will execute in whatever thread you called add_done_callback in. You can see this by looking at the add_done_callback source:
def add_done_callback(self, fn):
    """Attaches a callable that will be called when the future finishes.

    Args:
        fn: A callable that will be called with this future as its only
            argument when the future completes or is cancelled. The callable
            will always be called by a thread in the same process in which
            it was added. If the future has already completed or been
            cancelled then the callable will be called immediately. These
            callables are called in the order that they were added.
    """
    with self._condition:
        if self._state not in [CANCELLED, CANCELLED_AND_NOTIFIED, FINISHED]:
            self._done_callbacks.append(fn)
            return
    fn(self)
If the state of the Future indicates it's cancelled or finished, fn is just immediately called in the current thread of execution. Otherwise, it's added to an internal list of callbacks to run when the Future is complete.

For example:
>>> def func(*args):
...  time.sleep(5)
...  print("func {}".format(threading.current_thread()))
>>> def cb(a): print("cb {}".format(threading.current_thread()))
... 
>>> fut = ex.submit(func)
>>> func <Thread(Thread-1, started daemon 140084551563008)>
>>> fut = e.add_done_callback(cb)
cb <_MainThread(MainThread, started 140084622018368)>


                        
这篇关于Python ThreadPoolExecutor  - 是否保证在提交的func在同一个线程中运行的回调？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！