蚊章汇 - 并发
type: channel
link: https://linuxcn.bwsl.wang/tag/%E5%B9%B6%E5%8F%91/
items: 在 Go 中实现一个支持并发的 TCP 服务端
items: Go 并发编程中的经验教训
items: Pony 编程语言简介
items: 一些常见的并发编程错误
items: 并发服务器（五）：Redis 案例研究
items: 并发服务器（四）：libuv
items: 并发服务器（三）：事件驱动
items: 并发服务器（二）：线程
items: 并发服务器（一）：简介
items: 雾里看花之 Python Asyncio

在 Go 中实现一个支持并发的 TCP 服务端
type: item
date: 2022-05-22T19:55:00+08:00
description: 仅用大约 65 行代码，开发一个用于生成随机数、支持并发的 TCP 服务端。TCP 和 UDP 服务端随处可见，它们基于 TCP/IP 协议栈，通过网络为客户端提供服务。在这篇文章中，我将介绍如何使用 Go 语言 开发一个用于返回随机数、支持并发的 TCP 服务端。对于每一个来自 TCP 客户端的连接，它都会启动一个新的 goroutine（轻量级线程）来处理相应的请求。你可以在 GitHub 上找到本项目的源码：concTcp.go。处理 TCP 连接这个程序的主要逻辑在 handleConnection() 函数中，具体实现如下：func handleConnection(c net.Conn) { fmt.Printf("Serving %s\n", c.RemoteAddr().String()) for { netData, err := bufio.NewReader(c).ReadString('\n') if err != nil { fmt.Println(err) return } temp := strings.TrimSpace(string(netData)) if temp == "STOP" { break } result := strconv.Itoa(random()) + "\n" c.Write([]byte(string(result))) } c.Close() } 如果 TCP 客户端发送了一个 “STOP” 字符串，为它提供服务的 goroutine 就会终止；否则，TCP 服务端就会返回一个随机数给它。只要客户端不主动终止，服务端就会一直提供服务，这是由 for 循环保证的。具体来说，for 循环中的代码使用了 bufio.NewReader(c).ReadString('\n') 来逐行读取客户端发来的数据，并使用 c.Write([]byte(string(result))) 来返回数据（生成的随机数）。你可以在 Go 的 net 标准包 文档 中了解更多。支持并发在 main() 函数的实现部分，每当 TCP 服务端收到 TCP 客户端的连接请求，它都会启动一个新的 goroutine 来为这个请求提供服务。func main() { arguments := os.Args if len(arguments) == 1 { fmt.Println("Please provide a port number!") return } PORT := ":" + arguments[1] l, err := net.Listen("tcp4", PORT) if err != nil { fmt.Println(err) return } defer l.Close() rand.Seed(time.Now().Unix()) for { c, err := l.Accept() if err != nil { fmt.Println(err) return } go handleConnection(c) } } 首先，main() 确保程序至少有一个命令行参数。注意，现有代码并没有检查这个参数是否为有效的 TCP 端口号。不过，如果它是一个无效的 TCP 端口号，net.Listen() 就会调用失败，并返回一个错误信息，类似下面这样：$ go run concTCP.go 12a listen tcp4: lookup tcp4/12a: nodename nor servname provided, or not known $ go run concTCP.go -10 listen tcp4: address -10: invalid port net.Listen() 函数用于告诉 Go 接受网络连接，因而承担了服务端的角色。它的返回值类型是 net.Conn，后者实现了 io.Reader 和 io.Writer 接口。此外，main() 函数中还调用了 rand.Seed() 函数，用于初始化随机数生成器。最后，for 循环允许程序一直使用 Accept() 函数来接受 TCP 客户端的连接请求，并以 goroutine 的方式来运行 handleConnection(c) 函数，处理客户端的后续请求。net.Listen() 的第一个参数net.Listen() 函数的第一个参数定义了使用的网络类型，而第二个参数定义了服务端监听的地址和端口号。第一个参数的有效值为 tcp、tcp4、tcp6、udp、udp4、udp6、ip、ip4、ip6、Unix（Unix 套接字）、Unixgram 和 Unixpacket，其中：tcp4、udp4 和 ip4 只接…

