txHandler: Random Early Detection for backlog queue #4797
Conversation
This structure has not been tested in any way, it is commited for posterity as I work.
Codecov Report
@@ Coverage Diff @@
## master #4797 +/- ##
==========================================
+ Coverage 53.49% 53.56% +0.07%
==========================================
Files 432 433 +1
Lines 53624 53864 +240
==========================================
+ Hits 28687 28854 +167
- Misses 22708 22766 +58
- Partials 2229 2244 +15
📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more |
AlgoAxel
changed the title
AlgoAxel
changed the title
| ts := []time.Time{} | ||
| consumedByClient[e.c] = &ts | ||
| } | ||
| *(consumedByClient[e.c]) = append(*(consumedByClient[e.c]), e.t) |
There was a problem hiding this comment.
what is the upper bound for this time stamp arrays?
There was a problem hiding this comment.
Strictly, no upper bound, but the length of these lists will be proportional to the service rate:
- client submits
backlogSizetransactions, filling the backlog - txHandler is magic and can instantly dequeue all requests
- client fills the backlog again, repeat
However many times this can happen within the rolling window (10s) is how long the list could get, theoretically. Same answer for all-clients, since they'd share the ability to fill the backlog.
And obviously, window size is directly proportional to this as well, since a 20s window affords twice as much time to load and unload the backlog.
* package level `var` for common Errors * pruning optimizations * closeReservation unit tests * pointer to interface replaced by interface itself
algorandskiy
changed the title
| if err != nil { | ||
| handler.erl.EnableCongestionControl() |
There was a problem hiding this comment.
It looks strange to read that an error happened, and the response is to enable something.
It seems that the err is not necessarily an unexpected situation, but is a control parameter. ConsumeCapacity does not describe in the comments what errors are returned, and need digging into the internal functions to understand what is expected.
Either add a comment to describe what err entails, or use a different control parameter.
There was a problem hiding this comment.
That's a fair point -- using an error as a control is uncommon. But I think it makes sense here, so it does deserve a comment.
The current strategy for TX Handler to use the ERL is that it asks for capacity, and while it is available, everything is fine. However, if there should ever be an issue granting capacity, then TX Handler decides that the ERL should use congestion management.
That "Issue Granting Capacity" indicates that at the moment, the underlying TX Backlog is full. It is indeed an error, but it's an error we encounter any time the inputs overwhelm this handler, so it's not exactly "unexpected" like you mention. Pre-ERL this error was/is just as likely, but it happens lower down, when the TXHandler attempts to put the item onto the backlog and fails (and no sort of function calling happens inside the condition).
Another way of putting this -- enablement of congestion control is indeed the response to an error condition -- if the handler is being overwhelmed, that's an error. And if it's generating errors for being overwhelmed, that's our signal to turn on defensive measures (congestion management)
There was a problem hiding this comment.
That's a fair point -- using an error as a control is uncommon. But I think it makes sense here, so it does deserve a comment.
That is all I need :-)
| if wi.capguard != nil { | ||
| if err := wi.capguard.Release(); err != nil { | ||
| logging.Base().Warnf("Failed to release capacity to ElasticRateLimiter: %v", err) | ||
| } | ||
| } | ||
| if handler.checkAlreadyCommitted(wi) { | ||
| transactionMessagesAlreadyCommitted.Inc(nil) | ||
| if wi.capguard != nil { | ||
| wi.capguard.Served() | ||
| } | ||
| continue | ||
| } | ||
|
|
||
| // enqueue the task to the verification pool. | ||
| handler.txVerificationPool.EnqueueBacklog(handler.ctx, handler.asyncVerifySignature, wi, nil) | ||
| if wi.capguard != nil { | ||
| wi.capguard.Served() | ||
| } |
There was a problem hiding this comment.
Several points here:
- Served and Release both return err. One is checked, others are not. Why return an error if it is not important?
- When Release returns an error, it is of no consequence here. The logging of the warning can be inside Release.
- Why is it necessary to have 3 calls here? It is complicating the backlog code so much. There has to be another messaging mechanism.
There was a problem hiding this comment.
- That's a good point,
Servedprobably does not need to return an error. Served is what gets used to calculate Service Rates for theShouldDrop, so if data doesn't get to it, the math is a little wrong (which is okay). By contrast,Releaseis what puts capacity back into the ERL. If it fails, it means we're leaking capacity and in theory it would impact the txHandler over time.
EDIT: actually, Served() does not return an error, so as far as I can tell I am not dropping any errors on the ground here. The error check for Release is the only one
-
It can be moved, but I'd prefer to address this at a separate point from these other comments and the unit test fix that is ongoing. If we want I can cut a task to add logging to the ERL (it currently has no logging), which should be real easy.
-
Not sure what to tell you here, this is where request servicing happens, so it's the most appropriate spot to invoke the Return/Serve. I'll explain why they're needed where they are though:
-
wi.capguard.Release()on line 258 -- the message has been pulled off the backlog, meaning there is now more room in the backlog to service requests. Therefore, the capguard that is attached to this message should go and return itself. There's really no other place to put this. -
wi.capguard.Served()on line 265 -- the message is already committed, no further work is needed here. Since this request is now done, we callServed()to indicate our worker completed this work. -
wi.capguard.Served()on line 273 -- we have enqueued the work to the backlog, hooray! Since the request is now done, we callServed()to indicate this. We don't defer, since work done after this point doesn't count toward the time handling the request.
So all in all, I think this is right -- when the request is dequeued from the protected resource, the capacity is available and gets returned. When the handler fully "handles" the item, it gets served. The nil check on the capguard is needed since this whole feature is configurable on/off. Not having an ERL means you wouldn't have a capguard.
There was a problem hiding this comment.
(1) I must be mistaken here. I don't know what I was thinking...
for (2 & 3), here is my suggestion. It will only add a single line for handling capguard.
If will clean up more than 50% of the code in this segment: 18 lines -> 8 lines
| if wi.capguard != nil { | |
| if err := wi.capguard.Release(); err != nil { | |
| logging.Base().Warnf("Failed to release capacity to ElasticRateLimiter: %v", err) | |
| } | |
| } | |
| if handler.checkAlreadyCommitted(wi) { | |
| transactionMessagesAlreadyCommitted.Inc(nil) | |
| if wi.capguard != nil { | |
| wi.capguard.Served() | |
| } | |
| continue | |
| } | |
| // enqueue the task to the verification pool. | |
| handler.txVerificationPool.EnqueueBacklog(handler.ctx, handler.asyncVerifySignature, wi, nil) | |
| if wi.capguard != nil { | |
| wi.capguard.Served() | |
| } | |
| ac := handler.checkAlreadyCommitted(wi) | |
| handleCapguard(wi, ac) | |
| if ac { | |
| transactionMessagesAlreadyCommitted.Inc(nil) | |
| continue | |
| } | |
| // enqueue the task to the verification pool. | |
| handler.txVerificationPool.EnqueueBacklog(handler.ctx, handler.asyncVerifySignature, wi, nil) |
Then you can add this function wherever you please, outside txHandler.go
func handleCapguard(wi, alreadyCommitted) {
if wi.capguard != nil {
if err := wi.capguard.Release(); err != nil {
logging.Base().Warnf("Failed to release capacity to ElasticRateLimiter: %v", err)
}
}
if alreadyCommitted {
if wi.capguard != nil {
wi.capguard.Served()
}
}
}
| var capguard *util.ErlCapacityGuard | ||
| if handler.erl != nil { | ||
| // consume a capacity unit | ||
| capguard, err = handler.erl.ConsumeCapacity(rawmsg.Sender.(util.ErlClient)) |
There was a problem hiding this comment.
What makes you sure this will always return a valid pointer? rawmsg.Sender.(util.ErlClient)
There was a problem hiding this comment.
rawmsg.Sender is a Peer interface, and my understanding is that the only in-use implementation of a Peer is the wsPeer.
wsPeer satisfies ErlClient by implementing OnClose.
Since Peer is just an empty interface, it makes it hard to say I'm "sure" it will always be a valid pointer.
This looks like it might be a go pattern I have yet to learn(?)
There was a problem hiding this comment.
Since Peer is just an empty interface, it makes it hard to say I'm "sure" it will always be a valid pointer.
Yes, and we should not be adding more into this.
It is really not worth taking a chance here.
| capguard, err = handler.erl.ConsumeCapacity(rawmsg.Sender.(util.ErlClient)) | |
| capguard, err = handler.erl.ConsumeCapacity(rawmsg.Sender) |
Then, inside consume capacity:
ConsumeCapacity(p Peer) {
erlclient, ok := p.(util.ErlClient)
if !ok {
return nil, error("unsupported peer")
}
...
}
Addresses a flaky Unit Test by not overloading the component under test (previously, 5k messages were driven as-fast-as-possible. Now, 10k messages are driven in batches of 500 with a 100ms delay). The new behavior is more accurate to a realistic usage pattern, as well. This also addresses some deletable code and adds comments where requested from some post-merge requests on pull algorand#4797.
What
This PR introduces the following features and changes:
util, meant to manage mutli-client access to a single resource.OnClosefunctionality so the ERL can effectively close "Reservations" for peers when they disconnectThis Elastic Rate Limiter has been attached in front of TXHandler's adding to the Mempool.
Why
Today, peers send messages which are enqueued to our Mempool, with no regard for maximum bandwidth of the reciever. It is possible for a noisy peer to create difficulty for all peers adding to the Mempool. We would rather balance and selectively constrain peers based on their dominance in the messaging queue, and the service's ability to handle the load.
How
In TX Handler:
TX Handler now has an
erl, Elastic Rate Limiter, and a Random Early Detection Congestion Managercminside it. When a new request comes, the TX Handler will receive from the ERL for a "capacityGuard" object, which is then attached to the message being enqueued. When dequeued, theReleaseandServedare called on the guard, restoring capacity to the ERL, and notifying the congestion manager that the request has been servicedIn the Elastic Rate Limiter:
closeReservationto the client'sOnCloseIn the Congestion Manager:
RED Congestion manager implements the CongestionManager interface.
run()loop listens for newconsumed,servedandshouldDropQuerymessagesNticks, a new Target Service Rate is calculated using a sliding window of arrival and servicesThe "Random Early Detection" Congestion Management Algorithm
It was requested I write a short explainer for posterity, detailing the behavior of the congestion management algorithm. Wikipedia gives a good overview of how this works, but here's how it works specific to the txBacklog:
When a capacity unit is
consumed, it records the time in a slice dedicated to consumption by that client.When a capacity unit is
served, it records the time in a slice for all serves (serves are not tracked by client because it is not immediately useful to do so, but it could in the future)All consumption times and serve times are trimmed to a rolling window size (default 10s) and recalculated on an as-needed basis when a
shouldDroprequest arrives, or when the target service rate is being calculatedTo minimize excess computation, a service rate is calculated every N ticks (default is equal to 1/s at 100 requests per second). The service rate is equal to the
total number of serves/number of clients/window size. You can think of this as "Serves per client per second"When the Congestion manager is queried if it
shouldDrop:arrival rate.target service rate, and compare them as a ratio.Testing
Unit Tests
Unit tests have been written for the Elastic Rate Limiter itself, confirming it can safely vend and receive capacity.
Unit tests have also been written for the RED congestion manager
Manual Tests
Ran standard cluster tests, and originally observed a decrease in overall TPS, and a TPS over time which looked consistent with congestion control being active, and perhaps too punishing for the average case. I then introduced an exponential curve to the shouldDrop calculation, which mitigated most of the TPS reduction, which reflected in a less aggressive looking congestion management in the TPS graph.
Update: Since adding some fixes in the congestion management
served()reporting, there is no longer an impact on TPS. The rate limiting was likely being applied too often due to the misreporting of served events, which ironically shows us that the congestion management does what it's supposed to when it's active. However, with the correct served event reporting, TPS is unimpeded.Correctness test
Two clients sending in 15 sec, the second overflows the backlog in few threads.
client 1 sends 100k txns w/o dealay
client 1 with the 50us delay, runtime 15sec