Anti-pipelining (TODO better name) (AKA receiving despite having agency)

[nfrisby]

I'm currently upgrading the LeiosNotify mini protocol in the leios-prototype branch of ouroboros-consensus be closer to what I've been intended with the design so far.

This PR was the result of me suddenly remembering yesterday that typed-protocols doesn't already allow what I had in mind (but I was confident it was feasible).

The key idea: my intention has been for the LeiosNotify server to maintain a buffer of messages it would like to send. The capacity of that buffer should be NumRequestsThatHaveArrived - NumRepliesAlreadySent. If its capacity ever grows beyond some limit, we disconnect: they sent too many requests without waiting for responses.

Implementing that in typed-protocols requires letting the Peer that has agency receive a message (which can only happen if the other Peer is Pipelined). In other words, the send can be non-blocking. This is opposite/dual of what YieldPipelined already does, in which a receive is non-blocking. Hence YieldAntiPipelined etc (but I'd be happy adopt suggestions for a better name).

A few comments:

• I suspect there is a deadlock risk if one peer is AntiPipelined and the other is not Pipelined. That seems mild enough, though, since that's a pretty severe mistake to have made.
• I suspect a Peer could be both Pipelined and AntiPipelined at the same time. But I don't need that, so I didn't pursue it.
• The "simpler" and more expressive version of this PR would have the YieldAntiPipelined ctor carry the Sender and the driver would maintain a queue of Senders instead of just the counter this PR uses. But I'm anticipating quite big LeiosNotify pipeline depths (and this is on the upstream side, so there can be hundreds of these AntiPipelined Peers running), so I instead required that every Sender is the same, so that the driver state is merely a counter instead of a queue. (The LeiosNotify client also shares that same large pipelining depth, and it's currently using a queue of (). That's not ideal and maybe worth an eventual PR, but it's at least only one per upstream peer (tens), rather than one per downstream peer (hundreds).)

Lastly, an admission: this all turned out roughly how I thought it would. But then it occurred to me that perhaps there's a simpler approach to LeiosNotify. Since we're bounding how many outstanding requests there are, we could just use a normal NonPipelined Peer for the server, and have its request handler read from a bounded queue that's not allowed to grow beyond that pipelining depth (and that queue basically exists outside of the Peer EDSL/driver).

I can only see a couple benefits to AntiPipelined.

• With the bounded queue approach, the peer could have sent an arbitrary amount of requests and we might never punish them for it. Those requests would just build up in the "bearer" that the NonPipelined Peer pulls from whenever it calls Await. But suppose we had no responses to send for a while (ie the queue was empty). Then the only thing stopping the downstream peer from filling up that bearer's buffer is a low-level size limit on that buffer itself. Maybe that's tolerable?
• The other benefit is that the back-pressure is more "precise". It's not obvious to me that that's important. But it has an appeal.
  • The only example I've thought of (just now): a syncing node won't send any LeiosNotify requests until it's caught-up/nearly caught-up. But once it does, (the naive/"simple" implementation of) the bounded queue approach will have a queue full of arbitrarily-old notifications to instantly pull from. That seems undesirable, especially since LeiosNotify clients are going to punish peers that send messages that are "too old". So the fact that AntiPipelined server Peer automatically has a queue capacity of 0 before any requests arrive is quite preferable---this is "precision" that it seems nice to be able to take for granted.

[nfrisby]

I don't have it all wired up, so I haven't even seen this running in a prototype yet. But, as demonstration, here's my current best guess at the definition of the AntiPipelined LeiosNotify server Peer.

leiosNotifyServerPeerAntiPipelined ::
  forall m point announcement vote.
  MonadThrow m =>
  m Bool ->
  -- ^ increments the number of outstanding requests
  m (Message (LeiosNotify point announcement vote) StBusy StIdle) ->
  -- ^ blocks until the next reply (announcement\/offer\/vote) is ready
  PeerAntiPipelined (LeiosNotify point announcement vote) AsServer StIdle m ()
leiosNotifyServerPeerAntiPipelined incr next =
    PeerAntiPipelined responder (go Zero)
  where
    responder :: Sender (LeiosNotify point announcement vote) AsServer StBusy StIdle m
    responder = SenderEffect $ next <&> \msg -> SenderYield ReflServerAgency msg SenderDone

    go :: forall n.
      Nat n ->
      Peer (LeiosNotify point announcement vote) AsServer (AntiPipelined StBusy StIdle n) StIdle m ()
    go n =
      Await ReflClientAgency $ \case
        MsgDone                         -> drain n
        MsgLeiosNotificationRequestNext ->
          Effect $ do
            incr >>= \b -> when b $ throwIO MkExnLeiosNotifyExcessiveRequests
            pure $ YieldAntiPipelined ReflServerAgency (go (Succ n))

    -- on termination, flush the sends we've handed off, then Done.
    drain :: forall n.
      Nat n ->
      Peer (LeiosNotify point announcement vote) AsServer (AntiPipelined StBusy StIdle n) StDone m ()
    drain = \case
      Zero   -> Done ReflNobodyAgency ()
      Succ j -> AntiCollect (drain j) Nothing