Go 并发编程中的经验教训
type: item
date: 2020-01-13T23:12:00+08:00
description: 通过学习如何定位并发处理的陷阱来避免未来处理这些问题时的困境。在复杂的分布式系统进行任务处理时，你通常会需要进行并发的操作。在 Mode.net 公司，我们每天都要和实时、快速和灵活的软件打交道。而没有一个高度并发的系统，就不可能构建一个毫秒级的动态地路由数据包的全球专用网络。这个动态路由是基于网络状态的，尽管这个过程需要考虑众多因素，但我们的重点是链路指标。在我们的环境中，链路指标可以是任何跟网络链接的状态和当前属性（如链接延迟）有关的任何内容。并发探测链接监控我们的动态路由算法 H.A.L.O.（ 逐跳自适应链路状态最佳路由 Hop-by-Hop Adaptive Link-State Optimal Routing ）部分依赖于链路指标来计算路由表。这些指标由位于每个 PoP（ 存活节点 Point of Presence ）上的独立组件收集。PoP 是表示我们的网络中单个路由实体的机器，通过链路连接并分布在我们的网络拓扑中的各个位置。某个组件使用网络数据包探测周围的机器，周围的机器回复数据包给前者。从接收到的探测包中可以获得链路延迟。由于每个 PoP 都有不止一个临近节点，所以这种探测任务实质上是并发的：我们需要实时测量每个临近连接点的延迟。我们不能串行地处理；为了计算这个指标，必须尽快处理每个探测。序列号和重置：一个重新排列场景我们的探测组件互相发送和接收数据包，并依靠序列号进行数据包处理。这旨在避免处理重复的包或顺序被打乱的包。我们的第一个实现依靠特殊的序列号 0 来重置序列号。这个数字仅在组件初始化时使用。主要的问题是我们考虑了递增的序列号总是从 0 开始。在该组件重启后，包的顺序可能会重新排列，某个包的序列号可能会轻易地被替换成重置之前使用过的值。这意味着，后继的包都会被忽略掉，直到排到重置之前用到的序列值。UDP 握手和有限状态机这里的问题是该组件重启前后的序列号是否一致。有几种方法可以解决这个问题，经过讨论，我们选择了实现一个带有清晰状态定义的三步握手协议。这个握手过程在初始化时通过链接建立会话。这样可以确保节点通过同一个会话进行通信且使用了适当的序列号。为了正确实现这个过程，我们必须定义一个有清晰状态和过渡的有限状态机。这样我们就可以正确管理握手过程中的所有极端情况。会话 ID 由握手的初始化程序生成。一个完整的交换顺序如下：发送者发送一个 SYN(ID) 数据包。接收者存储接收到的 ID 并发送一个 SYN-ACK(ID)。发送者接收到 SYN-ACK(ID) 并发送一个 ACK(ID)。它还发送一个从序列号 0 开始的数据包。接收者检查最后接收到的 ID，如果 ID 匹配，则接受 ACK(ID)。它还开始接受序列号为 0 的数据包。处理状态超时基本上，每种状态下你都需要处理最多三种类型的事件：链接事件、数据包事件和超时事件。这些事件会并发地出现，因此你必须正确处理并发。链接事件包括网络连接或网络断开的变化，相应的初始化一个链接会话或断开一个已建立的会话。数据包事件是控制数据包（SYN/SYN-ACK/ACK）或只是探测响应。超时事件在当前会话状态的预定超时时间到期后触发。这里面临的最主要的问题是如何处理并发的超时到期和其他事件。这里很容易陷入死锁和资源竞争的陷阱。第一种方法本项目使用的语言是 Golang。它确实提供了原生的同步机制，如自带的通道和锁，并且能够使用轻量级线程来进行并发处理。gopher 们聚众狂欢首先，你可以设计两个分别表示我们的会话和超时处理程序的结构体。type Session struct { State SessionState Id SessionId RemoteIp string } type TimeoutHandler struct { callback func(Session) session Session duration int timer *timer.Timer }Session 标识连接会话，内有表示会话 ID、临近的连接点的 IP 和当前会话状态的字段。TimeoutHandler 包含回调函数、对应的会话、持续时间和指向调度计时器的指针。每一个临近连接点的会话都包含一个保存调度 TimeoutHandler 的全局映射。SessionTimeout map[Session]*TimeoutHandler下面方法注册和取消超时：// schedules the timeout callback function. func (timeout* TimeoutHandler) Register() { timeout.timer = time.AfterFunc(time.Duration(timeout.duration) * time.Second…

