这是我写的并发网络服务器系列文章的第五部分。在前四部分中我们讨论了并发服务器的结构，这篇文章我们将去研究一个在生产系统中大量使用的服务器的案例—— Redis。Redis 是一个非常有魅力的项目，我关注它很久了。它最让我着迷的一点就是它的 C 源代码非常清晰。它也是一个高性能、大并发的内存数据库服务器的非常好的例子，它是研究网络并发服务器的一个非常好的案例，因此，我们不能错过这个好机会。我们来看看前四部分讨论的概念在真实世界中的应用程序。本系列的所有文章有：第一节 - 简介第二节 - 线程第三节 - 事件驱动第四节 - libuv第五节 - Redis 案例研究事件处理库Redis 最初发布于 2009 年，它最牛逼的一件事情大概就是它的速度 —— 它能够处理大量的并发客户端连接。需要特别指出的是，它是用一个单线程来完成的，而且还不对保存在内存中的数据使用任何复杂的锁或者同步机制。Redis 之所以如此牛逼是因为，它在给定的系统上使用了其可用的最快的事件循环，并将它们封装成由它实现的事件循环库（在 Linux 上是 epoll，在 BSD 上是 kqueue，等等）。这个库的名字叫做 ae。ae 使得编写一个快速服务器变得很容易，只要在它内部没有阻塞即可，而 Redis 则保证 注1 了这一点。在这里，我们的兴趣点主要是它对文件事件的支持 —— 当文件描述符（如网络套接字）有一些有趣的未决事情时将调用注册的回调函数。与 libuv 类似，ae 支持多路事件循环（参阅本系列的第三节和第四节）和不应该感到意外的 aeCreateFileEvent 信号：int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask, aeFileProc *proc, void *clientData); 它在 fd 上使用一个给定的事件循环，为新的文件事件注册一个回调（proc）函数。当使用的是 epoll 时，它将调用 epoll_ctl 在文件描述符上添加一个事件（可能是 EPOLLIN、EPOLLOUT、也或许两者都有，取决于 mask 参数）。ae 的 aeProcessEvents 功能是 “运行事件循环和发送回调函数”，它在底层调用了 epoll_wait。处理客户端请求我们通过跟踪 Redis 服务器代码来看一下，ae 如何为客户端事件注册回调函数的。initServer 启动时，通过注册一个回调函数来读取正在监听的套接字上的事件，通过使用回调函数 acceptTcpHandler 来调用 aeCreateFileEvent。当新的连接可用时，这个回调函数被调用。它调用 accept 注2 ，接下来是 acceptCommonHandler，它转而去调用 createClient 以初始化新客户端连接所需要的数据结构。createClient 的工作是去监听来自客户端的入站数据。它将套接字设置为非阻塞模式（一个异步事件循环中的关键因素）并使用 aeCreateFileEvent 去注册另外一个文件事件回调函数以读取事件 —— readQueryFromClient。每当客户端发送数据，这个函数将被事件循环调用。readQueryFromClient 就让我们期望的那样 —— 解析客户端命令和动作，并通过查询和/或操作数据来回复。因为客户端套接字是非阻塞的，所以这个函数必须能够处理 EAGAIN，以及部分数据；从客户端中读取的数据是累积在客户端专用的缓冲区中，而完整的查询可能被分割在回调函数的多个调用当中。将数据发送回客户端在前面的内容中，我说到了 readQueryFromClient 结束了发送给客户端的回复。这在逻辑上是正确的，因为 readQueryFromClient 准备要发送回复，但它不真正去做实质的发送 —— 因为这里并不能保证客户端套接字已经准备好写入/发送数据。我们必须为此使用事件循环机制。Redis 是这样做的，它注册一个 beforeSleep 函数，每次事件循环即将进入休眠时，调用它去等待套接字变得可以读取/写入。beforeSleep 做的其中一件事情就是调用 handleClientsWithPendingWrites。它的作用是通过调用 writeToClient 去尝试立即发送所有可用的回复；如果一些套接字不可用时，那么当套接字可用时，它将注册一个事件循环去调用 sendReplyToClient。这可以被看作为一种优化 —— 如果套接字可用于立即发送数据（一般是 TCP 套接字），这时并不需要注册事件 ——直接发送数据。因为套接字是非阻塞的，它从不会去阻塞循环。为什么 Redis 要实现它自己的事件库？在 第四节 中我们讨论了使用 libuv 来构建一个异步并发服务器。需要注意的是，Redis 并没有使用 libuv，或者任何类似的事件库，而是它去实现自己的事件库 —— ae，用 ae 来封装 epoll、kqueue 和 select。事实上，Antirez（Redis 的创建者）恰好在 2011 年的一篇文章 中回答了这个问题。他的回答的要点是：ae 只有大约 770 行他理解的非常透彻的代码；而 libuv 代码量非常巨大，也没有提供 Redis 所需的额外功能。现在，ae 的代码大约增长到 1300 多行，比起 libuv 的 26000 行（这是在没有 Windows、测试、示例、文档的情况下的数据）来说那是小巫见大巫了。libuv 是一个非常综合的库，这使它更复杂，并且很难去适应其它项目的特殊需求；另一方面，ae 是专门为 Redis 设计的，与 Redis 共同演进，只包含 Redis 所需要的东西。这是我 前些年在一篇文章中 提到的软件项目依赖关系的另一个很好的示例：依赖的优势与在软件项目上花费的工作量成反比。在某种程度上，Antirez 在他的文章中也提到了这一点。他提到，提供大量附加价值（在我的文章中的“基础” 依赖）的依赖比像 libuv 这样的依赖更有意义（它的例子是 jemalloc 和 Lua），对于 Redis 特定需求，其功能的实现相当容易。Redis 中的多线程在 Redis 的绝大多数历史中，它都是一个不折不扣的单线程的东西。一些人觉得这太不可思议了，有这种想法完全可以理解。Redis 本质上是受网络束缚的 —— 只要数据库大小合理，对于任何给定的客户端请求，其大部分延时都是浪费在网络等待上，而不是在 Redis 的数据结构上。然而，现在事情已经不再那么简单了。Redis 现在有几个新功能都用到了线程：“惰性” 内存释放。在后台线程中使用 fsync 调用写一个 持久化日志。运行需要执行一个长周期运行的操作的用户定义模块。对于前两个特性，Redis 使用它自己的一个简单的 bio（它是 “Background I/O" 的首字母缩写）库。这个库是根据 Redis 的需要进行了硬编码，它不能用到其它的地方 —— 它运行预设数量的线程，每个 Redis 后台作业类型需要一个线程。而对于第三个特性，Redis 模块 可以定义新的 Redis 命令，并且遵循与普通 Redis 命令相同的标准，包括不阻塞主线程。如果在模块中自定义的一个 Redis 命令，希望去执行一个长周期运行的操作，这将创建一个线程在后台去运行它。在 Redis 源码树中的 src/modules/helloblock.c 提供了这样的一个示例。有了这些特性，Redis 使用线程将一个事件循环结合起来，在一般的案例中，Redis 具有了更快的速度和弹性，这有点类似于在本系统文章中 第四节 讨论的工作队列。注1： Redis 的一个核心部分是：它是一个 内存中 数据库；因此，查询从不会运行太长的时间。当然了，这将会带来各种各样的其它问题。在使用分区的情况下，服务器可能最终路由一个请求到另一个实例上；在这种情况下，将使用异步 I/O 来避免阻塞其它客户端。注2： 使用 anetAccept；anet 是 Redis 对 TCP 套接字代码的封装。via: https://eli.thegreenplace.net/2017/concurrent-servers-part-5-redis-case-study/作者：Eli Bendersky 译者：qhwdw 校对：wxy本文由 LCTT 原创编译，Linux中国 荣誉推出
这是并发网络服务器系列文章的第四部分。在这一部分中，我们将使用 libuv 再次重写我们的服务器，并且也会讨论关于使用一个线程池在回调中去处理耗时任务。最终，我们去看一下底层的 libuv，花一点时间去学习如何用异步 API 对文件系统阻塞操作进行封装。本系列的所有文章：第一节 - 简介第二节 - 线程第三节 - 事件驱动第四节 - libuv使用 libuv 抽象出事件驱动循环在 第三节 中，我们看到了基于 select 和 epoll 的服务器的相似之处，并且，我说过，在它们之间抽象出细微的差别是件很有吸引力的事。许多库已经做到了这些，所以在这一部分中我将去选一个并使用它。我选的这个库是 libuv，它最初设计用于 Node.js 底层的可移植平台层，并且，后来发现在其它的项目中也有使用。libuv 是用 C 写的，因此，它具有很高的可移植性，非常适用嵌入到像 JavaScript 和 Python 这样的高级语言中。虽然 libuv 为了抽象出底层平台细节已经变成了一个相当大的框架，但它仍然是以 事件循环 思想为中心的。在我们第三部分的事件驱动服务器中，事件循环是显式定义在 main 函数中的；当使用 libuv 时，该循环通常隐藏在库自身中，而用户代码仅需要注册事件句柄（作为一个回调函数）和运行这个循环。此外，libuv 会在给定的平台上使用更快的事件循环实现，对于 Linux 它是 epoll，等等。libuv 支持多路事件循环，因此事件循环在库中是非常重要的；它有一个句柄 —— uv_loop_t，以及创建/杀死/启动/停止循环的函数。也就是说，在这篇文章中，我将仅需要使用 “默认的” 循环，libuv 可通过 uv_default_loop() 提供它；多路循环大多用于多线程事件驱动的服务器，这是一个更高级别的话题，我将留在这一系列文章的以后部分。使用 libuv 的并发服务器为了对 libuv 有一个更深的印象，让我们跳转到我们的可靠协议的服务器，它通过我们的这个系列已经有了一个强大的重新实现。这个服务器的结构与第三部分中的基于 select 和 epoll 的服务器有一些相似之处，因为，它也依赖回调。完整的 示例代码在这里；我们开始设置这个服务器的套接字绑定到一个本地端口：int portnum = 9090; if (argc >= 2) { portnum = atoi(argv[1]); } printf("Serving on port %d\n", portnum); int rc; uv_tcp_t server_stream; if ((rc = uv_tcp_init(uv_default_loop(), &server_stream)) < 0) { die("uv_tcp_init failed: %s", uv_strerror(rc)); } struct sockaddr_in server_address; if ((rc = uv_ip4_addr("0.0.0.0", portnum, &server_address)) < 0) { die("uv_ip4_addr failed: %s", uv_strerror(rc)); } if ((rc = uv_tcp_bind(&server_stream, (const struct sockaddr*)&server_address, 0)) < 0) { die("uv_tcp_bind failed: %s", uv_strerror(rc)); } 除了它被封装进 libuv API 中之外，你看到的是一个相当标准的套接字。在它的返回中，我们取得了一个可工作于任何 libuv 支持的平台上的可移植接口。这些代码也展示了很认真负责的错误处理；多数的 libuv 函数返回一个整数状态，返回一个负数意味着出现了一个错误。在我们的服务器中，我们把这些错误看做致命问题进行处理，但也可以设想一个更优雅的错误恢复。现在，那个套接字已经绑定，是时候去监听它了。这里我们运行首个回调注册：// Listen on the socket for new peers to connect. When a new peer connects, // the on_peer_connected callback will be invoked. if ((rc = uv_listen((uv_stream_t*)&server_stream, N_BACKLOG, on_peer_connected)) < 0) { die("uv_listen failed: %s", uv_strerror(rc)); } uv_listen 注册一个事件回调，当新的对端连接到这个套接字时将会调用事件循环。我们的回调在这里被称为 on_peer_connected，我们一会儿将去查看它。最终，main 运行这个 libuv 循环，直到它被停止（uv_run 仅在循环被停止或者发生错误时返回）。// Run the libuv event loop. uv_run(uv_default_loop(), UV_RUN_DEFAULT); // If uv_run returned, close the default loop before exiting. return uv_loop_close(uv_default_loop()); 注意，在运行事件循环之前，只有一个回调是通过 main 注册的；我们稍后将看到怎么去添加更多的回调。在事件循环的整个运行过程中，添加和删除回调并不是一个问题 —— 事实上，大多数服务器就是这么写的。这是一个 on_peer_connected，它处理到服务器的新的客户端连接：void on_peer_connected(uv_stream_t* server_stream, int status) { if (status < 0) { fprintf(stderr, "Peer connection error: %s\n", uv_strerror(status)); return; } // client will represent this peer; it's allocated on the heap and only // released when the client disconnects. The client holds a pointer to // peer_state_t in its data field; this peer state tracks the protocol state // with this client throughout interaction. uv_tcp_t* client = (uv_tcp_t*)xmalloc(sizeof(*client)); int rc; if ((rc = uv_tcp_init(uv_default_loop(), client)) < 0) { die("uv_tcp_init failed: %s", uv_strerror(rc)); } client->data = NULL; if (uv_accept(server_stream, (uv_stream_t*)client) == 0) { struct sockaddr_storage peername; int namelen = sizeof(peername); if ((rc = uv_tcp_getpeername(client, (struct sockaddr*)&peername, &namelen)) < 0) { die("uv_tcp_getpeername failed: %s", uv_strerror(rc)); } report_peer_connected((const struct sockaddr_in*)&peername, namelen); // Initialize the peer state for a new client: we start by sending the peer // the initial '*' ack. peer_state_t* peerstate = (peer_state_t*)xmalloc(sizeof(*peerstate)); peerstate->state = INITIAL_ACK; peerstate->sendbuf[0] = '*'; peerstate->sendbuf_end = 1; peerstate->client = client; client->data = peerstate; // Enqueue the write request to send the ack; when it's done, // on_wrote_init_ack will be called. The peer state is passed to the write // request via the data pointer; the write request does not own this peer // state - it's owned by the client handle. uv_buf_t writebuf = uv_buf_init(peerstate->sendbuf, peerstate->sendbuf_end); uv_write_t* req = (uv_write_t*)xmalloc(sizeof(*req)); req->data = peerstate; if ((rc = uv_write(req, (uv_stream_t*)client, &writebuf, 1, on_wrote_init_ack)) < 0) { die("uv_write failed: %s", uv_strerror(rc)); } } else { uv_close((uv_handle_t*)client, on_client_closed); } } 这些代码都有很好的注释，但是，这里有一些重要的 libuv 语法我想去强调一下：传入自定义数据到回调中：因为 C 语言还没有闭包，这可能是个挑战，libuv 在它的所有的处理类型中有一个 void* data 字段；这些字段可以被用于传递用户数据。例如，注意 client->data 是如何指向到一个 peer_state_t 结构上，以便于 uv_write 和 uv_read_start 注册的回调可以知道它们正在处理的是哪个客户端的数据。内存管理：在带有垃圾回收的语言中进行事件驱动编程是非常容易的，因为，回调通常运行在一个与它们注册的地方完全不同的栈帧中，使得基于栈的内存管理很困难。它总是需要传递堆分配的数据到 libuv 回调中（当所有回调运行时，除了 main，其它的都运行在栈上），并且，为了避免泄漏，许多情况下都要求这些数据去安全释放（free()）。这些都是些需要实践的内容 注1 。这个服务器上对端的状态如下：typedef struct { ProcessingState state; char sendbuf[SENDBUF_SIZE]; int sendbuf_end; uv_tcp_t* client; } peer_state_t; 它与第三部分中的状态非常类似；我们不再需要 sendptr，因为，在调用 “done writing” 回调之前，uv_write 将确保发送它提供的整个缓冲。我们也为其它的回调使用保持了一个到客户端的指针。这里是 on_wrote_init_ack：void on_wrote_init_ack(uv_write_t* req, int status) { if (status) { die("Write error: %s\n", uv_strerror(status)); } peer_state_t* peerstate = (peer_state_t*)req->data; // Flip the peer state to WAIT_FOR_MSG, and start listening for incoming data // from this peer. peerstate->state = WAIT_FOR_MSG; peerstate->sendbuf_end = 0; int rc; if ((rc = uv_read_start((uv_stream_t*)peerstate->client, on_alloc_buffer, on_peer_read)) < 0) { die("uv_read_start failed: %s", uv_strerror(rc)); } // Note: the write request doesn't own the peer state, hence we only free the // request itself, not the state. free(req); } 然后，我们确信知道了这个初始的 '*' 已经被发送到对端，我们通过调用 uv_read_start 去监听从这个对端来的入站数据，它注册一个将被事件循环调用的回调（on_peer_read），不论什么时候，事件循环都在套接字上接收来自客户端的调用：void on_peer_read(uv_stream_t* client, ssize_t nread, const uv_buf_t* buf) { if (nread < 0) { if (nread != uv_eof) { fprintf(stderr, "read error: %s\n", uv_strerror(nread)); } uv_close((uv_handle_t*)client, on_client_closed); } else if (nread == 0) { // from the documentation of uv_read_cb: nread might be 0, which does not // indicate an error or eof. this is equivalent to eagain or ewouldblock // under read(2). } else { // nread > 0 assert(buf->len >= nread); peer_state_t* peerstate = (peer_state_t*)client->data; if (peerstate->state == initial_ack) { // if the initial ack hasn't been sent for some reason, ignore whatever // the client sends in. free(buf->base); return; } // run the protocol state machine. for (int i = 0; i < nread; ++i) { switch (peerstate->state) { case initial_ack: assert(0 && "can't reach here"); break; case wait_for_msg: if (buf->base[i] == '^') { peerstate->state = in_msg; } break; case in_msg: if (buf->base[i] == '$') { peerstate->state = wait_for_msg; } else { assert(peerstate->sendbuf_end < sendbuf_size); peerstate->sendbuf[peerstate->sendbuf_end++] = buf->base[i] + 1; } break; } } if (peerstate->sendbuf_end > 0) { // we have data to send. the write buffer will point to the buffer stored // in the peer state for this client. uv_buf_t writebuf = uv_buf_init(peerstate->sendbuf, peerstate->sendbuf_end); uv_write_t* writereq = (uv_write_t*)xmalloc(sizeof(*writereq)); writereq->data = peerstate; int rc; if ((rc = uv_write(writereq, (uv_stream_t*)client, &writebuf, 1, on_wrote_buf)) < 0) { die("uv_write failed: %s", uv_strerror(rc)); } } } free(buf->base); } 这个服务器的运行时行为非常类似于第三部分的事件驱动服务器：所有的客户端都在一个单个的线程中并发处理。并且类似的，一些特定的行为必须在服务器代码中维护：服务器的逻辑实现为一个集成的回调，并且长周期运行是禁止的，因为它会阻塞事件循环。这一点也很类似。让我们进一步探索这个问题。在事件驱动循环中的长周期运行的操作单线程的事件驱动代码使它先天就容易受到一些常见问题的影响：长周期运行的代码会阻塞整个循环。参见如下的程序：void on_timer(uv_timer_t* timer) { uint64_t timestamp = uv_hrtime(); printf("on_timer [%" PRIu64 " ms]\n", (timestamp / 1000000) % 100000); // "Work" if (random() % 5 == 0) { printf("Sleeping...\n"); sleep(3); } } int main(int argc, const char** argv) { uv_timer_t timer; uv_timer_init(uv_default_loop(), &timer); uv_timer_start(&timer, on_timer, 0, 1000); return uv_run(uv_default_loop(), UV_RUN_DEFAULT); } 它用一个单个注册的回调运行一个 libuv 事件循环：on_timer，它被每秒钟循环调用一次。回调报告一个时间戳，并且，偶尔通过睡眠 3 秒去模拟一个长周期运行。这是运行示例：$ ./uv-timer-sleep-demo on_timer [4840 ms] on_timer [5842 ms] on_timer [6843 ms] on_timer [7844 ms] Sleeping... on_timer [11845 ms] on_timer [12846 ms] Sleeping... on_timer [16847 ms] on_timer [17849 ms] on_timer [18850 ms] ... on_timer 忠实地每秒执行一次，直到随机出现的睡眠为止。在那个时间点，on_timer 不再被调用，直到睡眠时间结束；事实上，没有其它的回调 会在这个时间帧中被调用。这个睡眠调用阻塞了当前线程，它正是被调用的线程，并且也是事件循环使用的线程。当这个线程被阻塞后，事件循环也被阻塞。这个示例演示了在事件驱动的调用中为什么回调不能被阻塞是多少的重要。并且，同样适用于 Node.js 服务器、客户端侧的 Javascript、大多数的 GUI 编程框架、以及许多其它的异步编程模型。但是，有时候运行耗时的任务是不可避免的。并不是所有任务都有一个异步 API；例如，我们可能使用一些仅有同步 API 的库去处理，或者，正在执行一个可能的长周期计算。我们如何用事件驱动编程去结合这些代码？线程可以帮到你！“转换” 阻塞调用为异步调用的线程一个线程池可以用于转换阻塞调用为异步调用，通过与事件循环并行运行，并且当任务完成时去由它去公布事件。以阻塞函数 do_work() 为例，这里介绍了它是怎么运行的：不在一个回调中直接调用 do_work() ，而是将它打包进一个 “任务”，让线程池去运行这个任务。当任务完成时，我们也为循环去调用它注册一个回调；我们称它为 on_work_done()。在这个时间点，我们的回调就可以返回了，而事件循环保持运行；在同一时间点，线程池中的有一个线程运行这个任务。一旦任务运行完成，通知主线程（指正在运行事件循环的线程），并且事件循环调用 on_work_done()。让我们看一下，使用 libuv 的工作调度 API，是怎么去解决我们前面的计时器/睡眠示例中展示的问题的：void on_after_work(uv_work_t* req, int status) { free(req); } void on_work(uv_work_t* req) { // "Work" if (random() % 5 == 0) { printf("Sleeping...\n"); sleep(3); } } void on_timer(uv_timer_t* timer) { uint64_t timestamp = uv_hrtime(); printf("on_timer [%" PRIu64 " ms]\n", (timestamp / 1000000) % 100000); uv_work_t* work_req = (uv_work_t*)malloc(sizeof(*work_req)); uv_queue_work(uv_default_loop(), work_req, on_work, on_after_work); } int main(int argc, const char** argv) { uv_timer_t timer; uv_timer_init(uv_default_loop(), &timer); uv_timer_start(&timer, on_timer, 0, 1000); return uv_run(uv_default_loop(), UV_RUN_DEFAULT); } 通过一个 work_req 注2 类型的句柄，我们进入一个任务队列，代替在 on_timer 上直接调用 sleep，这个函数在任务中（on_work）运行，并且，一旦任务完成（on_after_work），这个函数被调用一次。on_work 是指 “work”（阻塞中的/耗时的操作）进行的地方。注意在这两个回调传递到 uv_queue_work 时的一个关键区别：on_work 运行在线程池中，而 on_after_work 运行在事件循环中的主线程上 —— 就好像是其它的回调一样。让我们看一下这种方式的运行：$ ./uv-timer-work-demo on_timer [89571 ms] on_timer [90572 ms] on_timer [91573 ms] on_timer [92575 ms] Sleeping... on_timer [93576 ms] on_timer [94577 ms] Sleeping... on_timer [95577 ms] on_timer [96578 ms] on_timer [97578 ms] ... 即便在 sleep 函数被调用时，定时器也每秒钟滴答一下，睡眠现在运行在一个单独的线程中，并且不会阻塞事件循环。一个用于练习的素数测试服务器因为通过睡眠去模拟工作并不是件让人兴奋的事，我有一个事先准备好的更综合的一个示例 —— 一个基于套接字接受来自客户端的数字的服务器，检查这个数字是否是素数，然后去返回一个 “prime" 或者 “composite”。完整的 服务器代码在这里 —— 我不在这里粘贴了，因为它太长了，更希望读者在一些自己的练习中去体会它。这个服务器使用了一个原生的素数测试算法，因此，对于大的素数可能花很长时间才返回一个回答。在我的机器中，对于 2305843009213693951，它花了 ~5 秒钟去计算，但是，你的方法可能不同。练习 1：服务器有一个设置（通过一个名为 MODE 的环境变量）要么在套接字回调（意味着在主线程上）中运行素数测试，要么在 libuv 工作队列中。当多个客户端同时连接时，使用这个设置来观察服务器的行为。当它计算一个大的任务时，在阻塞模式中，服务器将不回复其它客户端，而在非阻塞模式中，它会回复。练习 2：libuv 有一个缺省大小的线程池，并且线程池的大小可以通过环境变量配置。你可以通过使用多个客户端去实验找出它的缺省值是多少？找到线程池缺省值后，使用不同的设置去看一下，在重负载下怎么去影响服务器的响应能力。在非阻塞文件系统中使用工作队列对于只是呆板的演示和 CPU 密集型的计算来说，将可能的阻塞操作委托给一个线程池并不是明智的；libuv 在它的文件系统 API 中本身就大量使用了这种能力。通过这种方式，libuv 使用一个异步 API，以一个轻便的方式显示出它强大的文件系统的处理能力。让我们使用 uv_fs_read()，例如，这个函数从一个文件中（表示为一个 uv_fs_t 句柄）读取一个文件到一个缓冲中 注3，并且当读取完成后调用一个回调。换句话说， uv_fs_read() 总是立即返回，即使是文件在一个类似 NFS 的系统上，而数据到达缓冲区可能需要一些时间。换句话说，这个 API 与这种方式中其它的 libuv API 是异步的。这是怎么工作的呢？在这一点上，我们看一下 libuv 的底层；内部实际上非常简单，并且它是一个很好的练习。作为一个可移植的库，libuv 对于 Windows 和 Unix 系统在它的许多函数上有不同的实现。我们去看一下在 libuv 源树中的 src/unix/fs.c。这是 uv_fs_read 的代码：int uv_fs_read(uv_loop_t* loop, uv_fs_t* req, uv_file file, const uv_buf_t bufs[], unsigned int nbufs, int64_t off, uv_fs_cb cb) { if (bufs == NULL || nbufs == 0) return -EINVAL; INIT(READ); req->file = file; req->nbufs = nbufs; req->bufs = req->bufsml; if (nbufs > ARRAY_SIZE(req->bufsml)) req->bufs = uv__malloc(nbufs * sizeof(*bufs)); if (req->bufs == NULL) { if (cb != NULL) uv__req_unregister(loop, req); return -ENOMEM; } memcpy(req->bufs, bufs, nbufs * sizeof(*bufs)); req->off = off; POST; } 第一次看可能觉得很困难，因为它延缓真实的工作到 INIT 和 POST 宏中，以及为 POST 设置了一些本地变量。这样做可以避免了文件中的许多重复代码。这是 INIT 宏：#define INIT(subtype) \ do { \ req->type = UV_FS; \ if (cb != NULL) \ uv__req_init(loop, req, UV_FS); \ req->fs_type = UV_FS_ ## subtype; \ req->result = 0; \ req->ptr = NULL; \ req->loop = loop; \ req->path = NULL; \ req->new_path = NULL; \ req->cb = cb; \ } \ while (0) 它设置了请求，并且更重要的是，设置 req->fs_type 域为真实的 FS 请求类型。因为 uv_fs_read 调用 INIT(READ)，它意味着 req->fs_type 被分配一个常数 UV_FS_READ。这是 POST 宏：#define POST \ do { \ if (cb != NULL) { \ uv__work_submit(loop, &req->work_req, uv__fs_work, uv__fs_done); \ return 0; \ } \ else { \ uv__fs_work(&req->work_req); \ return req->result; \ } \ } \ while (0) 它做什么取决于回调是否为 NULL。在 libuv 文件系统 API 中，一个 NULL 回调意味着我们真实地希望去执行一个 同步 操作。在这种情况下，POST 直接调用 uv__fs_work（我们需要了解一下这个函数的功能），而对于一个非 NULL 回调，它把 uv__fs_work 作为一个工作项提交到工作队列（指的是线程池），然后，注册 uv__fs_done 作为回调；该函数执行一些登记并调用用户提供的回调。如果我们去看 uv__fs_work 的代码，我们将看到它使用很多宏按照需求将工作分发到实际的文件系统调用。在我们的案例中，对于 UV_FS_READ 这个调用将被 uv__fs_read 生成，它（最终）使用普通的 POSIX API 去读取。这个函数可以在一个 阻塞 方式中很安全地实现。因为，它通过异步 API 调用时被置于一个线程池中。在 Node.js 中，fs.readFile 函数是映射到 uv_fs_read 上。因此，可以在一个非阻塞模式中读取文件，甚至是当底层文件系统 API 是阻塞方式时。注1： 为确保服务器不泄露内存，我在一个启用泄露检查的 Valgrind 中运行它。因为服务器经常是被设计为永久运行，这是一个挑战；为克服这个问题，我在服务器上添加了一个 “kill 开关” —— 一个从客户端接收的特定序列，以使它可以停止事件循环并退出。这个代码在 theon_wrote_buf 句柄中。注2： 在这里我们不过多地使用 work_req；讨论的素数测试服务器接下来将展示怎么被用于去传递上下文信息到回调中。注3： uv_fs_read() 提供了一个类似于 preadv Linux 系统调用的通用 API：它使用多缓冲区用于排序，并且支持一个到文件中的偏移。基于我们讨论的目的可以忽略这些特性。via: https://eli.thegreenplace.net/2017/concurrent-servers-part-4-libuv/作者：Eli Bendersky 译者：qhwdw 校对：wxy本文由 LCTT 原创编译，Linux中国 荣誉推出
这是并发服务器系列的第三节。第一节 介绍了阻塞式编程，第二节：线程 探讨了多线程，将其作为一种可行的方法来实现服务器并发编程。另一种常见的实现并发的方法叫做 事件驱动编程，也可以叫做 异步 编程 注1 。这种方法变化万千，因此我们会从最基本的开始，使用一些基本的 API 而非从封装好的高级方法开始。本系列以后的文章会讲高层次抽象，还有各种混合的方法。本系列的所有文章：第一节 - 简介第二节 - 线程第三节 - 事件驱动阻塞式 vs. 非阻塞式 I/O作为本篇的介绍，我们先讲讲阻塞和非阻塞 I/O 的区别。阻塞式 I/O 更好理解，因为这是我们使用 I/O 相关 API 时的“标准”方式。从套接字接收数据的时候，调用 recv 函数会发生 阻塞，直到它从端口上接收到了来自另一端套接字的数据。这恰恰是第一部分讲到的顺序服务器的问题。因此阻塞式 I/O 存在着固有的性能问题。第二节里我们讲过一种解决方法，就是用多线程。哪怕一个线程的 I/O 阻塞了，别的线程仍然可以使用 CPU 资源。实际上，阻塞 I/O 通常在利用资源方面非常高效，因为线程就等待着 —— 操作系统将线程变成休眠状态，只有满足了线程需要的条件才会被唤醒。非阻塞式 I/O 是另一种思路。把套接字设成非阻塞模式时，调用 recv 时（还有 send，但是我们现在只考虑接收），函数返回的会很快，哪怕没有接收到数据。这时，就会返回一个特殊的错误状态 注2 来通知调用者，此时没有数据传进来。调用者可以去做其他的事情，或者尝试再次调用 recv 函数。示范阻塞式和非阻塞式的 recv 区别的最好方式就是贴一段示例代码。这里有个监听套接字的小程序，一直在 recv 这里阻塞着；当 recv 返回了数据，程序就报告接收到了多少个字节 注3 ：int main(int argc, const char** argv) { setvbuf(stdout, NULL, _IONBF, 0); int portnum = 9988; if (argc >= 2) { portnum = atoi(argv[1]); } printf("Listening on port %d\n", portnum); int sockfd = listen_inet_socket(portnum); struct sockaddr_in peer_addr; socklen_t peer_addr_len = sizeof(peer_addr); int newsockfd = accept(sockfd, (struct sockaddr*)&peer_addr, &peer_addr_len); if (newsockfd < 0) { perror_die("ERROR on accept"); } report_peer_connected(&peer_addr, peer_addr_len); while (1) { uint8_t buf[1024]; printf("Calling recv...\n"); int len = recv(newsockfd, buf, sizeof buf, 0); if (len < 0) { perror_die("recv"); } else if (len == 0) { printf("Peer disconnected; I'm done.\n"); break; } printf("recv returned %d bytes\n", len); } close(newsockfd); close(sockfd); return 0; } 主循环重复调用 recv 并且报告它返回的字节数（记住 recv 返回 0 时，就是客户端断开连接了）。试着运行它，我们会在一个终端里运行这个程序，然后在另一个终端里用 nc 进行连接，发送一些字符，每次发送之间间隔几秒钟：$ nc localhost 9988 hello # wait for 2 seconds after typing this socket world ^D # to end the connection> 监听程序会输出以下内容：$ ./blocking-listener 9988 Listening on port 9988 peer (localhost, 37284) connected Calling recv... recv returned 6 bytes Calling recv... recv returned 13 bytes Calling recv... Peer disconnected; I'm done. 现在试试非阻塞的监听程序的版本。这是代码：int main(int argc, const char** argv) { setvbuf(stdout, NULL, _IONBF, 0); int portnum = 9988; if (argc >= 2) { portnum = atoi(argv[1]); } printf("Listening on port %d\n", portnum); int sockfd = listen_inet_socket(portnum); struct sockaddr_in peer_addr; socklen_t peer_addr_len = sizeof(peer_addr); int newsockfd = accept(sockfd, (struct sockaddr*)&peer_addr, &peer_addr_len); if (newsockfd < 0) { perror_die("ERROR on accept"); } report_peer_connected(&peer_addr, peer_addr_len); // 把套接字设成非阻塞模式 int flags = fcntl(newsockfd, F_GETFL, 0); if (flags == -1) { perror_die("fcntl F_GETFL"); } if (fcntl(newsockfd, F_SETFL, flags | O_NONBLOCK) == -1) { perror_die("fcntl F_SETFL O_NONBLOCK"); } while (1) { uint8_t buf[1024]; printf("Calling recv...\n"); int len = recv(newsockfd, buf, sizeof buf, 0); if (len < 0) { if (errno == EAGAIN || errno == EWOULDBLOCK) { usleep(200 * 1000); continue; } perror_die("recv"); } else if (len == 0) { printf("Peer disconnected; I'm done.\n"); break; } printf("recv returned %d bytes\n", len); } close(newsockfd); close(sockfd); return 0; } 这里与阻塞版本有些差异，值得注意：accept 函数返回的 newsocktfd 套接字因调用了 fcntl， 被设置成非阻塞的模式。检查 recv 的返回状态时，我们对 errno 进行了检查，判断它是否被设置成表示没有可供接收的数据的状态。这时，我们仅仅是休眠了 200 毫秒然后进入到下一轮循环。同样用 nc 进行测试，以下是非阻塞监听器的输出：$ ./nonblocking-listener 9988 Listening on port 9988 peer (localhost, 37288) connected Calling recv... Calling recv... Calling recv... Calling recv... Calling recv... Calling recv... Calling recv... Calling recv... Calling recv... recv returned 6 bytes Calling recv... Calling recv... Calling recv... Calling recv... Calling recv... Calling recv... Calling recv... Calling recv... Calling recv... Calling recv... Calling recv... recv returned 13 bytes Calling recv... Calling recv... Calling recv... Peer disconnected; I'm done. 作为练习，给输出添加一个时间戳，确认调用 recv 得到结果之间花费的时间是比输入到 nc 中所用的多还是少（每一轮是 200 ms）。这里就实现了使用非阻塞的 recv 让监听者检查套接字变为可能，并且在没有数据的时候重新获得控制权。换句话说，用编程的语言说这就是 轮询 polling —— 主程序周期性的查询套接字以便读取数据。对于顺序响应的问题，这似乎是个可行的方法。非阻塞的 recv 让同时与多个套接字通信变成可能，轮询这些套接字，仅当有新数据到来时才处理。就是这样，这种方式 可以 用来写并发服务器；但实际上一般不这么做，因为轮询的方式很难扩展。首先，我在代码中引入的 200ms 延迟对于演示非常好（监听器在我输入 nc 之间只打印几行 “Calling recv...”，但实际上应该有上千行）。但它也增加了多达 200ms 的服务器响应时间，这无意是不必要的。实际的程序中，延迟会低得多，休眠时间越短，进程占用的 CPU 资源就越多。有些时钟周期只是浪费在等待，这并不好，尤其是在移动设备上，这些设备的电量往往有限。但是当我们实际这样来使用多个套接字的时候，更严重的问题出现了。想像下监听器正在同时处理 1000 个客户端。这意味着每一个循环迭代里面，它都得为 这 1000 个套接字中的每一个 执行一遍非阻塞的 recv，找到其中准备好了数据的那一个。这非常低效，并且极大的限制了服务器能够并发处理的客户端数。这里有个准则：每次轮询之间等待的间隔越久，服务器响应性越差；而等待的时间越少，CPU 在无用的轮询上耗费的资源越多。讲真，所有的轮询都像是无用功。当然操作系统应该是知道哪个套接字是准备好了数据的，因此没必要逐个扫描。事实上，就是这样，接下来就会讲一些 API，让我们可以更优雅地处理多个客户端。selectselect 的系统调用是可移植的（POSIX），是标准 Unix API 中常有的部分。它是为上一节最后一部分描述的问题而设计的 —— 允许一个线程可以监视许多文件描述符 注4 的变化，而不用在轮询中执行不必要的代码。我并不打算在这里引入一个关于 select 的全面教程，有很多网站和书籍讲这个，但是在涉及到问题的相关内容时，我会介绍一下它的 API，然后再展示一个非常复杂的例子。select 允许 多路 I/O，监视多个文件描述符，查看其中任何一个的 I/O 是否可用。int select(int nfds, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout); readfds 指向文件描述符的缓冲区，这个缓冲区被监视是否有读取事件；fd_set 是一个特殊的数据结构，用户使用 FD_* 宏进行操作。writefds 是针对写事件的。nfds 是监视的缓冲中最大的文件描述符数字（文件描述符就是整数）。timeout 可以让用户指定 select 应该阻塞多久，直到某个文件描述符准备好了（timeout == NULL 就是说一直阻塞着）。现在先跳过 exceptfds。select 的调用过程如下：在调用之前，用户先要为所有不同种类的要监视的文件描述符创建 fd_set 实例。如果想要同时监视读取和写入事件，readfds 和 writefds 都要被创建并且引用。用户可以使用 FD_SET 来设置集合中想要监视的特殊描述符。例如，如果想要监视描述符 2、7 和 10 的读取事件，在 readfds 这里调用三次 FD_SET，分别设置 2、7 和 10。select 被调用。当 select 返回时（现在先不管超时），就是说集合中有多少个文件描述符已经就绪了。它也修改 readfds 和 writefds 集合，来标记这些准备好的描述符。其它所有的描述符都会被清空。这时用户需要遍历 readfds 和 writefds，找到哪个描述符就绪了（使用 FD_ISSET）。作为完整的例子，我在并发的服务器程序上使用 select，重新实现了我们之前的协议。完整的代码在这里；接下来的是代码中的重点部分及注释。警告：示例代码非常复杂，因此第一次看的时候，如果没有足够的时间，快速浏览也没有关系。使用 select 的并发服务器使用 I/O 的多发 API 诸如 select 会给我们服务器的设计带来一些限制；这不会马上显现出来，但这值得探讨，因为它们是理解事件驱动编程到底是什么的关键。最重要的是，要记住这种方法本质上是单线程的 注5 。服务器实际上在 同一时刻只能做一件事。因为我们想要同时处理多个客户端请求，我们需要换一种方式重构代码。首先，让我们谈谈主循环。它看起来是什么样的呢？先让我们想象一下服务器有一堆任务，它应该监视哪些东西呢？两种类型的套接字活动：新客户端尝试连接。这些客户端应该被 accept。已连接的客户端发送数据。这个数据要用 第一节 中所讲到的协议进行传输，有可能会有一些数据要被回送给客户端。尽管这两种活动在本质上有所区别，我们还是要把它们放在一个循环里，因为只能有一个主循环。循环会包含 select 的调用。这个 select 的调用会监视上述的两种活动。这里是部分代码，设置了文件描述符集合，并在主循环里转到被调用的 select 部分。// “master” 集合存活在该循环中，跟踪我们想要监视的读取事件或写入事件的文件描述符（FD）。 fd_set readfds_master; FD_ZERO(&readfds_master); fd_set writefds_master; FD_ZERO(&writefds_master); // 监听的套接字一直被监视，用于读取数据，并监测到来的新的端点连接。 FD_SET(listener_sockfd, &readfds_master); // 要想更加高效，fdset_max 追踪当前已知最大的 FD；这使得每次调用时对 FD_SETSIZE 的迭代选择不是那么重要了。 int fdset_max = listener_sockfd; while (1) { // select() 会修改传递给它的 fd_sets，因此进行拷贝一下再传值。 fd_set readfds = readfds_master; fd_set writefds = writefds_master; int nready = select(fdset_max + 1, &readfds, &writefds, NULL, NULL); if (nready < 0) { perror_die("select"); } ... 这里的一些要点：由于每次调用 select 都会重写传递给函数的集合，调用器就得维护一个 “master” 集合，在循环迭代中，保持对所监视的所有活跃的套接字的追踪。注意我们所关心的，最开始的唯一那个套接字是怎么变成 listener_sockfd 的，这就是最开始的套接字，服务器借此来接收新客户端的连接。select 的返回值，是在作为参数传递的集合中，那些已经就绪的描述符的个数。select 修改这个集合，用来标记就绪的描述符。下一步是在这些描述符中进行迭代。... for (int fd = 0; fd <= fdset_max && nready > 0; fd++) { // 检查 fd 是否变成可读的 if (FD_ISSET(fd, &readfds)) { nready--; if (fd == listener_sockfd) { // 监听的套接字就绪了；这意味着有个新的客户端连接正在联系 ... } else { fd_status_t status = on_peer_ready_recv(fd); if (status.want_read) { FD_SET(fd, &readfds_master); } else { FD_CLR(fd, &readfds_master); } if (status.want_write) { FD_SET(fd, &writefds_master); } else { FD_CLR(fd, &writefds_master); } if (!status.want_read && !status.want_write) { printf("socket %d closing\n", fd); close(fd); } } 这部分循环检查 可读的 描述符。让我们跳过监听器套接字（要浏览所有内容，看这个代码） 然后看看当其中一个客户端准备好了之后会发生什么。出现了这种情况后，我们调用一个叫做 on_peer_ready_recv 的 回调 函数，传入相应的文件描述符。这个调用意味着客户端连接到套接字上，发送某些数据，并且对套接字上 recv 的调用不会被阻塞 注6 。这个回调函数返回结构体 fd_status_t。typedef struct { bool want_read; bool want_write; } fd_status_t; 这个结构体告诉主循环，是否应该监视套接字的读取事件、写入事件，或者两者都监视。上述代码展示了 FD_SET 和 FD_CLR 是怎么在合适的描述符集合中被调用的。对于主循环中某个准备好了写入数据的描述符，代码是类似的，除了它所调用的回调函数，这个回调函数叫做 on_peer_ready_send。现在来花点时间看看这个回调：typedef enum { INITIAL_ACK, WAIT_FOR_MSG, IN_MSG } ProcessingState; #define SENDBUF_SIZE 1024 typedef struct { ProcessingState state; // sendbuf 包含了服务器要返回给客户端的数据。on_peer_ready_recv 句柄填充这个缓冲， // on_peer_read_send 进行消耗。sendbuf_end 指向缓冲区的最后一个有效字节， // sendptr 指向下个字节 uint8_t sendbuf[SENDBUF_SIZE]; int sendbuf_end; int sendptr; } peer_state_t; // 每一端都是通过它连接的文件描述符（fd）进行区分。只要客户端连接上了，fd 就是唯一的。 // 当客户端断开连接，另一个客户端连接上就会获得相同的 fd。on_peer_connected 应该 // 进行初始化，以便移除旧客户端在同一个 fd 上留下的东西。 peer_state_t global_state[MAXFDS]; fd_status_t on_peer_ready_recv(int sockfd) { assert(sockfd < MAXFDs); peer_state_t* peerstate = &global_state[sockfd]; if (peerstate->state == INITIAL_ACK || peerstate->sendptr < peerstate->sendbuf_end) { // 在初始的 ACK 被送到了客户端，就没有什么要接收的了。 // 等所有待发送的数据都被发送之后接收更多的数据。 return fd_status_W; } uint8_t buf[1024]; int nbytes = recv(sockfd, buf, sizeof buf, 0); if (nbytes == 0) { // 客户端断开连接 return fd_status_NORW; } else if (nbytes < 0) { if (errno == EAGAIN || errno == EWOULDBLOCK) { // 套接字 *实际* 并没有准备好接收，等到它就绪。 return fd_status_R; } else { perror_die("recv"); } } bool ready_to_send = false; for (int i = 0; i < nbytes; ++i) { switch (peerstate->state) { case INITIAL_ACK: assert(0 && "can't reach here"); break; case WAIT_FOR_MSG: if (buf[i] == '^') { peerstate->state = IN_MSG; } break; case IN_MSG: if (buf[i] == '$') { peerstate->state = WAIT_FOR_MSG; } else { assert(peerstate->sendbuf_end < SENDBUF_SIZE); peerstate->sendbuf[peerstate->sendbuf_end++] = buf[i] + 1; ready_to_send = true; } break; } } // 如果没有数据要发送给客户端，报告读取状态作为最后接收的结果。 return (fd_status_t){.want_read = !ready_to_send, .want_write = ready_to_send}; } peer_state_t 是全状态对象，用来表示在主循环中两次回调函数调用之间的客户端的连接。因为回调函数在客户端发送的某些数据时被调用，不能假设它能够不停地与客户端通信，并且它得运行得很快，不能被阻塞。因为套接字被设置成非阻塞模式，recv 会快速的返回。除了调用 recv， 这个句柄做的是处理状态，没有其它的调用，从而不会发生阻塞。举个例子，你知道为什么这个代码需要一个额外的状态吗？这个系列中，我们的服务器目前只用到了两个状态，但是这个服务器程序需要三个状态。来看看 “套接字准备好发送” 的回调函数：fd_status_t on_peer_ready_send(int sockfd) { assert(sockfd < MAXFDs); peer_state_t* peerstate = &global_state[sockfd]; if (peerstate->sendptr >= peerstate->sendbuf_end) { // 没有要发送的。 return fd_status_RW; } int sendlen = peerstate->sendbuf_end - peerstate->sendptr; int nsent = send(sockfd, peerstate->sendbuf, sendlen, 0); if (nsent == -1) { if (errno == EAGAIN || errno == EWOULDBLOCK) { return fd_status_W; } else { perror_die("send"); } } if (nsent < sendlen) { peerstate->sendptr += nsent; return fd_status_W; } else { // 所有东西都成功发送；重置发送队列。 peerstate->sendptr = 0; peerstate->sendbuf_end = 0; // 如果我们现在是处于特殊的 INITIAL_ACK 状态，就转变到其他状态。 if (peerstate->state == INITIAL_ACK) { peerstate->state = WAIT_FOR_MSG; } return fd_status_R; } } 这里也一样，回调函数调用了一个非阻塞的 send，演示了状态管理。在异步代码中，回调函数执行的很快是受争议的，任何延迟都会阻塞主循环进行处理，因此也阻塞了整个服务器程序去处理其他的客户端。用脚步再来运行这个服务器，同时连接 3 个客户端。在一个终端中我们运行下面的命令：$ ./select-server 在另一个终端中：$ python3.6 simple-client.py -n 3 localhost 9090 INFO:2017-09-26 05:29:15,864:conn1 connected... INFO:2017-09-26 05:29:15,864:conn2 connected... INFO:2017-09-26 05:29:15,864:conn0 connected... INFO:2017-09-26 05:29:15,865:conn1 sending b'^abc$de^abte$f' INFO:2017-09-26 05:29:15,865:conn2 sending b'^abc$de^abte$f' INFO:2017-09-26 05:29:15,865:conn0 sending b'^abc$de^abte$f' INFO:2017-09-26 05:29:15,865:conn1 received b'bcdbcuf' INFO:2017-09-26 05:29:15,865:conn2 received b'bcdbcuf' INFO:2017-09-26 05:29:15,865:conn0 received b'bcdbcuf' INFO:2017-09-26 05:29:16,866:conn1 sending b'xyz^123' INFO:2017-09-26 05:29:16,867:conn0 sending b'xyz^123' INFO:2017-09-26 05:29:16,867:conn2 sending b'xyz^123' INFO:2017-09-26 05:29:16,867:conn1 received b'234' INFO:2017-09-26 05:29:16,868:conn0 received b'234' INFO:2017-09-26 05:29:16,868:conn2 received b'234' INFO:2017-09-26 05:29:17,868:conn1 sending b'25$^ab0000$abab' INFO:2017-09-26 05:29:17,869:conn1 received b'36bc1111' INFO:2017-09-26 05:29:17,869:conn0 sending b'25$^ab0000$abab' INFO:2017-09-26 05:29:17,870:conn0 received b'36bc1111' INFO:2017-09-26 05:29:17,870:conn2 sending b'25$^ab0000$abab' INFO:2017-09-26 05:29:17,870:conn2 received b'36bc1111' INFO:2017-09-26 05:29:18,069:conn1 disconnecting INFO:2017-09-26 05:29:18,070:conn0 disconnecting INFO:2017-09-26 05:29:18,070:conn2 disconnecting 和线程的情况相似，客户端之间没有延迟，它们被同时处理。而且在 select-server 也没有用线程！主循环 多路 处理所有的客户端，通过高效使用 select 轮询多个套接字。回想下 第二节中 顺序的 vs 多线程的客户端处理过程的图片。对于我们的 select-server，三个客户端的处理流程像这样：所有的客户端在同一个线程中同时被处理，通过乘积，做一点这个客户端的任务，然后切换到另一个，再切换到下一个，最后切换回到最开始的那个客户端。注意，这里没有什么循环调度，客户端在它们发送数据的时候被客户端处理，这实际上是受客户端左右的。同步、异步、事件驱动、回调select-server 示例代码为讨论什么是异步编程、它和事件驱动及基于回调的编程有何联系，提供了一个良好的背景。因为这些词汇在并发服务器的（非常矛盾的）讨论中很常见。让我们从一段 select 的手册页面中引用的一句话开始：select，pselect，FD\_CLR，FD\_ISSET，FD\_SET，FD\_ZERO - 同步 I/O 处理因此 select 是 同步 处理。但我刚刚演示了大量代码的例子，使用 select 作为 异步 处理服务器的例子。有哪些东西？答案是：这取决于你的观察角度。同步常用作阻塞处理，并且对 select 的调用实际上是阻塞的。和第 1、2 节中讲到的顺序的、多线程的服务器中对 send 和 recv 是一样的。因此说 select 是 同步的 API 是有道理的。可是，服务器的设计却可以是 异步的，或是 基于回调的，或是 事件驱动的，尽管其中有对 select 的使用。注意这里的 on_peer_* 函数是回调函数；它们永远不会阻塞，并且只有网络事件触发的时候才会被调用。它们可以获得部分数据，并能够在调用过程中保持稳定的状态。如果你曾经做过一些 GUI 编程，这些东西对你来说应该很亲切。有个 “事件循环”，常常完全隐藏在框架里，应用的 “业务逻辑” 建立在回调上，这些回调会在各种事件触发后被调用，用户点击鼠标、选择菜单、定时器触发、数据到达套接字等等。曾经最常见的编程模型是客户端的 JavaScript，这里面有一堆回调函数，它们在浏览网页时用户的行为被触发。select 的局限使用 select 作为第一个异步服务器的例子对于说明这个概念很有用，而且由于 select 是很常见、可移植的 API。但是它也有一些严重的缺陷，在监视的文件描述符非常大的时候就会出现。有限的文件描述符的集合大小。糟糕的性能。从文件描述符的大小开始。FD_SETSIZE 是一个编译期常数，在如今的操作系统中，它的值通常是 1024。它被硬编码在 glibc 的头文件里，并且不容易修改。它把 select 能够监视的文件描述符的数量限制在 1024 以内。曾有些人想要写出能够处理上万个并发访问的客户端请求的服务器，所以这个问题很有现实意义。有一些方法，但是不可移植，也很难用。糟糕的性能问题就好解决的多，但是依然非常严重。注意当 select 返回的时候，它向调用者提供的信息是 “就绪的” 描述符的个数，还有被修改过的描述符集合。描述符集映射着描述符“就绪/未就绪”，但是并没有提供什么有效的方法去遍历所有就绪的描述符。如果只有一个描述符是就绪的，最坏的情况是调用者需要遍历 整个集合 来找到那个描述符。这在监视的描述符数量比较少的时候还行，但是如果数量变的很大的时候，这种方法弊端就凸显出了 注7 。由于这些原因，为了写出高性能的并发服务器， select 已经不怎么用了。每一个流行的操作系统有独特的不可移植的 API，允许用户写出非常高效的事件循环；像框架这样的高级结构还有高级语言通常在一个可移植的接口中包含这些 API。epoll举个例子，来看看 epoll，Linux 上的关于高容量 I/O 事件通知问题的解决方案。epoll 高效的关键之处在于它与内核更好的协作。不是使用文件描述符，epoll_wait 用当前准备好的事件填满一个缓冲区。只有准备好的事件添加到了缓冲区，因此没有必要遍历客户端中当前 所有 监视的文件描述符。这简化了查找就绪的描述符的过程，把空间复杂度从 select 中的 O(N) 变为了 O(1)。关于 epoll API 的完整展示不是这里的目的，网上有很多相关资源。虽然你可能猜到了，我还要写一个不同的并发服务器，这次是用 epool 而不是 select。完整的示例代码 在这里。实际上，由于大部分代码和 用 select 的服务器相同，所以我只会讲要点，在主循环里使用 epoll：struct epoll_event accept_event; accept_event.data.fd = listener_sockfd; accept_event.events = EPOLLIN; if (epoll_ctl(epollfd, EPOLL_CTL_ADD, listener_sockfd, &accept_event) < 0) { perror_die("epoll_ctl EPOLL_CTL_ADD"); } struct epoll_event* events = calloc(MAXFDS, sizeof(struct epoll_event)); if (events == NULL) { die("Unable to allocate memory for epoll_events"); } while (1) { int nready = epoll_wait(epollfd, events, MAXFDS, -1); for (int i = 0; i < nready; i++) { if (events[i].events & EPOLLERR) { perror_die("epoll_wait returned EPOLLERR"); } if (events[i].data.fd == listener_sockfd) { // 监听的套接字就绪了；意味着新客户端正在连接。 ... } else { // A peer socket is ready. if (events[i].events & EPOLLIN) { // 准备好了读取 ... } else if (events[i].events & EPOLLOUT) { // 准备好了写入 ... } } } } 通过调用 epoll_ctl 来配置 epoll。这时，配置监听的套接字数量，也就是 epoll 监听的描述符的数量。然后分配一个缓冲区，把就绪的事件传给 epoll 以供修改。在主循环里对 epoll_wait 的调用是魅力所在。它阻塞着，直到某个描述符就绪了（或者超时），返回就绪的描述符数量。但这时，不要盲目地迭代所有监视的集合，我们知道 epoll_write 会修改传给它的 events 缓冲区，缓冲区中有就绪的事件，从 0 到 nready-1，因此我们只需迭代必要的次数。要在 select 里面重新遍历，有明显的差异：如果在监视着 1000 个描述符，只有两个就绪， epoll_waits 返回的是 nready=2，然后修改 events 缓冲区最前面的两个元素，因此我们只需要“遍历”两个描述符。用 select 我们就需要遍历 1000 个描述符，找出哪个是就绪的。因此，在繁忙的服务器上，有许多活跃的套接字时 epoll 比 select 更加容易扩展。剩下的代码很直观，因为我们已经很熟悉 “select 服务器” 了。实际上，“epoll 服务器” 中的所有“业务逻辑”和 “select 服务器” 是一样的，回调构成相同的代码。这种相似是通过将事件循环抽象分离到一个库/框架中。我将会详述这些内容，因为很多优秀的程序员曾经也是这样做的。相反，下一篇文章里我们会了解 libuv，一个最近出现的更加受欢迎的时间循环抽象层。像 libuv 这样的库让我们能够写出并发的异步服务器，并且不用考虑系统调用下繁琐的细节。注1：我试着在做网络浏览和阅读这两件事的实际差别中突显自己，但经常做得头疼。有很多不同的选项，从“它们是一样的东西”到“一个是另一个的子集”，再到“它们是完全不同的东西”。在面临这样主观的观点时，最好是完全放弃这个问题，专注特殊的例子和用例。注2：POSIX 表示这可以是 EAGAIN，也可以是 EWOULDBLOCK，可移植应用应该对这两个都进行检查。注3：和这个系列所有的 C 示例类似，代码中用到了某些助手工具来设置监听套接字。这些工具的完整代码在这个 仓库 的 utils 模块里。注4：select 不是网络/套接字专用的函数，它可以监视任意的文件描述符，有可能是硬盘文件、管道、终端、套接字或者 Unix 系统中用到的任何文件描述符。这篇文章里，我们主要关注它在套接字方面的应用。注5：有多种方式用多线程来实现事件驱动，我会把它放在稍后的文章中进行讨论。注6：由于各种非实验因素，它 仍然 可以阻塞，即使是在 select 说它就绪了之后。因此服务器上打开的所有套接字都被设置成非阻塞模式，如果对 recv 或 send 的调用返回了 EAGAIN 或者 EWOULDBLOCK，回调函数就装作没有事件发生。阅读示例代码的注释可以了解更多细节。注7：注意这比该文章前面所讲的异步轮询的例子要稍好一点。轮询需要 一直 发生，而 select 实际上会阻塞到有一个或多个套接字准备好读取/写入；select 会比一直询问浪费少得多的 CPU 时间。via: https://eli.thegreenplace.net/2017/concurrent-servers-part-3-event-driven/作者：Eli Bendersky 译者：GitFuture 校对：wxy本文由 LCTT 原创编译，Linux中国 荣誉推出
这是并发网络服务器系列的第二节。第一节 提出了服务端实现的协议，还有简单的顺序服务器的代码，是这整个系列的基础。这一节里，我们来看看怎么用多线程来实现并发，用 C 实现一个最简单的多线程服务器，和用 Python 实现的线程池。该系列的所有文章：第一节 - 简介第二节 - 线程第三节 - 事件驱动多线程的方法设计并发服务器说起第一节里的顺序服务器的性能，最显而易见的，是在服务器处理客户端连接时，计算机的很多资源都被浪费掉了。尽管假定客户端快速发送完消息，不做任何等待，仍然需要考虑网络通信的开销；网络要比现在的 CPU 慢上百万倍还不止，因此 CPU 运行服务器时会等待接收套接字的流量，而大量的时间都花在完全不必要的等待中。这里是一份示意图，表明顺序时客户端的运行过程：这个图片上有 3 个客户端程序。棱形表示客户端的“到达时间”（即客户端尝试连接服务器的时间）。黑色线条表示“等待时间”（客户端等待服务器真正接受连接所用的时间），有色矩形表示“处理时间”（服务器和客户端使用协议进行交互所用的时间）。有色矩形的末端表示客户端断开连接。上图中，绿色和橘色的客户端尽管紧跟在蓝色客户端之后到达服务器，也要等到服务器处理完蓝色客户端的请求。这时绿色客户端得到响应，橘色的还要等待一段时间。多线程服务器会开启多个控制线程，让操作系统管理 CPU 的并发（使用多个 CPU 核心）。当客户端连接的时候，创建一个线程与之交互，而在主线程中，服务器能够接受其他的客户端连接。下图是该模式的时间轴：每个客户端一个线程，在 C 语言里要用 pthread这篇文章的 第一个示例代码 是一个简单的 “每个客户端一个线程” 的服务器，用 C 语言编写，使用了 phtreads API 用于实现多线程。这里是主循环代码：while (1) { struct sockaddr_in peer_addr; socklen_t peer_addr_len = sizeof(peer_addr); int newsockfd = accept(sockfd, (struct sockaddr*)&peer_addr, &peer_addr_len); if (newsockfd < 0) { perror_die("ERROR on accept"); } report_peer_connected(&peer_addr, peer_addr_len); pthread_t the_thread; thread_config_t* config = (thread_config_t*)malloc(sizeof(*config)); if (!config) { die("OOM"); } config->sockfd = newsockfd; pthread_create(&the_thread, NULL, server_thread, config); // 回收线程 —— 在线程结束的时候，它占用的资源会被回收 // 因为主线程在一直运行，所以它比服务线程存活更久。 pthread_detach(the_thread); } 这是 server_thread 函数：void* server_thread(void* arg) { thread_config_t* config = (thread_config_t*)arg; int sockfd = config->sockfd; free(config); // This cast will work for Linux, but in general casting pthread_id to an 这个类型转换在 Linux 中可以正常运行，但是一般来说将 pthread_id 类型转换成整形不便于移植代码 // integral type isn't portable. unsigned long id = (unsigned long)pthread_self(); printf("Thread %lu created to handle connection with socket %d\n", id, sockfd); serve_connection(sockfd); printf("Thread %lu done\n", id); return 0; } 线程 “configuration” 是作为 thread_config_t 结构体进行传递的：typedef struct { int sockfd; } thread_config_t; 主循环中调用的 pthread_create 产生一个新线程，然后运行 server_thread 函数。这个线程会在 server_thread 返回的时候结束。而在 serve_connection 返回的时候 server_thread 才会返回。serve_connection 和第一节完全一样。第一节中我们用脚本生成了多个并发访问的客户端，观察服务器是怎么处理的。现在来看看多线程服务器的处理结果：$ python3.6 simple-client.py -n 3 localhost 9090 INFO:2017-09-20 06:31:56,632:conn1 connected... INFO:2017-09-20 06:31:56,632:conn2 connected... INFO:2017-09-20 06:31:56,632:conn0 connected... INFO:2017-09-20 06:31:56,632:conn1 sending b'^abc$de^abte$f' INFO:2017-09-20 06:31:56,632:conn2 sending b'^abc$de^abte$f' INFO:2017-09-20 06:31:56,632:conn0 sending b'^abc$de^abte$f' INFO:2017-09-20 06:31:56,633:conn1 received b'b' INFO:2017-09-20 06:31:56,633:conn2 received b'b' INFO:2017-09-20 06:31:56,633:conn0 received b'b' INFO:2017-09-20 06:31:56,670:conn1 received b'cdbcuf' INFO:2017-09-20 06:31:56,671:conn0 received b'cdbcuf' INFO:2017-09-20 06:31:56,671:conn2 received b'cdbcuf' INFO:2017-09-20 06:31:57,634:conn1 sending b'xyz^123' INFO:2017-09-20 06:31:57,634:conn2 sending b'xyz^123' INFO:2017-09-20 06:31:57,634:conn1 received b'234' INFO:2017-09-20 06:31:57,634:conn0 sending b'xyz^123' INFO:2017-09-20 06:31:57,634:conn2 received b'234' INFO:2017-09-20 06:31:57,634:conn0 received b'234' INFO:2017-09-20 06:31:58,635:conn1 sending b'25$^ab0000$abab' INFO:2017-09-20 06:31:58,635:conn2 sending b'25$^ab0000$abab' INFO:2017-09-20 06:31:58,636:conn1 received b'36bc1111' INFO:2017-09-20 06:31:58,636:conn2 received b'36bc1111' INFO:2017-09-20 06:31:58,637:conn0 sending b'25$^ab0000$abab' INFO:2017-09-20 06:31:58,637:conn0 received b'36bc1111' INFO:2017-09-20 06:31:58,836:conn2 disconnecting INFO:2017-09-20 06:31:58,836:conn1 disconnecting INFO:2017-09-20 06:31:58,837:conn0 disconnecting 实际上，所有客户端同时连接，它们与服务器的通信是同时发生的。每个客户端一个线程的难点尽管在现代操作系统中就资源利用率方面来看，线程相当的高效，但前一节中讲到的方法在高负载时却会出现纰漏。想象一下这样的情景：很多客户端同时进行连接，某些会话持续的时间长。这意味着某个时刻服务器上有很多活跃的线程。太多的线程会消耗掉大量的内存和 CPU 资源，而仅仅是用于上下文切换 注1 。另外其也可视为安全问题：因为这样的设计容易让服务器成为 DoS 攻击 的目标 —— 上百万个客户端同时连接，并且客户端都处于闲置状态，这样耗尽了所有资源就可能让服务器宕机。当服务器要与每个客户端通信，CPU 进行大量计算时，就会出现更严重的问题。这种情况下，容易想到的方法是减少服务器的响应能力 —— 只有其中一些客户端能得到服务器的响应。因此，对多线程服务器所能够处理的并发客户端数做一些 速率限制 就是个明智的选择。有很多方法可以实现。最容易想到的是计数当前已经连接上的客户端，把连接数限制在某个范围内（需要通过仔细的测试后决定）。另一种流行的多线程应用设计是使用 线程池。线程池线程池 很简单，也很有用。服务器创建几个任务线程，这些线程从某些队列中获取任务。这就是“池”。然后每一个客户端的连接被当成任务分发到池中。只要池中有空闲的线程，它就会去处理任务。如果当前池中所有线程都是繁忙状态，那么服务器就会阻塞，直到线程池可以接受任务（某个繁忙状态的线程处理完当前任务后，变回空闲的状态）。这里有个 4 线程的线程池处理任务的图。任务（这里就是客户端的连接）要等到线程池中的某个线程可以接受新任务。非常明显，线程池的定义就是一种按比例限制的机制。我们可以提前设定服务器所能拥有的线程数。那么这就是并发连接的最多的客户端数 —— 其它的客户端就要等到线程空闲。如果我们的池中有 8 个线程，那么 8 就是服务器可以处理的最多的客户端并发连接数，哪怕上千个客户端想要同时连接。那么怎么确定池中需要有多少个线程呢？通过对问题范畴进行细致的分析、评估、实验以及根据我们拥有的硬件配置。如果是单核的云服务器，答案只有一个；如果是 100 核心的多套接字的服务器，那么答案就有很多种。也可以在运行时根据负载动态选择池的大小 —— 我会在这个系列之后的文章中谈到这个东西。使用线程池的服务器在高负载情况下表现出 性能退化 —— 客户端能够以稳定的速率进行连接，可能会比其它时刻得到响应的用时稍微久一点；也就是说，无论多少个客户端同时进行连接，服务器总能保持响应，尽最大能力响应等待的客户端。与之相反，每个客户端一个线程的服务器，会接收多个客户端的连接直到过载，这时它更容易崩溃或者因为要处理所有客户端而变得缓慢，因为资源都被耗尽了（比如虚拟内存的占用）。在服务器上使用线程池为了改变服务器的实现，我用了 Python，在 Python 的标准库中带有一个已经实现好的稳定的线程池。（concurrent.futures 模块里的 ThreadPoolExecutor） 注2 。服务器创建一个线程池，然后进入循环，监听套接字接收客户端的连接。用 submit 把每一个连接的客户端分配到池中：pool = ThreadPoolExecutor(args.n) sockobj = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sockobj.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sockobj.bind(('localhost', args.port)) sockobj.listen(15) try: while True: client_socket, client_address = sockobj.accept() pool.submit(serve_connection, client_socket, client_address) except KeyboardInterrupt as e: print(e) sockobj.close() serve_connection 函数和 C 的那部分很像，与一个客户端交互，直到其断开连接，并且遵循我们的协议：ProcessingState = Enum('ProcessingState', 'WAIT_FOR_MSG IN_MSG') def serve_connection(sockobj, client_address): print('{0} connected'.format(client_address)) sockobj.sendall(b'*') state = ProcessingState.WAIT_FOR_MSG while True: try: buf = sockobj.recv(1024) if not buf: break except IOError as e: break for b in buf: if state == ProcessingState.WAIT_FOR_MSG: if b == ord(b'^'): state = ProcessingState.IN_MSG elif state == ProcessingState.IN_MSG: if b == ord(b'$'): state = ProcessingState.WAIT_FOR_MSG else: sockobj.send(bytes([b + 1])) else: assert False print('{0} done'.format(client_address)) sys.stdout.flush() sockobj.close() 来看看线程池的大小对并行访问的客户端的阻塞行为有什么样的影响。为了演示，我会运行一个池大小为 2 的线程池服务器（只生成两个线程用于响应客户端）。$ python3.6 threadpool-server.py -n 2 在另外一个终端里，运行客户端模拟器，产生 3 个并发访问的客户端：$ python3.6 simple-client.py -n 3 localhost 9090 INFO:2017-09-22 05:58:52,815:conn1 connected... INFO:2017-09-22 05:58:52,827:conn0 connected... INFO:2017-09-22 05:58:52,828:conn1 sending b'^abc$de^abte$f' INFO:2017-09-22 05:58:52,828:conn0 sending b'^abc$de^abte$f' INFO:2017-09-22 05:58:52,828:conn1 received b'b' INFO:2017-09-22 05:58:52,828:conn0 received b'b' INFO:2017-09-22 05:58:52,867:conn1 received b'cdbcuf' INFO:2017-09-22 05:58:52,867:conn0 received b'cdbcuf' INFO:2017-09-22 05:58:53,829:conn1 sending b'xyz^123' INFO:2017-09-22 05:58:53,829:conn0 sending b'xyz^123' INFO:2017-09-22 05:58:53,830:conn1 received b'234' INFO:2017-09-22 05:58:53,831:conn0 received b'2' INFO:2017-09-22 05:58:53,831:conn0 received b'34' INFO:2017-09-22 05:58:54,831:conn1 sending b'25$^ab0000$abab' INFO:2017-09-22 05:58:54,832:conn1 received b'36bc1111' INFO:2017-09-22 05:58:54,832:conn0 sending b'25$^ab0000$abab' INFO:2017-09-22 05:58:54,833:conn0 received b'36bc1111' INFO:2017-09-22 05:58:55,032:conn1 disconnecting INFO:2017-09-22 05:58:55,032:conn2 connected... INFO:2017-09-22 05:58:55,033:conn2 sending b'^abc$de^abte$f' INFO:2017-09-22 05:58:55,033:conn0 disconnecting INFO:2017-09-22 05:58:55,034:conn2 received b'b' INFO:2017-09-22 05:58:55,071:conn2 received b'cdbcuf' INFO:2017-09-22 05:58:56,036:conn2 sending b'xyz^123' INFO:2017-09-22 05:58:56,036:conn2 received b'234' INFO:2017-09-22 05:58:57,037:conn2 sending b'25$^ab0000$abab' INFO:2017-09-22 05:58:57,038:conn2 received b'36bc1111' INFO:2017-09-22 05:58:57,238:conn2 disconnecting 回顾之前讨论的服务器行为：在顺序服务器中，所有的连接都是串行的。一个连接结束后，下一个连接才能开始。前面讲到的每个客户端一个线程的服务器中，所有连接都被同时接受并得到服务。这里可以看到一种可能的情况：两个连接同时得到服务，只有其中一个结束连接后第三个才能连接上。这就是把线程池大小设置成 2 的结果。真实用例中我们会把线程池设置的更大些，取决于机器和实际的协议。线程池的缓冲机制就能很好理解了 —— 我 几个月前 更详细的介绍过这种机制，关于 Clojure 的 core.async 模块。总结与展望这篇文章讨论了在服务器中，用多线程作并发的方法。每个客户端一个线程的方法最早提出来，但是实际上却不常用，因为它并不安全。线程池就常见多了，最受欢迎的几个编程语言有良好的实现（某些编程语言，像 Python，就是在标准库中实现）。这里说的使用线程池的服务器，不会受到每个客户端一个线程的弊端。然而，线程不是处理多个客户端并行访问的唯一方法。下一节中我们会看看其它的解决方案，可以使用异步处理，或者事件驱动的编程。注1：老实说，现代 Linux 内核可以承受足够多的并发线程 —— 只要这些线程主要在 I/O 上被阻塞。这里有个示例程序，它产生可配置数量的线程，线程在循环体中是休眠的，每 50 ms 唤醒一次。我在 4 核的 Linux 机器上可以轻松的产生 10000 个线程；哪怕这些线程大多数时间都在睡眠，它们仍然消耗一到两个核心，以便实现上下文切换。而且，它们占用了 80 GB 的虚拟内存（Linux 上每个线程的栈大小默认是 8MB）。实际使用中，线程会使用内存并且不会在循环体中休眠，因此它可以非常快的占用完一个机器的内存。注2：自己动手实现一个线程池是个有意思的练习，但我现在还不想做。我曾写过用来练手的 针对特殊任务的线程池。是用 Python 写的；用 C 重写的话有些难度，但对于经验丰富的程序员，几个小时就够了。via: https://eli.thegreenplace.net/2017/concurrent-servers-part-2-threads/作者：Eli Bendersky 译者：GitFuture 校对：wxy本文由 LCTT 原创编译，Linux中国 荣誉推出
这是关于并发网络服务器编程的第一篇教程。我计划测试几个主流的、可以同时处理多个客户端请求的服务器并发模型，基于可扩展性和易实现性对这些模型进行评判。所有的服务器都会监听套接字连接，并且实现一些简单的协议用于与客户端进行通讯。该系列的所有文章：第一节 - 简介第二节 - 线程第三节 - 事件驱动协议该系列教程所用的协议都非常简单，但足以展示并发服务器设计的许多有趣层面。而且这个协议是 有状态的 —— 服务器根据客户端发送的数据改变内部状态，然后根据内部状态产生相应的行为。并非所有的协议都是有状态的 —— 实际上，基于 HTTP 的许多协议是无状态的，但是有状态的协议也是很常见，值得认真讨论。在服务器端看来，这个协议的视图是这样的：总之：服务器等待新客户端的连接；当一个客户端连接的时候，服务器会向该客户端发送一个 * 字符，进入“等待消息”的状态。在该状态下，服务器会忽略客户端发送的所有字符，除非它看到了一个 ^ 字符，这表示一个新消息的开始。这个时候服务器就会转变为“正在通信”的状态，这时它会向客户端回送数据，把收到的所有字符的每个字节加 1 回送给客户端 注1 。当客户端发送了 $ 字符，服务器就会退回到等待新消息的状态。^ 和 $ 字符仅仅用于分隔消息 —— 它们不会被服务器回送。每个状态之后都有个隐藏的箭头指向 “等待客户端” 状态，用于客户端断开连接。因此，客户端要表示“我已经结束”的方法很简单，关掉它那一端的连接就好。显然，这个协议是真实协议的简化版，真实使用的协议一般包含复杂的报文头、转义字符序列（例如让消息体中可以出现 $ 符号），额外的状态变化。但是我们这个协议足以完成期望。另一点：这个系列是介绍性的，并假设客户端都工作的很好（虽然可能运行很慢）；因此没有设置超时，也没有设置特殊的规则来确保服务器不会因为客户端的恶意行为（或是故障）而出现阻塞，导致不能正常结束。顺序服务器这个系列中我们的第一个服务端程序是一个简单的“顺序”服务器，用 C 进行编写，除了标准的 POSIX 中用于套接字的内容以外没有使用其它库。服务器程序是顺序，因为它一次只能处理一个客户端的请求；当有客户端连接时，像之前所说的那样，服务器会进入到状态机中，并且不再监听套接字接受新的客户端连接，直到当前的客户端结束连接。显然这不是并发的，而且即便在很少的负载下也不能服务多个客户端，但它对于我们的讨论很有用，因为我们需要的是一个易于理解的基础。这个服务器的完整代码在这里；接下来，我会着重于一些重点的部分。main 函数里面的外层循环用于监听套接字，以便接受新客户端的连接。一旦有客户端进行连接，就会调用 serve_connection，这个函数中的代码会一直运行，直到客户端断开连接。顺序服务器在循环里调用 accept 用来监听套接字，并接受新连接：while (1) { struct sockaddr_in peer_addr; socklen_t peer_addr_len = sizeof(peer_addr); int newsockfd = accept(sockfd, (struct sockaddr*)&peer_addr, &peer_addr_len); if (newsockfd < 0) { perror_die("ERROR on accept"); } report_peer_connected(&peer_addr, peer_addr_len); serve_connection(newsockfd); printf("peer done\n"); } accept 函数每次都会返回一个新的已连接的套接字，然后服务器调用 serve_connection；注意这是一个 阻塞式 的调用 —— 在 serve_connection 返回前，accept 函数都不会再被调用了；服务器会被阻塞，直到客户端结束连接才能接受新的连接。换句话说，客户端按 顺序 得到响应。这是 serve_connection 函数：typedef enum { WAIT_FOR_MSG, IN_MSG } ProcessingState; void serve_connection(int sockfd) { if (send(sockfd, "*", 1, 0) < 1) { perror_die("send"); } ProcessingState state = WAIT_FOR_MSG; while (1) { uint8_t buf[1024]; int len = recv(sockfd, buf, sizeof buf, 0); if (len < 0) { perror_die("recv"); } else if (len == 0) { break; } for (int i = 0; i < len; ++i) { switch (state) { case WAIT_FOR_MSG: if (buf[i] == '^') { state = IN_MSG; } break; case IN_MSG: if (buf[i] == '$') { state = WAIT_FOR_MSG; } else { buf[i] += 1; if (send(sockfd, &buf[i], 1, 0) < 1) { perror("send error"); close(sockfd); return; } } break; } } } close(sockfd); } 它完全是按照状态机协议进行编写的。每次循环的时候，服务器尝试接收客户端的数据。收到 0 字节意味着客户端断开连接，然后循环就会退出。否则，会逐字节检查接收缓存，每一个字节都可能会触发一个状态。recv 函数返回接收到的字节数与客户端发送消息的数量完全无关（^...$ 闭合序列的字节）。因此，在保持状态的循环中遍历整个缓冲区很重要。而且，每一个接收到的缓冲中可能包含多条信息，但也有可能开始了一个新消息，却没有显式的结束字符；而这个结束字符可能在下一个缓冲中才能收到，这就是处理状态在循环迭代中进行维护的原因。例如，试想主循环中的 recv 函数在某次连接中返回了三个非空的缓冲：^abc$de^abte$fxyz^12325$^ab$abab服务端返回的是哪些数据？追踪代码对于理解状态转变很有用。（答案见 注 2 ）多个并发客户端如果多个客户端在同一时刻向顺序服务器发起连接会发生什么事情？服务器端的代码（以及它的名字 “顺序服务器”）已经说的很清楚了，一次只能处理 一个 客户端的请求。只要服务器在 serve_connection 函数中忙于处理客户端的请求，就不会接受别的客户端的连接。只有当前的客户端断开了连接，serve_connection 才会返回，然后最外层的循环才能继续执行接受其他客户端的连接。为了演示这个行为，该系列教程的示例代码 包含了一个 Python 脚本，用于模拟几个想要同时连接服务器的客户端。每一个客户端发送类似之前那样的三个数据缓冲 注3 ，不过每次发送数据之间会有一定延迟。客户端脚本在不同的线程中并发地模拟客户端行为。这是我们的序列化服务器与客户端交互的信息记录：$ python3.6 simple-client.py -n 3 localhost 9090 INFO:2017-09-16 14:14:17,763:conn1 connected... INFO:2017-09-16 14:14:17,763:conn1 sending b'^abc$de^abte$f' INFO:2017-09-16 14:14:17,763:conn1 received b'b' INFO:2017-09-16 14:14:17,802:conn1 received b'cdbcuf' INFO:2017-09-16 14:14:18,764:conn1 sending b'xyz^123' INFO:2017-09-16 14:14:18,764:conn1 received b'234' INFO:2017-09-16 14:14:19,764:conn1 sending b'25$^ab0000$abab' INFO:2017-09-16 14:14:19,765:conn1 received b'36bc1111' INFO:2017-09-16 14:14:19,965:conn1 disconnecting INFO:2017-09-16 14:14:19,966:conn2 connected... INFO:2017-09-16 14:14:19,967:conn2 sending b'^abc$de^abte$f' INFO:2017-09-16 14:14:19,967:conn2 received b'b' INFO:2017-09-16 14:14:20,006:conn2 received b'cdbcuf' INFO:2017-09-16 14:14:20,968:conn2 sending b'xyz^123' INFO:2017-09-16 14:14:20,969:conn2 received b'234' INFO:2017-09-16 14:14:21,970:conn2 sending b'25$^ab0000$abab' INFO:2017-09-16 14:14:21,970:conn2 received b'36bc1111' INFO:2017-09-16 14:14:22,171:conn2 disconnecting INFO:2017-09-16 14:14:22,171:conn0 connected... INFO:2017-09-16 14:14:22,172:conn0 sending b'^abc$de^abte$f' INFO:2017-09-16 14:14:22,172:conn0 received b'b' INFO:2017-09-16 14:14:22,210:conn0 received b'cdbcuf' INFO:2017-09-16 14:14:23,173:conn0 sending b'xyz^123' INFO:2017-09-16 14:14:23,174:conn0 received b'234' INFO:2017-09-16 14:14:24,175:conn0 sending b'25$^ab0000$abab' INFO:2017-09-16 14:14:24,176:conn0 received b'36bc1111' INFO:2017-09-16 14:14:24,376:conn0 disconnecting 这里要注意连接名：conn1 是第一个连接到服务器的，先跟服务器交互了一段时间。接下来的连接 conn2 —— 在第一个断开连接后，连接到了服务器，然后第三个连接也是一样。就像日志显示的那样，每一个连接让服务器变得繁忙，持续了大约 2.2 秒的时间（这实际上是人为地在客户端代码中加入的延迟），在这段时间里别的客户端都不能连接。显然，这不是一个可扩展的策略。这个例子中，客户端中加入了延迟，让服务器不能处理别的交互动作。一个智能服务器应该能处理一堆客户端的请求，而这个原始的服务器在结束连接之前一直繁忙（我们将会在之后的章节中看到如何实现智能的服务器）。尽管服务端有延迟，但这不会过度占用 CPU；例如，从数据库中查找信息（时间基本上是花在连接到数据库服务器上，或者是花在硬盘中的本地数据库）。总结及期望这个示例服务器达成了两个预期目标：首先是介绍了问题范畴和贯彻该系列文章的套接字编程基础。对于并发服务器编程的抛砖引玉 —— 就像之前的部分所说，顺序服务器还不能在非常轻微的负载下进行扩展，而且没有高效的利用资源。在看下一篇文章前，确保你已经理解了这里所讲的服务器/客户端协议，还有顺序服务器的代码。我之前介绍过了这个简单的协议；例如 串行通信分帧 和 用协程来替代状态机。要学习套接字网络编程的基础，Beej 的教程 用来入门很不错，但是要深入理解我推荐你还是看本书。如果有什么不清楚的，请在评论区下进行评论或者向我发送邮件。深入理解并发服务器！注1：状态转变中的 In/Out 记号是指 Mealy machine。注2：回应的是 bcdbcuf23436bc。注3：这里在结尾处有一点小区别，加了字符串 0000 —— 服务器回应这个序列，告诉客户端让其断开连接；这是一个简单的握手协议，确保客户端有足够的时间接收到服务器发送的所有回复。via: https://eli.thegreenplace.net/2017/concurrent-servers-part-1-introduction/作者：Eli Bendersky 译者：GitFuture 校对：wxy本文由 LCTT 原创编译，Linux中国 荣誉推出
这是调试器的工作原理系列文章的第三篇。阅读这篇文章之前应当先阅读第一篇与第二篇。这篇文章的主要内容本文将解释调试器是如何在机器码中查找它将 C 语言源代码转换成机器语言代码时所需要的 C 语言函数、变量、与数据。调试信息现代编译器能够将有着各种缩进或嵌套的程序流程、各种数据类型的变量的高级语言代码转换为一大堆称之为机器码的 0/1 数据，这么做的唯一目的是尽可能快的在目标 CPU 上运行程序。通常来说一行 C 语言代码能够转换为若干条机器码。变量被分散在机器码中的各个部分，有的在堆栈中，有的在寄存器中，或者直接被优化掉了。数据结构与对象在机器码中甚至不“存在”，它们只是用于将数据按一定的结构编码存储进缓存。那么调试器怎么知道，当你需要在某个函数入口处暂停时，程序要在哪停下来呢？它怎么知道当你查看某个变量值时，它怎么找到这个值？答案是，调试信息。编译器在生成机器码时同时会生成相应的调试信息。调试信息代表了可执行程序与源代码之间的关系，并以一种提前定义好的格式，同机器码存放在一起。过去的数年里，人们针对不同的平台与可执行文件发明了很多种用于存储这些信息的格式。不过我们这篇文章不会讲这些格式的历史，而是将阐述这些调试信息是如何工作的，所以我们将专注于一些事情，比如 DWARF。DWARF 如今十分广泛的用作 Linux 和类 Unix 平台上的可执行文件的调试格式。ELF 中的 DWARF根据它的维基百科 所描述，虽然 DWARF 是同 ELF 一同设计的（DWARF 是由 DWARF 标准委员会推出的开放标准。上文中展示的图标就来自这个网站。），但 DWARF 在理论上来说也可以嵌入到其他的可执行文件格式中。DWARF 是一种复杂的格式，它吸收了过去许多年各种不同的架构与操作系统的格式的经验。正是因为它解决了一个在任何平台与 ABI （应用二进制接口）上为任意高级语言产生调试信息这样棘手的难题，它也必须很复杂。想要透彻的讲解 DWARF 仅仅是通过这单薄的一篇文章是远远不够的，说实话我也并没有充分地了解 DWARF 到每一个微小的细节，所以我也不能十分透彻的讲解 （如果你感兴趣的话，文末有一些能够帮助你的资源。建议从 DWARF 教程开始上手）。这篇文章中我将以浅显易懂的方式展示 DWARF，以说明调试信息是如何实际工作的。ELF 文件中的调试部分首先让我们看看 DWARF 处在 ELF 文件中的什么位置。ELF 定义了每一个生成的目标文件中的每一节。 节头表 section header table 声明并定义了每一节及其名字。不同的工具以不同的方式处理不同的节，例如连接器会寻找连接器需要的部分，调试器会查找调试器需要的部分。我们本文的实验会使用从这个 C 语言源文件构建的可执行文件，编译成 tracedprog2：#include <stdio.h> void do_stuff(int my_arg)、 { int my_local = my_arg + 2; int i; for (i = 0; i < my_local; ++i) printf("i = %d\n", i); } int main() { do_stuff(2); return 0; } 使用 objdump -h 命令检查 ELF 可执行文件中的 节头 section header ，我们会看到几个以 .debug_ 开头的节，这些就是 DWARF 的调试部分。26 .debug_aranges 00000020 00000000 00000000 00001037 CONTENTS, READONLY, DEBUGGING 27 .debug_pubnames 00000028 00000000 00000000 00001057 CONTENTS, READONLY, DEBUGGING 28 .debug_info 000000cc 00000000 00000000 0000107f CONTENTS, READONLY, DEBUGGING 29 .debug_abbrev 0000008a 00000000 00000000 0000114b CONTENTS, READONLY, DEBUGGING 30 .debug_line 0000006b 00000000 00000000 000011d5 CONTENTS, READONLY, DEBUGGING 31 .debug_frame 00000044 00000000 00000000 00001240 CONTENTS, READONLY, DEBUGGING 32 .debug_str 000000ae 00000000 00000000 00001284 CONTENTS, READONLY, DEBUGGING 33 .debug_loc 00000058 00000000 00000000 00001332 CONTENTS, READONLY, DEBUGGING 每个节的第一个数字代表了该节的大小，最后一个数字代表了这个节开始位置距离 ELF 的偏移量。调试器利用这些信息从可执行文件中读取节。现在让我们看看一些在 DWARF 中查找有用的调试信息的实际例子。查找函数调试器的最基础的任务之一，就是当我们在某个函数处设置断点时，调试器需要能够在入口处暂停。为此，必须为高级代码中的函数名称与函数在机器码中指令开始的地址这两者之间建立起某种映射关系。为了获取这种映射关系，我们可以查找 DWARF 中的 .debug_info 节。在我们深入之前，需要一点基础知识。DWARF 中每一个描述类型被称之为调试信息入口（DIE）。每个 DIE 都有关于它的类型、属性之类的标签。DIE 之间通过兄弟节点或子节点相互连接，属性的值也可以指向其它的 DIE。运行以下命令：objdump --dwarf=info tracedprog2 输出文件相当的长，为了方便举例我们只关注这些行（从这里开始，无用的冗长信息我会以 （...）代替，方便排版）：<1><71>: Abbrev Number: 5 (DW_TAG_subprogram) <72> DW_AT_external : 1 <73> DW_AT_name : (...): do_stuff <77> DW_AT_decl_file : 1 <78> DW_AT_decl_line : 4 <79> DW_AT_prototyped : 1 <7a> DW_AT_low_pc : 0x8048604 <7e> DW_AT_high_pc : 0x804863e <82> DW_AT_frame_base : 0x0 (location list) <86> DW_AT_sibling : <0xb3> <1><b3>: Abbrev Number: 9 (DW_TAG_subprogram) <b4> DW_AT_external : 1 <b5> DW_AT_name : (...): main <b9> DW_AT_decl_file : 1 <ba> DW_AT_decl_line : 14 <bb> DW_AT_type : <0x4b> <bf> DW_AT_low_pc : 0x804863e <c3> DW_AT_high_pc : 0x804865a <c7> DW_AT_frame_base : 0x2c (location list) 上面的代码中有两个带有 DW_TAG_subprogram 标签的入口，在 DWARF 中这是对函数的指代。注意，这是两个节的入口，其中一个是 do_stuff 函数的入口，另一个是主（main）函数的入口。这些信息中有很多值得关注的属性，但其中最值得注意的是 DW_AT_low_pc。它代表了函数开始处程序指针的值（在 x86 平台上是 EIP）。此处 0x8048604 代表了 do_stuff 函数开始处的程序指针。下面我们将利用 objdump -d 命令对可执行文件进行反汇编。来看看这块地址中都有什么：08048604 <do_stuff>: 8048604: 55 push ebp 8048605: 89 e5 mov ebp,esp 8048607: 83 ec 28 sub esp,0x28 804860a: 8b 45 08 mov eax,DWORD PTR [ebp+0x8] 804860d: 83 c0 02 add eax,0x2 8048610: 89 45 f4 mov DWORD PTR [ebp-0xc],eax 8048613: c7 45 (...) mov DWORD PTR [ebp-0x10],0x0 804861a: eb 18 jmp 8048634 <do_stuff+0x30> 804861c: b8 20 (...) mov eax,0x8048720 8048621: 8b 55 f0 mov edx,DWORD PTR [ebp-0x10] 8048624: 89 54 24 04 mov DWORD PTR [esp+0x4],edx 8048628: 89 04 24 mov DWORD PTR [esp],eax 804862b: e8 04 (...) call 8048534 <printf@plt> 8048630: 83 45 f0 01 add DWORD PTR [ebp-0x10],0x1 8048634: 8b 45 f0 mov eax,DWORD PTR [ebp-0x10] 8048637: 3b 45 f4 cmp eax,DWORD PTR [ebp-0xc] 804863a: 7c e0 jl 804861c <do_stuff+0x18> 804863c: c9 leave 804863d: c3 ret 显然，0x8048604 是 do_stuff 的开始地址，这样一来，调试器就可以建立函数与其在可执行文件中的位置间的映射关系。查找变量假设我们当前在 do_staff 函数中某个位置上设置断点停了下来。我们想通过调试器取得 my_local 这个变量的值。调试器怎么知道在哪里去找这个值呢？很显然这要比查找函数更为困难。变量可能存储在全局存储区、堆栈、甚至是寄存器中。此外，同名变量在不同的作用域中可能有着不同的值。调试信息必须能够反映所有的这些变化，当然，DWARF 就能做到。我不会逐一去将每一种可能的状况，但我会以调试器在 do_stuff 函数中查找 my_local 变量的过程来举个例子。下面我们再看一遍 .debug_info 中 do_stuff 的每一个入口，这次连它的子入口也要一起看。<1><71>: Abbrev Number: 5 (DW_TAG_subprogram) <72> DW_AT_external : 1 <73> DW_AT_name : (...): do_stuff <77> DW_AT_decl_file : 1 <78> DW_AT_decl_line : 4 <79> DW_AT_prototyped : 1 <7a> DW_AT_low_pc : 0x8048604 <7e> DW_AT_high_pc : 0x804863e <82> DW_AT_frame_base : 0x0 (location list) <86> DW_AT_sibling : <0xb3> <2><8a>: Abbrev Number: 6 (DW_TAG_formal_parameter) <8b> DW_AT_name : (...): my_arg <8f> DW_AT_decl_file : 1 <90> DW_AT_decl_line : 4 <91> DW_AT_type : <0x4b> <95> DW_AT_location : (...) (DW_OP_fbreg: 0) <2><98>: Abbrev Number: 7 (DW_TAG_variable) <99> DW_AT_name : (...): my_local <9d> DW_AT_decl_file : 1 <9e> DW_AT_decl_line : 6 <9f> DW_AT_type : <0x4b> <a3> DW_AT_location : (...) (DW_OP_fbreg: -20) <2><a6>: Abbrev Number: 8 (DW_TAG_variable) <a7> DW_AT_name : i <a9> DW_AT_decl_file : 1 <aa> DW_AT_decl_line : 7 <ab> DW_AT_type : <0x4b> <af> DW_AT_location : (...) (DW_OP_fbreg: -24) 看到每个入口处第一对尖括号中的数字了吗？这些是嵌套的等级，在上面的例子中，以 <2> 开头的入口是以 <1> 开头的子入口。因此我们得知 my_local 变量（以 DW_TAG_variable 标签标记）是 do_stuff 函数的局部变量。除此之外，调试器也需要知道变量的数据类型，这样才能正确的使用与显示变量。上面的例子中 my_local 的变量类型指向另一个 DIE <0x4b>。如果使用 objdump 命令查看这个 DIE 的话，我们会发现它是一个有符号 4 字节整型数据。而为了在实际运行的程序内存中查找变量的值，调试器需要使用到 DW_AT_location 属性。对于 my_local 而言，是 DW_OP_fbreg: -20。这个代码段的意思是说 my_local 存储在距离它所在函数起始地址偏移量为 -20 的地方。do_stuff 函数的 DW_AT_frame_base 属性值为 0x0 (location list)。这意味着这个属性的值需要在 location list 中查找。下面我们来一起看看。$ objdump --dwarf=loc tracedprog2 tracedprog2: file format elf32-i386 Contents of the .debug_loc section: Offset Begin End Expression 00000000 08048604 08048605 (DW_OP_breg4: 4 ) 00000000 08048605 08048607 (DW_OP_breg4: 8 ) 00000000 08048607 0804863e (DW_OP_breg5: 8 ) 00000000 <End of list> 0000002c 0804863e 0804863f (DW_OP_breg4: 4 ) 0000002c 0804863f 08048641 (DW_OP_breg4: 8 ) 0000002c 08048641 0804865a (DW_OP_breg5: 8 ) 0000002c <End of list> 我们需要关注的是第一列（do_stuff 函数的 DW_AT_frame_base 属性包含 location list 中 0x0 的偏移量。而 main 函数的相同属性包含 0x2c 的偏移量，这个偏移量是第二套地址列表的偏移量）。对于调试器可能定位到的每一个地址，它都会指定当前栈帧到变量间的偏移量，而这个偏移就是通过寄存器来计算的。对于 x86 平台而言，bpreg4 指向 esp，而 bpreg5 指向 ebp。让我们再看看 do_stuff 函数的头几条指令。08048604 <do_stuff>: 8048604: 55 push ebp 8048605: 89 e5 mov ebp,esp 8048607: 83 ec 28 sub esp,0x28 804860a: 8b 45 08 mov eax,DWORD PTR [ebp+0x8] 804860d: 83 c0 02 add eax,0x2 8048610: 89 45 f4 mov DWORD PTR [ebp-0xc],eax 只有当第二条指令执行后，ebp 寄存器才真正存储了有用的值。当然，前两条指令的基址是由上面所列出来的地址信息表计算出来的。一但 ebp 确定了，计算偏移量就十分方便了，因为尽管 esp 在操作堆栈的时候需要移动，但 ebp 作为栈底并不需要移动。究竟我们应该去哪里找 my_local 的值呢？在 0x8048610 这块地址后， my_local 的值经过在 eax 中的计算后被存在了内存中，从这里开始我们才需要关注 my_local 的值。调试器会利用 DW_OP_breg5: 8 这个栈帧来查找。我们回想下，my_local 的 DW_AT_location 属性值为 DW_OP_fbreg: -20。所以应当从基址中 -20 ，同时由于 ebp 寄存器需要 +8，所以最终结果为 ebp - 12。现在再次查看反汇编代码，来看看数据从 eax 中被移动到哪里了。当然，这里 my_local 应当被存储在了 ebp - 12 的地址中。查看行号当我们谈到在调试信息寻找函数的时候，我们利用了些技巧。当调试 C 语言源代码并在某个函数出放置断点的时候，我们并不关注第一条“机器码”指令（函数的调用准备工作已经完成而局部变量还没有初始化）。我们真正关注的是函数的第一行“C 代码”。这就是 DWARF 完全覆盖映射 C 源代码中的行与可执行文件中机器码地址的原因。下面是 .debug_line 节中所包含的内容，我们将其转换为可读的格式展示如下。$ objdump --dwarf=decodedline tracedprog2 tracedprog2: file format elf32-i386 Decoded dump of debug contents of section .debug_line: CU: /home/eliben/eli/eliben-code/debugger/tracedprog2.c: File name Line number Starting address tracedprog2.c 5 0x8048604 tracedprog2.c 6 0x804860a tracedprog2.c 9 0x8048613 tracedprog2.c 10 0x804861c tracedprog2.c 9 0x8048630 tracedprog2.c 11 0x804863c tracedprog2.c 15 0x804863e tracedprog2.c 16 0x8048647 tracedprog2.c 17 0x8048653 tracedprog2.c 18 0x8048658 很容易就可以看出其中 C 源代码与反汇编代码之间的对应关系。第 5 行指向 do_stuff 函数的入口，0x8040604。第 6 行，指向 0x804860a ，正是调试器在调试 do_stuff 函数时需要停下来的地方。这里已经完成了函数调用的准备工作。上面的这些信息形成了行号与地址间的双向映射关系。当在某一行设置断点的时候，调试器会利用这些信息去查找相应的地址来做断点工作（还记得上篇文章中的 int 3 指令吗？）当指令造成段错误时，调试器会利用这些信息来查看源代码中发生问题的行。libdwarf - 用 DWARF 编程尽管使用命令行工具来获得 DWARF 很有用，但这仍然不够易用。作为程序员，我们希望知道当我们需要这些调试信息时应当怎么编程来获取这些信息。自然我们想到的第一种方法就是阅读 DWARF 规范并按规范操作阅读使用。有句话说的好，分析 HTML 应当使用库函数，永远不要手工分析。对于 DWARF 来说正是如此。DWARF 比 HTML 要复杂得多。上面所展示出来的只是冰山一角。更糟糕的是，在实际的目标文件中，大部分信息是以非常紧凑的压缩格式存储的，分析起来更加复杂（信息中的某些部分，例如位置信息与行号信息，在某些虚拟机下是以指令的方式编码的）。所以我们要使用库来处理 DWARF。下面是两种我熟悉的主要的库（还有些不完整的库这里没有写）BFD （libbfd），包含了 objdump （对，就是这篇文章中我们一直在用的这货），ld（GNU 连接器）与 as（GNU 编译器）。BFD 主要用于 GNU binutils。libdwarf ，同它的哥哥 libelf 一同用于 Solaris 与 FreeBSD 中的调试信息分析。相比较而言我更倾向于使用 libdwarf，因为我对它了解的更多，并且 libdwarf 的开源协议更开放（LGPL 对比 GPL）。因为 libdwarf 本身相当复杂，操作起来需要相当多的代码，所以我在这不会展示所有代码。你可以在 这里 下载代码并运行试试。运行这些代码需要提前安装 libelfand 与 libdwarf ，同时在使用连接器的时候要使用参数 -lelf 与 -ldwarf。这个示例程序可以接受可执行文件并打印其中的函数名称与函数入口地址。下面是我们整篇文章中使用的 C 程序经过示例程序处理后的输出。$ dwarf_get_func_addr tracedprog2 DW_TAG_subprogram: 'do_stuff' low pc : 0x08048604 high pc : 0x0804863e DW_TAG_subprogram: 'main' low pc : 0x0804863e high pc : 0x0804865a libdwarf 的文档很棒，如果你花些功夫，利用 libdwarf 获得这篇文章中所涉及到的 DWARF 信息应该并不困难。结论与计划原理上讲，调试信息是个很简单的概念。尽管实现细节可能比较复杂，但经过了上面的学习我想你应该了解了调试器是如何从可执行文件中获取它需要的源代码信息的了。对于程序员而言，程序只是代码段与数据结构；对可执行文件而言，程序只是一系列存储在内存或寄存器中的指令或数据。但利用调试信息，调试器就可以将这两者连接起来，从而完成调试工作。此文与这系列的前两篇，一同介绍了调试器的内部工作过程。利用这里所讲到的知识，再敲些代码，应该可以完成一个 Linux 中最简单、基础但也有一定功能的调试器。下一步我并不确定要做什么，这个系列文章可能就此结束，也有可能我要讲些堆栈调用的事情，又或者讲 Windows 下的调试。你们有什么好的点子或者相关材料，可以直接评论或者发邮件给我。参考objdump 参考手册ELF 与 DWARF 的维基百科Dwarf Debugging Standard 主页，这里有很棒的 DWARF 教程与 DWARF 标准，作者是 Michael Eager。第二版基于 GCC 也许更能吸引你。libdwarf 主页，这里可以下载到 libwarf 的完整库与参考手册BFD 文档via: http://eli.thegreenplace.net/2011/02/07/how-debuggers-work-part-3-debugging-information作者：Eli Bendersky 译者：YYforymj 校对：wxy本文由 LCTT 原创编译，Linux中国 荣誉推出
这是调试器工作原理系列文章的第二部分，阅读本文前，请确保你已经读过第一部分。关于本文我将会演示如何在调试器中实现断点。断点是调试的两大利器之一，另一个是可以在被调试进程的内存中检查变量值。我们在系列的第一部分已经了解过值检查，但是断点对我们来说依然神秘。不过本文过后，它们就不再如此了。软件中断为了在 x86 架构机器上实现断点，软件中断（也被称作“陷阱”）被会派上用场。在我们深入细节之前，我想先大致解释一下中断和陷阱的概念。CPU 有一条单独的执行流，一条指令接一条的执行（在更高的层面看是这样的，但是在底层的细节上来说，现在的许多 CPU 都会并行执行多个指令，这其中的一些指令就不是按照原本的顺序执行的）。为了能够处理异步的事件，如 IO 和 硬件定时器，CPU 使用了中断。硬件中断通常是一个特定的电子信号，并附加了一个特别的”响应电路”。该电路通知中断激活，并让 CPU 停止当前执行，保存状态，然后跳转到一个预定义的地址，也就是中断处理程序的位置。当处理程序完成其工作后，CPU 又从之前停止的地方重新恢复运行。软件中断在规则上与硬件相似，但实际操作中有些不同。CPU 支持一些特殊的指令，来允许软件模拟出一个中断。当这样的一个指令被执行时，CPU 像对待一个硬件中断那样 —— 停止正常的执行流，保存状态，然后跳转到一个处理程序。这种“中断”使得许多现代 OS 的惊叹设计得以高效地实现（如任务调度，虚拟内存，内存保护，调试）。许多编程错误（如被 0 除）也被 CPU 当做中断对待，常常也叫做“异常”， 这时候硬件和软件中断之间的界限就模糊了，很难说这种异常到底是硬件中断还是软件中断。但我已经偏离今天主题太远了，所以现在让我们回到断点上来。int 3 理论前面说了很多，现在简单来说断点就是一个部署在 CPU 上的特殊中断，叫 int 3。int 是一个 “中断指令”的 x86 术语，该指令是对一个预定义中断处理的调用。x86 支持 8 位的 int 指令操作数，这决定了中断的数量，所以理论上可以支持 256 个中断。前 32 个中断为 CPU 自己保留，而 int 3 就是本文关注的 —— 它被叫做 “调试器专用中断”。避免更深的解释，我将引用“圣经”里一段话（这里说的“圣经”，当然指的是英特尔的体系结构软件开发者手册, 卷 2A）。INT 3 指令生成一个以字节操作码（CC），用于调用该调试异常处理程序。（这个一字节格式是非常有用的，因为它可以用于使用断点来替换任意指令的第一个字节 ，包括哪些一字节指令，而不会覆写其它代码）上述引用非常重要，但是目前去解释它还是为时过早。本文后面我们会回过头再看。int 3 实践没错，知道事物背后的理论非常不错，不过，这些理论到底意思是啥？我们怎样使用 int 3 部署断点？或者怎么翻译成通用的编程术语 —— 请给我看代码！实际上，实现非常简单。一旦你的程序执行了 int 3 指令， OS 就会停止程序（ OS 是怎么做到像这样停止进程的？ OS 注册其 int 3 的控制程序到 CPU 即可，就这么简单）。在 Linux（这也是本文比较关心的地方） 上， OS 会发送给进程一个信号 —— SIGTRAP。就是这样，真的。现在回想一下本系列的第一部分, 追踪进程（调试程序） 会得到其子进程（或它所连接的被调试进程）所得到的所有信号的通知，接下来你就知道了。就这样, 没有更多的电脑架构基础术语了。该是例子和代码的时候了。手动设置断点现在我要演示在程序里设置断点的代码。我要使用的程序如下：section .text ; The _start symbol must be declared for the linker (ld) global _start _start: ; Prepare arguments for the sys_write system call: ; - eax: system call number (sys_write) ; - ebx: file descriptor (stdout) ; - ecx: pointer to string ; - edx: string length mov edx, len1 mov ecx, msg1 mov ebx, 1 mov eax, 4 ; Execute the sys_write system call int 0x80 ; Now print the other message mov edx, len2 mov ecx, msg2 mov ebx, 1 mov eax, 4 int 0x80 ; Execute sys_exit mov eax, 1 int 0x80 section .data msg1 db 'Hello,', 0xa len1 equ $ - msg1 msg2 db 'world!', 0xa len2 equ $ - msg2 我现在在使用汇编语言，是为了当我们面对 C 代码的时候，能清楚一些编译细节。上面代码做的事情非常简单，就是在一行打印出 “hello,”，然后在下一行打印出 “world!”。这与之前文章中的程序非常类似。现在我想在第一次打印和第二次打印之间设置一个断点。我们看到在第一条 int 0x80 ，其后指令是 mov edx, len2。（等等，再次 int？是的，Linux 使用 int 0x80 来实现用户进程到系统内核的系统调用。用户将系统调用的号码及其参数放到寄存器，并执行 int 0x80。然后 CPU 会跳到相应的中断处理程序，其中， OS 注册了一个过程，该过程查看寄存器并决定要执行的系统调用。）首先，我们需要知道该指令所映射的地址。运行 objdump -d:traced_printer2: file format elf32-i386 Sections: Idx Name Size VMA LMA File off Algn 0 .text 00000033 08048080 08048080 00000080 2**4 CONTENTS, ALLOC, LOAD, READONLY, CODE 1 .data 0000000e 080490b4 080490b4 000000b4 2**2 CONTENTS, ALLOC, LOAD, DATA Disassembly of section .text: 08048080 <.text>: 8048080: ba 07 00 00 00 mov $0x7,%edx 8048085: b9 b4 90 04 08 mov $0x80490b4,%ecx 804808a: bb 01 00 00 00 mov $0x1,%ebx 804808f: b8 04 00 00 00 mov $0x4,%eax 8048094: cd 80 int $0x80 8048096: ba 07 00 00 00 mov $0x7,%edx 804809b: b9 bb 90 04 08 mov $0x80490bb,%ecx 80480a0: bb 01 00 00 00 mov $0x1,%ebx 80480a5: b8 04 00 00 00 mov $0x4,%eax 80480aa: cd 80 int $0x80 80480ac: b8 01 00 00 00 mov $0x1,%eax 80480b1: cd 80 int $0x80 所以，我们要设置断点的地址是 0x8048096。等等，这不是调试器工作的真实姿势，对吧？真正的调试器是在代码行和函数上设置断点，而不是赤裸裸的内存地址？完全正确，但是目前我们仍然还没到那一步，为了更像真正的调试器一样设置断点，我们仍不得不首先理解一些符号和调试信息。所以现在，我们就得面对内存地址。在这点上，我真想又偏离一下主题。所以现在你有两个选择，如果你真的感兴趣想知道为什么那个地址应该是 0x8048096，它代表着什么，那就看下面的部分。否则你只是想了解断点，你可以跳过这部分。题外话 —— 程序地址和入口坦白说，0x8048096 本身没多大意义，仅仅是可执行程序的 text 部分开端偏移的一些字节。如果你看上面导出来的列表，你会看到 text 部分从地址 0x08048080 开始。这告诉 OS 在分配给进程的虚拟地址空间里，将该地址映射到 text 部分开始的地方。在 Linux 上面，这些地址可以是绝对地址（例如，当可执行程序加载到内存中时它不做重定位），因为通过虚拟地址系统，每个进程获得自己的一块内存，并且将整个 32 位地址空间看做自己的（称为 “线性” 地址）。如果我们使用 readelf 命令检查 ELF 文件头部（ELF，可执行和可链接格式，是 Linux 上用于对象文件、共享库和可执行程序的文件格式），我们会看到：$ readelf -h traced_printer2 ELF Header: Magic: 7f 45 4c 46 01 01 01 00 00 00 00 00 00 00 00 00 Class: ELF32 Data: 2's complement, little endian Version: 1 (current) OS/ABI: UNIX - System V ABI Version: 0 Type: EXEC (Executable file) Machine: Intel 80386 Version: 0x1 Entry point address: 0x8048080 Start of program headers: 52 (bytes into file) Start of section headers: 220 (bytes into file) Flags: 0x0 Size of this header: 52 (bytes) Size of program headers: 32 (bytes) Number of program headers: 2 Size of section headers: 40 (bytes) Number of section headers: 4 Section header string table index: 3 注意头部里的 Entry point address，它同样指向 0x8048080。所以我们在系统层面解释该 elf 文件的编码信息，它意思是：映射 text 部分（包含所给的内容）到地址 0x8048080从入口 —— 地址 0x8048080 处开始执行但是，为什么是 0x8048080 呢？事实证明是一些历史原因。一些 Google 的结果把我引向源头，宣传每个进程的地址空间的前 128M 是保留在栈里的。128M 对应为 0x8000000，该地址是可执行程序其他部分可以开始的地方。而 0x8048080，比较特别，是 Linux ld 链接器使用的默认入口地址。该入口可以通过给 ld 传递 -Ttext 参数改变。总结一下，这地址没啥特别的，我们可以随意修改它。只要 ELF 可执行文件被合理的组织，并且头部里的入口地址与真正的程序代码（text 部分）开始的地址匹配，一切都没问题。用 int 3 在调试器中设置断点为了在被追踪进程的某些目标地址设置一个断点，调试器会做如下工作：记住存储在目标地址的数据用 int 指令替换掉目标地址的第一个字节然后，当调试器要求 OS 运行该进程的时候（通过上一篇文章中提过的 PTRACE_CONT），进程就会运行起来直到遇到 int 3，此处进程会停止运行，并且 OS 会发送一个信号给调试器。调试器会收到一个信号表明其子进程（或者说被追踪进程）停止了。调试器可以做以下工作：在目标地址，用原来的正常执行指令替换掉 int 3 指令将被追踪进程的指令指针回退一步。这是因为现在指令指针位于刚刚执行过的 int 3 之后。允许用户以某些方式与进程交互，因为该进程仍然停止在特定的目标地址。这里你的调试器可以让你取得变量值，调用栈等等。当用户想继续运行，调试器会小心地把断点放回目标地址去（因为它在第 1 步时被移走了），除非用户要求取消该断点。让我们来看看，这些步骤是如何翻译成具体代码的。我们会用到第一篇里的调试器 “模板”（fork 一个子进程并追踪它）。无论如何，文末会有一个完整样例源代码的链接/* Obtain and show child's instruction pointer */ ptrace(PTRACE_GETREGS, child_pid, 0, &regs); procmsg("Child started. EIP = 0x%08x\n", regs.eip); /* Look at the word at the address we're interested in */ unsigned addr = 0x8048096; unsigned data = ptrace(PTRACE_PEEKTEXT, child_pid, (void*)addr, 0); procmsg("Original data at 0x%08x: 0x%08x\n", addr, data); 这里调试器从被追踪的进程中取回了指令指针，也检查了在 0x8048096 的字。当开始追踪运行文章开头的汇编代码，将会打印出：[13028] Child started. EIP = 0x08048080 [13028] Original data at 0x08048096: 0x000007ba 目前为止都看起来不错。接下来：/* Write the trap instruction 'int 3' into the address */ unsigned data_with_trap = (data & 0xFFFFFF00) | 0xCC; ptrace(PTRACE_POKETEXT, child_pid, (void*)addr, (void*)data_with_trap); /* See what's there again... */ unsigned readback_data = ptrace(PTRACE_PEEKTEXT, child_pid, (void*)addr, 0); procmsg("After trap, data at 0x%08x: 0x%08x\n", addr, readback_data); 注意到 int 3 是如何被插入到目标地址的。此处打印：[13028] After trap, data at 0x08048096: 0x000007cc 正如预料的那样 —— 0xba 被 0xcc 替换掉了。现在调试器运行子进程并等待它在断点处停止：/* Let the child run to the breakpoint and wait for it to ** reach it */ ptrace(PTRACE_CONT, child_pid, 0, 0); wait(&wait_status); if (WIFSTOPPED(wait_status)) { procmsg("Child got a signal: %s\n", strsignal(WSTOPSIG(wait_status))); } else { perror("wait"); return; } /* See where the child is now */ ptrace(PTRACE_GETREGS, child_pid, 0, &regs); procmsg("Child stopped at EIP = 0x%08x\n", regs.eip); 这里打印出：Hello, [13028] Child got a signal: Trace/breakpoint trap [13028] Child stopped at EIP = 0x08048097 注意到 “Hello,” 在断点前打印出来了 —— 完全如我们计划的那样。同时注意到子进程停止的地方 —— 刚好就是单字节中断指令后面。最后，如早先诠释的那样，为了让子进程继续运行，我们得做一些工作。我们用原来的指令替换掉中断指令，并且让进程从这里继续之前的运行。/* Remove the breakpoint by restoring the previous data ** at the target address, and unwind the EIP back by 1 to ** let the CPU execute the original instruction that was ** there. */ ptrace(PTRACE_POKETEXT, child_pid, (void*)addr, (void*)data); regs.eip -= 1; ptrace(PTRACE_SETREGS, child_pid, 0, &regs); /* The child can continue running now */ ptrace(PTRACE_CONT, child_pid, 0, 0); 这会使子进程继续打印出 “world！”，然后退出。注意，我们在这里没有恢复断点。通过在单步调试模式下，运行原来的指令，然后将中断放回去，并且只在运行 PTRACE\_CONT 时做到恢复断点。文章稍后会展示 debuglib 如何做到这点。更多关于 int 3现在可以回过头去看看 int 3 和因特尔手册里那个神秘的说明，原文如下：这个一字节格式是非常有用的，因为它可以用于使用断点来替换任意指令的第一个字节 ，包括哪些一字节指令，而不会覆写其它代码int 指令在 x86 机器上占两个字节 —— 0xcd 紧跟着中断数（细心的读者可以在上面列出的转储中发现 int 0x80 翻译成了 cd 80）。int 3 被编码为 cd 03，但是为其还保留了一个单字节指令 —— 0xcc。为什么这样呢？因为这可以允许我们插入一个断点，而不需要重写多余的指令。这非常重要，考虑下面的代码： .. some code .. jz foo dec eax foo: call bar .. some code .. 假设你想在 dec eax 这里放置一个断点。这对应一个单字节指令（操作码为 0x48）。由于替换断点的指令长于一个字节，我们不得不强制覆盖掉下个指令（call）的一部分，这就会篡改 call 指令，并很可能导致一些完全不合理的事情发生。这样一来跳转到 foo 分支的 jz foo 指令会导致什么？就会不在 dec eax 这里停止，CPU 径直去执行后面一些无效的指令了。而有了单字节的 int 3 指令，这个问题就解决了。 1 字节是在 x86 上面所能找到的最短指令，这样我们可以保证仅改变我们想中断的指令。封装一些晦涩的细节很多上述章节样例代码的底层细节，都可以很容易封装在方便使用的 API 里。我已经做了很多封装的工作，将它们都放在一个叫做 debuglib 的通用库里 —— 文末可以去下载。这里我仅仅是想展示它的用法示例，但是绕了一圈。下面我们将追踪一个用 C 写的程序。追踪一个 C 程序地址和入口目前为止，为了简单，我把注意力放在了目标汇编代码。现在是时候往上一个层次，去看看我们如何追踪一个 C 程序。事实证明并不是非常难 —— 找到放置断点位置有一点难罢了。考虑下面样例程序：#include <stdio.h> void do_stuff() { printf("Hello, "); } int main() { for (int i = 0; i < 4; ++i) do_stuff(); printf("world!\n"); return 0; } 假设我想在 do_stuff 入口处放置一个断点。我会先使用 objdump 反汇编一下可执行文件，但是打印出的东西太多。尤其看到很多无用，也不感兴趣的 C 程序运行时的初始化代码。所以我们仅看一下 do_stuff 部分：080483e4 <do_stuff>: 80483e4: 55 push %ebp 80483e5: 89 e5 mov %esp,%ebp 80483e7: 83 ec 18 sub $0x18,%esp 80483ea: c7 04 24 f0 84 04 08 movl $0x80484f0,(%esp) 80483f1: e8 22 ff ff ff call 8048318 <puts@plt> 80483f6: c9 leave 80483f7: c3 ret 那么，我们将会把断点放在 0x080483e4，这是 do_stuff 第一条指令执行的地方。而且，该函数是在循环里面调用的，我们想要在断点处一直停止执行直到循环结束。我们将会使用 debuglib 来简化该流程，下面是完整的调试函数：void run_debugger(pid_t child_pid) { procmsg("debugger started\n"); /* Wait for child to stop on its first instruction */ wait(0); procmsg("child now at EIP = 0x%08x\n", get_child_eip(child_pid)); /* Create breakpoint and run to it*/ debug_breakpoint* bp = create_breakpoint(child_pid, (void*)0x080483e4); procmsg("breakpoint created\n"); ptrace(PTRACE_CONT, child_pid, 0, 0); wait(0); /* Loop as long as the child didn't exit */ while (1) { /* The child is stopped at a breakpoint here. Resume its ** execution until it either exits or hits the ** breakpoint again. */ procmsg("child stopped at breakpoint. EIP = 0x%08X\n", get_child_eip(child_pid)); procmsg("resuming\n"); int rc = resume_from_breakpoint(child_pid, bp); if (rc == 0) { procmsg("child exited\n"); break; } else if (rc == 1) { continue; } else { procmsg("unexpected: %d\n", rc); break; } } cleanup_breakpoint(bp); } 为了避免修改 EIP 标志位和目的进程的内存空间的麻烦，我们仅需要调用 create_breakpoint，resume_from_breakpoint 和 cleanup_breakpoint。让我们来看看追踪上面的 C 代码样例会输出什么：$ bp_use_lib traced_c_loop [13363] debugger started [13364] target started. will run 'traced_c_loop' [13363] child now at EIP = 0x00a37850 [13363] breakpoint created [13363] child stopped at breakpoint. EIP = 0x080483E5 [13363] resuming Hello, [13363] child stopped at breakpoint. EIP = 0x080483E5 [13363] resuming Hello, [13363] child stopped at breakpoint. EIP = 0x080483E5 [13363] resuming Hello, [13363] child stopped at breakpoint. EIP = 0x080483E5 [13363] resuming Hello, world! [13363] child exited 如预期一样！样例代码这里是本文用到的完整源代码文件。在归档中你可以找到：debuglib.h 和 debuglib.c - 封装了调试器的一些内部工作的示例库bp\_manual.c - 这篇文章开始部分介绍的“手动”设置断点的方法。一些样板代码使用了 debuglib 库。bpuselib.c - 大部分代码使用了 debuglib 库，用于在第二个代码范例中演示在 C 程序的循环中追踪。引文在准备本文的时候，我搜集了如下的资源和文章：How debugger worksUnderstanding ELF using readelf and objdumpImplementing breakpoints on x86 LinuxNASM manualSO discussion of the ELF entry pointThis Hacker News discussion of the first part of the seriesGDB Internalsvia: http://eli.thegreenplace.net/2011/01/27/how-debuggers-work-part-2-breakpoints作者：Eli Bendersky 译者：wi-cuckoo 校对：wxy本文由 LCTT 原创编译，Linux中国 荣誉推出
这是调试器工作原理系列文章的第一篇，我不确定这个系列会有多少篇文章，会涉及多少话题，但我仍会从这篇基础开始。这一篇会讲什么我将为大家展示 Linux 中调试器的主要构成模块 - ptrace 系统调用。这篇文章所有代码都是基于 32 位 Ubuntu 操作系统。值得注意的是，尽管这些代码是平台相关的，将它们移植到其它平台应该并不困难。缘由为了理解我们要做什么，让我们先考虑下调试器为了完成调试都需要什么资源。调试器可以开始一个进程并调试这个进程，又或者将自己同某个已经存在的进程关联起来。调试器能够单步执行代码，设定断点并且将程序执行到断点，检查变量的值并追踪堆栈。许多调试器有着更高级的特性，例如在调试器的地址空间内执行表达式或者调用函数，甚至可以在进程执行过程中改变代码并观察效果。尽管现代的调试器都十分的复杂（我没有检查，但我确信 gdb 的代码行数至少有六位数），但它们的工作的原理却是十分的简单。调试器的基础是操作系统与编译器 / 链接器提供的一些基础服务，其余的部分只是简单的编程而已。Linux 的调试 - ptraceLinux 调试器中的瑞士军刀便是 ptrace 系统调用（使用 man 2 ptrace 命令可以了解更多）。这是一种复杂却强大的工具，可以允许一个进程控制另外一个进程并从 内部替换 Peek and poke 被控制进程的内核镜像的值（Peek and poke 在系统编程中是很知名的叫法，指的是直接读写内存内容）。接下来会深入分析。执行进程的代码我将编写一个示例，实现一个在“跟踪”模式下运行的进程。在这个模式下，我们将单步执行进程的代码，就像机器码（汇编代码）被 CPU 执行时一样。我将分段展示、讲解示例代码，在文章的末尾也有完整 c 文件的下载链接，你可以编译、执行或者随心所欲的更改。更进一步的计划是实现一段代码，这段代码可以创建可执行用户自定义命令的子进程，同时父进程可以跟踪子进程。首先是主函数：int main(int argc, char** argv) { pid_t child_pid; if (argc < 2) { fprintf(stderr, "Expected a program name as argument\n"); return -1; } child_pid = fork(); if (child_pid == 0) run_target(argv[1]); else if (child_pid > 0) run_debugger(child_pid); else { perror("fork"); return -1; } return 0; } 看起来相当的简单：我们用 fork 创建了一个新的子进程（这篇文章假定读者有一定的 Unix/Linux 编程经验。我假定你知道或至少了解 fork、exec 族函数与 Unix 信号）。if 语句的分支执行子进程（这里称之为 “target”），else if 的分支执行父进程（这里称之为 “debugger”）。下面是 target 进程的代码：void run_target(const char* programname) { procmsg("target started. will run '%s'\n", programname); /* Allow tracing of this process */ if (ptrace(PTRACE_TRACEME, 0, 0, 0) < 0) { perror("ptrace"); return; } /* Replace this process's image with the given program */ execl(programname, programname, 0); } 这段代码中最值得注意的是 ptrace 调用。在 sys/ptrace.h 中，ptrace 是如下定义的：long ptrace(enum __ptrace_request request, pid_t pid, void *addr, void *data); 第一个参数是 _request_，这是许多预定义的 PTRACE_* 常量中的一个。第二个参数为请求分配进程 ID。第三个与第四个参数是地址与数据指针，用于操作内存。上面代码段中的 ptrace 调用发起了 PTRACE_TRACEME 请求，这意味着该子进程请求系统内核让其父进程跟踪自己。帮助页面上对于 request 的描述很清楚：意味着该进程被其父进程跟踪。任何传递给该进程的信号（除了 SIGKILL）都将通过 wait() 方法阻塞该进程并通知其父进程。此外，该进程的之后所有调用 exec() 动作都将导致 SIGTRAP 信号发送到此进程上，使得父进程在新的程序执行前得到取得控制权的机会。如果一个进程并不需要它的的父进程跟踪它，那么这个进程不应该发送这个请求。（pid、addr 与 data 暂且不提）我高亮了这个例子中我们需要注意的部分。在 ptrace 调用后，run_target 接下来要做的就是通过 execl 传参并调用。如同高亮部分所说明，这将导致系统内核在 execl 创建进程前暂时停止，并向父进程发送信号。是时候看看父进程做什么了。void run_debugger(pid_t child_pid) { int wait_status; unsigned icounter = 0; procmsg("debugger started\n"); /* Wait for child to stop on its first instruction */ wait(&wait_status); while (WIFSTOPPED(wait_status)) { icounter++; /* Make the child execute another instruction */ if (ptrace(PTRACE_SINGLESTEP, child_pid, 0, 0) < 0) { perror("ptrace"); return; } /* Wait for child to stop on its next instruction */ wait(&wait_status); } procmsg("the child executed %u instructions\n", icounter); } 如前文所述，一旦子进程调用了 exec，子进程会停止并被发送 SIGTRAP 信号。父进程会等待该过程的发生并在第一个 wait() 处等待。一旦上述事件发生了，wait() 便会返回，由于子进程停止了父进程便会收到信号（如果子进程由于信号的发送停止了，WIFSTOPPED 就会返回 true）。父进程接下来的动作就是整篇文章最需要关注的部分了。父进程会将 PTRACE_SINGLESTEP 与子进程 ID 作为参数调用 ptrace 方法。这就会告诉操作系统，“请恢复子进程，但在它执行下一条指令前阻塞”。周而复始地，父进程等待子进程阻塞，循环继续。当 wait() 中传出的信号不再是子进程的停止信号时，循环终止。在跟踪器（父进程）运行期间，这将会是被跟踪进程（子进程）传递给跟踪器的终止信号（如果子进程终止 WIFEXITED 将返回 true）。icounter 存储了子进程执行指令的次数。这么看来我们小小的例子也完成了些有用的事情 - 在命令行中指定程序，它将执行该程序并记录它从开始到结束所需要的 cpu 指令数量。接下来就让我们这么做吧。测试我编译了下面这个简单的程序并利用跟踪器运行它：#include <stdio.h> int main() { printf("Hello, world!\n"); return 0; } 令我惊讶的是，跟踪器花了相当长的时间，并报告整个执行过程共有超过 100,000 条指令执行。仅仅是一条输出语句？什么造成了这种情况？答案很有趣（至少你同我一样痴迷与机器/汇编语言）。Linux 的 gcc 默认会动态的将程序与 c 的运行时库动态地链接。这就意味着任何程序运行前的第一件事是需要动态库加载器去查找程序运行所需要的共享库。这些代码的数量很大 - 别忘了我们的跟踪器要跟踪每一条指令，不仅仅是主函数的，而是“整个进程中的指令”。所以当我将测试程序使用静态编译时（通过比较，可执行文件会多出 500 KB 左右的大小，这部分是 C 运行时库的静态链接），跟踪器提示只有大概 7000 条指令被执行。这个数目仍然不小，但是考虑到在主函数执行前 libc 的初始化以及主函数执行后的清除代码，这个数目已经是相当不错了。此外，printf 也是一个复杂的函数。仍然不满意的话，我需要的是“可以测试”的东西 - 例如可以完整记录每一个指令运行的程序执行过程。这当然可以通过汇编代码完成。所以我找到了这个版本的 “Hello, world!” 并编译了它。section .text ; The _start symbol must be declared for the linker (ld) global _start _start: ; Prepare arguments for the sys_write system call: ; - eax: system call number (sys_write) ; - ebx: file descriptor (stdout) ; - ecx: pointer to string ; - edx: string length mov edx, len mov ecx, msg mov ebx, 1 mov eax, 4 ; Execute the sys_write system call int 0x80 ; Execute sys_exit mov eax, 1 int 0x80 section .data msg db 'Hello, world!', 0xa len equ $ - msg 当然，现在跟踪器提示 7 条指令被执行了，这样一来很容易区分它们。深入指令流上面那个汇编语言编写的程序使得我可以向你介绍 ptrace 的另外一个强大的用途 - 详细显示被跟踪进程的状态。下面是 run_debugger 函数的另一个版本：void run_debugger(pid_t child_pid) { int wait_status; unsigned icounter = 0; procmsg("debugger started\n"); /* Wait for child to stop on its first instruction */ wait(&wait_status); while (WIFSTOPPED(wait_status)) { icounter++; struct user_regs_struct regs; ptrace(PTRACE_GETREGS, child_pid, 0, &regs); unsigned instr = ptrace(PTRACE_PEEKTEXT, child_pid, regs.eip, 0); procmsg("icounter = %u. EIP = 0x%08x. instr = 0x%08x\n", icounter, regs.eip, instr); /* Make the child execute another instruction */ if (ptrace(PTRACE_SINGLESTEP, child_pid, 0, 0) < 0) { perror("ptrace"); return; } /* Wait for child to stop on its next instruction */ wait(&wait_status); } procmsg("the child executed %u instructions\n", icounter); } 不同仅仅存在于 while 循环的开始几行。这个版本里增加了两个新的 ptrace 调用。第一条将进程的寄存器值读取进了一个结构体中。 sys/user.h 定义有 user_regs_struct。如果你查看头文件，头部的注释这么写到：/* 这个文件只为了 GDB 而创建 不用详细的阅读.如果你不知道你在干嘛， 不要在除了 GDB 以外的任何地方使用此文件 */ 不知道你做何感想，但这让我觉得我们找对地方了。回到例子中，一旦我们在 regs 变量中取得了寄存器的值，我们就可以通过将 PTRACE_PEEKTEXT 作为参数、 regs.eip（x86 上的扩展指令指针）作为地址，调用 ptrace ，读取当前进程的当前指令（警告：如同我上面所说，文章很大程度上是平台相关的。我简化了一些设定 - 例如，x86 指令集不需要调整到 4 字节，我的32位 Ubuntu unsigned int 是 4 字节。事实上，许多平台都不需要。从内存中读取指令需要预先安装完整的反汇编器。我们这里没有，但实际的调试器是有的）。下面是新跟踪器所展示出的调试效果：$ simple_tracer traced_helloworld [5700] debugger started [5701] target started. will run 'traced_helloworld' [5700] icounter = 1\. EIP = 0x08048080\. instr = 0x00000eba [5700] icounter = 2\. EIP = 0x08048085\. instr = 0x0490a0b9 [5700] icounter = 3\. EIP = 0x0804808a. instr = 0x000001bb [5700] icounter = 4\. EIP = 0x0804808f. instr = 0x000004b8 [5700] icounter = 5\. EIP = 0x08048094\. instr = 0x01b880cd Hello, world! [5700] icounter = 6\. EIP = 0x08048096\. instr = 0x000001b8 [5700] icounter = 7\. EIP = 0x0804809b. instr = 0x000080cd [5700] the child executed 7 instructions 现在，除了 icounter，我们也可以观察到指令指针与它每一步所指向的指令。怎么来判断这个结果对不对呢？使用 objdump -d 处理可执行文件：$ objdump -d traced_helloworld traced_helloworld: file format elf32-i386 Disassembly of section .text: 08048080 <.text>: 8048080: ba 0e 00 00 00 mov $0xe,%edx 8048085: b9 a0 90 04 08 mov $0x80490a0,%ecx 804808a: bb 01 00 00 00 mov $0x1,%ebx 804808f: b8 04 00 00 00 mov $0x4,%eax 8048094: cd 80 int $0x80 8048096: b8 01 00 00 00 mov $0x1,%eax 804809b: cd 80 int $0x80 这个结果和我们跟踪器的结果就很容易比较了。 将跟踪器关联到正在运行的进程如你所知，调试器也能关联到已经运行的进程。现在你应该不会惊讶，ptrace 通过以 PTRACE_ATTACH 为参数调用也可以完成这个过程。这里我不会展示示例代码，通过上文的示例代码应该很容易实现这个过程。出于学习目的，这里使用的方法更简便（因为我们在子进程刚开始就可以让它停止）。代码上文中的简单的跟踪器（更高级的，可以打印指令的版本）的完整c源代码可以在这里找到。它是通过 4.4 版本的 gcc 以 -Wall -pedantic --std=c99 编译的。结论与计划诚然，这篇文章并没有涉及很多内容 - 我们距离亲手完成一个实际的调试器还有很长的路要走。但我希望这篇文章至少可以使得调试这件事少一些神秘感。ptrace 是功能多样的系统调用，我们目前只展示了其中的一小部分。单步调试代码很有用，但也只是在一定程度上有用。上面我通过 C 的 “Hello World!” 做了示例。为了执行主函数，可能需要上万行代码来初始化 C 的运行环境。这并不是很方便。最理想的是在 main 函数入口处放置断点并从断点处开始分步执行。为此，在这个系列的下一篇，我打算展示怎么实现断点。参考撰写此文时参考了如下文章：Playing with ptrace, Part IHow debugger worksvia: http://eli.thegreenplace.net/2011/01/23/how-debuggers-work-part-1作者：Eli Bendersky 译者：YYforymj 校对：wxy本文由 LCTT 原创编译，Linux中国 荣誉推出