title: Trying to create an asynchronous example using Core.Async.Pipe, but Pipe.write seems to block awaiting a Pipe.read
body: <p>I've created the following toy example that counts in a loop and writes the value to an Async.Pipe:</p>
  
  <pre><code>open Sys
  open Unix
  open Async.Std
  let (r,w) = Pipe.create ()
  
  let rec readloop r = 
    Pipe.read r &gt;&gt;=
    function
    | `Eof -&gt; return ()
    | `Ok v -&gt; return (printf "Got %d\n" v) &gt;&gt;=
    fun () -&gt; after (Core.Time.Span.of_sec 0.5) &gt;&gt;=
    fun () -&gt; readloop r 
  
  let countup hi w =
    let rec loop i = 
      printf "i=%d\n" i ;
      if (i &lt; hi &amp;&amp;( not (Pipe.is_closed w))) then 
         Pipe.write w i &gt;&gt;&gt;
         fun () -&gt; loop (i+1)
       else Pipe.close w
    in 
    loop 0 
  
  let () =
    countup 10 w;
    ignore(readloop r);;
    Core.Never_returns.never_returns (Scheduler.go ()) 
  </code></pre>
  
  <p>Notice the <em>readloop</em> function is recursive - it just continuously reads values from the Pipe as they are available. However, I've added a delay there of 0.5 sec between each read. The <em>countup</em> function is kind of similar but it loops and does a write to the same Pipe.</p>
  
  <p>When I run this I get:</p>
  
  <pre><code>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
  </code></pre>
  
  <p>Aside from the first three lines of output above, all the rest of the output lines seem to need to wait the half second. So it seems that the Pipe is blocked after a write until there is a read from the Pipe. (<em>Pipe.write w data</em> appears to block waiting for a <em>Pipe.read r</em> )
  What I thought should happen (since this is an Async Pipe of some sort) is that values would be queued up in the Pipe until the reads take place, something like:</p>
  
  <pre><code>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
  </code></pre>
  
  <p>I'm wondering if there is a way to get the behavior using Async?</p>
  
  <p>My real usecase is that I've got a Tcp socket open (as a client) and if I were using threads after some setup between the client and the server I would start a thread that just sits and reads data coming in from the socket from the server and put that data into a queue of messages that can be examined in the main thread of the program when it's ready. However, instead of using threads I want to use Core.Async to achieve the same thing: Read data from the socket as it comes in from the server and when data is available, examine the message and do something based on it's content. There could be other things going on as well, so this is simulated by the "wait half a second" in the code above. I thought Pipe would queue up the messages so that they could be read when the reader side was ready, but that doesn't seem to be the case.</p>