Applying backpressure to a flow by blocking in the MessageEventHandler

Ah ok, so if I understand you correctly that's the setting for the buffered-ack timer inside the solace library code (that actually acks back to the VPN), and not related to the time between a client receiving a message and calling IFlow.Ack(). Probably a good one to leave as default I expect.

For technical completeness, as a slight improvement to the above sample code (given that posting to the channel is expected to work most of the time), I decided to modify the implementation as below, but the question itself is unchanged:

messageEventHandler: (_, args) =>
{
   // deserialization to a known msg type omitted
   if (!buffer.Writer.TryWrite(msg))
   {
      buffer.Writer.WriteAsync(msg).AsTask().Wait();
   }
}

Thanks Aaron, great info.

Yeah, the subscriber wrapper exposes the broker as a pair of custom interfaces, IProducer (of messages) and an IConsumer (of acknowledgements), with the idea being that the devs just wire up a processing pipeline between those two endpoints... probably using a weak-and-linear-but-hey-its-easy-to-use fluent wrapper I wrote for System.Threading.Tasks.Dataflow. If they want branching they can write their own damn wrapper

I do have a convenient point to call stop (basically, just after the try_write in my example code above), detecting that it is safe to restart is not quite as easy, but not very very difficult.

On the buffer, yeah, I have the default size configured to be max unacked + receive window size, because that just makes everything easy, no need to stop or start anything or worry about backpressure since the VPN will "self-backpressure" while waiting for acks. Right now I do expose the ability for a subscriber to change that buffer size in the constructor of the wrapper implementation, but I think I might just remove that ability to make my life easier Then I don't ever have to worry about manipulating the flow.

Thanks again for going deep on the implementation details! Always a pleasure.

Haha, that description is probably good for any internal marketing hype I want to create, but the implementation is rather more modest XD

I was thinking that the best place to restart the flow (if I go with that model at all) would be inside the acknowledgement wrapper. My reasoning here is that from the client's point of view, they don't really care about messages in flight on the network or the internal constraints of the system. What they care about when it comes to buffering is the amount of memory they're using. Being able to specify a known maximum capacity for their processing pipeline "as a whole" is all they're really going to care about, and since messages effectively leave their memory space at acknowledgement time (with the small caveat of some timed buffering of acknowledgement inside the solace library itself, but that's just a set of long for the ADMessageId right? So, pretty small), that's the specification they might want to provide. The slight pain here is more about the inability to read the current count inside a Channel, since it's more like an IEnumerable than an IList, but I could just switch to a ConcurrentQueue, or Interlocked.Increment / Interlocked.Decrement on a private field.

The "subscriber" object is a per-topic entity in my model, and encapsulates a flow. That is to say, the flow is instantiated at construction time, so I do have the option of setting the max unacked messages in the ISession.CreateFlow call in the subscription constructor, so that works - if I decide not to just hardcode it as you mentioned. Perhaps there is still some benefit it having both controls in the implementation - ie, stopping and starting flows as well as letting the unacked message count provide flow control.

Edit:

Thinking about that a little more, there's an off-by-one error in my thoughts there, because if the flow is stopped only when the buffer refuses a message, then that message is effectively lost to the ether. So the flow has to be stopped at the message that fills the buffer, not at the message that would overflow the buffer. But in that case the flow could be "bouncing off the limiter" a lot, which would be inefficient. So it does seem sensible to make the buffer size one larger than max unacked plus ad window size. But then it follows that stopping the flow isn't really needed, since the flow would already be throttled by the broker. So explicit flow control only seems necessary if the buffer is too small, and since I have control over all of the parameters, that shouldn't be a problem.

Hehe, I actually ended up with almost exactly the same logic a few days ago. Here, _carrier is a thin wrapper around bits of the IFlow. The MessageEventHandler forwards to:

_carrier.ItemAvailable += (_, message) =>
{
   if (!_buffer.Writer.TryWrite(message)) throw // ...
   if (_buffer.Reader.Count > maxUnackedMessageCount) // > rather than >= since the flow will typically have taken over anyway
   {
      lock (_carrier)
      {
         if (!_paused)
         {
            _carrier.Stop();
            _paused = true;
            _logger?.LogWarning("Backpressure on {subscription}", this);
         }
      }
   }
};

Then on the other side, the IFlow.Ack is wrapped like this:

public async Task AcknowledgeAsync(T payload, CancellationToken ct)
{
   if (_buffer.Reader.Count < _resumeAtCount)
   {
      lock (_carrier)
      {
         if (_paused)
         {
            _carrier.Start();
            _paused = false;
         }
      }
   }
   await _ackRetryPolicy.RetryForever(() => _carrier.TryAck(payload), ct);
}

There is some inefficiency here, in that full locks are sometimes taken unnecessarily. However, that only happens either when we actually need to stop the flow anyway or the flow is already stopped (in the message event handler), or when there no real penalty because the message rate is low, since I have resumeAtCount set to 10% of whatever the max unacked count is. Enough to reduce periodic latencies, but not so much that the condition is regularly hit except when there is no performance issue anyway.

I also set the buffer size to double the max unacked count to be completely safe even if messages dribble in after the flow is stopped, and of course I actually do set the max unacked on the flow, so typically the flow control is really being done by the Solace IFlow anyway (hence the name, I suppose!)

There can theoretically be a race for the _buffer.Reader.Count, and as you mentioned keeping the max and min threshold apart reduces the likeliness. But the way I figure, even if things happen out of the expected order, it can never result in a long term problem since the buffer count eventually becomes stable no matter what.