系统编程

概述

本章讲解Cython在系统编程领域的应用。文件I/O、网络通信、进程管理是Cython的传统强项，可显著提升性能。

学习路径：文件I/O → 网络编程 → 进程管理 → 硬件交互

核心应用：

• 缓冲I/O：减少系统调用次数
• 异步编程：asyncio集成提升并发
• 原子操作：stdatomic实现无锁并发

---

15.1 文件I/O

缓冲区管理

功能说明：带缓冲的文件写入，减少系统调用。

cimport sys

cdef class BufferedWriter:
    cdef list _buffer
    cdef int _buffer_size
    cdef str _filename

    def __init__(self, str filename, int buffer_size=8192):
        self._filename = filename
        self._buffer_size = buffer_size
        self._buffer = []

    cpdef void write(self, str data):
        self._buffer.append(data)
        if len(self._buffer) >= self._buffer_size:
            self._flush()

    cdef void _flush(self):
        cdef str content = ''.join(self._buffer)
        with open(self._filename, 'a') as f:
            f.write(content)
        self._buffer = []

输出示例：

>>> writer = BufferedWriter("output.txt")
>>> writer.write("Line 1\n")
>>> writer.write("Line 2\n")
>>> # 缓冲区未满，文件尚未写入
>>> # 关闭或缓冲区满时写入

内存映射文件

功能说明：使用mmap实现高效随机访问。

import mmap

cdef class MemoryMappedFile:
    cdef object _mmap
    cdef object _file

    def __init__(self, str filename, int size):
        self._file = open(filename, 'r+b')
        self._mmap = mmap.mmap(
            self._file.fileno(),
            size,
            access=mmap.ACCESS_WRITE
        )

    cpdef void write_int(self, int offset, int value):
        self._mmap.seek(offset)
        self._mmap.write_int(value)

    cpdef int read_int(self, int offset):
        self._mmap.seek(offset)
        return self._mmap.read_int()

    def __dealloc__(self):
        if self._mmap is not None:
            self._mmap.close()
        if self._file is not None:
            self._file.close()

性能说明：内存映射文件访问速度接近内存，避免传统read/write系统调用。

异步I/O

功能说明：集成asyncio实现异步文件操作。

import asyncio

cpdef async void async_read(str filename):
    """异步文件读取"""
    await asyncio.sleep(0)  # 允许其他任务运行
    with open(filename, 'r') as f:
        return f.read()

async def main():
    content = await async_read("large_file.txt")
    print(len(content))

---

15.2 网络编程

套接字封装

功能说明：TCP服务器基本实现。

cimport socket

cdef class TCPServer:
    cdef int _sockfd
    cdef int _port
    cdef bint _running

    def __init__(self, int port):
        self._port = port
        self._running = False

    cpdef void start(self):
        cdef int sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        sock.bind(('0.0.0.0', self._port))
        sock.listen(5)
        self._sockfd = sock
        self._running = True

    cpdef void handle_connection(self):
        cdef int clientfd
        cdef bytes data

        while self._running:
            clientfd, addr = self._sockfd.accept()
            data = clientfd.recv(1024)
            clientfd.send(b"HTTP/1.1 200 OK\r\n\r\n")
            clientfd.close()

高性能服务器

功能说明：asyncio实现高并发服务器。

cimport asyncio

cdef class AsyncServer:
    cdef object _server

    async def handle_client(self, reader, writer):
        data = await reader.read(1024)
        message = data.decode()
        writer.write(b"HTTP/1.1 200 OK\r\n\r\nHello")
        await writer.drain()
        writer.close()

    async def start(self, int port):
        self._server = await asyncio.start_server(
            self.handle_client,
            '0.0.0.0',
            port
        )
        async with self._server:
            await self._server.serve_forever()

协议解析

功能说明：HTTP请求解析示例。

cpdef dict parse_http_request(bytes data):
    """简单HTTP请求解析"""
    cdef str s = data.decode('utf-8', errors='ignore')
    cdef list lines = s.split('\r\n')
    cdef str first_line = lines[0]
    cdef list parts = first_line.split(' ')
    cdef dict result = {
        'method': parts[0] if len(parts) > 0 else '',
        'path': parts[1] if len(parts) > 1 else '',
        'version': parts[2] if len(parts) > 2 else '',
    }
    return result

输出示例：

>>> parse_http_request(b"GET /index.html HTTP/1.1\r\nHost: example.com")
{'method': 'GET', 'path': '/index.html', 'version': 'HTTP/1.1'}

---

15.3 进程管理

子进程

功能说明：运行外部命令并获取输出。

