Linux网络优化之内核epoll/io_uring 到Python(ASGI/WSGI) Java（BIO/NIO) 模型认知

// 简化的socket内核结构关系
struct socket {
    struct file *file;       // 关联的文件对象（内核用文件系统统一管理资源）
    struct sock *sk;         // 核心套接字对象（包含协议相关状态）
    struct proto_ops *ops;   // 协议操作集（如TCP的连接、读写方法）
};

struct sock {
    struct proto *sk_prot;   // 协议处理函数（如TCP的发送/接收逻辑）
    struct sk_buff_head sk_receive_queue;// 接收队列（存储待处理的数据包）
    struct socket_wq *sk_wq;// 等待队列（阻塞时进程挂起的地方）
    void (*sk_data_ready)(struct sock *sk， int len); // 数据就绪回调函数
    // 其他字段：状态（如ESTABLISHED）、窗口大小、超时时间等
};

>>> from socket import socket， AF_INET， SOCK_STREAM
>>> s = socket(AF_INET， SOCK_STREAM)
>>> s.connect(('localhost'， 20000))
>>> s.send(b'Hello')
5
>>> s.recv(8192)
b'Hello'
>>>

// inet_create关键代码（net/ipv4/af_inet.c）
static int inet_create(...) {
    // 遍历协议表，找到SOCK_STREAM对应的TCP配置
    list_for_each_entry_rcu(answer， &inetsw[sock->type]， list) {
        if (protocol == answer->protocol) {
            sock->ops = answer->ops;  // 绑定inet_stream_ops
            answer_prot = answer->prot;  // 绑定tcp_prot
            break;
        }
    }
    // 分配并初始化sock对象
    sk = sk_alloc(net， PF_INET， GFP_KERNEL， answer_prot);
    sock_init_data(sock， sk);  // 设置sk_data_ready等回调
}

from socketserver import BaseRequestHandler， TCPServer

class EchoHandler(BaseRequestHandler):
    def handle(self):
        print('Got connection from'， self.client_address)
        whileTrue:
            #接收客户端发送的数据， 这次接收数据的最大字节数是8192
            msg = self.request.recv(8192)
            # 接收的到数据在发送回去
            ifnot msg:
                break
            self.request.send(msg)

if __name__ == '__main__':
    # 20000端口，默认IP为本地IP，监听到消息交个EchoHandler处理器
    serv = TCPServer((''， 20000)， EchoHandler)
    serv.serve_forever()

int tcp_recvmsg(...) {
    int copied = 0;
    int target = len;  // len是用户传入的期望读取量（如8192）
    struct sock *sk = ...;

    do {
        // 遍历接收队列，拷贝所有可用数据（不超过target）
        skb_queue_walk(&sk->sk_receive_queue， skb) {
            int chunk = min(skb->len， target - copied);  // 本次拷贝的字节数（不超过剩余需求）
            memcpy(buf， skb->data， chunk);  // 拷贝到用户态缓冲区
            copied += chunk;
            skb_pull(skb， chunk);  // 从skb中移除已拷贝的数据
            if (copied >= target) break;  // 达到期望量，退出
        }

        // 关键判断：若已收到FIN包（发送方关闭），即使数据不足也退出
        if (sk->sk_state == TCP_CLOSE_WAIT || sk->sk_state == TCP_CLOSED) {
            break;  // 无更多数据，退出循环
        }

        // 若未收到FIN，且数据不足，才阻塞等待
        if (copied < target) {
            sk_wait_data(sk， &timeo);  // 阻塞等待新数据
        }
    } while (copied < target && !timeo_expired);

    return copied;  // 返回实际拷贝的字节数（如100）
}

// sk_wait_data核心代码（net/core/sock.c）
int sk_wait_data(struct sock *sk, long *timeo) {
    DEFINE_WAIT(wait);  // 初始化等待项,关联current
    prepare_to_wait(sk_sleep(sk), &wait, TASK_INTERRUPTIBLE);
    // 检查是否有数据,若无则让出CPU
    rc = sk_wait_event(sk, timeo, !skb_queue_empty(&sk->sk_receive_queue));
    finish_wait(sk_sleep(sk), &wait);
    return rc;
}

