Kotlin协程：快速和可扩展的会合和缓冲通道算法

107×→Kotlin协程NikitaKovalnikita.koval@jetbrains.com荷兰摘要Dan Alistarhdan.ist.ac.atIST奥地利奥地利1引言RomanElizarovroman.elizarov@jetbrains.comJetBrains荷兰在过去的十年中，异步编程已经获得了显著的流行：通过库和本地语言实现，在许多流行的语言中都可以获得对这种编程模式的支持，通常以协程或JavaScript/await构造的形式。 与通过共享内存编程不同，这个概念假定通过消息传递进行隐式同步。实现这种通信的关键数据结构是会合信道。粗略地说，会合通道是一个大小为零的阻塞队列，所以send（e）和receive（）操作都在等待对方，当它们相遇时执行会合为了优化消息传递模式，通道通常配备了一个固定大小的缓冲区，因此发送方不会挂起并将元素放入缓冲区，直到超过其容量。这种原语称为缓冲通道。本文提出了一种快速和可扩展的算法，会合和缓冲通道。类似于现代队列，我们的解决方案是基于一个无限数组，有两个位置计数器用于send（e）和receive（）操作，调用无条件的Fetch-And-Add指令来更新它们。然而，该算法需要对该经典模式进行重要修改，以便支持完整的通道语义，例如缓冲和取消等待请求。 我们将我们的解决方案的性能与Kotlin实现以及其他学术建议进行了比较，显示出高达9.8的加速比。 为了展示其表现力和性能，我们还将所提出的算法集成到标准Kotlin Coroutines中，取代了之前的通道实现。CCS概念：·计算方法学并发算法;共享内存算法。关键词：并发，集合通道，同步队列，缓冲通道，协程，Kotlin允许免费制作本作品的全部或部分的数字或硬拷贝，以供个人或课堂使用，前提是制作或分发副本的目的不是为了盈利或商业利益，并且副本的第一页上有本声明和完整的引用。本作品的版权归ACM以外的其他人所有，必须予以尊重。允许使用学分进行摘要以其他方式复制、重新发布、在服务器上发布或重新分发到列表中，需要事先获得特定许可和/或付费。从permissions@acm.org请求权限。PPoPP '23，2023年2月25日至3月1日，加拿大魁北克省蒙特利尔市© 2023版权归所有者/作者所有授权给ACM的出版权。ACM 979-8-4007-0015-6/23/https://doi.org/10.1145/3572848.3577481异步编程是一种经典的构造，它允许将代码单元的执行与主应用程序分离。当执行延迟差异非常大的代码时，例如I/O或网络时，它非常有用协程[16]是一种支持异步编程的通用而有效的方法。它们可以暂停和恢复代码执行，从而允许非抢占式多任务处理。 使用协程编程类似于标准的基于线程的编程，主要区别在于协程是轻量级的，并且重用本机线程。因此，程序员可以为每个小任务创建一个协程，并使用结构化并发将这些任务与线程相反，线程被调度程序抢先多任务并需要额外的同步，协同进程是协同多任务的-它们显式地产生对执行流的控制，并且不需要额外的同步。许多现代编程语言，如Kotlin、Go和C++，都足够灵活和高效，可以为协程提供原生支持。协程背后的关键同步原语是会合通道，也称为同步队列。在其核心，会合通道是一个阻塞的零容量有界队列。 它支持send（e）和receive（）请求：send（e）检查队列是否包含任何接收者，并删除第一个（即， 执行“会合”）或将自己作为等待发送者添加到队列中;receive（）操作对称地工作。请注意，这些操作在设计上是阻塞的。 会合通道作为主要的同步原语，因此是异步编程的关键，为其他消息传递协议提供了基础，例如通信顺序进程（CSP）[14]和演员[9]。为了更好地了解通道语义，举个例子。下图显示了经典的生产者-消费者场景，多个生产者通过共享的集合通道向消费者发送任务。在这里，通道类似于队列：任务被添加到一端，并从另一端接收让第一个消费者首先调用receive（）来检索一个任务：当通道为空时，它挂起，等待生产者。当108×PPoPP当生产者到来时，它与挂起的消费者会合，将任务直接转移给它。但是，如果又有一个生产者进来，而没有消费者在通道上等待，则send（e）调用挂起。消费者的下一个receive（）调用与挂起的生产者会合，处理任务。当使用通道传递消息时，如上面的示例，程序员通常使用更有效的缓冲通道原语。粗略地说，它是一个配备有固定大小缓冲区的会合通道：生产者将元件放置在该缓冲区中，直到其容量被超过。 一旦通道满了，新的send（e）调用就会挂起，直到缓冲区中的空间变得可用。语义类似于有限容量的阻塞队列。 对缓冲区大小进行限制可以防止生产者比消费者快，从而无限期地增加缓冲区。缓冲通道的高效和可伸缩实现对于高性能消息传递至关重要。渠道是有用的，大量的研究已投入资金提供快速实施。Java提供了一个有效的无锁SynchronousQueue数据结构，具有会合通道语义[24];本质上，它将挂起的请求存储在Michael-Scott队列中[20]。Kotlin提供了一个基于无锁双向链表[3]的优化版本的实现，并带有描述符[11]以确保原子性。大多数其他语言，如Go和Rust，利用粗粒度的锁定设计[5，6]。Izraelevitz和Scott 1 [15]以及Koval等人的近期作品。 [17]提出了会合通道的新实现，利用了非阻塞队列设计中的创新[21，27]。这些作品不支持缓冲。在建设高效通道时，有两个关键挑战。第一个是有效地维护一个挂起操作的队列，并实现允许每个操作检查是否有相反类型的等待操作的过程，它可以与之会合;如果不可能重定向，操作应该保留一个新的队列单元，它可以在其中挂起。当将集合语义扩展到缓冲通道时，中断支持成为主要问题-实际上，可以取消挂起的请求，从而使集合不可用。支持中断和缓冲语义是非常重要的：所有实际的解决方案，例如Kotlin [3]和Go [5]中的官方缓冲通道实现，仍然使用朴素的粗粒度锁定设计进行同步。我们的贡献。 本文提出了一种新的可扩展的和有效的方法来实现会合和缓冲通道。我们从构建一个逻辑无限数组的想法开始以及两个原子计数器，其跟踪曾经执行的send（e）和receive（）调用的总数。计数器应该在每个操作开始时递增-之后，算法能够决定当前操作是否应该暂停或与相反类型的操作会合当一个操作使计数器递增时，它也会在无限数组中“保留”相应的单元格。剩余的同步在此小区中执行，只能由一个发送方和一个接收方处理。到目前为止，我们的设计遵循的模式所提出的辛勤工作。 我们的主要贡献是在支持缓冲通道语义，引入非平凡的困难。粗略地说，在逻辑层，我们需要添加一个额外的计数器，它应该指示阵列中逻辑缓冲区的结束，所以CPU可以检查保留单元是否在缓冲区中。 与此同时，接收器应向前移动此计数器，如果需要，则恢复暂停的计数器。尽管这种逻辑看起来很简单，但是缓冲通道的同步模式必须非常小心地实现，以保持send（e）和receive（）操作的正确性，特别是当挂起的操作可能被中断时。我们在Kotlin和Inte中实现了我们的通道设计Grated it to implement communication mechanismsunderlying the Kotlin Coroutines library. 与原生Kotlin解决方案和之前的同步队列实现[17，24]相比，我们的算法更具可扩展性，并且在吞吐量方面优于之前的提议高达9.8。值得注意的是，我们的方法是可移植的，所以它应该扩展到其他语言，如Go或Rust。全纸。本文的扩展版本在附录部分包含了许多有用的实现细节，可在arXiv [18]上公开获取。2 环境协同程序管理。本文提出的算法可以同时操作线程和协程。由于通道通常用于异步编程，我们将重点放在协程作为用例。由于send（e）和receive（）操作都是阻塞的，所以应该有一种机制来挂起和调度协程。我们在本文中使用的API如清单1所示，可以很容易地适应大多数编程语言，如Kotlin、Go或Java。interfaceCoroutine{有趣try Unpark（）： Boolean有趣interrupt（）funpark（o n Interpretation：lambda（）->Unit）1Izraelevitz和Scott [15]的同步队列算法在操作暂停的条件方面打破了会合通道语义;我们在本文的完整版本中给出了这样一个例子[18]。}funcur Corr（）：输出清单1.协同程序管理API109Kotlin Coroutines中的快速和可扩展通道PPoPP就像在操作系统中一样，我们称之为park（..） 用于挂起可通过curCor（）调用获得的正在运行的协程。当停驻一个协程时，相应的本机线程不会停止，而是调度另一个协程，这使得协程挂起机制相对便宜。当被停驻时，我们假设协程可以通过中断（）调用取消，变得无法重新停驻。请注意，协同程序中断与线程中断不同，它发生得相对频繁，因此应该是高效的。 当一个挂 起 的 协 程 被 中 断 后 ， 在 park （ .. ） 中 设 置 的oncodeaction将被调用。 执行（第4行）。 如果协程在活动状态下被中断，则中断将通过以下park（..） 召唤为了恢复协程，应该调用tryUnpark（） 如果中断成功 ， 此 函 数 将 返 回 true; 如 果 协 程 已 被 中 断 ，则 返 回false。注意，tryUnpark（）可以在park（..）之前调用。 - 在这种情况下，以下公园（.） 调用完成而不挂起。内存模型和内存回收。 为了简单起见，我们假设顺序一致的内存模型。除了普通的读和写，我们使用原子的比较和交换（CAS）和提取和添加（FAA）指令。 我们还假设运行时环境支持垃圾收集（GC）。 回收技术，如危险点[19]，可以在没有GC的环境中使用。3 缓冲通道算法我们的算法的高级结构将基于无限数组数据结构的共同抽象这个无限数组存储元素和协程，并为所有执行过的send（e）和receive（）调用使用单独的计数器。此外，我们有一个计数器，它指示无限数组中逻辑缓冲区的结束。send（e）和receive（）操作都是通过递增计数器开始的，保留计数器引用的数组单元。 对于会合通道，操作either挂起，存储其协程在细胞中，或使会合与相反类型的请求。每个信元可以由一个send（e）和一个receive（）处理。对于缓冲通道，如果信元在逻辑缓冲区中，则send（s）添加元素而不挂起，而receive（）负责更新缓冲区的结尾右边嵌入的图形显示了这样一个通道的一个小例子，其中缓冲区的容量为2，一个receive（）操作等待对于一个元素。下面的发送（e）调用增加计数器S，并与存储在单元中的挂起接收器进行会合（在图中标记为« R»）。接下来的两个send（e）-s将继续执行而不挂起因为单元在逻辑缓冲器中（缓冲器单元以黄色突出显示）。但是，第四个send（e）操作将不得不挂起，因为缓冲区已经满了。清单2给出了上面讨论的缓冲通道结构。 S、R和B 64位计数器存储send（e）和receive（）调用的总数以及逻辑缓冲区的结束。 我们假设send（e）在s {2varS、R、B：长3varA：无限数组服务员><4fun send（element：E）{. }5有趣public void run（）{ ... 个文件夹（六）7struct Waiter （state：Any？，elem：E？）的方式清单2. 渠道结构。3.1交会信道我们开始一个算法，只支持重定向通道语义，我们稍后将扩展这种设计缓冲通道。send（e）和receive（）都是通过Fetch-And-Add递增相应的计数器开始的，因此，保留信元-最多一个发送方和一个接收方可以处理每个信元。接下来，操作读取相反操作类型的计数器，以决定是进行会合还是暂停;在这一部分，我们忽略计数器B，只操作S和R。当相反的计数器“覆盖”保留的单元（大于保留的单元索引）时，操作进行会合。否则，它将当前协程安装到状态字段并挂起。细胞生命周期。每个发送-接收对工作在同一个细胞上的状态字段，其状态机，中国是在图1。为了传输元素，send（e）在同步之前将其放置在elem字段中，因此receive（）可以在同步之后安全地检索元素，遵循安全发布模式。最初，每个单元都处于EMPTY状态（在代码中表示为null）。通常，要挂起的操作会将其更改为Coroutine。之后，一个相反类型的操作来到单元，通过恢复存储的协程并将单元状态更新为DONE来进行会合。但是，如果要挂起的操作尚未存储其协程，则应该恢复等待请求的操作可能会进入EMPTY状态。我们使用不同110≥PPoPP方法send（e）和receive（）来处理竞争。send（e）操作通过将信元状态更改为BUFFERED来进行消除;它知道接收器已经到来，因此没有理由等待它。之后，一个接收器来了，发现消除已经发生，采取的元素和完成没有暂停。不幸的是，我们不能对receive（）使用类似的逻辑，因为它不仅要恢复相反的操作，还要检索元素。相反，它这个解决方案让人想起LCRQ队列[22]。重新启动操作的另一个原因是发现相反请求的协程已经中断;在这种情况下，要么对应的tryUnpark（）失败，要么单元处于INTERRUPTED状态。 当一个协程被中断时，它应该将单元状态更改为INTERRUPTED以避免内存泄漏-这个逻辑可以通过online lambda参数park（..）.send（e）的算法。 清单3给出了send（e）和receive（）的伪代码。 要发送一个元素，操作首先递增它的S计数器，并通过Fetch-And-Add（第2行）在递增之前获取值。接下来，它将元素放入保留单元A[s]（第3行）。最后，它根据图1中的图表通过调用updCellSend（..） 函数（第4行）。为了与生命周期图保持一致，updCellSend（..）是以一种基于状态的方式编写的。逻辑被包装在无限循环中，在开始时获得当前单元状态和R值（第8图1.会合通道的细胞生命周期。每个状态都列出了一组可以发现它的操作 如果操作应该执行某个动作（例如，完成或重新启动）时，在方括号中指定该动作否则，操作执行到另一状态的转换。当可能有多个转换时，每个转换的条件在圆括号中指定 与状态中的规范一样，当转换成功时，操作可以执行方括号中指定的操作。参数s和r（小写）指的是send（e）和receive（）的工作小区索引。当单元状态为EMPTY并且没有接收器到达（s r）时，操作决定挂起。因此，它获得当前的协程引用（第13行），通过原子CAS指令用它替换EMPTY（第14行），并通过cor.park（.）调用（第15 - 16行）。如果安装协程的CAS失败，操作将重新启动。操作恢复后通过接收器，它成功地完成（第17行）。发送（e）也可能在被停放时被中断 我们将一个特殊的中断处理程序传递给cor.park（.） 将单元格状态移动到INTERRUPTED并清除元素字段的调用（第16行）。当计算单元存储一个Coroutine实例时，我们知道这是一个等待的receive（）操作，应该使用它来进行重定向因此，send（e）试图恢复它，调用tryUnpark（）（第20行）。成功后，状态转移到DONE，操作完成（第21行）。否则，如果接收器已经被中断，操作将清除A[s].elem以避免内存泄漏并重新启动（第23行）。也有可能信元已经为了解决竞争，send（e）将单元更新为BUFFERED，通知接收方该元素已经可供检索（第26行）;因此，发生了消除最后，当发现状态是INTERRUPTED或BROKEN时，send（e）清除单元中的元素字段并重新启动（第28receive（）的算法。receive（）实现是对称的，并根据图1中的 生 命 周 期 图 通 过 updCellRcv （ .. ） 功 能 与updCellSend（..）的本质区别是它(1) 当一个单元是空的但应该发生重新加载时，它标记单元BROKEN，以及（2）检查BUFFERED标记。虽然该算法适用于额外的次要优化，但我们保留了其结构，使其保持一致具有接下来呈现的缓冲通道变体3.2缓冲通道本质上，缓冲通道是具有用于存储元素的固定大小的缓冲区的会合通道一个容量缓冲通道允许缓存在挂起之前发送缓存元素，将它们存储在一个缓冲区中，类似于有界阻塞队列。当缓冲区已满时，send（e）挂起。 receive（）操作要么在缓冲区为空（如果 > 0）时挂起，要么检索第一个元素并恢复第一个等待的发送者，将其元素添加到缓冲区。高水平的想法。 为了支持上面的语义，我们维护一个附加的计数器B，初始化为缓冲区容量C，它指示单元阵列中“假想缓冲区”的结束。send（e）操作将保留的索引s与B进行比较，如果s为B，则缓冲元素<，否则挂起。receive（）操作操作缓冲区的第一个单元格，要么检索元素，要么安装其协程以供挂起--在这两种情况下，缓冲区容量111Kotlin Coroutines中的快速和可扩展通道PPoPP1 funsend（elemen t： E）=while（true）{2s：=FAA（&S，+1）//ree3A [s]。elem=元素4ifupd Cell Send（ s）：return（五）6 //返回`false`如果这个7 funupdCel lSen d（ s：In t）：Boo l=while（true）{8状态：= A [s]。state//读取当前状态9r：=R//读取接收方的计数器10当11//为空且没有接收器到来=> suspend12状态e==null s>= r：13cor：=cur Cor（）//获取当前协程14ifCAS（&A [s]. 统计，数字，更正）：15林前park（//等待集合16中断={A[ s]=（IN TERRU PTED，nul1）}）17返回true18//等待接收器=>尝试恢复它19状态为C或输出：20如果国家。try Unpark（）：21A[s]。state=DONE;return true22else：//interrupted，cleanecellanddfail23A[s]。em=null;returnfalse24//空但接收器即将到来=>消除25状态e==null s r：26ifCAS（&A[s]. state，null，BU FF ERED）：恢复正常27//中断接收器或中毒=>失败28状态e==INTERRU PTED||state==BR OKEN:29A[s]。em=null//清除以避免存储器泄漏30returnfalse31 个文件夹32}33 funreceive（）： E=whilee（true）{34r：=FAA（& R1+ 1）//ree35如果upd小区Rcv（ r）：36e：=A[r]。elem;A[r]. em=null;returne37}38 //返回`false`如果这个39 funupdCel lRc v（ r：In t）：Boo l=while（true）{40状态：= A [r]。state//读取当前状态41S：=S//读取发送者的计数器42当43//为空且没有发送者到来=> suspend44状态e==null r>= s：45cor：=cur Cor（）//获取当前协程46ifCAS（&A [r]. 统计，数字，更正）：47林前park（//等待集合48o n Interpret ={A [r]. state=INTERRU PTED}）49返回true50//等待发送者=>尝试恢复它51状态为C或输出：52如果国家。 try Unpark（）：53A[r]. state=DONE;return true54else：//interrupted55returnfalse56//空的，但发送者来了=>毒药57状态e==null r s：58ifCAS（&A[r]. state，null，BR OKEN）：返回59//发生了一次淘汰=> finish60state ==BUFFERED：return true61//中断sender => fail62state==INTERRU PTED：恢复正常63 个文件夹64}清单3. 会合通道算法。 updCellSend（..）和updCellReceive（..） 函数根据图1中的图表更新单元状态。与消除和细胞中毒有关的部分用黄色和红色突出显示减少。 为了恢复它，receive（）递增计数器B（获得缓冲区后第一个单元的索引b），并在需要时恢复挂起在单元A[b]中的发送器。右图显示了容量为2的缓冲通道的示例状态。这里，两个元素在缓冲区中，一个发送者是wait-在它之后的第一个单元格中（该单元格标记为“S”）。下面的receive（）调用增加R并重新获取第一个元素。之后，它通过递增B并恢复在相应单元A[b]中等待的发送方来扩展缓冲区（其中b是递增之前很容易假设缓冲器的逻辑结束总是在位置R+C，其中C是信道容量，在这种情况下，将不需要附加计数器B. 不幸的是，由于中断，这是不正确的考虑容量为1的通道 两个线程：第一个线程将其元素插入缓冲区，而第二个线程挂起并中断。所得到的信道状态在右侧嵌入图中示出，其中RB参考单元由虚线箭头表示 。 之后，receive（）来了，递增R并检索第一个元素。如果缓冲区的新末端如果是R+C，则缓冲区（R和R+C之间的信元）将覆盖发送方被中断的唯一信元。下面的send（e）调用将挂起，这是不正确的，因为通道已经为空。保持缓冲区的结尾明确解决了这个问题-如果单元A[b]存储中断的发送器，则缓冲区扩展过程重新启动，再次递增计数器B，这样就解决了这个问题。缓冲区扩展。 我们将缓冲区扩展逻辑提取到一个特殊的expandBuffer（）过程中，该过程在receive（）成功执行其同步后调用，要么检索第一个元素，要么存储其协程以供暂停。 当提到receive（）时，我们通常只提到它的同步阶段，而没有扩展。因此，send（e）、receive（）和expandBuffer（）操纵它们自己的计数器，并且每个信元可以由单个这样的调用处理，除了中断的情况扩大112≥≥PPoPP在缓冲区中，我们首先递增计数器B，在递增之前获得索引b如果单元A[b]没有被发送方（b S）覆盖，则保证稍后将与单元一起工作的发送方（e）将观察到s B并将缓冲其元素。否则，如果bS，则单元格A[b]存储了一个发送者，或者有一个传入的发送者。<我们现在简要讨论可能的情况。通常，单元格存储一个挂起的发送者，expandBuffer（）尝试恢复它，并在成功时结束。在发送方恢复失败的情况下，缓冲区扩展过程重新开始，因为向缓冲区添加已经“损坏”的为了通知即将到来的发送者它不应该挂起，expandBuffer（）将单元格移动到一个特殊的IN_BUFFER状态并返回。当send（e）已经观察到信元是缓冲区的一部分并将其移动到BUFFERED状态时，会发生另一个竞争（这种竞争是可能的，因为expandBuffer（）在B递增之后读取S计数器，所以send（e）可能已经观察到单元已经在缓冲区中;类似的竞争发生在send（e）和receive（）之间。由于单元格已经在缓冲区中，因此expandBuffer（）过程返回。最后，receive（）可以更早到达并处理单元格。 如果单元格中存储了发送方，receive（）会帮助expandBuffer（）恢复它。成功后，receive（）和expandBuffer（）都会结束，如果发送方已经被中断，则会重新启动。 如果接收者在 发 送 者 之 前 到 达 ， 它 将占 用 第 一 个缓 冲 单 元 ， 而expandBuffer（）过程可以合法地返回。不可区分的协程。为了简单起见，我们假设可以区分存储在单元中的协程是发送方还是接收方。虽然有些语言（如Go）提供了这种支持，但许多其他语言（如Kotlin或Java）需要更通用的实现。我们将在全文中讨论如何克服这一限制[18]。细胞生命周期。 图2显示了缓冲通道的细胞生命周期图;相应的updCellSend（.）和updCellRcv（..） 实现在列表4中给出。 抽象的send（e）和receive（）操作与清单3中的集合通道相同。send（e）的算法。最初，每个单元格都处于EMPTY状态。当send（e）到达一个空单元格时，它决定是缓冲元素还是挂起。 如果信元是缓冲器的一部分（s buffer6状态e==null&&（s suspend54funexpandBuffer（）=while（true）{55b：=FAA（&B，+1）56如果b>=S：返回//不包含在send（），finish中57if upd单元格EB（b）：return//更新单元格状态58}59 //返回`true`如果expandBuffer（）应该60 // finish和`false`当它应该重新启动61 funupdateCel lE B（ b：In t）：Boo l=while（true）{62状态：=A[ b]63当11状态e==null&&12cor：= cur Cor（）&& S>=r：64//一个暂停的发送器被存储=>尝试恢复它65状态为发送C或O输出：13141516171819202122232425}ifCAS（&A[s]. 统计，数字，更正）：林前park（{on park（s）}）; return true// Waiting receiver => try to resumeitstateisCoroutineRCV：如果国家。try Unpark（）：A[s]。state=DONERCV;return trueelse：//interrupted，cleanecellanddfailA[s].em=null;returnfalse//接收器中断或中毒=>故障状态e==INTERRU PTED RCV||state==BR OKEN:A[s]。em=null//清除以避免存储器泄漏returnfalse66ifCAS（&A [s]. 状态，状态，S_RESU MIN G EB）：67如果国家。try Unpark（）：68A[s]。state=返回true69else：70A[s]。state=INTERRU PTED SEND;returnfalse71//元素已经被缓冲=> finish72state ==BUFFERED：return true73//发送者被中断=>重新启动74state==INTERRU PTED SEND：returnfalse75//单元格为空，send（）即将到来=>76//将单元格标记为“in the buffer“并结束77状态e==null：78ifCAS（&A[b]. state，null，IN_BU FF ER）：26 funupdCel lRc v（ r：In t）：Boo l=while（true）{27状态：= A [r]。状态; s：= S7980//返回true接收器存储在单元格中=>完成28当29//为空且没有发送者到来=> suspend30（状态e==null||state==IN_BU FFER)&&r>=s:31cor：=cur Cor（）32ifCAS（&A [r]. 统计，统计，订正）：33expand（）34林前park（{on String（r）}）; return true35//为空，但发送者即将到来=>毒药重启8182838485868788}状态是RCV中的核心||state==INTERRU PTED RCV: return true// Poisoned cell => finish，receive（）is inchargestate==BROKEN：return true//接收方正在恢复发送方=>waitstate==S_RESU MIN G RCV：continuee}363738394041424344454647484950515253}（状态e==null||state==IN_BU FFER)&&r

下载后可阅读完整内容，剩余1页未读，立即下载