Python文件描述符与操作系统交互

文件描述符是操作系统分配给每个打开文件的整数标识符。Python的open返回的文件对象内部封装了一个文件描述符。

文件描述符的基本操作：

import os

# 打开文件获取文件描述符
fd = os.open('/tmp/test.txt', os.O_CREAT | os.O_WRONLY, 0o644)
os.write(fd, b'Hello World')
os.close(fd)

os.open直接与系统调用open(2)交互，返回int类型的文件描述符。

Python文件对象到文件描述符的转换：

with open('/tmp/test.txt', 'w') as f:
fd = f.fileno()
print(f"File descriptor: {fd}") # 3

fileno返回文件对象的底层文件描述符。文件对象关闭时文件描述符也关闭。

文件描述符的复制：

import os

fd = os.open('/tmp/test.txt', os.O_RDWR)
fd2 = os.dup(fd) # 复制文件描述符

os.write(fd, b'Data from fd')
os.lseek(fd2, 0, os.SEEK_SET)
print(os.read(fd2, 100)) # b'Data from fd'

os.close(fd)
os.close(fd2)

dup创建新的文件描述符指向同一个文件表项。两个文件描述符共享文件偏移量。

dup2重定向到指定fd：

import os

null_fd = os.open(os.devnull, os.O_WRONLY)
os.dup2(null_fd, 1) # 将标准输出重定向到/dev/null
print("This won't be visible") # 输出到/dev/null

dup2将标准输出fd(1)重定向到null_fd。此后所有print输出消失。

pipe创建管道：

import os

r, w = os.pipe()
pid = os.fork()

if pid == 0:
# 子进程
os.close(r)
os.write(w, b'Hello from child')
os.close(w)
os._exit(0)
else:
# 父进程
os.close(w)
data = os.read(r, 1024)
print(f"Parent received: {data}")
os.close(r)
os.waitpid(pid, 0)

pipe创建一对文件描述符，r用于读取，w用于写入。fork后父子进程通过管道通信。

select监听多个文件描述符：

import select
import sys

# 监控标准输入的可读事件
readable, _, _ = select.select([sys.stdin], [], [], 5.0)
if readable:
data = sys.stdin.read()
print(f"Read: {data}")
else:
print("Timeout")

select同时监控多个文件描述符的状态。第三个参数是超时秒数，0表示非阻塞。

非阻塞IO模式：

import os
import fcntl

fd = os.open('/tmp/test.txt', os.O_RDONLY | os.O_NONBLOCK)
try:
data = os.read(fd, 4096)
except BlockingIOError:
print("No data available, would block")

os.O_NONBLOCK设置非阻塞模式。在无数据时立即返回BlockingIOError而不是阻塞。

fcntl设置文件描述符属性：

import fcntl

fd = os.open('/tmp/test.txt', os.O_RDONLY)
flags = fcntl.fcntl(fd, fcntl.F_GETFL)
fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)

fcntl.F_GETFL获取当前标志，fcntl.F_SETFL设置新标志。

文件锁flock：

import fcntl

with open('/tmp/lock.txt', 'w') as f:
try:
fcntl.flock(f.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
# 独占锁
except IOError:
print("File already locked")

fcntl.LOCK_EX排他锁，LOCK_SH共享锁，LOCK_NB非阻塞。LOCK_UN释放锁。

sendfile零拷贝传输：

import os

with open('source.txt', 'rb') as src, open('dest.txt', 'wb') as dst:
os.sendfile(dst.fileno(), src.fileno(), 0, os.fstat(src.fileno()).st_size)

sendfile在内核空间直接从一个fd向另一个fd复制数据，无需经过用户空间缓冲区。适用于文件传输服务器。

epoll事件循环：

import select
import socket

epoll = select.epoll()
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(('localhost', 8888))
server.listen(10)
server.setblocking(False)

epoll.register(server.fileno(), select.EPOLLIN)

connections = {}

while True:
events = epoll.poll(timeout=1)
for fd, event in events:
if fd == server.fileno():
conn, addr = server.accept()
conn.setblocking(False)
epoll.register(conn.fileno(), select.EPOLLIN)
connections[conn.fileno()] = conn
elif event & select.EPOLLIN:
data = connections[fd].recv(1024)
if data:
print(f"Received: {data}")
else:
epoll.unregister(fd)
connections[fd].close()

epoll支持百万级并发连接，使用事件驱动避免轮询开销。

os.fstat获取文件描述符信息：

import os

fd = os.open('/tmp/test.txt', os.O_RDONLY)
stat = os.fstat(fd)
print(f"Size: {stat.st_size}")
print(f"Inode: {stat.st_ino}")
print(f"Permissions: {oct(stat.st_mode)}")
os.close(fd)

fstat直接从文件描述符获取文件信息，无需通过路径查找。