// tcp_rcv_established中的唤醒逻辑（net/ipv4/tcp_input.c）
int tcp_rcv_established(...) {
    // 数据入队
    eaten = tcp_queue_rcv(sk， skb， ...);
    // 触发数据就绪回调
    sk->sk_data_ready(sk， 0);
}

static void sock_def_readable(struct sock *sk， int len) {
    struct socket_wq *wq = rcu_dereference(sk->sk_wq);
    if (wq_has_sleeper(wq)) {
        // 唤醒等待队列上的进程
        wake_up_interruptible_sync_poll(&wq->wait， POLL_IN | POLLPRI | ...);
    }
}

import socket
import select

# 创建监听套接字
server = socket.socket(socket.AF_INET， socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET， socket.SO_REUSEADDR， 1)
server.bind(('127.0.0.1'， 8888))
server.listen(5)
server.setblocking(False)  # 设置非阻塞

# 创建 epoll 对象
epoll = select.epoll()
epoll.register(server.fileno()， select.EPOLLIN)  # 监听服务器套接字的读事件

connections = {}  # 保存客户端连接 {文件描述符: 套接字}

try:
    whileTrue:
        # 等待事件发生 (阻塞，直到有事件)
        events = epoll.poll()
        for fileno， event in events:
            # 服务器套接字描述符有读事件 (新连接)
            if fileno == server.fileno():
                client， addr = server.accept()
                client.setblocking(False)
                epoll.register(client.fileno()， select.EPOLLIN)  # 注册客户端读事件
                connections[client.fileno()] = client
                print(f"新连接: {addr}")
            # 客户端有读事件 (发来了数据)
            elif event & select.EPOLLIN:
                client = connections[fileno]
                data = client.recv(1024)
                if data:
                    print(f"收到 {client.getpeername()}: {data.decode()}")
                    client.send(f"已收到: {data}".encode())  # 回传数据
                else:  # 客户端断开
                    epoll.unregister(fileno)
                    client.close()
                    del connections[fileno]
                    print(f"连接断开: {client.getpeername()}")
finally:
    epoll.unregister(server.fileno())
    epoll.close()
    server.close()

struct eventpoll {
    wait_queue_head_t wq;       // epoll_wait 的等待队列（进程阻塞在这里）
    struct list_head rdllist;   // 就绪事件链表（已准备好的socket）
    struct rb_root rbr;         // 红黑树（管理所有注册的socket）
    // 其他字段：唤醒掩码、用户态地址等
};

struct epitem {
    struct rb_node rbn;         // 红黑树节点
    struct epoll_filefd ffd;    // 关联的文件描述符（socket）
    struct eventpoll *ep;       // 所属的eventpoll对象
    struct list_head rdllink;   // 就绪链表的节点（用于加入rdllist）
    struct list_head pwqlist;   // 等待队列项链表（关联socket的等待队列）
};

// ep_insert核心代码（fs/eventpoll.c）
static int ep_insert(...) {
    // 分配epitem
    struct epitem *epi = kmem_cache_alloc(epi_cache, GFP_KERNEL);
    epi->ep = ep;  // 关联eventpoll
    ep_set_ffd(&epi->ffd, tfile, fd);  // 关联socket的文件对象

    // 设置socket等待队列的回调
    struct ep_pqueue epq;
    epq.epi = epi;
    init_poll_funcptr(&epq.pt, ep_ptable_queue_proc);  // 回调设为ep_ptable_queue_proc
    revents = ep_item_poll(epi, &epq.pt);  // 触发回调注册

    // 插入红黑树
    ep_rbtree_insert(ep, epi);
}

