试图使用Core.Async.Pipe创建异步示例，但Pipe.write似乎阻止了等待Pipe.read

Question

我创建了下面的玩具示例，该示例在循环中进行计数，并将值写入Async.Pipe：

open Sys
open Unix
open Async.Std
let (r,w) = Pipe.create ()

let rec readloop r = 
  Pipe.read r >>=
  function
  | `Eof -> return ()
  | `Ok v -> return (printf "Got %d\n" v) >>=
  fun () -> after (Core.Time.Span.of_sec 0.5) >>=
  fun () -> readloop r 

let countup hi w =
  let rec loop i = 
    printf "i=%d\n" i ;
    if (i < hi &&( not (Pipe.is_closed w))) then 
       Pipe.write w i >>>
       fun () -> loop (i+1)
     else Pipe.close w
  in 
  loop 0 

let () =
  countup 10 w;
  ignore(readloop r);;
  Core.Never_returns.never_returns (Scheduler.go ()) 

请注意readloop函数是递归的-它只是不断地从管道中读取值，因为它们是可用的。但是，我在每次读取之间增加了0.5秒的延迟。计数函数有点类似，但它循环并对同一个管道进行写入。

当我运行这个时，我得到：

i=0
i=1
Got 0
i=2
Got 1
i=3
Got 2
i=4
Got 3
i=5
Got 4
i=6
Got 5
i=7
Got 6
i=8
Got 7
i=9
Got 8   
i=10
Got 9

除了上面的前三行输出之外，其余的输出行似乎都需要等待半秒。因此，似乎在写完之后，管道就会被阻塞，直到从管道中读出为止。(Pipe.write w数据似乎阻止了等待Pipe.read r)我认为应该发生的事情(因为这是某种异步管道)是值将在管道中排队，直到读取发生，如下所示：

i=0
Got 0 (* now reader side waits for 1/2 second before reading again *) 
i=1   (* meanwhile writer side keeps running *)
i=2
i=3
i=4
i=5
i=6
i=7
i=8  
i=9 (* up till here, all output happens pretty much simultaneously *)
Got 1 (* 1/2 second between these messages *)
Got 2
Got 3
Got 4
Got 5
Got 6
Got 7
Got 8
Got 9

我想知道是否有一种方法可以使用异步获得行为？

我真正的用法是我打开了一个Tcp套接字(作为一个客户端)，如果我在客户机和服务器之间设置了一些线程之后使用线程，我会启动一个线程，这个线程只会从服务器上读取从套接字传入的数据，并将这些数据放入一个消息队列中，这些消息可以在程序的主线程准备好时进行检查。但是，我不想使用线程来实现同样的目的:从套接字读取来自服务器的数据，当数据可用时，检查消息并根据其内容执行一些操作。可能还会发生其他事情，所以上面的代码中的“等待半秒”对此进行了模拟。我原以为管道会排好队，以便在读者准备好的时候阅读这些信息，但情况似乎并非如此。

Answer 1

实际上，管道是一个队列，但默认情况下它的长度设置为0。所以，当你推倒时，制片人会立即停下来等待。您可以使用set_size_budget函数来控制大小。