Asynchronous Data Movement in Numaflow Core (Source, Map, Sink)

[vigith]

Summary

The current implementation of Numaflow expects the data read from the (N-1)th vertex to be forwarded to the (N)th vertex as a batch. We need to make this asynchronous and out of order sending so that the slower processing items in the read batch will not hold up resources.

The main aim is to make sure we always have X elements being processed at anytime

Use Cases

This is quintessential for ML workloads or any other workload where we have varying distribution of processing latency (in other words, not a good gaussian distribution on processing latency)

CAVEATS

• Watermark needs to be implemented via min-heap
• Reduce will be tracked via a followup issue

Tasks

Preview Give feedback

1. Source should be able to do out of order writes to Sink #2250
   +0
   
2. single ack for flatmap on transform
3. feat: Conditional Forwarding In Asynchronous Pipeline #2270
   +0
4. feat: Asynchronous Map Implementation for Pipeline #2295
   +0
5. builtin transformers
6. multi-partition for pipeline
7. option to start each vertex in Rust or Golang code
8. buffer-full strategy
9. cherry-pick fix: Fix Sink Config to respect Fallback #2261
10. ordering should not be lost in transformer
11. Watermark changes to support async data movement #2365
    performance +1
    
12. total size and metric for per operation in tracker
13. use Bytes in Message for cheap cloning #2267
    +0
    
14. tune tokio runtime to avoid over consumption of CPU
15. switch e2e tests from go code to rust code (source, map, sink) #2300
    +0