data ExnLeiosNotifyExcessiveRequests = MkExnLeiosNotifyExcessiveRequests
  deriving Show

instance Exception ExnLeiosNotifyExcessiveRequests

[nfrisby]

Oh, another thing: suppose the client has pipelined 300 requests and then also pipelines MsgDone.

This LeiosNotify server as-is cannot terminate until it generates 300 responses, which might take a while (if the protocol happens to be idle).

But: @coot has long been saying we need to add a "heart beat" message: an empty response that the server sends on some period of inactivity and that the client accepts if it's not sent too often (with a slightly asymmetric send rate and allowed receive rate for clockskew/transmission delay/buffering/etc).

If we were to add that, then the AntiPipelined server could just instantly dump one heart beat response for each outstanding request; and the client should allow them to arrive in such frequency because the client has already sent MsgDone.

(edit: this "prompt-drain" sounds quite useful, and applicable to other client-Pipelined mini protocols, eg ChainSync)

[nfrisby]

Ah, another possible reason to consider AntiPipelining unnecessary:

just use a NonPipelined server and whenever it's processing a request it queries the state in order to find the best response it can send and sends that

Sounds nice and simple, but the query can be non-trivial to define, might be expensive, and require maintaining extra state. Even worse: it might require maintaining extra state per peer---recall again that one LeiosNotify server runs per downstream peer and there can be hundreds of those. So I'm trying to avoid this.

It's worth emphasizing that all of our existing mini protocol servers do this. However, my intuition (which I admittedly haven't already unpacked) is that LeiosNotify is less compatible with it than the Praos mini protocols are: ChainSync and BlockFetch both have very very simple server logic (the thing we run hundreds of). I think that was also true for old TxSubmission; I'm not sure about new TxSubmission (b/c it has centralized decision logic now).

[coot]

Have you tried to extend proofs so Pipelined and AntiPiplined peers can run against each other? That's the fun/challenging part of typed-protocols 😁

[coot]

Why not require Sender here, as the Receiver is passed in YeildPipelined?

That's probably why you said in the PR description that all Senders in your use cases are the same.

[nfrisby]

Correct: yeah, excluding the Sender here means we don't need the driver to maintain a queue.

[coot]

The AntiPipelined Peer allows to handle responses in parallel, something that is common in http servers where each requests are handled by a pool of threads. There's no chance for a deadlock, unless the Sender is observing the requests, and only start replying if there were enough of them. But we should be able to provide a function:

forgetAnitPipelined :: [Bool]
                    -> PeerAntiPipelined ps pr st m a
                    -> Peer ps pr m a

which inlines the Sender, in similar way that the forgetPipelined inlines the Reciever.

[nfrisby]

You posted a top-level question here about "proofs". #92 (review)

• I'm posting it here (arbitrary) so we can have a threaded conversation about it.
• Do you mean the Agda? I seem to recall some (previous version?) Haskell type classes whose methods were "proofs", and i was expecting GHC to force me to incorporate AntiPipelined into them at some point, but it never did.
• I would love to be paid to work with Agda, but I never have been; so I'm probably not the most efficient person to tackle these proofs. (... take it with a grain of salt, but my intuition is that it should be simple/very similar to the existing Pipelined proof).

[coot]

No, in our most loved theorem prover - Haskell 😉. If we implement forgetAntiPielined then we can have a connectAntiPipelined similar to connectPipeliend

[coot]

Do you mean the Agda? I seem to recall some type classes whose methods were "proofs", and i was expecting GHC to force me to incorporate AntiPipelined into them at some point, but it never did.

That's because you extended the type and guarded it at the type level with AniPipelined constructor.

[coot]

🤔 if our proofs where polymorphic in IsPipelined argument, then it would force you for write a proof, but connect requires NonPipelined peers, and connectPipelined requires Pipelined ones.

[coot]

What do you think about these names:

• YieldAntiPipelined -> YieldNonBlocking
• CollectAntiPipelined -> CollectSent or Flush (both suggested by claude)

Please also add pattern synonyms in Network.TypedProtocols.Peer.Client and Network.TypedProtocols.Peer.Server modules.