Tornado PipeIOStream报OSError(9) Bad file descriptor问题求助

看起来你遇到的问题是因为文件描述符的生命周期管理不当，加上错误地使用了PipeIOStream处理普通文件导致的。让我一步步帮你分析和修复：

问题根源拆解 #

1. 错误的Stream类型选择：PipeIOStream是专门为管道、套接字这类双向字节流设计的，用来处理普通磁盘文件并不合适，它的初始化逻辑会尝试以r+模式操作文件描述符，这和你打开文件的ab模式冲突，而且容易引发描述符管理问题。
2. 文件描述符被提前关闭：你用open()拿到文件对象fd后，只把它的文件描述符传给了Stream，但Python的垃圾回收机制可能会在Stream还在使用这个描述符的时候，自动关闭fd对应的文件句柄——因为此时fd变量在当前作用域里已经没有被引用了。这就导致第二次处理队列项时，拿到的是已经失效的描述符，抛出Bad file descriptor错误。
3. 队列任务未标记完成：你的代码里没有调用self.queue.task_done()，这会导致队列无法正确跟踪任务状态，长期运行可能引发任务积压。

修复后的完整代码 #

from tornado import web, ioloop, gen
from tornado.queues import Queue
from tornado.iostream import FileIOStream  # 替换成FileIOStream

class Store:  # 修正类名，原代码里实例化用了Store但类定义是Sample
    def __init__(self):
        self.queue = Queue()

    @gen.coroutine
    def write_queue(self):
        while True:
            item = yield self.queue.get()
            print("Message with id %s stored" % item[0])
            # 直接将打开的文件对象传给FileIOStream，让Stream管理文件生命周期
            stream = FileIOStream(open(item[0], 'ab'))
            try:
                yield stream.write(item[1])
            finally:
                # 确保Stream被关闭，自动清理底层文件描述符
                yield stream.close()
            # 标记队列任务完成，这是Tornado队列的最佳实践
            self.queue.task_done()

class MainHandler(web.RequestHandler):
    def initialize(self, store):
        self.store = store

    @gen.coroutine
    def put(self, id):
        yield self.store.queue.put((id, self.request.body))

def start(store):
    return web.Application([
        (r"/(.*)", MainHandler, {"store": store})
    ])

if __name__ == '__main__':
    store = Store()
    app = start(store)
    app.listen(8888)
    ioloop.IOLoop.current().add_callback(store.write_queue)
    ioloop.IOLoop.current().start()

关键修改说明 #

• 替换Stream类型：用FileIOStream替代PipeIOStream，它是Tornado专门为普通文件设计的异步IO流，能正确处理文件的读写模式和生命周期。
• 让Stream管理文件对象：直接把open()返回的文件对象传给FileIOStream，而不是提取文件描述符。这样Stream会持有文件对象的引用，避免Python垃圾回收提前关闭文件。
• 确保Stream被关闭：用try-finally块包裹写入操作，保证无论写入是否成功，Stream都会被关闭，防止文件描述符泄漏。
• 标记队列任务完成：添加self.queue.task_done()，告诉队列当前任务已处理完毕，有助于队列的状态管理和资源回收。

额外注意点 #

原代码里还有个小笔误：类定义是Sample，但实例化的时候用了store = Store()，我已经把类名修正为Store，避免运行时出现NameError。

内容的提问来源于stack exchange，提问作者boofar