static void ep_ptable_queue_proc(...) {
    struct eppoll_entry *pwq = kmem_cache_alloc(pwq_cache, GFP_KERNEL);
    init_waitqueue_func_entry(&pwq->wait, ep_poll_callback);  // 回调设为ep_poll_callback
    add_wait_queue(whead, &pwq->wait);  // 加入socket的等待队列
    list_add(&pwq->llink, &epi->pwqlist);  // 关联epitem
}

// ep_poll核心代码（fs/eventpoll.c）
static int ep_poll(...) {
    fetch_events:
    if (!ep_events_available(ep)) {  // 无就绪事件
        init_waitqueue_entry(&wait， current);  // 等待项关联当前进程
        __add_wait_queue_exclusive(&ep->wq， &wait);  // 加入eventpoll的等待队列

        // 阻塞等待
        for (;;) {
            set_current_state(TASK_INTERRUPTIBLE);  // 设为可中断睡眠
            if (ep_events_available(ep)) break;  // 有事件则退出循环
            if (!schedule_hrtimeout_range(...)) break;  // 超时退出
        }

        __remove_wait_queue(&ep->wq， &wait);  // 从等待队列移除
        set_current_state(TASK_RUNNING);  // 恢复运行状态
    }

    // 收集就绪事件并返回
    return ep_send_events(ep， events， maxevents);
}

// ep_poll_callback核心代码（fs/eventpoll.c）
static int ep_poll_callback(...) {
    struct epitem *epi = ep_item_from_wait(wait);
    struct eventpoll *ep = epi->ep;

    // 将epitem加入就绪链表
    list_add_tail(&epi->rdllink， &ep->rdllist);

    // 唤醒阻塞在epoll_wait的进程
    if (waitqueue_active(&ep->wq))
        wake_up_locked(&ep->wq);
    return 1;
}

应用程序（Application）
  ├─ 将新请求放入提交队列的队尾
  └─ 从完成队列的队头获取响应
       队头（head）                                             队头（head）
提交队列（Submission Queue）   共享内存（Shared Memory）     完成队列（Completion Queue）
       队尾（tail）                                             队尾（tail）

内核（Kernel）
  ├─ 从提交队列的队头获取任务并执行
  └─ 将响应结果放入完成队列的队尾

// 初始化 io_uring 实例
int io_uring_setup(unsigned int entries, struct io_uring_params *params);

// 提交 I/O 请求并/或等待完成事件
int io_uring_enter(unsigned int fd, u32 to_submit, u32 min_complete, u32 flags);

// 注册资源（文件描述符、缓冲区等）
int io_uring_register(unsigned int fd, unsigned int opcode, void *arg, unsigned int nr_args);

//#用户进程创建 io_uring 的代码示例如下
struct io_uring ring;
io_uring_queue_init(QUEUE_DEPTH, &ring, 0);
//向 io_uring 提交队列提交操作的代码示例：
struct io_uring_sqe *sqe = io_uring_get_sqe(&ring);
io_uring_prep_readv(sqe, client_socket, iov, 1, 0);
io_uring_sqe_set_data(sqe, user_data);
io_uring_submit(&ring);
//进程等待操作完成的代码示例：
struct io_uring_cqe *cqe;
int ret = io_uring_wait_cqe(&ring, &cqe);
//获取并处理响应结果的代码示例：
user_data = io_uring_cqe_get_data(cqe);
if (cqe->res < 0) {
    // 处理错误
} else {
    // 处理响应
}
io_uring_cqe_seen(&ring, cqe);

client_fd = accept(listen_fd, &client_addr, &client_addrlen);
for (;;) {
    numRead = read(client_fd, buf, BUF_SIZE);
    if (numRead <= 0)   // 遇到 EOF 或错误时退出循环
        break;
    if (write(client_fd, buf, numRead) != numRead)
        // 处理写操作错误
}
close(client_fd);

add_accept_request(listen_socket, &client_addr, &client_addr_len);

// 见了连接一次提交
io_uring_submit(&ring);