import subprocess

cpdef str run_command(str cmd):
    """运行命令并返回输出"""
    result = subprocess.run(
        cmd,
        shell=True,
        capture_output=True,
        text=True
    )
    return result.stdout

cpdef void run_parallel_commands(list commands):
    """并行运行多个命令"""
    processes = []
    for cmd in commands:
        p = subprocess.Popen(
            cmd,
            shell=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE
        )
        processes.append(p)

    for p in processes:
        p.wait()

输出示例：

>>> run_command("echo 'Hello'")
"Hello\n"

管道通信

功能说明：进程间管道通信。

cimport os

cdef class Pipe:
    cdef int _read_fd
    cdef int _write_fd

    def __init__(self):
        cdef int[2] fds
        os.pipe(fds)
        self._read_fd = fds[0]
        self._write_fd = fds[1]

    cpdef void write_data(self, bytes data):
        os.write(self._write_fd, data)

    cpdef bytes read_data(self, int size):
        return os.read(self._read_fd, size)

    def __dealloc__(self):
        os.close(self._read_fd)
        os.close(self._write_fd)

信号处理

功能说明：捕获和处理系统信号。

import signal

cdef int received_sigint = 0

def sigint_handler(signum, frame):
    global received_sigint
    received_sigint = 1

cpdef void install_signal_handler():
    signal.signal(signal.SIGINT, sigint_handler)

cpdef int check_signal():
    return received_sigint

---

15.4 硬件交互

内存屏障

功能说明：C11原子操作实现线程安全。

cdef extern from "stdatomic.h":
    ctypedef struct atomic_int:
        int _value

    int atomic_load_explicit(atomic_int*, int memory_order)
    void atomic_store_explicit(atomic_int*, int, int memory_order)

cdef int memory_order_seq_cst = 5  # memory_order_seq_cst

cdef class AtomicCounter:
    cdef atomic_int _counter

    def __init__(self, int initial=0):
        atomic_store_explicit(&self._counter, initial, memory_order_seq_cst)

    cpdef int get(self):
        return atomic_load_explicit(&self._counter, memory_order_seq_cst)

    cpdef void increment(self):
        cdef int current = atomic_load_explicit(&self._counter, memory_order_seq_cst)
        atomic_store_explicit(&self._counter, current + 1, memory_order_seq_cst)

输出示例：

>>> counter = AtomicCounter(0)
>>> counter.increment()
>>> counter.get()
1

原子操作

功能说明：无锁原子加法和减法。

cdef extern from "stdatomic.h":
    ctypedef struct atomic_int:
        int _value

    int atomic_fetch_add_explicit(atomic_int*, int, int)
    int atomic_fetch_sub_explicit(atomic_int*, int, int)

cdef class FastAtomic:
    cdef atomic_int _value

    def __init__(self, int initial=0):
        atomic_store_explicit(&self._value, initial, 5)

    cpdef int add_and_fetch(self, int delta):
        return atomic_fetch_add_explicit(&self._value, delta, 5)

    cpdef int sub_and_fetch(self, int delta):
        return atomic_fetch_sub_explicit(&self._value, delta, 5)

DMA传输

功能说明：直接内存访问示例（平台相关）。

cdef class DMATransfer:
    cdef void* _buffer
    cdef int _size
    cdef int _channel

    def __init__(self, int size):
        self._size = size
        self._buffer = malloc(size)
        self._channel = 0

    cpdef void start_transfer(self, int device_addr):
        # 配置DMA控制器
        self._configure_channel(self._channel, device_addr, self._buffer, self._size)
        self._start_channel(self._channel)

    cdef void _configure_channel(self, int ch, int dest, void* src, int size):
        pass  # 平台特定实现

    cdef void _start_channel(self, int ch):
        pass

    def __dealloc__(self):
        if self._buffer != NULL:
            free(self._buffer)

---

本章小结

领域对比

领域	Cython优势	注意事项
文件I/O	C级缓冲	GIL影响I/O
网络编程	异步支持	选择器优化
进程管理	subprocess集成	平台差异
硬件交互	原子操作	需C库支持

最佳实践

1. 批量写入用缓冲减少系统调用
2. 内存映射文件适合大文件随机访问
3. asyncio集成提升网络并发
4. 原子操作避免锁竞争

---

练习

1. 实现带缓冲的文件写入器，对比无缓冲性能
2. 创建TCP回显服务器，支持多客户端
3. 使用subprocess并行执行多个命令
4. 实现原子计数器，测试多线程安全
5. 创建内存映射文件类，实现高效随机读写
6. 实现进程间管道通信