We have added new logic for DVT in this PR:
One of the goals was to keep the non-DVT code as similar as possible to its previous incarnation. However, this has lead to quite a bit of duplication. This issue is to track the de-duplication of this logic, by harmonising the distributed and non-distributed cases. E.g. by using FuturesUnordered for both.