while (1) {
    int ret = io_uring_wait_cqe(&ring, &cqe);

    struct request *req = (struct request *) cqe->user_data;
    switch (req->type) {
    case ACCEPT:
        add_accept_request(listen_socket,
                          &client_addr, &client_addr_len);
        add_read_request(cqe->res);
        io_uring_submit(&ring);
        break;
    case READ:
        if (cqe->res <= 0) {
            add_close_request(req);
        } else {
            add_write_request(req);
        }
        
        io_uring_submit(&ring);
        break;
    case WRITE:
        add_read_request(req->socket);
        io_uring_submit(&ring);
        break;
    case CLOSE:
        free_request(req);
        break;
    default:
        fprintf(stderr, "Unexpected req type %d\n", req->type);
        break;
    }

    io_uring_cqe_seen(&ring, cqe);
}

while (1) {
    int submissions = 0;
    int ret = io_uring_wait_cqe(&ring, &cqe);

    while (1) {
        struct request *req = (struct request *) cqe->user_data;
        switch (req->type) {
        case ACCEPT:
            add_accept_request(listen_socket,
                              &client_addr, &client_addr_len);
            add_read_request(cqe->res);
            submissions += 2;
            break;
        case READ:
            if (cqe->res <= 0) {
                add_close_request(req);
                submissions += 1;
            } else {
                add_write_request(req);
                add_read_request(req->socket);
                submissions += 2;
            }
            break;
        case WRITE:
          break;
        case CLOSE:
            free_request(req);
            break;
        default:
            fprintf(stderr, "Unexpected req type %d\n", req->type);
            break;
        }

        io_uring_cqe_seen(&ring, cqe);

        if (io_uring_sq_space_left(&ring) < MAX_SQE_PER_LOOP) {
            break;     // 提交队列已满
        }

        ret = io_uring_peek_cqe(&ring, &cqe);
        if (ret == -EAGAIN) {
            break;     // 完成队列中无剩余待处理任务
        }
    }

    if (submissions > 0) {
        io_uring_submit(&ring);
    }
}

[root@developer ~]# sysctl -a | grep mem
....
net.core.optmem_max = 81920  # 套接字选项缓冲区的最大大小（字节），用于存储套接字选项相关数据，超过此值会返回错误
net.core.rmem_default = 212992  # 所有协议（如TCP、UDP）的默认接收缓冲区大小（字节），新建套接字默认使用此值
net.core.rmem_max = 212992  # 所有协议的最大接收缓冲区大小（字节），应用程序可设置的接收缓冲区上限（不能超过此值）
net.core.wmem_default = 212992  # 所有协议的默认发送缓冲区大小（字节），新建套接字默认使用此值
net.core.wmem_max = 212992  # 所有协议的最大发送缓冲区大小（字节），应用程序可设置的发送缓冲区上限（不能超过此值）
net.ipv4.tcp_mem = 78777105039157554  # TCP协议的整体内存管理阈值（单位：页，1页=4KB），三个值分别为：最小内存页、压力模式阈值、最大内存页。内核会根据TCP总内存使用量动态调整行为（如触发内存回收）
net.ipv4.tcp_rmem = 40961310726291456  # TCP接收缓冲区的内存范围（字节），三个值分别为：最小大小、默认大小、最大大小。应用程序可在范围内调整，内核也可能根据拥塞控制动态调整
net.ipv4.tcp_wmem = 4096163844194304  # TCP发送缓冲区的内存范围（字节），三个值分别为：最小大小、默认大小、最大大小。作用类似tcp_rmem，用于控制发送缓冲区的动态调整范围
net.ipv4.udp_mem = 157557210078315114  # UDP协议的整体内存管理阈值（单位：页），三个值分别为：最小内存页、压力模式阈值、最大内存页。控制UDP总内存使用量，避免占用过多系统内存
net.ipv4.udp_rmem_min = 4096  # UDP接收缓冲区的最小大小（字节），应用程序设置的接收缓冲区不能小于此值
net.ipv4.udp_wmem_min = 4096  # UDP发送缓冲区的最小大小（字节），应用程序设置的发送缓冲区不能小于此值
[root@developer ~]# 