Pony 编程语言简介
type: item
date: 2019-02-16T23:39:16+08:00
description: Pony，一种“Rust 遇上 Erlang”的语言，让开发快捷、安全、高效、高并发的程序更简单。在 Wallaroo Labs，我是工程副总裁，我们正在构建一个用 Pony 编程语言编写的 高性能分布式流处理器。大多数人没有听说过 Pony，但它一直是 Wallaroo 的最佳选择，它也可能成为你的下一个项目的最佳选择。“一门编程语言只是另一种工具。与语法无关，与表达性无关，与范式或模型无关，仅与解决难题有关。” —Sylvan Clebsch，Pony 的创建者我是 Pony 项目的贡献者，但在这里我要谈谈为什么 Pony 对于像 Wallaroo 这样的应用是个好选择，并分享我使用 Pony 的方式。如果你对我们为什么使用 Pony 来编写 Wallaroo 甚感兴趣，我们有一篇关于它的 博文。Pony 是什么？你可以把 Pony 想象成某种“Rust 遇上 Erlang”的东西。Pony 有着最引人注目的特性，它们是：类型安全存储安全异常安全无数据竞争无死锁此外，它可以被编译为高效的本地代码，它是在开放的情况下开发的，在两句版 BSD 许可证下发布。以上说的功能不少，但在这里我将重点关注那些对我们公司来说采用 Pony 至关重要的功能。为什么使用 Pony？使用大多数我们现有的工具编写快速、安全、高效、高并发的程序并非易事。“快速、高效、高并发”是可实现的目标，但加入“安全”之后，就困难了许多。对于 Wallaroo，我们希望同时实现四个目标，而 Pony 让实现它们更加简单。高并发Pony 让并发变得简单。部分是通过提供一个固执的并发方式实现的。在 Pony 语言中，所有的并发都是通过 Actor 模型 进行的。Actor 模型以在 Erlang 和 Akka 中的实现最为著名。Actor 模型出现于上世纪 70 年代，细节因实现方式而异。不变的是，所有计算都由通过异步消息进行通信的 actor 来执行。你可以用这种方式来看待 Actor 模型：面向对象中的对象是状态 + 同步方法，而 actor 是状态 + 异步方法。当一个 actor 收到一个消息时，它执行相应的方法。该方法可以在只有该 actor 可访问的状态下运行。Actor 模型允许我们以并发安全的方式使用可变状态。每个 actor 都是单线程的。一个 actor 中的两个方法绝不会并发运行。这意味着，在给定的 actor 中，数据更新不会引起数据竞争或通常与线程和可变状态相关的其他问题。快速高效Pony actor 通过一个高效的工作窃取调度程序来调度。每个可用的 CPU 都有一个单独 Pony 调度程序。这种每个核心一个线程的并发模型是 Pony 尝试与 CPU 协同工作以尽可能高效运行的一部分。Pony 运行时尝试尽可能利用 CPU 缓存。代码越少干扰缓存，运行得越好。Pony 意在帮你的代码与 CPU 缓存友好相处。Pony 的运行时还会有每个 actor 的堆，因此在垃圾收集期间，没有 “停止一切” 的垃圾收集步骤。这意味着你的程序总是至少能做一点工作。因此 Pony 程序最终具有非常一致的性能和可预测的延迟。安全Pony 类型系统引入了一个新概念：引用能力，它使得数据安全成为类型系统的一部分。Pony 语言中每种变量的类型都包含了有关如何在 actor 之间分享数据的信息。Pony 编译器用这些信息来确认，在编译时，你的代码是无数据竞争和无死锁的。如果这听起来有点像 Rust，那是因为本来就是这样的。Pony 的引用功能和 Rust 的借用检查器都提供数据安全性；它们只是以不同的方式来接近这个目标，并有不同的权衡。Pony 适合你吗？决定是否要在一个非业余爱好的项目上使用一门新的编程语言是困难的。与其他方法想比，你必须权衡工具的适当性和不成熟度。那么，Pony 和你搭不搭呢？如果你有一个困难的并发问题需要解决，那么 Pony 可能是一个好选择。解决并发应用问题是 Pony 之所以存在的理由。如果你能用一个单线程的 Python 脚本就完成所需操作，那你大概不需要它。如果你有一个困难的并发问题，你应该考虑 Pony 及其强大的无数据竞争、并发感知类型系统。你将获得一个这样的编译器，它将阻止你引入许多与并发相关的错误，并在运行时为你提供出色的性能特征。开始使用 Pony如果你准备好开始使用 Pony，你需要先在 Pony 的网站上访问 学习部分。在这里你会找到安装 Pony 编译器的步骤和学习这门语言的资源。如果你愿意为你正在使用的这个语言做出贡献，我们会在 GitHub 上为你提供一些 初学者友好的问题。同时，我迫不及待地想在 我们的 IRC 频道 和 Pony 邮件列表 上与你交谈。要了解更多有关 Pony 的消息，请参阅 Sean Allen…

