如何让s3fs与ProcessPoolExecutor协同工作?新版死锁问题求助
s3fs与ProcessPoolExecutor协同工作死锁问题求助
我一直没法让s3fs和ProcessPoolExecutor正常配合使用,核心问题是s3fs默认会保留连接会话信息,这和fork进程的机制不兼容。
我知道s3fs.S3FileSystem的构造函数支持asynchronous=True, loop=这类参数,而且大部分实用调用都有前缀为_的异步版本,但找不到任何将s3fs与ProcessPoolExecutor结合使用的示例,自己也理不清实现方式,所以来求助。
下面是我写的最简测试用例,这个用例在s3fs早期版本(0.4.x,还没和fsspec集成的时候)能正常运行,但在更新的版本(我测试的是2023.10.0)里,ProcessPoolExecutor.map()调用会出现死锁。虽然这个用例很简单没什么实际用途,但能体现我的需求方向,只要能让它正常运行,我就能推进后续工作。
import concurrent.futures import s3fs def get_tags(uri): s3 = s3fs.S3FileSystem() return s3.get_tags(uri) def get_files(filepath): s3 = s3fs.S3FileSystem() return s3.ls(filepath) if __name__ == "__main__": files = get_files('s3://<bucket name>/<prefix>') with concurrent.futures.ProcessPoolExecutor() as executor: results = executor.map(get_tags, files) print([r for r in results])
补充说明:我使用的是Python 3.12,不确定是否有影响,还是提一下。
内容的提问来源于stack exchange,提问作者David Moye
相关产品推荐
相关产品推荐

