我有一个活动的concurrent.futures.ThreadPoolExecutor
。我想要检查它的状态,想知道有多少个线程,有多少个处理任务和哪些任务,有多少个是空闲的,以及哪些任务在队列中。我该如何找出这些信息?
我有一个活动的concurrent.futures.ThreadPoolExecutor
。我想要检查它的状态,想知道有多少个线程,有多少个处理任务和哪些任务,有多少个是空闲的,以及哪些任务在队列中。我该如何找出这些信息?
可以查看池和待处理工作项队列的一些可见性。要了解可用内容,请打印poolx.__dict__
以查看结构。阅读ThreadPool代码,它很好:concurrent.futures.thread
以下创建一个带有一个线程的池。然后创建两个作业:一个睡眠3秒,另一个立即返回。然后打印出池中待处理工作项的数量。
接下来,我们从工作队列中打印出项目。在这种情况下,一个线程已经执行time.sleep(3)
函数,因此不在队列中。打印具有参数[0]
和kwargs {}
的sleep
函数,因为那是池要运行的下一个工作项。
感谢@dano提供非破坏性队列洞察力,以及@abarnert。
import concurrent.futures, time
poolx = concurrent.futures.ThreadPoolExecutor(max_workers=1)
poolx.submit(time.sleep, 3)
poolx.submit(time.sleep, 0) # very fast
print('pending:', poolx._work_queue.qsize(), 'jobs')
print('threads:', len(poolx._threads))
print()
# TODO: make thread safe; work on copy of queue?
print('Estimated Pending Work Queue:')
for num,item in enumerate(poolx._work_queue.queue):
print('{}\t{}\t{}\t{}'.format(
num+1, item.fn, item.args, item.kwargs,
))
poolx.shutdown(wait=False)
pending: 1 jobs
threads: 1
Pending Work Queue:
1 <built-in function sleep> (0,) {}
for item in poolx._work_queue.queue: print(item.fn, item.args, item.kwargs)
。 - danopool._work_queue
这一部分没问题,但是对 pool._work_queue.queue
进行迭代是有问题的。这个队列成员只应该在适当同步时使用;你的迭代器可能会被另一个线程推入或弹出而无效。在低争用场景下通常可以在 CPython 中成功运行,但在没有 GIL 的实现或高争用情况下(即你真正需要队列的情况)会失败。 - abarnertpool._work_queue.mutex
。这是 queue.Queue
用于同步的全部内容。 - dano没有非常干净和可靠的方法来查找pending
futures,但我是这样做的:
if 'state=pending' in str(future):
logger.debug('PENDING')
elif future.running():
logger.debug('RUNNING')
elif future.cancelled():
logger.debug('CANCELLED')
elif future.exception():
logger.debug('EXCEPTION')
elif future.done():
logger.debug('DONE')