Allow task-based subscription processing via the subscription manager

[fselmo]

What was wrong?

Closes #3672

How was it fixed?

• For high-throughput subscriptions, where order of processing is not important, we should allow a configuration to be set to process tasks in a task-based manner - creating a task for each subscription and processing them in parallel.

• Add a parallelize flag to the subscription manager, defaulting to False, but allowing this to be set to True to enable task-based, parallelized processing.

• This should be able to be controlled on a per-subscription basis as well. If sub-second blocks need to be processed in order but, say, processing some information about transactions in the mempool (newPendingTransactions) does not, this should be able to be accounted for as well. Adds a parallelize flag to subscriptions which overrides whatever the manager setting is for that particular subscription - defaults to None which means the manager controls the global processing state.

Todo:

• Clean up commit history
• Add or update documentation related to these changes
• Add entry to the release notes

Cute Animal Picture

[kclowes]

LGTM! 🚀 (after tests pass :) )

[fselmo]

I cleaned this up a bit but I while I think this mostly keeps FIFO, I think the correct implementation here is setting off a task for each subscription to process its own loop in the background but to keep messages in order for that subscription. I think for most tasks this won't matter but either way it would be the right way to do this.

I'm going to put this away for a little bit as there are more pressing tasks ahead of this one but I already have a clean idea for how to handle that with the current manager structure.

---

also, rip py38 🙁

[fselmo]

OK nevermind. I thought about it some more and I think this maybe is the only way to go. I don't see enough subscriptions on a single web3 instance to where they each need their own queue. That feels like over-engineering for a niche case that I don't even think exists yet today. If a user is subscribing across many chains they will have many different web3 instances so they can parallelize each instance and they should be within their own contexts with their own managers.

I do think it's useful to parallelize on a per-subscription basis as you may want to parallelize newPendingTransactions, for example, but not necessarily newHeads. Something like newHeads on a chain that does sub-second blocks would matter and you likely want to process blocks in order.

So instead, I built on the manager configuration as a global flag and allowed per-subscription management as well. I added some good tests for this but it's till missing the docs. I will try to update the docs soon but marking this Ready for review so the code can be reviewed for now.

[fubuloubu]

This PR looks good (will try to test it early next week if I can), however I'm a bit confused about the disclaimer concerning OOO processing of subscription data. Shouldn't it be fairly transparent based on the behavior of the node, regardless of whether its processed sequentially or thrown into a background task queue?

At least in my experience the data flows in based on the order the node provides it. I don't see anything in this update that would modify that behavior, so I want to make sure I'm not missing something about the change.

Is the disclaimer more that you cannot guarantee the order in which the background tasks get executed/completed once the callback is made on new message data? If so, that's okay for our use case, because we are using the callback to populate a message queue which maintains ordered execution.

[fselmo]

Is the disclaimer more that you cannot guarantee the order in which the background tasks get executed/completed once the callback is made on new message data?

Yeah, I just mean that you can't guarantee the order of when the task processing finishes. They will still all start in the order they were received, but web3.py can't guarantee the amount of logic you're throwing at each task in the handler. So let's say you are trying to keep track of balances and putting this data in a database, if you end up processing a future block faster than the previous block because you had to do more calculations on that block, you can run into a race condition where your shared state is not being updated sequentially, which could lead to some miscalculations.

That's just one example but it really depends on the conditionals defined in the handler of the task. If tasks are running through the same logic, they'll still all likely be processed in order, it's just no longer a guarantee when you're yielding control across tasks.

Does that make sense? The documentation for this change should be quite clear on this and I haven't started on that yet. But, for this reason, the default setting should be that all processing is done within the main loop imo.

[fubuloubu]

Is the disclaimer more that you cannot guarantee the order in which the background tasks get executed/completed once the callback is made on new message data?

Yeah, I just mean that you can't guarantee the order of when the task processing finishes. They will still all start in the order they were received, but web3.py can't guarantee the amount of logic you're throwing at each task in the handler. So let's say you are trying to keep track of balances and putting this data in a database, if you end up processing a future block faster than the previous block because you had to do more calculations on that block, you can run into a race condition where your shared state is not being updated sequentially, which could lead to some miscalculations.

That's just one example but it really depends on the conditionals defined in the handler of the task. If tasks are running through the same logic, they'll still all likely be processed in order, it's just no longer a guarantee when you're yielding control across tasks.

Does that make sense? The documentation for this change should be quite clear on this and I haven't started on that yet. But, for this reason, the default setting should be that all processing is done within the main loop imo.

Okay, yeah that makes sense, I agree with the default behevaior being essentially synchronous, and this feature having the disclaimer that even though the subscriptions loop is single-threaded/single-process, it does pay to understand OOO execution across coroutines may affect when the tasks complete with parallelize=True