Allow TCPConnectionNotify to cause connection to yield while receiving (ponylang#1355)

This comes from direct experience using the existing `TCPConnection`
functionality at Sendence. We are heavy users of `expect` on `TCPConnection` in
order to support framed protocols. Our `received` methods on notifiers are
generally of the following form:

```pony
  fun ref received(conn: TCPConnection ref, data: Array[U8] iso) =>
    if _header then
      // convert the 4 byte header into a value for expect, aka payload length
      let expect = Bytes.to_u32(data(0), data(1), data(2), data(3)).usize()

      conn.expect(expect)
      _header = false
    else
      // do something with payload
      ...

      // reset expect for next 4 byte header
        conn.expect(4)
        _header = true
    end
```

This short of usage is why `expect` was initially added to `TCPConnection`.
Upon usage, we found a serious drawback with this approach. `TCPConnection`
will read up to 4k of data on a single behavior run and if there is still data
available, it will then send itself a `_read_again` message to trigger more
reading of additional data. It does this so that it doesn't hogged the
scheduler while reading from the socket. This can work reasonably well in some
scenarios but not others.

In the framed protocol example above, if the message payloads are small then
4k of data can result in a lot of messages being sent from our `received`
method to other actors in the application. In an application that is
continously receiving data, this results in a very bursty scheduling experience.

After consulting with Sylvan, we changed `received` and `TCPConnection` to
allow `received` to return a Boolean to indicate whether `TCPConnection`
should continue sending more data on this behavior run.

We've found that for some workloads, we are able to get equal performance
while greatly lowering latency by having `TCPConnection` call `_read_again`
earlier than it otherwise would.

Closes RFC ponylang#19
Closes ponylang#1343

Author: SeanTAllen

CHANGELOG.md

@@ -11,6 +11,7 @@ All notable changes to the Pony compiler and standard library will be documented
 ### Added
 
 - TCP read and write backpressure hooks in `TCPConnection` (issue #1311)
+- Allow TCP notifiers to cause connections to yield while receiving (issue #1343)
 
 ### Changed
 

examples/echo/echo.pony

@@ -41,10 +41,11 @@ class Server is TCPConnectionNotify
   fun ref accepted(conn: TCPConnection ref) =>
     _out.print("connection accepted")
 
-  fun ref received(conn: TCPConnection ref, data: Array[U8] iso) =>
+  fun ref received(conn: TCPConnection ref, data: Array[U8] iso): Bool =>
     _out.print("data received, looping it back")
     conn.write("server says: ")
     conn.write(consume data)
+    true
 
   fun ref closed(conn: TCPConnection ref) =>
     _out.print("server closed")

examples/net/client.pony

@@ -21,8 +21,9 @@ class ClientSide is TCPConnectionNotify
   fun ref connect_failed(conn: TCPConnection ref) =>
     _env.out.print("connect failed")
 
-  fun ref received(conn: TCPConnection ref, data: Array[U8] iso) =>
+  fun ref received(conn: TCPConnection ref, data: Array[U8] iso): Bool =>
     _env.out.print(consume data)
+    true
 
   fun ref closed(conn: TCPConnection ref) =>
     _env.out.print("client closed")

examples/net/server.pony

@@ -13,9 +13,10 @@ class ServerSide is TCPConnectionNotify
       conn.write("server says hi")
     end
 
-  fun ref received(conn: TCPConnection ref, data: Array[U8] iso) =>
+  fun ref received(conn: TCPConnection ref, data: Array[U8] iso): Bool =>
     _env.out.print(consume data)
     conn.dispose()
+    true
 
   fun ref closed(conn: TCPConnection ref) =>
     _env.out.print("server closed")

packages/net/_test.pony

@@ -225,7 +225,7 @@ class _TestTCPExpectNotify is TCPConnectionNotify
     _h.complete_action("expect received")
     qty
 
-  fun ref received(conn: TCPConnection ref, data: Array[U8] val) =>
+  fun ref received(conn: TCPConnection ref, data: Array[U8] val): Bool =>
     if _frame then
       _frame = false
       _expect = 0
@@ -250,6 +250,7 @@ class _TestTCPExpectNotify is TCPConnectionNotify
     end
 
     conn.expect(_expect)
+    true
 
   fun ref _send(conn: TCPConnection ref, data: String) =>
     let len = data.size()
@@ -301,7 +302,7 @@ class _TestTCPWritevNotifyServer is TCPConnectionNotify
   new iso create(h: TestHelper) =>
     _h = h
 
-  fun ref received(conn: TCPConnection ref, data: Array[U8] iso) =>
+  fun ref received(conn: TCPConnection ref, data: Array[U8] iso): Bool =>
     _buffer.append(consume data)
 
     let expected = "hello, hello (from client)"
@@ -311,6 +312,7 @@ class _TestTCPWritevNotifyServer is TCPConnectionNotify
       _h.assert_eq[String](expected, consume buffer)
       _h.complete_action("server receive")
     end
+    true
 
 class iso _TestTCPMute is UnitTest
   """
@@ -360,8 +362,9 @@ class _TestTCPMuteReceiveNotify is TCPConnectionNotify
     _h.complete_action("receiver asks for data")
     _h.dispose_when_done(conn)
 
-  fun ref received(conn: TCPConnection ref, data: Array[U8] val) =>
+  fun ref received(conn: TCPConnection ref, data: Array[U8] val): Bool =>
     _h.complete(false)
+    true
 
 class _TestTCPMuteSendNotify is TCPConnectionNotify
   """
@@ -383,9 +386,10 @@ class _TestTCPMuteSendNotify is TCPConnectionNotify
   fun ref connect_failed(conn: TCPConnection ref) =>
     _h.fail_action("sender connected")
 
-   fun ref received(conn: TCPConnection ref, data: Array[U8] val) =>
+   fun ref received(conn: TCPConnection ref, data: Array[U8] val): Bool =>
      conn.write("it's sad that you won't ever read this")
      _h.complete_action("sender sent data")
+     true
 
 class iso _TestTCPUnmute is UnitTest
   """
@@ -434,8 +438,9 @@ class _TestTCPUnmuteReceiveNotify is TCPConnectionNotify
     conn.unmute()
     _h.complete_action("receiver unmuted")
 
-  fun ref received(conn: TCPConnection ref, data: Array[U8] val) =>
+  fun ref received(conn: TCPConnection ref, data: Array[U8] val): Bool =>
     _h.complete(true)
+    true
 
 class iso _TestTCPThrottle is UnitTest
   """
@@ -505,9 +510,10 @@ class _TestTCPThrottleSendNotify is TCPConnectionNotify
   fun ref connect_failed(conn: TCPConnection ref) =>
     _h.fail_action("sender connected")
 
-  fun ref received(conn: TCPConnection ref, data: Array[U8] val) =>
+  fun ref received(conn: TCPConnection ref, data: Array[U8] val): Bool =>
     conn.write("it's sad that you won't ever read this")
     _h.complete_action("sender sent data")
+    true
 
   fun ref throttled(conn: TCPConnection ref) =>
     _throttled_yet = true

packages/net/http/_request_builder.pony

@@ -34,7 +34,7 @@ class _RequestBuilder is TCPConnectionNotify
 
     _server = _ServerConnection(_handler, _logger, conn, host)
 
-  fun ref received(conn: TCPConnection ref, data: Array[U8] iso) =>
+  fun ref received(conn: TCPConnection ref, data: Array[U8] iso): Bool =>
     """
     Assemble chunks of data into a request. When we have a whole request,
     dispatch it.
@@ -56,3 +56,4 @@ class _RequestBuilder is TCPConnectionNotify
         break
       end
     end
+    true

packages/net/http/_response_builder.pony

@@ -33,14 +33,15 @@ class _ResponseBuilder is TCPConnectionNotify
     """
     _client._auth_failed(conn)
 
-  fun ref received(conn: TCPConnection ref, data: Array[U8] iso) =>
+  fun ref received(conn: TCPConnection ref, data: Array[U8] iso): Bool =>
     """
     Assemble chunks of data into a response. When we have a whole response,
     give it to the client and start a new one.
     """
     // TODO: inactivity timer
     _buffer.append(consume data)
     _dispatch(conn)
+    true
 
   fun ref closed(conn: TCPConnection ref) =>
     """

packages/net/ssl/ssl_connection.pony

@@ -69,13 +69,14 @@ class SSLConnection is TCPConnectionNotify
 
   recover val Array[ByteSeq] end
 
-  fun ref received(conn: TCPConnection ref, data: Array[U8] iso) =>
+  fun ref received(conn: TCPConnection ref, data: Array[U8] iso): Bool =>
     """
     Pass the data to the SSL session and check for both new application data
     and new destination data.
     """
     _ssl.receive(consume data)
     _poll(conn)
+    true
 
   fun ref expect(conn: TCPConnection ref, qty: USize): USize =>
     """

packages/net/tcp_connection.pony

@@ -616,8 +616,13 @@ actor TCPConnection
             data.truncate(_read_len)
             _read_len = 0
 
-            _notify.received(this, consume data)
-            _read_buf_size()
+            if not _notify.received(this, consume data) then
+              _read_buf_size()
+              _read_again()
+              return
+            else
+              _read_buf_size()
+            end
           end
 
           sum = sum + len

packages/net/tcp_connection_notify.pony

@@ -58,11 +58,14 @@ interface TCPConnectionNotify
     """
     data
 
-  fun ref received(conn: TCPConnection ref, data: Array[U8] iso) =>
+  fun ref received(conn: TCPConnection ref, data: Array[U8] iso): Bool =>
     """
-    Called when new data is received on the connection.
+    Called when new data is received on the connection. Return true if you
+    want to continue receiving messages without yielding until you read
+    max_size on the TCPConnection. Return false to cause the TCPConnection
+    to yield now.
     """
-    None
+    true
 
   fun ref expect(conn: TCPConnection ref, qty: USize): USize =>
     """