# 
#!/usr/bin/env python
# -*- encoding: utf-8 -*-
"""
@File    :   app.py
@Time    :   2022/05/03 14:43:56
@Author  :   Li Ruilong
@Version :   1.0
@Contact :   1224965096@qq.com
@Desc    :   None
"""
# here put the import lib

import time
import cgi

def notfound_404(environ， start_response):
    start_response('404 Not Found'， [('Content-type'， 'text/plain')])
    return [b'Not Found']
# 核心控制器，用于路由注册
class PathDispatcher:
    def __init__(self):
        # 映射字典
        self.pathmap = {}
    # 核心控制器的回调
    def __call__(self， environ， start_response):
        # 获取路由
        path = environ['PATH_INFO']
        # 获取请求参数
        params = cgi.FieldStorage(environ['wsgi.input']，
                                  environ=environ)
        # 获取请求方法
        method = environ['REQUEST_METHOD'].lower()
        environ['params'] = {key: params.getvalue(key) for key in params}
        # 找到映射的函数
        handler = self.pathmap.get((method， path)， notfound_404)
        # 返回函数
        return handler(environ， start_response)

    def register(self， method， path， function):
        # 请求方法和路由作为K，执行函数为V
        self.pathmap[method.lower()， path] = function
        return function

_hello_resp = "wo jiao {name}"

def hello_world(environ， start_response):
    start_response('200 OK'， [('Content-type'， 'text/html')])
    params = environ['params']
    resp = _hello_resp.format(name=params.get('name'))
    yield resp.encode('utf-8')

_localtime_resp = "dang qian shjian {t}"
# 路由的回调
def localtime(environ， start_response):
    start_response('200 OK'， [('Content-type'， 'application/xml')])
    resp = _localtime_resp.format(t=time.localtime())
    yield resp.encode('utf-8')

if __name__ == '__main__':
    from wsgiref.simple_server import make_server
    # 创建一个核心控制器，用于路由注册
    dispatcher = PathDispatcher()
    # 注册路由，对应的回调方法
    dispatcher.register('GET'， '/hello'， hello_world)
    dispatcher.register('GET'， '/localtime'， localtime)
    # Launch a basic server 监听8080端口，注入核心控制器
    httpd = make_server(''， 8080， dispatcher)
    print('Serving on port 8080...')
    httpd.serve_forever()

# Flask（WSGI框架）的同步阻塞示例
from flask import Flask
import time

app = Flask(__name__)

@app.route('/')
def index():
    time.sleep(1)  # 模拟IO阻塞（如数据库查询）
    return "Hello， Flask"
if __name__ == '__main__':
    app.run()  # 内置服务器使用同步模型，单线程处理请求

# ASGI应用示例
async def application(scope， receive， send):
    # scope：连接元数据（如类型、方法、路径）
    # receive：异步函数，接收事件（如请求体）
    # send：异步函数，发送事件（如响应）
    await send({
        'type': 'http.response.start'，
        'status': 200，
        'headers': [(b'content-type'， b'text/plain')]，
    })
    await send({
        'type': 'http.response.body'，
        'body': b'Hello， ASGI'，
    })

# FastAPI（ASGI框架）的异步示例
from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.get("/")
async def index():
    await asyncio.sleep(1)  # 异步IO操作（非阻塞）
    return {"message": "Hello， FastAPI"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app)  # Uvicorn使用事件循环和epoll

# 简化的事件循环伪代码（模拟uvloop）
import select

epoll = select.epoll()
server_socket = ...  # 监听socket
epoll.register(server_socket.fileno()， select.EPOLLIN)

whileTrue:
    events = epoll.poll()  # 等待事件（epoll_wait）
    for fileno， event in events:
        if fileno == server_socket.fileno():
            # 新连接
            client_socket， addr = server_socket.accept()
            epoll.register(client_socket.fileno()， select.EPOLLIN)
        elif event & select.EPOLLIN:
            # 数据可读
            data = client_socket.recv(1024)
            # 创建协程处理请求
            loop.create_task(handle_request(client_socket， data))