一些常见的并发编程错误
type: item
date: 2018-06-08T08:52:05+08:00
description: Go 是一个内置支持并发编程的语言。借助使用 go 关键字去创建 协程 goroutine （轻量级线程）和在 Go 中提供的 使用 信道 和 其它的并发 同步方法，使得并发编程变得很容易、很灵活和很有趣。另一方面，Go 并不会阻止一些因 Go 程序员粗心大意或者缺乏经验而造成的并发编程错误。在本文的下面部分将展示一些在 Go 编程中常见的并发编程错误，以帮助 Go 程序员们避免再犯类似的错误。需要同步的时候没有同步代码行或许 不是按出现的顺序运行的。在下面的程序中有两个错误。第一，在 main 协程中读取 b 和在新的 协程 中写入 b 可能导致数据争用。第二，条件 b == true 并不能保证在 main 协程 中的 a != nil。在新的协程中编译器和 CPU 可能会通过 重排序指令 进行优化，因此，在运行时 b 赋值可能发生在 a 赋值之前，在 main 协程 中当 a 被修改后，它将会让部分 a 一直保持为 nil。package main import ( "time" "runtime" ) func main() { var a []int // nil var b bool // false // a new goroutine go func () { a = make([]int, 3) b = true // write b }() for !b { // read b time.Sleep(time.Second) runtime.Gosched() } a[0], a[1], a[2] = 0, 1, 2 // might panic } 上面的程序或者在一台计算机上运行的很好，但是在另一台上可能会引发异常。或者它可能运行了 N 次都很好，但是可能在第 (N+1) 次引发了异常。我们将使用 sync 标准包中提供的信道或者同步方法去确保内存中的顺序。例如，package main func main() { var a []int = nil c := make(chan struct{}) // a new goroutine go func () { a = make([]int, 3) c <- struct{}{} }() <-c a[0], a[1], a[2] = 0, 1, 2 } 使用 time.Sleep 调用去做同步我们先来看一个简单的例子。package main import ( "fmt" "time" ) func main() { var x = 123 go func() { x = 789 // write x }() time.Sleep(time.Second) fmt.Println(x) // read x } 我们预期程序将打印出 789。如果我们运行它，通常情况下，它确定打印的是 789。但是，这个程序使用的同步方式好吗？No！原因是 Go 运行时并不保证 x 的写入一定会发生在 x 的读取之前。在某些条件下，比如在同一个操作系统上，大部分 CPU 资源被其它运行的程序所占用的情况下，写入 x 可能就会发生在读取 x 之后。这就是为什么我们在正式的项目中，从来不使用 time.Sleep 调用去实现同步的原因。我们来看一下另外一个示例。package main import ( "fmt" "time" ) var x = 0 func main() { var num = 123 var p = &num c := make(chan int) go func() { c <- *p + x }() time.Sleep(time.Second) num = 789 fmt.Println(<-c) } 你认为程序的预期输出是什么？123 还是 789？事实上它的输出与编译器有关。对于标准的 Go 编译器 1.10 来说，这个程序很有可能输出是 123。但是在理论上，它可能输出的是 789，或者其它的随机数。现在，我们来改变 c <- *p + x 为 c <- *p，然后再次运行这个程序。你将会发现输出变成了 789 （使用标准的 Go 编译器 1.10）。这再次说明它的输出是与编译器相关的。是的，在上面的程序中存在数据争用。表达式 *p 可能会被先计算、后计算、或者在处理赋值语句 num = 789 时计算。time.Sleep 调用并不能保证 *p 发生在赋值语句处理之前进行。对于这个特定的示例，我们将在新的协程创建之前，将值保存到一个临时值中，然后在新的协程中使用临时值去消除数据争用。... tmp := *p + x go func() { c <- tmp }() ... 使协程挂起挂起协程是指让协程一直处于阻塞状态。导致协程被挂起的原…

并发服务器（五）：Redis 案例研究
type: item
date: 2018-03-08T03:31:28+08:00
description: 这是我写的并发网络服务器系列文章的第五部分。在前四部分中我们讨论了并发服务器的结构，这篇文章我们将去研究一个在生产系统中大量使用的服务器的案例—— 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 要实现它自己的事件库？在 第四节 中我们讨论了使用 li…

并发服务器（四）：libuv
type: item
date: 2018-03-02T09:20:53+08:00
description: 这是并发网络服务器系列文章的第四部分。在这一部分中，我们将使用 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 注册一个事件回…

并发服务器（三）：事件驱动
type: item
date: 2017-12-08T18:20:00+08:00
description: 这是并发服务器系列的第三节。第一节 介绍了阻塞式编程，第二节：线程 探讨了多线程，将其作为一种可行的方法来实现服务器并发编程。另一种常见的实现并发的方法叫做 事件驱动编程，也可以叫做 异步 编程 注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…

并发服务器（二）：线程
type: item
date: 2017-10-26T16:19:00+08:00
description: 这是并发网络服务器系列的第二节。第一节 提出了服务端实现的协议，还有简单的顺序服务器的代码，是这整个系列的基础。这一节里，我们来看看怎么用多线程来实现并发，用 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 返回的时…

并发服务器（一）：简介
type: item
date: 2017-10-25T17:46:00+08:00
description: 这是关于并发网络服务器编程的第一篇教程。我计划测试几个主流的、可以同时处理多个客户端请求的服务器并发模型，基于可扩展性和易实现性对这些模型进行评判。所有的服务器都会监听套接字连接，并且实现一些简单的协议用于与客户端进行通讯。该系列的所有文章：第一节 - 简介第二节 - 线程第三节 - 事件驱动协议该系列教程所用的协议都非常简单，但足以展示并发服务器设计的许多有趣层面。而且这个协议是 有状态的 —— 服务器根据客户端发送的数据改变内部状态，然后根据内部状态产生相应的行为。并非所有的协议都是有状态的 —— 实际上，基于 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)…

雾里看花之 Python Asyncio
type: item
date: 2016-12-22T02:27:00+08:00
description: 最近我开始发力钻研 Python 的新 asyncio 模块。原因是我需要做一些事情，使用事件 IO 会使这些事情工作得更好，炙手可热的 asynio 正好可以用来牛刀小试。从试用的经历来看，该模块比我预想的复杂许多，我现在可以非常肯定地说，我不知道该如何恰当地使用 asyncio。从 Twisted 框架借鉴一些经验来理解 asynio 并非难事，但是，asyncio 包含众多的元素，我开始动摇，不知道如何将这些孤立的零碎拼图组合成一副完整的图画。我已没有足够的智力提出任何更好的建议，在这里，只想分享我的困惑，求大神指点。原语asyncio 通过 协程 coroutines 的帮助来实现异步 IO。最初它是通过 yield 和 yield from 表达式实现的一个库，因为 Python 语言本身演进的缘故，现在它已经变成一个更复杂的怪兽。所以，为了在同一个频道讨论下去，你需要了解如下一些术语：事件循环事件循环策略awaitable协程函数老式协程函数协程协程封装 生成器 （ generator ） future并发的future 任务 （ task ） 句柄 执行器 （ executor ） 传输 （ transport ） 协议此外，Python 还新增了一些新的特殊方法：__aenter__ 和 __aenter__，用于异步块操作__aiter__ 和 __anext__，用于异步迭代器（异步循环和异步推导）。为了更强大些，协议已经改变过一次了。 在 Python 3.5 它返回一个 awaitable（这是个协程）；在 3.6它返回一个新的异步生成器。__await__，用于自定义的 awaitable你还需要了解相当多的内容，文档涵盖了那些部分。尽管如此，我做了一些额外说明以便对其有更好的理解：事件循环asyncio 事件循环和你第一眼看上去的略有不同。表面看，每个线程都有一个事件循环，然而事实并非如此。我认为它们应该按照如下的方式工作：如果是主线程，当调用 asyncio.get_event_loop() 时创建一个事件循环。如果是其它线程，当调用 asyncio.get_event_loop() 时返回运行时错误。当前线程可以使用 asyncio.set_event_loop() 在任何时间节点绑定事件循环。该事件循环可由 asyncio.new_evet_loop() 函数创建。事件循环可以在不绑定到当前线程的情况下使用。asyncio.get_event_loop() 返回绑定线程的事件循环，而非当前运行的事件循环。这些行为的组合是超混淆的，主要有以下几个原因。 首先，你需要知道这些函数被委托到全局设置的底层事件循环策略。 默认是将事件循环绑定到线程。 或者，如果需要的话，可以在理论上将事件循环绑定到一个 greenlet 或类似的。 然而，重要的是要知道库代码不控制策略，因此不能推断 asyncio 将适用于线程。其次，asyncio 不需要通过策略将事件循环绑定到上下文。 事件循环可以单独工作。 但是这正是库代码的第一个问题，因为协同程序或类似的东西并不知道哪个事件循环负责调度它。 这意味着，如果从协程中调用 asyncio.get_event_loop()，你可能没有机会取得事件循环。 这也是所有 API 均采用可选的显式事件循环参数的原因。 举例来说，要弄清楚当前哪个协程正在运行，不能使用如下调用：def get_task(): loop = asyncio.get_event_loop() try: return asyncio.Task.get_current(loop) except RuntimeError: return None 相反，必须显式地传递事件循环。 这进一步要求你在库代码中显式地遍历事件循环，否则可能发生很奇怪的事情。 我不知道这种设计的思想是什么，但如果不解决这个问题（例如 get_event_loop() 返回实际运行的事件循环），那么唯一有意义的其它方案是明确禁止显式事件循环传递，并要求它绑定到当前上下文（线程等）。由于事件循环策略不提供当前上下文的标识符，因此库也不可能以任何方式“索引”到当前上下文。 也没有回调函数用来监视这样的上下文的拆除，这进一步限制了实际可以开展的操作。awaitable 与 协程 coroutine 以我的愚见，Python 最大的设计错误是过度重载迭代器。它们现在不仅用于迭代，而且用于各种类型的协程。 Python 中迭代器最大的设计错误之一是如果 StopIteration 没有被捕获形成的空泡。 这可能导致非常令人沮丧的问题，其中某处的异常可能导致其它地方的生成器或协同程序中止。 这是一个长期存在的问题，基于 Python 的模板引擎…