// BIO模型的简化流程
publicclass BioConnector {
    private ServerSocket serverSocket;
    private ExecutorService workerPool = Executors.newFixedThreadPool(100);

    public void start(int port) throws IOException {
        serverSocket = new ServerSocket(port);
        // Acceptor线程：接收新连接
        new Thread(() -> {
            while (true) {
                Socket socket = serverSocket.accept();  // 阻塞等待新连接
                // 提交给Worker线程池处理
                workerPool.submit(() -> handle(socket));
            }
        }).start();
    }

    private void handle(Socket socket) {
        try (InputStream in = socket.getInputStream();
             OutputStream out = socket.getOutputStream()) {
            // 读取请求（阻塞）
            byte[] buffer = newbyte[1024];
            in.read(buffer);  // 阻塞等待数据
            // 处理请求（调用Servlet）
            String response = "HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nHello";
            // 发送响应（阻塞）
            out.write(response.getBytes());
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

// NIO模型的简化流程
publicclass NioConnector {
    private ServerSocketChannel serverChannel;
    private Selector selector;
    private ExecutorService workerPool = Executors.newFixedThreadPool(10);

    public void start(int port) throws IOException {
        serverChannel = ServerSocketChannel.open();
        serverChannel.bind(new InetSocketAddress(port));
        serverChannel.configureBlocking(false);  // 非阻塞模式
        selector = Selector.open();
        serverChannel.register(selector, SelectionKey.OP_ACCEPT);  // 注册接受事件

        // Selector线程：监控IO事件
        new Thread(() -> {
            while (true) {
                try {
                    selector.select();  // 阻塞等待事件（底层epoll_wait）
                    Set<SelectionKey> keys = selector.selectedKeys();
                    for (Iterator<SelectionKey> it = keys.iterator(); it.hasNext(); ) {
                        SelectionKey key = it.next();
                        it.remove();
                        if (key.isAcceptable()) {
                            // 处理新连接
                            accept(key);
                        } elseif (key.isReadable()) {
                            // 处理读事件
                            workerPool.submit(() -> read(key));
                        }
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

    private void accept(SelectionKey key) throws IOException {
        ServerSocketChannel server = (ServerSocketChannel) key.channel();
        SocketChannel client = server.accept();  // 非阻塞,立即返回
        client.configureBlocking(false);
        // 注册读事件
        client.register(selector, SelectionKey.OP_READ);
    }

    private void read(SelectionKey key) {
        try (SocketChannel channel = (SocketChannel) key.channel()) {
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            channel.read(buffer);  // 非阻塞读
            buffer.flip();
            // 处理请求...
            String response = "HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nHello";
            channel.write(ByteBuffer.wrap(response.getBytes()));  // 非阻塞写
            key.cancel();  // 关闭连接
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

// NIO2模型的简化流程
publicclass Nio2Connector {
    public void start(int port) throws IOException {
        AsynchronousServerSocketChannel server = AsynchronousServerSocketChannel.open();
        server.bind(new InetSocketAddress(port));
        // 接受连接,指定回调
        server.accept(null, new CompletionHandler<AsynchronousSocketChannel, Void>() {
            @Override
            public void completed(AsynchronousSocketChannel client, Void attachment) {
                server.accept(null, this);  // 继续接受新连接
                // 读取数据,指定回调
                ByteBuffer buffer = ByteBuffer.allocate(1024);
                client.read(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() {
                    @Override
                    public void completed(Integer bytesRead, ByteBuffer buf) {
                        buf.flip();
                        // 处理请求...
                        String response = "HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nHello";
                        client.write(ByteBuffer.wrap(response.getBytes()));
                    }

                    @Override
                    public void failed(Throwable exc, ByteBuffer buf) {
                        exc.printStackTrace();
                    }
                });
            }

            @Override
            public void failed(Throwable exc, Void attachment) {
                exc.printStackTrace();
            }
        });
    }
}